ten31-database/backend/redaction/test_scrub_leak.py

#!/usr/bin/env python3
"""Golden-file LEAK TEST for the redaction boundary, hardened across two adversarial
leak-hunts. Synthetic fixtures only (guardrail #9).

Per case: must_vanish (never reach Claude), tier1_excluded (also not in the map),
substance (survives verbatim), perfect inverse, leak-proof audit. Plus a round-2
"hardening vectors" section that regression-locks: NFD/ligature unicode names,
slash/comma SSN + SWIFT + passport Tier-1 drops, sentence-final bare digits, the
rehydrate collision fix, and the FALSE-POSITIVE survival of non-money quantities /
version numbers / ISINs (we de-identify, we don't destroy substance).

Deterministic + offline (the dictionary is each case's own lists; the unknown-name
NER backstop is exercised in test_grounding_boundary.py). Currency-CUED amounts are
caught here; bare magnitudes ('5MM') are left to minimize-first + NER by design.
Run: cd backend && python3 redaction/test_scrub_leak.py
"""
import json
import os
import re
import sqlite3
import sys
import unicodedata

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import scrub as R  # noqa: E402

CASES = [
    {
        "name": "labeled-tier1 + core tier2",
        "raw": ("Jonathan Reyes (jon@cedarpoint.example) at Cedar Point Capital is cooling on Fund III. "
                "Reyes would commit $5,000,000. Wire to acct 000123456789 spooked compliance. Met 1986-03-12. "
                "Substance: the objection is fee load and lock-up; sentiment negative on the energy thesis."),
        "known": {"persons": ["Jonathan Reyes", "Reyes"], "orgs": ["Cedar Point Capital"],
                  "funds": ["Fund III"], "emails": ["jon@cedarpoint.example"]},
        "must_vanish": ["Jonathan Reyes", "Reyes", "jon@cedarpoint.example", "Cedar Point Capital",
                        "Fund III", "$5,000,000", "1986-03-12", "000123456789"],
        "tier1_excluded": ["000123456789"],
        "substance": ["the objection is fee load and lock-up", "sentiment negative on the energy thesis"],
    },
    {
        "name": "worded/coded amounts, intl phone, urls, non-iso dates",
        "raw": ("He would commit five million dollars; a $5MM ticket, USD 5,000,000, and a $3-5M range. "
                "Reach +44 20 7946 0958 or www.cedarpoint.example; profile linkedin.com/in/jreyes. "
                "Met March 12, 1986 and again 3/12/86. Concern: liquidity timeline only."),
        "known": {"persons": [], "orgs": [], "funds": [], "emails": []},
        "must_vanish": ["five million dollars", "$5MM", "USD 5,000,000", "$3-5M", "+44 20 7946 0958",
                        "www.cedarpoint.example", "linkedin.com/in/jreyes", "March 12, 1986", "3/12/86"],
        "tier1_excluded": [],
        "substance": ["Concern: liquidity timeline only"],
    },
    {
        "name": "diacritics + hyphenated + short surnames",
        "raw": ("Spoke to Jonathán Reyés about the thesis. Reyes-Castellanos co-invests. "
                "Wu is warm; Li wants a side letter on fees."),
        "known": {"persons": ["Jonathan Reyes", "Reyes", "Li Wu", "Li", "Wu"], "orgs": [], "funds": [], "emails": []},
        "must_vanish": ["Jonathán", "Reyés", "Castellanos", "Wu", "Li"],
        "tier1_excluded": [],
        "substance": ["wants a side letter on fees"],
    },
    {
        "name": "tier1 separators (slash/comma/space) + swift + address + ext",
        "raw": ("Wire to acct # 1234-5678-9012 spooked compliance. SSN 123/45/6789 and 123 45 6789 on file. "
                "Via SWIFT CHASUS33XXX. Lives at 42 Maple Avenue, Greenwich, CT 06830. Office 212-555-0188 x4021. "
                "Substance: wants a co-investment right."),
        "known": {"persons": [], "orgs": [], "funds": [], "emails": []},
        "must_vanish": ["1234-5678-9012", "123/45/6789", "123 45 6789", "CHASUS33XXX", "42 Maple Avenue",
                        "212-555-0188", "x4021", "06830"],
        "tier1_excluded": ["1234-5678-9012", "123/45/6789", "123 45 6789", "CHASUS33XXX"],
        "substance": ["wants a co-investment right"],
    },
]

FAILS = []


def check(cond, msg):
    print(("  PASS " if cond else "  FAIL ") + msg)
    if not cond:
        FAILS.append(msg)


def tier1_redacted(raw):
    s = unicodedata.normalize("NFKC", raw)
    for _, pat in R.TIER1_PATTERNS:
        s = pat.sub("[redacted]", s)
    return s


def main():
    db = os.path.join(__import__("tempfile").mkdtemp(), "log.db")
    conn = sqlite3.connect(db)
    conn.execute("""CREATE TABLE interaction_log (id TEXT PRIMARY KEY, ts TEXT, actor_type TEXT, actor_id TEXT,
        action TEXT, target_type TEXT, target_id TEXT, payload TEXT, source TEXT, created_at TEXT)""")

    for case in CASES:
        raw, known = case["raw"], case["known"]
        print(f"\n[{case['name']}]")
        check(not R.residual_tokens(raw), "raw fixture has no [TYPE_N]-shaped strings")
        outbound, tmap, audit = R.scrub(raw, known_entities=known, bucket=False)
        for v in case["must_vanish"]:
            check(v not in outbound, f"identifier {v!r} absent from outbound")
        for v in case["tier1_excluded"]:
            check(all(v not in mv for mv in tmap.values()), f"Tier-1 {v!r} excluded, not tokenized")
        for s in case["substance"]:
            check(s in outbound, f"substance survives: {s!r}")
        check(len(set(tmap.values())) == len(tmap), "map injective")
        check(R.rehydrate(outbound, tmap) == tier1_redacted(raw), "rehydrate == raw w/ Tier-1 redacted (perfect inverse)")
        check(not R.residual_tokens(R.rehydrate(outbound, tmap)), "no placeholder survives rehydrate")
        R.log_scrub(conn, "architect", audit, task="g", session_id="t", source="mcp")
        conn.commit()
        blob = " ".join(r[0] for r in conn.execute("SELECT payload FROM interaction_log"))
        check(all(v not in blob for v in case["must_vanish"]), "audit log carries NO sensitive value")

    # ── round-2 hardening vectors ──
    def out(raw, known=None):
        o, _m, _a = R.scrub(raw, known_entities=known or {}, bucket=False)
        return o

    print("\n[unicode — NFD / ligature names]")
    nfd = unicodedata.normalize("NFD", "Jonathan Reyés is cooling.")
    check("Reyés" not in unicodedata.normalize("NFKC", out(nfd, {"persons": ["Jonathan Reyes", "Reyes"]})),
          "NFD-decomposed accented name does not leak")
    check("Steffen" not in out("LP Steffen is cooling.", {"persons": ["Steffen"]}),
          "ligature name (Steffen) does not leak")

    print("\n[tier1 — slash/comma/swift/passport]")
    o, m, _ = R.scrub("Reyes SSN 123/45/6789 and 123,45,6789 on the W9.", known_entities={}, bucket=False)
    check("123/45/6789" not in o and "123,45,6789" not in o, "slash/comma SSN dropped")
    check(all("123/45/6789" not in v and "123,45,6789" not in v for v in m.values()), "SSN not in map (excluded)")
    check("CHASUS33XXX" not in out("Wire via SWIFT CHASUS33XXX today."), "SWIFT/BIC dropped")
    check("a1234567" not in out("Passport number a1234567 expires 2030."), "passport-with-'number' dropped")

    print("\n[bare digits at sentence end]")
    check("123456789012" not in out("The security ID is 123456789012."), "9+ digit run at sentence end tokenized")

    print("\n[FALSE-POSITIVE survival — substance preserved]")
    check("3m tall" in out("The wall is 3m tall."), "'3m tall' (meters) NOT eaten as money")
    check("250k followers" in out("She has 250k followers on X."), "'250k followers' NOT eaten as money")
    check("3.14.159" in out("Pi is roughly 3.14.159 here."), "version-ish number NOT eaten as a date")
    check("US0378331005" in out("We hold ISIN US0378331005 in the sleeve."), "ISIN preserved (substance, not dropped)")
    check("2019-2024" in out("Track record spans 2019-2024."), "year range NOT mislabeled as a phone")

    print("\n[integrity — rehydrate single-pass, no cascade]")
    raw = "Refer to [MISC_2] then [PERSON_9]."
    oo, mm, _ = R.scrub(raw, known_entities={}, bucket=False)
    check(R.rehydrate(oo, mm) == raw, "same-length placeholder literals round-trip without cascade")

    print("\n[round-4 — alpha-prefixed accounts, MM, zero-width]")
    o, m, _ = R.scrub("Acct A123456789012 flagged. Member ID: X4451200931 noted. Wire to GB123456789012 today.",
                      known_entities={}, bucket=False)
    for v in ["A123456789012", "X4451200931", "GB123456789012"]:
        check(v not in o, f"alpha-prefixed labelled identifier {v!r} dropped")
        check(all(v not in mv for mv in m.values()), f"{v!r} excluded, not tokenized")
    o2 = out("Commit of $5MM and €10MM confirmed.")
    check("$5MM" not in o2 and "5M " not in o2 and "MM" not in o2, "double-magnitude $5MM fully tokenized (no stray 'M')")
    zw = "LP Rey​es is cooling."   # zero-width space splitting the surname
    check("Rey​es" not in out(zw, {"persons": ["Reyes"]}) and "Reyes" not in out(zw, {"persons": ["Reyes"]}),
          "zero-width-split known name does not leak")

    print("\n[round-5 — magnitude suffix must not eat a following word]")
    # A single-letter magnitude (k/m/b) immediately before a real word must NOT be
    # consumed as a suffix: '$5,000,000 but' -> the 'b' of 'but' was being eaten,
    # yielding '[AMOUNT_1]ut'. A \b after the magnitude fixes it. Money still vanishes,
    # the following word survives intact, and legitimate suffixes still tokenize.
    for raw, word in [("$5,000,000 but he hesitates", "but he hesitates"),
                      ("committed $250,000 because timing", "because timing"),
                      ("USD 5,000,000 but capped", "but capped"),
                      ("between $3-5M but capped", "but capped")]:
        o = out(raw)
        check("[AMOUNT_1]ut" not in o and "[AMOUNT_1]ecause" not in o, f"magnitude does not bleed into next word: {raw!r}")
        check(word in o, f"following word survives intact: {word!r}")
        check("$" not in o and "USD 5" not in o, f"amount still tokenized: {raw!r}")
    check(out("raised $5m but later") == "raised [AMOUNT_1] but later", "real 'm' suffix still tokenizes ($5m)")
    check(out("about $5b in assets") == "about [AMOUNT_1] in assets", "real 'b' suffix still tokenizes ($5b)")

    conn.close()
    print()
    if FAILS:
        print(f"FAILED ({len(FAILS)}):")
        for f in FAILS:
            print(f"  - {f}")
        sys.exit(1)
    print("ALL PASS (redaction leak test — hardened x2)")


if __name__ == "__main__":
    main()