++ add verify_secret_contract.py for v2 secret acceptance testing

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://<path>#<field> 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()