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

1#!/usr/bin/env python3 

2"""Verify that CI-built container tools report their reviewed source pins.""" 

3 

4from __future__ import annotations 

5 

6import argparse 

7import json 

8import re 

9import subprocess 

10from collections.abc import Callable, Sequence 

11from pathlib import Path 

12 

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) 

24 

25 

26class VerificationError(RuntimeError): 

27 """A pin is missing, output is malformed, or a runtime version drifted.""" 

28 

29 

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} 

38 

39 

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() 

50 

51 

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 } 

67 

68 

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 

82 

83 

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 

96 

97 

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) 

103 

104 

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 

113 

114 

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}") 

119 

120 

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 

160 

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 

171 

172 

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) 

184 

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} 

195 

196 

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) 

203 

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 

212 

213 

214if __name__ == "__main__": 

215 raise SystemExit(main())