← back to Unclaimed Property Platform
tests/test_cycle2_hardening.py
241 lines
"""Cycle 2 hardening tests — each asserts a specific finding from the adversarial review
(Cody + security-auditor + code-reviewer) is actually fixed. Stdlib only, $0.
Run: python -m tests.test_cycle2_hardening
"""
from __future__ import annotations
import sys
from pathlib import Path
from uuid import UUID, uuid4
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from services.common.normalize import mask_name, phonetic_key
from services.common.public_projection import assert_public_safe
from services.common.sqlite_repo import FileObjectStore, SqliteRepository
from services.ingestion.ingest import FeedDefinition, ingest_authorized_feed
from services.claims.claim_workflow import (
Claim, ClaimStatus, complete_state_submission, queue_state_submission, transition_claim,
)
REPO_ROOT = Path(__file__).resolve().parents[1]
SAMPLE = REPO_ROOT / "data" / "sample" / "sample_state_feed.csv"
def ok(msg: str) -> None:
print(f" ✓ {msg}")
def _fresh_repo(tmp: Path):
store = FileObjectStore(tmp / "objstore")
(tmp / "objstore").mkdir(parents=True, exist_ok=True)
store.write_bytes("incoming/sample.csv", SAMPLE.read_bytes())
repo = SqliteRepository(str(tmp / "proto.db"))
return store, repo
# --- 1. Masking: constant width, no length leak (Security 1.1) --------------------------
def test_mask_no_length_leak() -> None:
print("1) Masking hides token length (no confirmation oracle)")
a = mask_name("Jo Li")
b = mask_name("Catherine Oneill")
# both tokens masked to identical width regardless of true length
assert a == "J••• L•••", a
assert b == "C••• O•••", b
# you cannot infer length: a short and a long surname produce the same mask shape
assert len(a.split()[0]) == len(b.split()[0]), (a, b)
ok(f"'{a}' and '{b}' — identical mask width, length not revealed")
# --- 2. Projection allowlist fails closed (Security 1.4/1.6) ----------------------------
def test_projection_fails_closed() -> None:
print("2) Public projection is fail-closed")
assert_public_safe({"public_reference": "x", "jurisdiction": "SAMPLE",
"owner_name_masked": "C•••", "owner_city": "SPRINGFIELD",
"holder_name": "Bank", "property_type": "check", "amount_band": "Under $50"})
for leak in ({"owner_name_raw": "Catherine Oneill"},
{"postal_code": "00001"},
{"amount": "42.50"},
{"owner_ssn": "x"}):
try:
assert_public_safe({"public_reference": "x", **leak})
raise AssertionError(f"assert_public_safe FAILED to catch leak: {leak}")
except ValueError:
pass
ok("allowlisted row passes; raw/postal/amount/ssn leaks all rejected")
# --- 3. masked_search rejects enumeration + returns only public fields (m2, 1.4) ---------
def test_masked_search_guards(tmp: Path) -> None:
print("3) masked_search: no empty-query enumeration, only public fields")
_, repo = _fresh_repo(tmp)
ingest_authorized_feed(FeedDefinition("SAMPLE", "incoming/sample.csv"),
FileObjectStore(tmp / "objstore"), repo)
for bad in ("", " ", "!!!"):
try:
repo.masked_search(bad)
raise AssertionError(f"empty/punct query {bad!r} should have been rejected")
except ValueError:
pass
hits = repo.masked_search("Oneill")
assert hits, "expected a masked hit"
for h in hits:
assert_public_safe(h) # would raise if a raw field leaked
ok(f"empty queries rejected; {len(hits)} hit(s), all pass assert_public_safe")
# --- 4. Idempotent RE-DELIVERY preserves owner_id + versions (C4/Cody#3, M3, m1) --------
def test_redelivery_preserves_history(tmp: Path) -> None:
print("4) Re-delivery: owner_id survives, version history grows, count stable")
store, repo = _fresh_repo(tmp)
ingest_authorized_feed(FeedDefinition("SAMPLE", "incoming/sample.csv"), store, repo)
# find SP-0001's property + owner, and simulate an entity_link on it
prop = repo.conn.execute(
"SELECT property_id FROM property WHERE source_property_id='SP-0001'").fetchone()["property_id"]
owner_id = repo.conn.execute(
"SELECT owner_id FROM owner WHERE property_id=?", (prop,)).fetchone()["owner_id"]
repo.conn.execute("INSERT INTO canonical_entity(entity_id, entity_type) VALUES (?, 'person')",
("E1",))
repo.conn.execute(
"INSERT INTO entity_link(entity_link_id, owner_id, entity_id, score, model_version) "
"VALUES (?,?,?,?,?)", ("L1", owner_id, "E1", 0.99, "test"))
repo.conn.commit()
count_before = repo.count_properties("SAMPLE")
# a NEW file (different checksum) with SP-0001's amount changed 42.50 -> 99.99
changed = SAMPLE.read_bytes().replace(b"42.50", b"99.99", 1)
store.write_bytes("incoming/sample_v2.csv", changed)
r = ingest_authorized_feed(FeedDefinition("SAMPLE", "incoming/sample_v2.csv"), store, repo)
assert r["status"] in ("completed", "completed_with_errors"), r
# property count did NOT grow (upsert in place)
assert repo.count_properties("SAMPLE") == count_before, "re-delivery duplicated a property!"
# owner_id preserved -> entity_link still valid (C4 fixed: no DELETE+FK break)
still = repo.conn.execute(
"SELECT owner_id FROM owner WHERE property_id=?", (prop,)).fetchone()["owner_id"]
assert still == owner_id, "owner_id changed on re-delivery — entity links would break"
link = repo.conn.execute("SELECT 1 FROM entity_link WHERE owner_id=?", (owner_id,)).fetchone()
assert link is not None, "entity_link destroyed by re-delivery"
# two versions now, old one closed with effective_to (M3)
versions = repo.conn.execute(
"SELECT effective_to FROM property_version WHERE property_id=? ORDER BY effective_from",
(prop,)).fetchall()
assert len(versions) == 2, f"expected 2 versions, got {len(versions)}"
assert versions[0]["effective_to"] is not None, "prior version not closed (effective_to null)"
assert versions[1]["effective_to"] is None, "current version should be open"
ok(f"owner_id preserved, entity_link survived, {len(versions)} versions (prior closed)")
# --- 5. No-scrape red line enforced in code (Security 2.1) ------------------------------
def test_ingestion_source_allowlist(tmp: Path) -> None:
print("5) Ingestion refuses network/non-synthetic sources")
store, repo = _fresh_repo(tmp)
for bad in (FeedDefinition("SAMPLE", "https://ca.gov/portal/data.csv"), # network
FeedDefinition("CA", "incoming/real_ca.csv"), # real jurisdiction
FeedDefinition("SAMPLE", "/etc/passwd")): # outside allowlist
try:
ingest_authorized_feed(bad, store, repo)
raise AssertionError(f"ingestion should have refused {bad}")
except PermissionError:
pass
ok("URL source, non-SAMPLE jurisdiction, and out-of-allowlist path all refused")
# --- 6. Bad amount does not kill the batch (C2) ----------------------------------------
def test_bad_amount_tolerated(tmp: Path) -> None:
print("6) A malformed amount is tolerated, not batch-fatal")
store = FileObjectStore(tmp / "obj6")
(tmp / "obj6").mkdir(parents=True, exist_ok=True)
csv = (b"property_id,holder_name,owner_name,address,city,state,zip,property_type,amount\n"
b"BAD-1,Bank,Test Owner,1 St,Town,SAMPLE,00001,check,not-a-number\n"
b"OK-1,Bank,Other Owner,2 St,Town,SAMPLE,00001,check,50.00\n")
store.write_bytes("incoming/bad.csv", csv)
repo = SqliteRepository(str(tmp / "p6.db"))
r = ingest_authorized_feed(FeedDefinition("SAMPLE", "incoming/bad.csv"), store, repo)
assert r["accepted"] == 2, f"expected both rows accepted, got {r}"
amt = repo.conn.execute(
"SELECT amount FROM property WHERE source_property_id='BAD-1'").fetchone()["amount"]
assert amt is None, f"malformed amount should store NULL, got {amt}"
ok("bad-amount row accepted with NULL amount; batch not failed")
# --- 7. Claim workflow: state-only guard + idempotency non-collision (C3, 2.3) ----------
class InMemoryClaimRepo:
"""Minimal ClaimRepository that ENFORCES UNIQUE(claim_id, idempotency_key) so the C3
collision would actually raise if unfixed."""
def __init__(self, claim: Claim) -> None:
self._claim = claim
self._event_keys: set[tuple] = set()
self.outbox: list[dict] = []
def get_for_update(self, claim_id: UUID) -> Claim:
return self._claim
def save(self, claim: Claim) -> None:
self._claim = claim
def append_event(self, claim_id, event_type, payload, idempotency_key) -> None:
key = (claim_id, idempotency_key)
if key in self._event_keys:
raise ValueError(f"UNIQUE(claim_id, idempotency_key) violation: {key}")
self._event_keys.add(key)
def add_outbox_event(self, event_type, aggregate_id, payload) -> None:
self.outbox.append({"event_type": event_type, "payload": payload})
class FakeStateAdapter:
def submit_claim(self, claim, idempotency_key) -> str:
return "STATE-CASE-123"
def test_claim_workflow() -> None:
print("7) Claim workflow: state-only guard + idempotency non-collision")
claim = Claim(claim_id=uuid4(), jurisdiction="SAMPLE", public_property_reference="ref",
claimant_id=uuid4(), status=ClaimStatus.SUBMITTED_TO_STATE, version=1)
repo = InMemoryClaimRepo(claim)
# a non-state actor must NOT be able to approve
try:
transition_claim(repo, claim.claim_id, ClaimStatus.APPROVED,
actor_id="claimant:self", idempotency_key="ik-approve")
raise AssertionError("claimant self-approval should be refused")
except PermissionError:
pass
# a state actor CAN approve
transition_claim(repo, claim.claim_id, ClaimStatus.APPROVED,
actor_id="state:CA-reviewer", idempotency_key="ik-approve")
assert repo._claim.status == ClaimStatus.APPROVED
ok("claimant approval refused; state actor approval allowed")
# idempotency non-collision across the two-step submission (C3)
claim2 = Claim(claim_id=uuid4(), jurisdiction="SAMPLE", public_property_reference="ref",
claimant_id=uuid4(), status=ClaimStatus.READY_FOR_SUBMISSION, version=1)
repo2 = InMemoryClaimRepo(claim2)
queue_state_submission(repo2, claim2.claim_id, actor_id="worker", idempotency_key="ik-sub")
complete_state_submission(repo2, FakeStateAdapter(), claim2.claim_id, idempotency_key="ik-sub")
assert repo2._claim.status == ClaimStatus.SUBMITTED_TO_STATE
assert repo2._claim.state_case_id == "STATE-CASE-123"
ok("queue+complete submission used distinct idempotency keys (no collision)")
def main() -> int:
import tempfile
tmp = Path(tempfile.mkdtemp(prefix="upp-cycle2-"))
test_mask_no_length_leak()
test_projection_fails_closed()
test_masked_search_guards(tmp / "a")
test_redelivery_preserves_history(tmp / "b")
test_ingestion_source_allowlist(tmp / "c")
test_bad_amount_tolerated(tmp / "d")
test_claim_workflow()
print("\nALL CYCLE-2 HARDENING ASSERTIONS PASSED ✅")
return 0
if __name__ == "__main__":
raise SystemExit(main())