Add k8s test: init container can reach HTTPS endpoints (CA trust store)

tests/k8s_lib_injection/test_k8s_lib_injection.py

@@ -1,5 +1,10 @@
 from utils import scenarios, features, context, bug, logger
-from tests.k8s_lib_injection.utils import get_dev_agent_traces, get_cluster_info
+from tests.k8s_lib_injection.utils import (
+    get_dev_agent_traces,
+    get_cluster_info,
+    get_library_init_image,
+    run_https_probe_pod,
+)
 from utils.onboarding.weblog_interface import make_get_request, warmup_weblog
 from utils.onboarding.backend_interface import wait_backend_trace_id
 from utils.onboarding.wait_for_tcp_port import wait_for_port
@@ -19,6 +24,20 @@ def test_k8s_lib_injection(self):
         traces_json = get_dev_agent_traces(get_cluster_info())
         assert len(traces_json) > 0, "No traces found"
 
+    def test_k8s_init_container_https_egress(self):
+        """The init image must be able to complete a TLS-verified HTTPS request.
+
+        The init container ships a CA trust store so admission-controller wrappers that dial
+        HTTPS from inside it (e.g. bank-vaults' vault-env) keep working. This runs a throwaway
+        pod from the init image doing `curl -fsS https://...` (verification on) and asserts it
+        succeeds; a missing or broken cert store makes curl exit non-zero.
+        """
+        image = get_library_init_image()
+        phase, logs = run_https_probe_pod(get_cluster_info(), image, "https://app.datadoghq.com")
+        if phase != "Succeeded":
+            logger.error(f"[HTTPS probe] init image {image} failed HTTPS egress; pod logs:\n{logs}")
+        assert phase == "Succeeded", f"init image could not complete a verified HTTPS request (pod phase: {phase})"
+
 
 @features.k8s_admission_controller
 @scenarios.k8s_lib_injection_operator

tests/k8s_lib_injection/utils.py

@@ -1,9 +1,11 @@
 import time
 
 import requests
+from kubernetes import client
+
 from utils import logger, context
 from utils.k8s_lib_injection.k8s_cluster_provider import K8sClusterInfo
-from utils._context._scenarios.k8s_lib_injection import K8sScenarioWithClusterProvider
+from utils._context._scenarios.k8s_lib_injection import K8sScenario, K8sScenarioWithClusterProvider
 
 
 def get_dev_agent_traces(k8s_cluster_info: K8sClusterInfo, retry: int = 10) -> list:
@@ -24,3 +26,58 @@ def get_cluster_info() -> K8sClusterInfo:
     assert isinstance(context.scenario, K8sScenarioWithClusterProvider)
 
     return context.scenario.k8s_cluster_provider.get_cluster_info()
+
+
+def get_library_init_image() -> str:
+    """Return the init image under test (the image the scenario deploys as the init container)."""
+    assert isinstance(context.scenario, K8sScenario)
+
+    return context.scenario.test_weblog.library_init_image
+
+
+def run_https_probe_pod(
+    k8s_cluster_info: K8sClusterInfo, image: str, url: str, *, namespace: str = "default", timeout: int = 120
+) -> tuple[str | None, str]:
+    """Run a one-shot pod from `image` performing a TLS-verified HTTPS GET to `url`.
+
+    Uses the image's curl with verification on (no -k), so a missing or invalid CA trust
+    store makes curl exit non-zero. Returns (pod_phase, logs); phase "Succeeded" means the
+    image completed a verified HTTPS request. The probe pod is always cleaned up.
+    """
+    api = k8s_cluster_info.core_v1_api()
+    name = "init-https-probe"
+    pod = client.V1Pod(
+        metadata=client.V1ObjectMeta(name=name),
+        spec=client.V1PodSpec(
+            restart_policy="Never",
+            containers=[
+                client.V1Container(
+                    name="probe",
+                    image=image,
+                    image_pull_policy="Always",
+                    command=["/usr/bin/curl", "-fsS", "-o", "/dev/null", "--max-time", "15", url],
+                )
+            ],
+        ),
+    )
+    logger.info(f"[HTTPS probe] running {image} -> curl {url}")
+    api.create_namespaced_pod(namespace=namespace, body=pod)
+    try:
+        phase: str | None = None
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            phase = api.read_namespaced_pod_status(name=name, namespace=namespace).status.phase
+            if phase in ("Succeeded", "Failed"):
+                break
+            time.sleep(2)
+        logs = ""
+        try:
+            logs = api.read_namespaced_pod_log(name=name, namespace=namespace)
+        except Exception as e:
+            logger.warning(f"[HTTPS probe] could not read probe pod logs: {e}")
+        return phase, logs
+    finally:
+        try:
+            api.delete_namespaced_pod(name=name, namespace=namespace)
+        except Exception as e:
+            logger.warning(f"[HTTPS probe] could not delete probe pod: {e}")