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

Companion to DataDog/libdatadog-build#235 (init images busybox -> scratch +
dash/toybox/curl). Asserts the init image can complete a TLS-verified HTTPS
request via a throwaway probe pod, guarding the CA trust store against
regression on the base-image change.

Author: knusbaum

tests/k8s_lib_injection/test_k8s_lib_injection.py

@@ -1,5 +1,5 @@
 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, 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 +19,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 = context.scenario.test_weblog.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,6 +1,8 @@
 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
@@ -24,3 +26,51 @@ def get_cluster_info() -> K8sClusterInfo:
     assert isinstance(context.scenario, K8sScenarioWithClusterProvider)
 
     return context.scenario.k8s_cluster_provider.get_cluster_info()
+
+
+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:  # noqa: BLE001
+            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:  # noqa: BLE001
+            logger.warning(f"[HTTPS probe] could not delete probe pod: {e}")