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