|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Diff-ratchet deployment setting classifications without inspecting values. |
| 3 | +
|
| 4 | +The checker deliberately reads only names in checked-in deployment wiring. It |
| 5 | +never loads environment variables, Secret Manager data, or rendered manifests. |
| 6 | +""" |
| 7 | + |
| 8 | +from __future__ import annotations |
| 9 | + |
| 10 | +import argparse |
| 11 | +import datetime as dt |
| 12 | +import json |
| 13 | +import os |
| 14 | +import re |
| 15 | +import subprocess |
| 16 | +import sys |
| 17 | +from dataclasses import dataclass |
| 18 | +from pathlib import Path |
| 19 | +from typing import Iterable |
| 20 | + |
| 21 | + |
| 22 | +ROOT = Path(__file__).resolve().parents[2] |
| 23 | +DEFAULT_POLICY = ROOT / "config" / "deployment-setting-classification.json" |
| 24 | +EXPRESSION = re.compile(r"\$\{\{\s*(secrets|vars)\.([A-Z][A-Z0-9_]*)\s*}}") |
| 25 | +UPPER_VALUE = r"([A-Z][A-Z0-9_]*)" |
| 26 | +KEY_VALUE = re.compile(rf"^\s*(?:secretKey|remoteKey):\s*{UPPER_VALUE}\s*(?:#.*)?$") |
| 27 | +CLOUD_RUN_SECRET = re.compile(rf"^\s*secret:\s*{UPPER_VALUE}\s*(?:#.*)?$") |
| 28 | +YAML_KEY = re.compile(r"^(\s*)([A-Z][A-Z0-9_]*):\s*(?:#.*)?$") |
| 29 | +YAML_VALUE = re.compile(r"^\s*(?:value|env_var):") |
| 30 | +HELM_ENV_NAME = re.compile(rf"^\s*-\s*name:\s*{UPPER_VALUE}\s*(?:#.*)?$") |
| 31 | +HELM_SECRET_KEY = re.compile(rf"^\s*key:\s*{UPPER_VALUE}\s*(?:#.*)?$") |
| 32 | +CLOUD_RUN_SECRET_LINE = re.compile(rf"^\s*{UPPER_VALUE}={UPPER_VALUE}(?::[^\s#]+)?\s*(?:#.*)?$") |
| 33 | +SECRET_MANAGER_OUTPUT = re.compile( |
| 34 | + rf'^\s*echo\s+["\']?({UPPER_VALUE})=\$\(gcloud\s+secrets\s+versions\s+access\b.*?--secret={UPPER_VALUE}' |
| 35 | +) |
| 36 | + |
| 37 | + |
| 38 | +@dataclass(frozen=True, order=True) |
| 39 | +class Binding: |
| 40 | + path: str |
| 41 | + source: str |
| 42 | + name: str |
| 43 | + |
| 44 | + |
| 45 | +def _value_from_match(match: re.Match[str], group: int = 1) -> str: |
| 46 | + return match.group(group) |
| 47 | + |
| 48 | + |
| 49 | +def _deployment_paths(root: Path) -> set[str]: |
| 50 | + paths = { |
| 51 | + str(path.relative_to(root)) |
| 52 | + for path in (root / ".github" / "workflows").glob("*.y*ml") |
| 53 | + if path.is_file() |
| 54 | + } |
| 55 | + for relative in ("backend/deploy/runtime_env.yaml",): |
| 56 | + if (root / relative).is_file(): |
| 57 | + paths.add(relative) |
| 58 | + charts = root / "backend" / "charts" |
| 59 | + if charts.is_dir(): |
| 60 | + paths.update( |
| 61 | + str(path.relative_to(root)) |
| 62 | + for path in charts.glob("*/*_values.yaml") |
| 63 | + if path.is_file() |
| 64 | + ) |
| 65 | + return paths |
| 66 | + |
| 67 | + |
| 68 | +def _git_environment() -> dict[str, str]: |
| 69 | + """Avoid hook-scoped Git paths when inspecting an explicit repository root.""" |
| 70 | + return {name: value for name, value in os.environ.items() if not name.startswith("GIT_")} |
| 71 | + |
| 72 | + |
| 73 | +def _base_paths(root: Path, base: str) -> set[str]: |
| 74 | + result = subprocess.run( |
| 75 | + ["git", "ls-tree", "-r", "--name-only", base], |
| 76 | + cwd=root, |
| 77 | + check=True, |
| 78 | + text=True, |
| 79 | + stdout=subprocess.PIPE, |
| 80 | + env=_git_environment(), |
| 81 | + ) |
| 82 | + paths = set(result.stdout.splitlines()) |
| 83 | + return { |
| 84 | + path |
| 85 | + for path in paths |
| 86 | + if path.startswith(".github/workflows/") and path.endswith((".yml", ".yaml")) |
| 87 | + or path == "backend/deploy/runtime_env.yaml" |
| 88 | + or re.fullmatch(r"backend/charts/[^/]+/[^/]+_values\.yaml", path) is not None |
| 89 | + } |
| 90 | + |
| 91 | + |
| 92 | +def _read_base_file(root: Path, base: str, relative_path: str) -> str | None: |
| 93 | + result = subprocess.run( |
| 94 | + ["git", "show", f"{base}:{relative_path}"], |
| 95 | + cwd=root, |
| 96 | + text=True, |
| 97 | + stdout=subprocess.PIPE, |
| 98 | + stderr=subprocess.DEVNULL, |
| 99 | + env=_git_environment(), |
| 100 | + ) |
| 101 | + return result.stdout if result.returncode == 0 else None |
| 102 | + |
| 103 | + |
| 104 | +def _extract_bindings(relative_path: str, text: str) -> set[Binding]: |
| 105 | + bindings: set[Binding] = set() |
| 106 | + helm_env_name: str | None = None |
| 107 | + helm_secret_indent: int | None = None |
| 108 | + cloud_run_secrets_indent: int | None = None |
| 109 | + yaml_keys: list[tuple[int, str]] = [] |
| 110 | + lines = text.splitlines() |
| 111 | + |
| 112 | + for line in lines: |
| 113 | + indent = len(line) - len(line.lstrip()) |
| 114 | + stripped = line.strip() |
| 115 | + for expression in EXPRESSION.finditer(line): |
| 116 | + source, name = expression.groups() |
| 117 | + bindings.add(Binding(relative_path, f"github_{source}", name)) |
| 118 | + |
| 119 | + secret_manager_output = SECRET_MANAGER_OUTPUT.match(line) |
| 120 | + if secret_manager_output: |
| 121 | + bindings.add(Binding(relative_path, "secret_manager", secret_manager_output.group(1))) |
| 122 | + |
| 123 | + key_value = KEY_VALUE.match(line) |
| 124 | + if key_value: |
| 125 | + bindings.add(Binding(relative_path, "external_secret", _value_from_match(key_value))) |
| 126 | + |
| 127 | + cloud_run_secret = CLOUD_RUN_SECRET.match(line) |
| 128 | + if cloud_run_secret: |
| 129 | + bindings.add(Binding(relative_path, "cloud_run_secret", _value_from_match(cloud_run_secret))) |
| 130 | + |
| 131 | + if helm_secret_indent is not None and stripped and indent <= helm_secret_indent: |
| 132 | + helm_secret_indent = None |
| 133 | + if stripped == "secretKeyRef:": |
| 134 | + helm_secret_indent = indent |
| 135 | + elif helm_secret_indent is not None: |
| 136 | + helm_secret_key = HELM_SECRET_KEY.match(line) |
| 137 | + if helm_secret_key: |
| 138 | + bindings.add(Binding(relative_path, "helm_secret", _value_from_match(helm_secret_key))) |
| 139 | + |
| 140 | + if cloud_run_secrets_indent is not None and stripped and indent <= cloud_run_secrets_indent: |
| 141 | + cloud_run_secrets_indent = None |
| 142 | + if stripped == "secrets:": |
| 143 | + cloud_run_secrets_indent = indent |
| 144 | + elif cloud_run_secrets_indent is not None: |
| 145 | + cloud_run_secret_line = CLOUD_RUN_SECRET_LINE.match(line) |
| 146 | + if cloud_run_secret_line: |
| 147 | + remote_name = cloud_run_secret_line.group(2) |
| 148 | + bindings.add(Binding(relative_path, "cloud_run_secret", remote_name)) |
| 149 | + |
| 150 | + helm_env = HELM_ENV_NAME.match(line) |
| 151 | + if helm_env: |
| 152 | + helm_env_name = _value_from_match(helm_env) |
| 153 | + elif stripped.startswith("- "): |
| 154 | + helm_env_name = None |
| 155 | + elif helm_env_name is not None and re.match(r"^\s*value:\s*", line): |
| 156 | + bindings.add(Binding(relative_path, "normal_env", helm_env_name)) |
| 157 | + |
| 158 | + while yaml_keys and indent <= yaml_keys[-1][0]: |
| 159 | + yaml_keys.pop() |
| 160 | + yaml_key = YAML_KEY.match(line) |
| 161 | + if yaml_key: |
| 162 | + yaml_keys.append((len(yaml_key.group(1)), yaml_key.group(2))) |
| 163 | + elif yaml_keys and indent > yaml_keys[-1][0] and YAML_VALUE.match(line): |
| 164 | + bindings.add(Binding(relative_path, "normal_env", yaml_keys[-1][1])) |
| 165 | + |
| 166 | + return bindings |
| 167 | + |
| 168 | + |
| 169 | +def extract_current_bindings(root: Path) -> set[Binding]: |
| 170 | + bindings: set[Binding] = set() |
| 171 | + for relative_path in _deployment_paths(root): |
| 172 | + bindings.update(_extract_bindings(relative_path, (root / relative_path).read_text(encoding="utf-8"))) |
| 173 | + return bindings |
| 174 | + |
| 175 | + |
| 176 | +def extract_base_bindings(root: Path, base: str) -> set[Binding]: |
| 177 | + bindings: set[Binding] = set() |
| 178 | + for relative_path in _base_paths(root, base): |
| 179 | + content = _read_base_file(root, base, relative_path) |
| 180 | + if content is not None: |
| 181 | + bindings.update(_extract_bindings(relative_path, content)) |
| 182 | + return bindings |
| 183 | + |
| 184 | + |
| 185 | +def load_policy(path: Path) -> dict[str, object]: |
| 186 | + with path.open(encoding="utf-8") as handle: |
| 187 | + policy = json.load(handle) |
| 188 | + if not isinstance(policy, dict): |
| 189 | + raise ValueError(f"{path} must contain a JSON object") |
| 190 | + return policy |
| 191 | + |
| 192 | + |
| 193 | +def _policy_kinds(policy: dict[str, object]) -> dict[str, set[str]]: |
| 194 | + raw_kinds = policy.get("kinds") |
| 195 | + if not isinstance(raw_kinds, dict): |
| 196 | + raise ValueError("policy.kinds must be an object") |
| 197 | + kinds: dict[str, set[str]] = {} |
| 198 | + for kind in ("secret", "config", "public_build"): |
| 199 | + names = raw_kinds.get(kind) |
| 200 | + if not isinstance(names, list) or not all(isinstance(name, str) for name in names): |
| 201 | + raise ValueError(f"policy.kinds.{kind} must be a list of names") |
| 202 | + kinds[kind] = set(names) |
| 203 | + return kinds |
| 204 | + |
| 205 | + |
| 206 | +def validate_policy(policy: dict[str, object]) -> list[str]: |
| 207 | + errors: list[str] = [] |
| 208 | + try: |
| 209 | + kinds = _policy_kinds(policy) |
| 210 | + except ValueError as exc: |
| 211 | + return [str(exc)] |
| 212 | + seen: dict[str, str] = {} |
| 213 | + for kind, names in kinds.items(): |
| 214 | + for name in names: |
| 215 | + if not re.fullmatch(r"[A-Z][A-Z0-9_]*", name): |
| 216 | + errors.append(f"{kind} classification has invalid name {name!r}") |
| 217 | + if name in seen: |
| 218 | + errors.append(f"{name} is classified as both {seen[name]} and {kind}") |
| 219 | + seen[name] = kind |
| 220 | + |
| 221 | + exceptions = policy.get("exceptions", {}) |
| 222 | + if not isinstance(exceptions, dict): |
| 223 | + return errors + ["policy.exceptions must be an object"] |
| 224 | + today = dt.date.today() |
| 225 | + for name, raw_metadata in sorted(exceptions.items()): |
| 226 | + if name not in seen: |
| 227 | + errors.append(f"exception {name} must also be classified") |
| 228 | + if not isinstance(raw_metadata, dict): |
| 229 | + errors.append(f"exception {name} must be an object") |
| 230 | + continue |
| 231 | + for field in ("owner", "reason", "expires", "allowed_sources"): |
| 232 | + if not raw_metadata.get(field): |
| 233 | + errors.append(f"exception {name} is missing {field}") |
| 234 | + expiry = raw_metadata.get("expires") |
| 235 | + if isinstance(expiry, str): |
| 236 | + try: |
| 237 | + if dt.date.fromisoformat(expiry) < today: |
| 238 | + errors.append(f"exception {name} expired on {expiry}") |
| 239 | + except ValueError: |
| 240 | + errors.append(f"exception {name} has invalid ISO expiry {expiry!r}") |
| 241 | + allowed_sources = raw_metadata.get("allowed_sources") |
| 242 | + if not isinstance(allowed_sources, list) or not all(isinstance(source, str) for source in allowed_sources): |
| 243 | + errors.append(f"exception {name} allowed_sources must be a list") |
| 244 | + return errors |
| 245 | + |
| 246 | + |
| 247 | +def _classification(policy: dict[str, object], name: str) -> str | None: |
| 248 | + for kind, names in _policy_kinds(policy).items(): |
| 249 | + if name in names: |
| 250 | + return kind |
| 251 | + return None |
| 252 | + |
| 253 | + |
| 254 | +def _exception_allows(policy: dict[str, object], binding: Binding) -> bool: |
| 255 | + exceptions = policy.get("exceptions", {}) |
| 256 | + metadata = exceptions.get(binding.name) if isinstance(exceptions, dict) else None |
| 257 | + return isinstance(metadata, dict) and binding.source in metadata.get("allowed_sources", []) |
| 258 | + |
| 259 | + |
| 260 | +def _expected_kinds(source: str) -> set[str]: |
| 261 | + if source == "github_vars": |
| 262 | + return {"config", "public_build"} |
| 263 | + if source == "normal_env": |
| 264 | + return {"config", "public_build"} |
| 265 | + return {"secret"} |
| 266 | + |
| 267 | + |
| 268 | +def validate_bindings( |
| 269 | + policy: dict[str, object], current_bindings: Iterable[Binding], base_bindings: Iterable[Binding] |
| 270 | +) -> list[str]: |
| 271 | + errors: list[str] = [] |
| 272 | + current = set(current_bindings) |
| 273 | + new_or_changed = current - set(base_bindings) |
| 274 | + |
| 275 | + # Public build configuration is an explicit migration target, not legacy |
| 276 | + # debt. Reject it even when a pre-ratchet line still exists in the base. |
| 277 | + to_validate = new_or_changed | { |
| 278 | + binding for binding in current if binding.source == "github_secrets" and _classification(policy, binding.name) == "public_build" |
| 279 | + } |
| 280 | + for binding in sorted(to_validate): |
| 281 | + kind = _classification(policy, binding.name) |
| 282 | + if kind is None: |
| 283 | + errors.append(f"{binding.path}: {binding.source} binding {binding.name} is unclassified") |
| 284 | + continue |
| 285 | + expected = _expected_kinds(binding.source) |
| 286 | + if kind in expected or _exception_allows(policy, binding): |
| 287 | + continue |
| 288 | + if binding.source == "github_secrets" and kind == "public_build": |
| 289 | + errors.append(f"{binding.path}: public_build setting {binding.name} must use vars.{binding.name}, not secrets.{binding.name}") |
| 290 | + else: |
| 291 | + expected_text = " or ".join(sorted(expected)) |
| 292 | + errors.append( |
| 293 | + f"{binding.path}: {binding.source} binding {binding.name} is {kind}; expected {expected_text}" |
| 294 | + ) |
| 295 | + return errors |
| 296 | + |
| 297 | + |
| 298 | +def main() -> int: |
| 299 | + parser = argparse.ArgumentParser(description="Diff-ratchet deployment setting classifications.") |
| 300 | + parser.add_argument("--base", required=True, help="Git ref used as the legacy binding baseline.") |
| 301 | + parser.add_argument("--root", type=Path, default=ROOT) |
| 302 | + parser.add_argument("--policy", type=Path) |
| 303 | + args = parser.parse_args() |
| 304 | + |
| 305 | + root = args.root.resolve() |
| 306 | + policy_path = (args.policy or root / DEFAULT_POLICY.relative_to(ROOT)).resolve() |
| 307 | + try: |
| 308 | + policy = load_policy(policy_path) |
| 309 | + errors = validate_policy(policy) |
| 310 | + errors.extend(validate_bindings(policy, extract_current_bindings(root), extract_base_bindings(root, args.base))) |
| 311 | + except (OSError, ValueError, subprocess.CalledProcessError) as exc: |
| 312 | + print(f"deployment secret-boundary check failed: {exc}", file=sys.stderr) |
| 313 | + return 1 |
| 314 | + if errors: |
| 315 | + print("deployment secret-boundary check failed:", file=sys.stderr) |
| 316 | + print("\n".join(f"- {error}" for error in errors), file=sys.stderr) |
| 317 | + return 1 |
| 318 | + print("deployment secret-boundary check passed") |
| 319 | + return 0 |
| 320 | + |
| 321 | + |
| 322 | +if __name__ == "__main__": |
| 323 | + raise SystemExit(main()) |
0 commit comments