Coverage for .github / scripts / verify_container_tool_versions.py: 100.00%
96 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1#!/usr/bin/env python3
2"""Verify that CI-built container tools report their reviewed source pins."""
4from __future__ import annotations
6import argparse
7import json
8import re
9import subprocess
10from collections.abc import Callable, Sequence
11from pathlib import Path
13CommandRunner = Callable[[list[str]], str]
14DEV_PIN_NAMES = (
15 "NODE_VERSION",
16 "NPM_VERSION",
17 "CDK_VERSION",
18 "KUBECTL_VERSION",
19 "AWSCLI_VERSION",
20 "DOCKER_VERSION",
21 "BUILDX_VERSION",
22 "UV_VERSION",
23 "SESSION_MANAGER_PLUGIN_VERSION",
24)
27class VerificationError(RuntimeError):
28 """A pin is missing, output is malformed, or a runtime version drifted."""
31def parse_dev_pins(dockerfile: Path) -> dict[str, str]:
32 """Read the exact tool-version ARGs from ``Dockerfile.dev``."""
33 content = dockerfile.read_text(encoding="utf-8")
34 pins = dict(re.findall(r"^ARG ([A-Z0-9_]+)=([^\s#]+)\s*$", content, re.MULTILINE))
35 missing = [name for name in DEV_PIN_NAMES if not pins.get(name)]
36 if missing:
37 raise VerificationError(f"missing Dockerfile.dev pins: {', '.join(missing)}")
38 return {name: pins[name] for name in DEV_PIN_NAMES}
41def _single_release_pin(content: str, pattern: str, label: str) -> str:
42 matches: set[str] = set()
43 for match in re.findall(pattern, content):
44 if not isinstance(match, str):
45 raise VerificationError(f"{label} pin pattern must contain exactly one capture group")
46 matches.add(match)
47 if len(matches) != 1:
48 values = ", ".join(sorted(matches)) or "none"
49 raise VerificationError(f"expected one {label} pin, found: {values}")
50 return matches.pop()
53def parse_helm_installer_pins(dockerfile: Path) -> dict[str, str]:
54 """Read Helm and kubectl versions from their authenticated asset URLs."""
55 content = dockerfile.read_text(encoding="utf-8")
56 return {
57 "HELM_VERSION": _single_release_pin(
58 content,
59 r"helm-(v\d+\.\d+\.\d+)-linux-amd64\.tar\.gz",
60 "Helm",
61 ),
62 "KUBECTL_VERSION": _single_release_pin(
63 content,
64 r"release/(v\d+\.\d+\.\d+)/bin/linux/amd64/kubectl",
65 "kubectl",
66 ),
67 }
70def run_command(command: list[str]) -> str:
71 """Run one bounded command and return its primary output stream."""
72 completed = subprocess.run(
73 command,
74 check=True,
75 capture_output=True,
76 text=True,
77 timeout=60,
78 )
79 output = completed.stdout.strip() or completed.stderr.strip()
80 if not output:
81 raise VerificationError(f"command produced no version output: {command!r}")
82 return output
85def _docker_command(
86 image: str,
87 command: Sequence[str],
88 *,
89 entrypoint: str | None = None,
90) -> list[str]:
91 result = ["docker", "run", "--rm"]
92 if entrypoint is not None:
93 result.extend(["--entrypoint", entrypoint])
94 result.append(image)
95 result.extend(command)
96 return result
99def _extract_version(output: str, pattern: str, tool: str) -> str:
100 match = re.search(pattern, output, re.MULTILINE)
101 if match is None:
102 raise VerificationError(f"could not parse {tool} version from: {output!r}")
103 return match.group(1)
106def _kubectl_version(output: str) -> str:
107 try:
108 value = json.loads(output)["clientVersion"]["gitVersion"]
109 except (KeyError, TypeError, json.JSONDecodeError) as exc:
110 raise VerificationError(f"could not parse kubectl version from: {output!r}") from exc
111 if not isinstance(value, str) or not re.fullmatch(r"v\d+\.\d+\.\d+", value):
112 raise VerificationError(f"invalid kubectl version: {value!r}")
113 return value
116def _require_version(tool: str, expected: str, actual: str) -> None:
117 if actual != expected:
118 raise VerificationError(f"expected {tool} {expected}, got {actual}")
119 print(f"{tool}: {actual}")
122def verify_dev_image(
123 image: str,
124 dockerfile: Path,
125 *,
126 runner: CommandRunner = run_command,
127) -> dict[str, str]:
128 """Compare every contributor-tool runtime version with ``Dockerfile.dev``."""
129 pins = parse_dev_pins(dockerfile)
130 checks = (
131 ("Node.js", "NODE_VERSION", ("node", "--version"), r"^(v\d+\.\d+\.\d+)\b"),
132 ("npm", "NPM_VERSION", ("npm", "--version"), r"^(\d+\.\d+\.\d+)\b"),
133 ("CDK", "CDK_VERSION", ("cdk", "--version"), r"^(\d+\.\d+\.\d+)\b"),
134 (
135 "AWS CLI",
136 "AWSCLI_VERSION",
137 ("aws", "--version"),
138 r"\baws-cli/(\d+\.\d+\.\d+)\b",
139 ),
140 (
141 "Docker CLI",
142 "DOCKER_VERSION",
143 ("docker", "--version"),
144 r"\bDocker version (\d+\.\d+\.\d+),",
145 ),
146 (
147 "Buildx",
148 "BUILDX_VERSION",
149 ("docker", "buildx", "version"),
150 r"\b(v\d+\.\d+\.\d+)\b",
151 ),
152 ("uv", "UV_VERSION", ("uv", "--version"), r"^uv (\d+\.\d+\.\d+)\b"),
153 ("uvx", "UV_VERSION", ("uvx", "--version"), r"^uvx (\d+\.\d+\.\d+)\b"),
154 (
155 "Session Manager plugin",
156 "SESSION_MANAGER_PLUGIN_VERSION",
157 ("session-manager-plugin", "--version"),
158 r"^(\d+\.\d+\.\d+\.\d+)\b",
159 ),
160 )
161 actual_versions: dict[str, str] = {}
162 for tool, pin_name, command, pattern in checks:
163 output = runner(_docker_command(image, command))
164 actual = _extract_version(output, pattern, tool)
165 _require_version(tool, pins[pin_name], actual)
166 actual_versions[tool] = actual
168 kubectl_output = runner(
169 _docker_command(
170 image,
171 ("kubectl", "version", "--client=true", "--output=json"),
172 )
173 )
174 kubectl_actual = _kubectl_version(kubectl_output)
175 _require_version("kubectl", pins["KUBECTL_VERSION"], kubectl_actual)
176 actual_versions["kubectl"] = kubectl_actual
177 return actual_versions
180def verify_helm_installer_image(
181 image: str,
182 dockerfile: Path,
183 *,
184 runner: CommandRunner = run_command,
185) -> dict[str, str]:
186 """Compare Helm-installer runtime binaries with authenticated URL pins."""
187 pins = parse_helm_installer_pins(dockerfile)
188 helm_output = runner(_docker_command(image, ("version", "--short"), entrypoint="helm"))
189 helm_actual = _extract_version(helm_output, r"\b(v\d+\.\d+\.\d+)\b", "Helm")
190 _require_version("Helm", pins["HELM_VERSION"], helm_actual)
192 kubectl_output = runner(
193 _docker_command(
194 image,
195 ("version", "--client=true", "--output=json"),
196 entrypoint="kubectl",
197 )
198 )
199 kubectl_actual = _kubectl_version(kubectl_output)
200 _require_version("kubectl", pins["KUBECTL_VERSION"], kubectl_actual)
201 return {"Helm": helm_actual, "kubectl": kubectl_actual}
204def main(argv: list[str] | None = None) -> int:
205 parser = argparse.ArgumentParser(description=__doc__)
206 parser.add_argument("profile", choices=("dev", "helm-installer"))
207 parser.add_argument("--image", required=True)
208 parser.add_argument("--dockerfile", type=Path)
209 args = parser.parse_args(argv)
211 root = Path(__file__).resolve().parents[2]
212 if args.profile == "dev":
213 dockerfile = args.dockerfile or root / "Dockerfile.dev"
214 verify_dev_image(args.image, dockerfile)
215 else:
216 dockerfile = args.dockerfile or root / "lambda" / "helm-installer" / "Dockerfile"
217 verify_helm_installer_image(args.image, dockerfile)
218 return 0
221if __name__ == "__main__":
222 raise SystemExit(main())