terraform-contour-mirror/scripts/verify_secret_contract.py

#!/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()