Add regression tests for v74 fixes; close soft-delete leak in list-view aggregates
Lock in the three v0.1.0:74 security/privacy fixes with regression tests, and fix a same-class soft-delete leak surfaced while writing them. - backend/test_assets_traversal.py: boots the real server, proves /assets/ path-traversal vectors (incl. a real decoy file and the live crm.db, plain and URL-encoded) 404 and leak nothing, while a legit asset still serves 200. - backend/test_soft_delete_reads.py: get-by-id 404s soft-deleted rows and nested + list-view aggregates exclude soft-deleted children. - backend/mcp/test_outreach_redaction.py: an unknown free-prose name is tokenized away from the Claude payload but re-hydrated locally, and the path fails closed (no Claude call) when the local NER model is down. - backend/run_tests.py: aggregate runner (each backend/**/test_*.py in its own subprocess); replaces the manual for-loop. 16/16 green. A reviewer pass on the tests confirmed the soft-delete filter was missing from list-view aggregate sub-selects: org contact_count/total_funded and contacts comm_count/last_contact_date counted soft-deleted rows. Add `deleted_at IS NULL` to those four (server.py) and regression-cover them. The reports subsystem (dashboard/pipeline/LP-breakdown, ~16 aggregate queries) has the same leak and is logged as P2 for a dedicated pass. Not yet built or deployed — bump the package version before the next s9pk build.
This commit is contained in:
Keysat
@@ -0,0 +1,123 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Regression test for the outreach NER-backstop wiring (v0.1.0:74).
|
||||
|
||||
The outreach draft path scrubs free-prose LP context (CRM notes + email bodies) before
|
||||
it reaches Claude. The dictionary+regex floor only tokenizes KNOWN CRM entities, so an
|
||||
UNKNOWN person/firm mentioned in an email body would otherwise reach Claude in the clear.
|
||||
The v74 fix wired the local-Qwen NER backstop into draft_outreach (outreach_agent.py:
|
||||
`Boundary(..., ner_fn=_ner_local)`) and made it FAIL CLOSED when the local model is down.
|
||||
|
||||
This drives the real draft_outreach with Claude and the NER model stubbed (offline,
|
||||
synthetic — guardrail #9) and proves:
|
||||
(1) an unknown name in an email body is tokenized AWAY from the Claude payload;
|
||||
(2) it is re-hydrated locally so the human still sees the real name;
|
||||
(3) the interaction_log captures no sensitive value;
|
||||
(4) when the local NER model raises (unreachable), the path returns scrub_unavailable
|
||||
and Claude is never called.
|
||||
|
||||
Run: cd backend && python3 mcp/test_outreach_redaction.py
|
||||
"""
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
sys.path.insert(0, _HERE) # backend/mcp
|
||||
sys.path.insert(0, os.path.dirname(_HERE)) # backend (for the redaction package)
|
||||
|
||||
import outreach_agent as oa # noqa: E402
|
||||
import architect_grounding as G # noqa: E402
|
||||
import architect_agent as aa # noqa: E402 (imports OK offline; client is lazy)
|
||||
|
||||
FAILS = []
|
||||
|
||||
UNKNOWN = "Penelope Ashworth-Vane" # a person in NO CRM table -> only NER can catch her
|
||||
INVESTOR = "Harbor & Vine" # a known org (fundraising_investors) -> dictionary floor
|
||||
|
||||
|
||||
def check(cond, msg):
|
||||
print((" PASS " if cond else " FAIL ") + msg)
|
||||
if not cond:
|
||||
FAILS.append(msg)
|
||||
|
||||
|
||||
def make_db():
|
||||
path = os.path.join(tempfile.mkdtemp(), "crm.db")
|
||||
c = sqlite3.connect(path)
|
||||
c.row_factory = sqlite3.Row
|
||||
c.executescript("""
|
||||
CREATE TABLE fundraising_investors (id TEXT PRIMARY KEY, investor_name TEXT, notes TEXT);
|
||||
CREATE TABLE emails (id TEXT PRIMARY KEY, subject TEXT, body_text TEXT, snippet TEXT, sent_at TEXT,
|
||||
from_email TEXT, to_emails_json TEXT, thread_id TEXT, is_matched INT);
|
||||
CREATE TABLE email_investor_links (id TEXT, email_id TEXT, fundraising_investor_id TEXT);
|
||||
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);
|
||||
""")
|
||||
c.execute("INSERT INTO fundraising_investors VALUES ('inv1',?,?)",
|
||||
(INVESTOR, "Warm on Fund III; weighing lock-up terms."))
|
||||
# The active-thread email body names an UNKNOWN person in free prose.
|
||||
c.execute("INSERT INTO emails (id,subject,body_text,sent_at,thread_id,is_matched) VALUES "
|
||||
"('e1','Re: Fund III',?,?,'t1',1)",
|
||||
(f"Thanks for the call. My partner {UNKNOWN} still has a lock-up objection.", "2026-06-02T10:00:00"))
|
||||
c.execute("INSERT INTO email_investor_links (id,email_id,fundraising_investor_id) VALUES ('l1','e1','inv1')")
|
||||
c.commit()
|
||||
return path, c
|
||||
|
||||
|
||||
def main():
|
||||
db_path, conn = make_db()
|
||||
|
||||
# Stub the thesis fetch (avoid the thesis DB dependency) and Claude. The NER stub stands
|
||||
# in for the local-Qwen model; _draft_with_claude echoes the de-identified text back so
|
||||
# re-hydration is exercised and we can inspect exactly what would have reached Claude.
|
||||
aa.at.get_thesis = lambda *a, **k: {}
|
||||
captured = {}
|
||||
|
||||
def fake_claude(aa_mod, thesis, type_desc, deident_target, deident_voice, guidance):
|
||||
captured["target"] = deident_target
|
||||
return deident_target # passthrough -> rehydrate must restore the real name
|
||||
|
||||
oa._draft_with_claude = fake_claude
|
||||
G._ner_local = lambda text: [(UNKNOWN, "PERSON")] # local model UP, finds the unknown name
|
||||
|
||||
# ── A) unknown name is tokenized away from Claude, restored locally ──
|
||||
print("\n[A — NER backstop tokenizes an unknown name in outreach]")
|
||||
res = oa.draft_outreach(conn, "inv1", "follow_up", "", db_path, sender_email=None)
|
||||
check(res.get("status") == "ok", f"draft ok (status={res.get('status')})")
|
||||
sent = captured.get("target", "")
|
||||
check(UNKNOWN not in sent, "unknown name absent from the Claude payload (NER tokenized it)")
|
||||
check(INVESTOR not in sent, "known investor org absent from the Claude payload (dictionary floor)")
|
||||
check("lock-up" in sent, "objection substance survives to Claude")
|
||||
check(UNKNOWN in res.get("draft", ""), "unknown name re-hydrated locally for the human")
|
||||
|
||||
blob = " ".join(r[0] for r in conn.execute("SELECT payload FROM interaction_log WHERE payload IS NOT NULL"))
|
||||
check(UNKNOWN not in blob and INVESTOR not in blob, "interaction_log carries NO sensitive value")
|
||||
|
||||
# ── B) FAIL CLOSED: local NER model unreachable -> no Claude call ──
|
||||
print("\n[B — fail closed: local NER model down]")
|
||||
called = {"claude": False}
|
||||
|
||||
def boom(text):
|
||||
raise RuntimeError("Spark Control unreachable")
|
||||
|
||||
G._ner_local = boom
|
||||
oa._draft_with_claude = lambda *a, **k: called.__setitem__("claude", True) or a[3]
|
||||
res2 = oa.draft_outreach(conn, "inv1", "follow_up", "", db_path, sender_email=None)
|
||||
check(res2.get("status") == "scrub_unavailable", f"status scrub_unavailable (got {res2.get('status')})")
|
||||
check(bool(res2.get("reason")), "scrub_unavailable carries the propagated NER failure reason (non-vacuous)")
|
||||
check(called["claude"] is False, "Claude was NOT called when the NER model is down (fail closed)")
|
||||
check("draft" not in res2, "no draft returned when scrub fails closed")
|
||||
|
||||
conn.close()
|
||||
print()
|
||||
if FAILS:
|
||||
print(f"FAILED ({len(FAILS)}):")
|
||||
for f in FAILS:
|
||||
print(f" - {f}")
|
||||
sys.exit(1)
|
||||
print("ALL PASS (outreach NER-backstop wiring + fail-closed)")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user