diff --git a/scripts/verify_secret_contract.py b/scripts/verify_secret_contract.py new file mode 100755 index 0000000..2f348b2 --- /dev/null +++ b/scripts/verify_secret_contract.py @@ -0,0 +1,207 @@ +#!/usr/bin/env python3 +"""Verify a v2 secret contract resolved identically across every declared +target (and, for ownership=referenced, that the target matches its Vault +source). + +Never prints secret values - only key names, byte lengths, and sha256 +prefixes. + +Usage: + python3 scripts/verify_secret_contract.py --env brusnika-stage + python3 scripts/verify_secret_contract.py --env brusnika-stage --id regcred + +Env vars: + KUBECONFIG, KUBE_CONTEXT - kubeconfig used by kubectl for kubernetes targets + VAULT_ADDR, VAULT_TOKEN - required for vault targets and secret_ref sources + INFRA_SECRET_VALUES_FILE - pre-decrypted infrastructure-secrets.yaml; if unset, + `sops --decrypt` is called on the file directly +""" +import argparse +import base64 +import hashlib +import json +import os +import subprocess +import sys +import urllib.request + +import yaml + +CANONICAL_SCHEMAS = { + "postgresql.v1": {"host", "port", "database", "username", "password", "sslmode", "ca", "url"}, + "kafka.v1": {"bootstrap_servers", "username", "password", "sasl_mechanism", "security_protocol", "ca"}, + "rabbitmq.v1": {"host", "port", "hostname", "username", "password", "vhost", "uri", "management_endpoint"}, + "s3.v1": {"endpoint", "region", "bucket", "access_key_id", "secret_access_key"}, + "valkey.v1": {"host", "port", "username", "password", "tls", "ca", "url"}, +} + + +def sha256_12(raw: bytes) -> str: + return hashlib.sha256(raw).hexdigest()[:12] + + +def load_secrets_config(repo_root: str) -> dict: + path = os.environ.get("INFRA_SECRET_VALUES_FILE", "") + if path: + with open(path) as f: + raw = f.read() + else: + raw = subprocess.run( + ["sops", "--decrypt", os.path.join(repo_root, "infrastructure-secrets.yaml")], + capture_output=True, text=True, check=True, + ).stdout + return yaml.safe_load(raw) + + +def kubernetes_secret_report(namespace: str, name: str) -> dict: + out = subprocess.run( + ["kubectl", "get", "secret", name, "-n", namespace, "-o", "json"], + capture_output=True, text=True, + ) + if out.returncode != 0: + return {"found": False, "error": out.stderr.strip().splitlines()[-1] if out.stderr else "unknown error"} + doc = json.loads(out.stdout) + fields = {} + for key, v_b64 in (doc.get("data") or {}).items(): + raw = base64.b64decode(v_b64) + fields[key] = {"len": len(raw), "sha256_12": sha256_12(raw)} + return {"found": True, "type": doc.get("type"), "fields": fields} + + +def vault_kv_report(vault_addr: str, vault_token: str, mount: str, path: str) -> dict: + if not vault_addr or not vault_token: + return {"found": False, "error": "VAULT_ADDR/VAULT_TOKEN not set"} + req = urllib.request.Request(f"{vault_addr.rstrip('/')}/v1/{mount}/data/{path}") + req.add_header("X-Vault-Token", vault_token) + try: + resp = urllib.request.urlopen(req, timeout=10) + except Exception as e: # noqa: BLE001 - report and continue, don't crash the whole run + return {"found": False, "error": str(e)} + body = json.loads(resp.read()) + data = (body.get("data") or {}).get("data") or {} + fields = {} + for key, value in data.items(): + raw = value.encode() if isinstance(value, str) else json.dumps(value, sort_keys=True).encode() + fields[key] = {"len": len(raw), "sha256_12": sha256_12(raw)} + return {"found": True, "fields": fields, "version": (body.get("data") or {}).get("metadata", {}).get("version")} + + +def parse_vault_ref(ref: str): + prefix = "vault://" + if not ref.startswith(prefix): + return None, None + rest = ref[len(prefix):] + if "#" in rest: + path, field = rest.split("#", 1) + return path, field + return rest, None + + +def verify_secret(secret: dict, vault_addr: str, vault_token: str, default_kv_mount: str) -> bool: + sid = secret.get("id", "?") + schema = secret.get("schema", "") + print(f"=== {sid} (schema={schema or 'legacy'}) ===") + + reports = [] + + ownership = secret.get("ownership", "") + source = secret.get("source", {}) + if ownership == "referenced" and source.get("kind") == "secret_ref": + src_path, src_field = parse_vault_ref(source.get("ref", "")) + if src_path is None: + print(f" [FAIL] source.ref '{source.get('ref')}' is not a valid vault://# reference") + print(" --> FAIL\n") + return False + r = vault_kv_report(vault_addr, vault_token, default_kv_mount, src_path) + r["target"] = f"vault(source):{default_kv_mount}/{src_path}" + if r["found"] and src_field and src_field not in r["fields"]: + r["found"] = False + r["error"] = f"field '{src_field}' not present in source secret" + reports.append(r) + + for t in secret.get("targets", []): + kind = t.get("kind") + if kind == "kubernetes": + ns = t.get("namespace", secret.get("namespace")) + name = t.get("name", sid) + r = kubernetes_secret_report(ns, name) + r["target"] = f"kubernetes:{ns}/{name}" + elif kind == "vault": + mount = t.get("mount", default_kv_mount) + path = t.get("path", sid) + r = vault_kv_report(vault_addr, vault_token, mount, path) + r["target"] = f"vault:{mount}/{path}" + else: + r = {"found": False, "error": f"unknown target kind '{kind}'", "target": str(kind)} + reports.append(r) + + ok = True + for r in reports: + if not r["found"]: + print(f" [FAIL] {r['target']}: {r.get('error')}") + ok = False + continue + keys = sorted(r["fields"].keys()) + print(f" [OK] {r['target']}: keys={keys}") + for key, f in sorted(r["fields"].items()): + print(f" {key}: len={f['len']} sha256_12={f['sha256_12']}") + + if schema in CANONICAL_SCHEMAS and all(r["found"] for r in reports): + expected = CANONICAL_SCHEMAS[schema] + for r in reports: + got = set(r["fields"].keys()) + missing = expected - got + extra = got - expected + if missing: + print(f" [FAIL] {r['target']}: missing schema field(s) {sorted(missing)}") + ok = False + if extra: + print(f" [WARN] {r['target']}: field(s) not in canonical schema {sorted(extra)}") + + if all(r["found"] for r in reports) and len(reports) > 1: + by_field = {} + for r in reports: + for key, f in r["fields"].items(): + by_field.setdefault(key, []).append((r["target"], f["sha256_12"])) + for key, entries in by_field.items(): + hashes = {h for _, h in entries} + if len(hashes) > 1 and len(entries) > 1: + print(f" [FAIL] field '{key}' differs across targets/source: {entries}") + ok = False + + print(f" --> {'PASS' if ok else 'FAIL'}\n") + return ok + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--env", required=True, help="environment name, e.g. brusnika-stage") + ap.add_argument("--id", action="append", dest="ids", help="limit to this secret id (repeatable); default = every v2 (schema-bearing) secret in the env") + args = ap.parse_args() + + repo_root = subprocess.run( + ["git", "rev-parse", "--show-toplevel"], capture_output=True, text=True, check=True + ).stdout.strip() + + config = load_secrets_config(repo_root) + env = (config.get("environments") or {}).get(args.env) + if env is None: + sys.exit(f"environment '{args.env}' not found in infrastructure-secrets.yaml") + + vault_addr = os.environ.get("VAULT_ADDR", "") + vault_token = os.environ.get("VAULT_TOKEN", "") + default_kv_mount = (env.get("vault") or {}).get("kv_mount", "secrets") + + secrets = [s for s in env.get("secrets", []) if "schema" in s] + if args.ids: + secrets = [s for s in secrets if s.get("id") in args.ids] + if not secrets: + sys.exit("no v2 (schema-bearing) secrets found to verify (check --env / --id)") + + results = [verify_secret(s, vault_addr, vault_token, default_kv_mount) for s in secrets] + print(f"{sum(results)}/{len(results)} secrets passed") + sys.exit(0 if all(results) else 1) + + +if __name__ == "__main__": + main()