#!/usr/bin/env python3 # SPDX-License-Identifier: MIT-0 """Toy check: a fake secret must not appear, raw or as a digest, in any output. Everything here is invented for illustration: the secret, the output shapes, the field names. Standard library only; no network, files, accounts or services. Passing this script says nothing about the security of any real system. It only shows the shape of a test you could adapt to your own outputs. Run: python3 secret_metadata_check.py exit 0: every check behaved python3 secret_metadata_check.py --demo-leak exit 1: the "clean" outputs carry the short digest in the legacy record, so the run must fail """ import hashlib import json import sys # An explicitly fake value. It is not a credential for anything. FAKE_SECRET = "FAKE-DOCS-ONLY-not-a-real-key-0123456789" FAKE_DIGEST = hashlib.sha256(FAKE_SECRET.encode("utf-8")).hexdigest() SHORT_DIGEST = FAKE_DIGEST[:12] # the kind of prefix often shown as an "id" # Each representation we treat as forbidden in any output. FORBIDDEN = { "raw value": FAKE_SECRET, "sha256 hex": FAKE_DIGEST, "sha256 prefix": SHORT_DIGEST, } # ---------------------------------------------------------------- outputs -- # Four invented output surfaces an agent platform might produce after an agent # is given a scoped secret. `leak` lets a negative control plant one string at # one named placement; with no leak, every surface is clean. def build_response(leak=None, where=None): return { "status": "ok", "binding": {"service": "example-mail", "scopes": ["send"], "label": "outbound-mail"}, "meta": {"cache": {"hit": False, "key": leak if where == "meta.cache.key" else "c-001"}}, } def build_activity(leak=None, where=None): return [ {"event": "binding.created", "actor": "agent-a", "at": "2000-01-01T00:00:00Z"}, { "event": "binding.updated", "actor": "agent-a", "details": {"changes": [{"attr": "secret", "from": "(set)", "to": leak if where == "details.changes[0].to" else "(set)"}]}, }, ] def build_audit(leak=None, where=None): # The second record is legacy-shaped: flat keys, and its extra metadata is # a JSON document stored as a string rather than a nested object. legacy_extra = {"prev": {"hint": leak if where == "legacy.extra.prev.hint" else "n/a"}} return [ { "schema": 2, "action": "bind", "target": {"service": "example-mail", "labels": {"owner": "team-x", "note": leak if where == "current.target.labels.note" else "rotated monthly"}}, }, {"v": 1, "msg": "bind ok", "svc": "example-mail", "extra": json.dumps(legacy_extra)}, ] def build_receipt(leak=None, where=None): items_by = {"binding": 1} if where == "by_kind": items_by = {leak: 1} # a secret-derived string used as a dict key return { "receipt": "r-0001", "items": [{"kind": "binding", "proof": {"sealed_by": "example-vault", "ref": leak if where == "items[0].proof.ref" else "sealed"}}], "by_kind": items_by, } SURFACES = { "response": (build_response, ["meta.cache.key"]), "activity": (build_activity, ["details.changes[0].to"]), "audit": (build_audit, ["current.target.labels.note", "legacy.extra.prev.hint"]), "receipt": (build_receipt, ["items[0].proof.ref", "by_kind"]), } def build_all(leak_surface=None, leak=None, where=None): return {name: (fn(leak, where) if name == leak_surface else fn()) for name, (fn, _) in SURFACES.items()} # ---------------------------------------------------------------- checker -- def walk(node, path="$"): """Yield (path, text) for every string key and string value, decoding strings that are themselves JSON documents (legacy records do this).""" if isinstance(node, dict): for k, v in node.items(): if isinstance(k, str): yield f"{path}.", k yield from walk(v, f"{path}.{k}") elif isinstance(node, list): for i, v in enumerate(node): yield from walk(v, f"{path}[{i}]") elif isinstance(node, str): yield path, node s = node.strip() if s[:1] in "{[": try: inner = json.loads(s) except ValueError: return yield from walk(inner, f"{path}") def find_leaks(outputs, forbidden=FORBIDDEN): """Return [(surface, path, representation)] for every forbidden hit. Hex digests are matched case-insensitively; the raw value exactly.""" hits = [] for surface, data in outputs.items(): for path, text in walk(data): for rep, needle in forbidden.items(): hay, pin = (text, needle) if rep == "raw value" else (text.lower(), needle.lower()) if pin in hay: hits.append((surface, path, rep)) return hits def naive_find_leaks(outputs): """The check this example argues is not enough: raw value only.""" return find_leaks(outputs, {"raw value": FAKE_SECRET}) # ------------------------------------------------------------------- main -- def main(argv): demo_leak = "--demo-leak" in argv failures = [] print(f"fake secret : {FAKE_SECRET}") print(f"sha256 of fake : {FAKE_DIGEST}") print(f"sha256 prefix (12) : {SHORT_DIGEST}") print() print("negative controls (plant one representation at one placement):") total = caught = naive_missed_digest = digest_controls = 0 for surface, (_, placements) in SURFACES.items(): for where in placements: for rep, needle in FORBIDDEN.items(): total += 1 planted = f"sha256:{needle.upper()}" if rep != "raw value" else needle outputs = build_all(surface, planted, where) hits = find_leaks(outputs) ok = any(h[0] == surface and h[2] == rep for h in hits) caught += ok if rep != "raw value": digest_controls += 1 if not naive_find_leaks(outputs): naive_missed_digest += 1 mark = "DETECTED" if ok else "MISSED" print(f" {mark:8} {surface:8} {where:32} {rep}") if not ok: failures.append(f"missed {rep} at {surface}:{where}") print(f" full check detected {caught}/{total}") print(f" raw-only check missed {naive_missed_digest}/{digest_controls} digest plants") if naive_missed_digest != digest_controls: failures.append("raw-only check unexpectedly caught a digest plant") print() if demo_leak: print("--demo-leak: planting the sha256 prefix in the legacy audit record") clean = build_all("audit", SHORT_DIGEST, "legacy.extra.prev.hint") else: clean = build_all() clean_hits = find_leaks(clean) print(f"clean outputs: {len(clean_hits)} forbidden hits across {len(clean)} surfaces") for surface, path, rep in clean_hits: failures.append(f"clean output leaked {rep} at {surface} {path}") # Guard against a vacuous pass: the checker must actually see the legacy # record's decoded inner document. paths = [p for p, _ in walk(clean["audit"])] if "$[1].extra.prev.hint" not in paths: failures.append("walker did not descend into the legacy JSON string") print() if failures: for f in failures: print(f"FAIL: {f}") print("RESULT: FAIL") return 1 print("RESULT: PASS (toy example; not evidence about any real system)") return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))