Coverage for dockerfiles / runtime_smoke.py: 100.00%
111 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Runtime smoke for the distroless service images.
3Runs as the final Dockerfile stage's only RUN — exec form, as the runtime
4user — with the service entry module as its argument, plus ``--tracing`` in
5the four traced API images. It proves builder-to-scratch parity
6programmatically instead of via a hand-maintained module list:
8- every stdlib C extension that was importable in the builder stage must
9 import here too. ``build_scratch_rootfs.py`` records that set in
10 ``runtime_smoke_manifest.json`` next to this file, derived by actually
11 importing everything under ``lib-dynload`` in the builder. Import parity
12 is a strictly stronger completeness proof than the ldd closure alone: it
13 also catches libraries reached only via ``dlopen``, which ldd cannot see.
14- the service entry module must import (the full third-party graph);
15- ``getpass`` must resolve the synthesized passwd identity for the runtime
16 uid (NSS wiring);
17- OpenSSL's default trust store must load CA certificates (TLS to AWS);
18- ``zoneinfo`` must resolve from the staged tzdata.
20``--tracing`` adds the tracing stack. ``gco.services.tracing`` imports
21OpenTelemetry, botocore's SigV4 signer and httpx2 only once tracing is
22enabled, so importing the entry module never loads them: a package missing
23from a traced image would surface only at runtime, as one warning and a
24service that exports nothing. With the flag, the smoke parses that module's
25source as the image ships it and performs every import its functions
26contain, each the way its statement runs (``from package import name`` needs
27``name``), and httpx2's default TLS context (the public trust the X-Ray
28exporter uses, which httpx2 builds with truststore) must load the image's CA
29anchors. Neither check opens a connection. Like the extension set, the import
30list is derived rather than kept here, so it cannot drift from the module.
32Every failure is collected and reported, then the process exits non-zero so
33the image build — including CDK deploys — fails instead of the pod. The
34script and its manifest live in the builder stage's ``/opt/build`` and reach
35the final stage through a BuildKit bind mount scoped to the smoke RUN alone:
36the deployed image ships neither of them (deleting files in a later layer
37would only hide them — layers are additive — so they are never written into
38the image at all). The manifest is located relative to ``__file__``, so the
39pair works from any mount target.
41Only ``json``, ``sys``, ``importlib``, and ``pathlib`` are imported at module
42scope; everything under test (``ssl``, ``getpass``, ``zoneinfo``, the stdlib
43extensions, the entry module, the tracing stack and the ``ast`` parse that
44finds it) is imported inside guarded sections so a single breakage cannot
45mask the rest of the report.
46"""
48from __future__ import annotations
50import importlib
51import json
52import sys
53from pathlib import Path
55TRACING_FLAG = "--tracing"
57#: The module whose deferred imports are the tracing stack: nothing its
58#: functions import loads until ``GCO_TRACING_ENABLED=true``.
59TRACING_MODULE = "gco.services.tracing"
61# ``wrap_bio`` needs a server name to satisfy check_hostname. No handshake
62# runs, so the name is never resolved or contacted.
63_TRUST_PROBE_HOSTNAME = "localhost"
66def function_scope_imports(source: str, package: str | None) -> list[tuple[str, str | None]]:
67 """Every import inside a function of ``source``, as ``(module, name)`` pairs.
69 ``import a.b`` gives ``("a.b", None)``; ``from a import b, c`` gives
70 ``("a", "b")`` and ``("a", "c")``; relative imports resolve against
71 ``package``. Imports that run at module level (``TYPE_CHECKING`` blocks
72 and class bodies included) are left out: importing the module already
73 proved them.
74 """
75 import ast
76 import importlib.util
78 found: set[tuple[str, str | None]] = set()
79 for function in ast.walk(ast.parse(source)):
80 if not isinstance(function, ast.FunctionDef | ast.AsyncFunctionDef):
81 continue
82 for node in ast.walk(function):
83 if isinstance(node, ast.Import):
84 found.update((alias.name, None) for alias in node.names)
85 elif isinstance(node, ast.ImportFrom):
86 module = importlib.util.resolve_name(
87 "." * node.level + (node.module or ""), package
88 )
89 found.update((module, alias.name) for alias in node.names)
90 return sorted(found, key=lambda pair: (pair[0], pair[1] or ""))
93def tracing_imports(failures: list[str]) -> list[tuple[str, str | None]]:
94 """What :data:`TRACING_MODULE` imports inside its functions, from its source here.
96 Parsed, not imported: the module defers these imports until tracing is
97 enabled, which is why importing the entry module cannot prove them.
98 Appends a failure when the source cannot be found or parsed, or holds no
99 function-scope import at all, so a broken derivation never passes
100 vacuously.
101 """
102 try:
103 import importlib.util
105 spec = importlib.util.find_spec(TRACING_MODULE)
106 if spec is None or spec.origin is None:
107 raise ModuleNotFoundError(f"no source file for {TRACING_MODULE}")
108 found = function_scope_imports(Path(spec.origin).read_text(encoding="utf-8"), spec.parent)
109 except BaseException as exc: # aggregate every breakage
110 failures.append(f"tracing imports of {TRACING_MODULE}: {type(exc).__name__}: {exc}")
111 return []
112 if not found:
113 failures.append(
114 f"{TRACING_MODULE} has no function-scope imports; the tracing derivation broke"
115 )
116 return found
119def check_tracing_stack(failures: list[str]) -> tuple[int, int]:
120 """Perform the tracing module's deferred imports and check httpx2's default trust.
122 Each import runs the way its statement does: ``from package import name``
123 needs ``name``, found as an attribute or else as the submodule
124 ``package.name``. truststore needs no entry of its own: httpx2 imports it
125 to build the context the trust check builds. Appends every problem to
126 ``failures`` and returns the number of imports checked and the number of
127 CA certificates httpx2's default context loaded (0 when that check
128 failed).
129 """
130 imports = tracing_imports(failures)
131 for module_name, name in imports:
132 statement = f"from {module_name} import {name}" if name else f"import {module_name}"
133 try:
134 # The names come from the tracing module's own source, never input.
135 # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
136 module = importlib.import_module(module_name)
137 if name is not None and not hasattr(module, name):
138 # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
139 importlib.import_module(f"{module_name}.{name}")
140 except BaseException as exc: # aggregate every breakage
141 failures.append(f"tracing import ({statement}): {type(exc).__name__}: {exc}")
143 try:
144 import ssl
146 import httpx2
148 # The context the X-Ray exporter's client builds (trust_env=False) is
149 # a truststore one, which loads the system trust store when a
150 # connection is wrapped rather than when it is built. Wrapping two
151 # in-memory BIOs does that without a socket or a handshake; the SSL
152 # object's context is the OpenSSL one truststore configured.
153 context = httpx2.create_ssl_context(trust_env=False)
154 tls = context.wrap_bio(
155 ssl.MemoryBIO(), ssl.MemoryBIO(), server_hostname=_TRUST_PROBE_HOSTNAME
156 )
157 anchors = tls.context.cert_store_stats()["x509_ca"]
158 except BaseException as exc:
159 failures.append(f"httpx2 default trust: {type(exc).__name__}: {exc}")
160 return len(imports), 0
161 if anchors <= 0:
162 failures.append("httpx2 default trust store (truststore) loaded zero CA certificates")
163 return len(imports), anchors
166def main() -> int:
167 arguments = sys.argv[1:]
168 tracing = arguments[1:] == [TRACING_FLAG]
169 if len(arguments) != 1 and not tracing:
170 print(
171 f"usage: runtime_smoke.py <service-entry-module> [{TRACING_FLAG}]",
172 file=sys.stderr,
173 )
174 return 2
175 entry_module = arguments[0]
177 manifest_path = Path(__file__).with_name("runtime_smoke_manifest.json")
178 manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
179 extensions: list[str] = manifest["stdlib_extensions"]
180 expected_user: str = manifest["runtime_user"]
182 failures: list[str] = []
183 if not extensions:
184 failures.append("manifest lists no stdlib extensions; the builder probe broke")
186 for name in extensions:
187 try:
188 # Dynamic import IS the check: the names come from the manifest
189 # this build's rootfs script derived (root-written, read-only at
190 # runtime), never from untrusted input.
191 # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
192 importlib.import_module(name)
193 except BaseException as exc: # aggregate every breakage
194 failures.append(f"stdlib extension {name}: {type(exc).__name__}: {exc}")
196 try:
197 # The entry module is a literal baked into each Dockerfile's smoke
198 # RUN, not user input; importing it dynamically is this script's job.
199 # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
200 importlib.import_module(entry_module)
201 except BaseException as exc:
202 failures.append(f"entry module {entry_module}: {type(exc).__name__}: {exc}")
204 actual_user = "<unresolved>"
205 try:
206 import getpass
208 actual_user = getpass.getuser()
209 if actual_user != expected_user:
210 failures.append(f"runtime user: expected {expected_user!r}, got {actual_user!r}")
211 except BaseException as exc:
212 failures.append(f"runtime identity lookup: {type(exc).__name__}: {exc}")
214 try:
215 import ssl
217 ca_count = ssl.create_default_context().cert_store_stats()["x509_ca"]
218 if ca_count <= 0:
219 failures.append("OpenSSL default trust store loaded zero CA certificates")
220 except BaseException as exc:
221 failures.append(f"CA trust store: {type(exc).__name__}: {exc}")
223 try:
224 import zoneinfo
226 zoneinfo.ZoneInfo("UTC")
227 except BaseException as exc:
228 failures.append(f"zoneinfo/tzdata: {type(exc).__name__}: {exc}")
230 tracing_imported, trust_anchors = check_tracing_stack(failures) if tracing else (0, 0)
232 if failures:
233 print(f"distroless runtime smoke FAILED ({len(failures)} problem(s)):", file=sys.stderr)
234 for failure in failures:
235 print(f" - {failure}", file=sys.stderr)
236 return 1
238 summary = (
239 f"distroless runtime smoke OK: {len(extensions)} stdlib extensions, "
240 f"entry module {entry_module}, user {actual_user}, CA trust and tzdata present"
241 )
242 if tracing:
243 summary += (
244 f"; {tracing_imported} deferred imports of {TRACING_MODULE} present, "
245 f"httpx2 default trust loads {trust_anchors} CA certificates"
246 )
247 print(summary)
248 return 0
251if __name__ == "__main__":
252 raise SystemExit(main())