unbroker

1"""Case ledger: opt-out state machine + append-only audit log.2 3A "case" is one (subject x broker) record. State changes are validated against4TRANSITIONS and mirrored into audit.jsonl so every action is auditable.5"""6from __future__ import annotations7 8import datetime as _dt9from pathlib import Path10 11import paths12import storage13 14STATES = [15    "new", "searching", "not_found", "found", "indirect_exposure", "action_selected", "submitted",16    "verification_pending", "awaiting_processing", "confirmed_removed", "reappeared",17    "human_task_queued", "blocked",18]19 20TRANSITIONS: dict[str, set[str]] = {21    "new": {"searching", "found", "not_found", "indirect_exposure", "blocked"},22    "searching": {"not_found", "found", "indirect_exposure", "blocked"},23    "not_found": {"searching", "found", "indirect_exposure", "blocked"},24    # found -> not_found: a parent re-verification (or re-scan) found the "found" was a false25    # positive (namesake, or an address-only property-record match) -- retract it with evidence.26    "found": {"action_selected", "submitted", "human_task_queued", "indirect_exposure", "blocked",27              "not_found"},28    # indirect_exposure: subject's PII (email/phone/name) sits on a THIRD PARTY's record. The29    # self-service opt-out form does not apply; the lever is a targeted CCPA/GDPR delete-my-PII30    # request (-> submitted) or a human task. Re-scan can clear it (-> not_found) or upgrade it to a31    # direct listing (-> found).32    "indirect_exposure": {"submitted", "human_task_queued", "not_found", "found", "blocked"},33    "action_selected": {"submitted", "human_task_queued", "blocked"},34    "submitted": {"verification_pending", "awaiting_processing", "human_task_queued", "blocked"},35    # verification_pending -> awaiting_processing: the verify link was opened/acknowledged and the36    # broker is now processing the removal (their stated window). confirmed_removed still requires a37    # verifying re-scan, never the submission flow's own say-so.38    "verification_pending": {"awaiting_processing", "confirmed_removed", "human_task_queued", "blocked"},39    "awaiting_processing": {"confirmed_removed", "human_task_queued", "blocked"},40    "confirmed_removed": {"reappeared", "confirmed_removed"},41    "reappeared": {"found", "indirect_exposure"},42    "human_task_queued": {43        "found", "indirect_exposure", "action_selected", "submitted", "verification_pending",44        "awaiting_processing", "confirmed_removed", "blocked",45    },46    # blocked: automated tools (web_extract/proxyless browser) couldn't read the site. A later pass47    # -- a stealth/cloud browser OR guiding the operator's own (residential) browser -- can resolve it48    # to any real scan verdict, so blocked reaches not_found / indirect_exposure too, not just found.49    # blocked -> human_task_queued: some blocked sites need an operator step to proceed at all50    # (face-recognition sites needing a selfie/gov-ID, etc.), so route them to the digest.51    "blocked": {"searching", "found", "not_found", "indirect_exposure", "action_selected",52                "human_task_queued"},53}54 55 56def now() -> str:57    return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")58 59 60def load(subject_id: str) -> dict:61    return storage.read_json(paths.ledger_path(subject_id), {}) or {}62 63 64def save(subject_id: str, ledger: dict) -> Path:65    return storage.write_json(paths.ledger_path(subject_id), ledger)66 67 68def new_case(subject_id: str, broker_id: str) -> dict:69    return {70        "case_id": f"case_{subject_id}_{broker_id}",71        "subject_id": subject_id,72        "broker_id": broker_id,73        "state": "new",74        "found": None,75        "evidence": {},76        "disclosure_log": [],77        "history": [],78    }79 80 81def get_case(subject_id: str, broker_id: str) -> dict:82    return load(subject_id).get(broker_id) or new_case(subject_id, broker_id)83 84 85def can_transition(old: str, new: str) -> bool:86    return new == old or new in TRANSITIONS.get(old, set())87 88 89def transition(subject_id: str, broker_id: str, new_state: str, **fields) -> dict:90    if new_state not in STATES:91        raise ValueError(f"unknown state {new_state!r}")92    # Lock the whole load-modify-save so a concurrent cron re-scan / other tenant93    # can't read a stale ledger and clobber this transition.94    with storage.locked(paths.ledger_path(subject_id)):95        ledger = load(subject_id)96        case = ledger.get(broker_id) or new_case(subject_id, broker_id)97        old = case.get("state", "new")98        if not can_transition(old, new_state):99            raise ValueError(f"illegal transition {old!r} -> {new_state!r} for broker {broker_id!r}")100        case["state"] = new_state101        for key, value in fields.items():102            case[key] = value103        stamp = now()104        case.setdefault("history", []).append({"at": stamp, "from": old, "to": new_state})105        ledger[broker_id] = case106        save(subject_id, ledger)107        storage.append_jsonl(108            paths.audit_path(subject_id),109            {"at": stamp, "broker_id": broker_id, "event": "transition", "from": old, "to": new_state},110        )111        return case112 113 114DEFAULT_PROCESSING_DAYS = 14   # when a broker record doesn't state est_processing_days115VERIFICATION_POLL_DAYS = 1     # how soon to re-poll for an unarrived verification email116 117 118def _plus_days(days: int, start: str | None = None) -> str:119    base = _dt.datetime.strptime(start, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=_dt.timezone.utc) \120        if start else _dt.datetime.now(_dt.timezone.utc)121    return (base + _dt.timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%SZ")122 123 124def followup_fields(new_state: str, broker: dict | None = None,125                    dossier: dict | None = None) -> dict:126    """Auto-scheduling stamps for a transition, so nobody has to remember follow-ups.127 128    submitted / awaiting_processing -> recheck after the broker's stated processing window;129    verification_pending            -> re-poll the inbox quickly;130    confirmed_removed               -> periodic reappearance re-scan per subject preference.131    """132    if new_state in ("submitted", "awaiting_processing"):133        days = ((broker or {}).get("optout") or {}).get("est_processing_days") or DEFAULT_PROCESSING_DAYS134        return {"next_recheck_at": _plus_days(int(days))}135    if new_state == "verification_pending":136        return {"next_recheck_at": _plus_days(VERIFICATION_POLL_DAYS)}137    if new_state == "confirmed_removed":138        interval = ((dossier or {}).get("preferences") or {}).get("rescan_interval_days") or 120139        return {"removal_confirmed_at": now(), "next_recheck_at": _plus_days(int(interval))}140    return {}141 142 143def due(subject_id: str, at: str | None = None, ledger: dict | None = None) -> list[dict]:144    """Cases whose next_recheck_at has arrived - the autonomous follow-up queue."""145    stamp = at or now()146    out = []147    for case in (ledger if ledger is not None else load(subject_id)).values():148        when = case.get("next_recheck_at")149        if when and when <= stamp:150            out.append(case)151    out.sort(key=lambda c: c.get("next_recheck_at") or "")152    return out153 154 155def log_disclosure(subject_id: str, broker_id: str, fields: list[str], channel: str) -> dict:156    """Record exactly which PII field *names* were disclosed to a broker."""157    with storage.locked(paths.ledger_path(subject_id)):158        ledger = load(subject_id)159        case = ledger.get(broker_id) or new_case(subject_id, broker_id)160        stamp = now()161        record = {"at": stamp, "fields": sorted(fields), "channel": channel}162        case.setdefault("disclosure_log", []).append(record)163        ledger[broker_id] = case164        save(subject_id, ledger)165        storage.append_jsonl(166            paths.audit_path(subject_id),167            {"at": stamp, "broker_id": broker_id, "event": "disclosure",168             "fields": record["fields"], "channel": channel},169        )170        return record171