DESIGN.md

@@ -121,9 +121,13 @@ means a future route cannot pass a string where a proof belongs.
 
 ### Suspension is not death
 
-`revoke <agent>` with no end time is permanent: the identity is dead until
-someone re-enrols it. `revoke <agent> --until <unix>` is a pause, and it lapses
-on its own.
+`revoke <agent>` with no end time is permanent: the enrolled public key is
+cleared and the identity stays dead until someone bootstraps a fresh secret and
+calls `enrol_rekey` (then `reinstate` if still banned). `enrol` will not work —
+the agent row still exists — and would only burn the seed. `reinstate` alone
+lifts the ban flags but does **not** restore the compromised key — that was the
+footgun. `revoke <agent> --until <unix>` is a pause that keeps the key, and it
+lapses on its own.
 
 The distinction matters because the two failure modes are different. A key
 compromise wants death — a lapsing revocation would hand the attacker back in
@@ -131,8 +135,18 @@ when the timer ran out, which is a revocation that silently fails at exactly the
 wrong moment. A noisy neighbour or a holiday wants a pause, and re-enrolling for
 that would destroy the audit history that says it was deliberate.
 
-Revoking kills live sessions and drops outstanding challenges, so a revocation
-that races an in-flight authentication cannot be overtaken by it.
+Revoking kills live sessions, drops outstanding challenges, and invalidates
+unused view codes. Rekey invalidates codes the same way: rotating a key must
+close every door the old key could open.
+
+A narrow check-then-insert race can still insert a session or code after the
+revoke/rekey statement has run. The choke point is `sessions.resolve`: every
+use re-checks that the agent is not revoked, still has a live key, and that
+the session fingerprint (when present — always for agent sessions, and for
+human_via_agent sessions minted with a fingerprint) matches the current key
+epoch. View codes store the issuing fingerprint and refuse redeem after rekey
+or key wipe. That is what makes revocation overtake in-flight
+verify/issue/redeem, not the first-pass invalidation alone.
 
 ### Existing controls stay
 
@@ -190,9 +204,15 @@ history, forges replacements, and rebuilds the chain — `verify_chain` passes.
 So the head digest also gets **signed checkpoints**, by an enrolled agent whose
 private key the database does not contain. `verify-chain` runs both checks, and
 the rewrite is caught by the signature the attacker cannot produce. A forged
-checkpoint fails signature verification, a checkpoint from a stranger is refused
-at signing time, and a rekey invalidates every earlier checkpoint — which is
-correct, and says so rather than failing quietly.
+checkpoint fails signature verification, and a checkpoint from a stranger is
+refused at signing time.
+
+Public keys are kept in an append-only `agent_keys` history. Revoke and rekey
+*retire* a key rather than destroy it, so honest pre-compromise checkpoints and
+off-box anchors stay verifiable against the fingerprint they name. Checkpoints
+timestamped after a key's `retired_at` are flagged (not failed). Prefer a
+**dedicated checkpoint-signer identity** that is never a working agent, so a
+routine agent compromise-revoke does not touch the tip-signing key at all.
 
 What this still does not do is survive an attacker who can also delete the
 checkpoints, because they live in the same database. Set `anchor_path` /
@@ -211,7 +231,7 @@ because the review looked for it.
 
 ## Status
 
-106 passing tests, and two hardening passes that closed real defects. The worst were
+158 passing tests, and two hardening passes that closed real defects. The worst were
 concurrency: one-time codes were redeemable repeatedly (753 sessions from 200
 codes under load), and a bootstrap secret could enrol more than one key. Those
 are now atomic compare-and-set operations, each covered by a test that races
@@ -224,7 +244,7 @@ sessions mintable from a bare agent name, no revocation operation at all, an
 unkeyed chain that a database-level attacker could rebuild, and SQLite without
 a busy timeout.
 
-Leora's contracted prod pass (Quarry 18a5d762) closed three remaining review blockers on this branch: rekey no longer un-revokes, dual-bootstrap enrol is exactly-one under race, and redeem rate limits are per-agent (anonymous fails isolated from success). Off-box checkpoint tip publish/verify ships when `anchor_path` / `SIGIL_ANCHOR` is set. Independent review, sealed store, and platform glue remain.
+Leora's contracted prod pass (Quarry 18a5d762) closed three remaining review blockers on this branch: rekey no longer un-revokes, dual-bootstrap enrol is exactly-one under race, and redeem rate limits are per-agent (anonymous fails isolated from success). Off-box checkpoint tip publish/verify ships when `anchor_path` / `SIGIL_ANCHOR` is set. SealedStore / `SIGIL_SEAL_PASS` seals the DB file with AES-256-GCM (incremental path short of SQLCipher). Independent review returned GO-WITH-FIXES; R1–R8/R11 fixes on this branch. Platform glue remains off.
 
 **Still not production ready**, and the reason is not a to-do list:
 
@@ -235,7 +255,9 @@ Leora's contracted prod pass (Quarry 18a5d762) closed three remaining review blo
   signed tip outside the SQLite file and `verify_external_anchors` catches a DB
   rewrite that in-DB checkpoints alone would miss after deletion. Leave it unset
   and the tip stays in-DB only.
-- **The service is unwritten.** TLS, CSRF, transport, secret storage, routes.
+- **The service is unwritten.** TLS, CSRF, transport, routes. DB file seal is
+  available (`SealedStore`, plaintext working copy `0600`); SQLCipher
+  page-level encryption is deferred (not yet available).
 - **Rate limits are per-process**, so N processes means N times the quota.
 - **No key recovery.** Rotate and old sessions die. Deliberate.
 - **Humans remain unattributable** to themselves, which is accepted and is a
@@ -245,11 +267,27 @@ The tests show the logic is correct. They do not make it safe.
 
 ## Next
 
-1. Independent security review before this guards anything real. Two hardening
-   passes made it reviewable; neither substituted for one.
+1. ~~Independent security review~~ — independent review returned GO-WITH-FIXES
+   (tip c0027bb); fix PRs for R1–R8/R11 landed on `sigil-initial`. Platform glue
+   remains off until integration review. See `docs/PROD_RUNBOOK.md`.
 2. ~~Anchor the checkpoint head digest outside the database~~ — done via
    `Store(..., anchor_path=)` / `SIGIL_ANCHOR` (signed tip file + `.log`).
+   **Prod must set it** (see runbook).
 3. Secret broker on top — scoped, short-lived leases bound to a key fingerprint.
    Its issuance and redemption logs *are* the work-attestation substrate, so it
    is the natural second piece.
 4. Deprecate passcode login across the platform, then rotate live credentials.
+
+### Info / follow-ups (no code change required yet)
+
+- **R9 — `issue_agent_session`:** library helper mints `agent:act` from a name
+  after the caller claims to have verified. Keep internal; integration must
+  never mint from headers/names. Consider renaming to `_issue_agent_session`.
+- **R10 — challenge domain separation:** signatures currently cover
+  `nonce.encode()` with no context prefix. Adding
+  `b"sigil-challenge-v1\0" + nonce` is not backward-compatible with in-flight
+  challenges; schedule with a protocol version bump (follow-up).
+- **R12 — multi-process rate limits:** quotas are per-`Store` / per-process.
+  Single writer until a shared limiter exists (documented in README + runbook).
+- **R13 — opt-in anchor:** without `SIGIL_ANCHOR`, deleting in-DB checkpoints
+  allows silent history rewrite. Runbook requires the anchor on separate storage.

README.md

@@ -28,6 +28,7 @@ is **not independently reviewed** — see *Status* below.
 
 ```sh
 export SIGIL_DB=/tmp/demo.db
+# optional: export SIGIL_SEAL_PASS='...'  # AES-GCM seal the DB file at rest
 
 python3 sigil.py seed bigpickle                       # -> one-time secret
 python3 sigil.py keygen /tmp/demo.key                 # -> public_b64, fingerprint
@@ -54,10 +55,10 @@ python3 sigil.py sweep                                # prune stale rows
 python3 -m unittest discover -s tests -v
 ```
 
-119 tests. They cover the refusals, not just the happy paths: replayed
+158 tests. They cover the refusals, not just the happy paths: replayed
 signatures, wrong keys, expired nonces, single-use codes, scope escalation, and
-the invariant below. `test_hardening.py`, `test_production.py` and
-`test_anchor.py` have one test per defect found in review, so any of them
+the invariant below. `test_hardening.py`, `test_production.py`, `test_anchor.py` and
+`test_sealed.py` have one test per defect found in review, so any of them
 regressing brings that defect back.
 
 ## The invariant
@@ -125,6 +126,7 @@ A hardening pass closed the defects below. Each is now covered by a test.
 | Dual-bootstrap enrol race (last writer won) | `put_agent` is INSERT OR IGNORE; exactly one enrol wins |
 | Redeem fail limit was one global bucket | Per-agent `code_redeem` / `code_redeem_fail`; anonymous fails isolated |
 | Checkpoints lived only inside the same DB | Optional off-box signed tip (`SIGIL_ANCHOR` / `anchor_path`) |
+| DB file held roster/audit in cleartext | Optional `SealedStore` / `SIGIL_SEAL_PASS` (AES-256-GCM file seal) |
 
 Still open, and these are the reasons not to ship it yet:
 
@@ -132,10 +134,13 @@ Still open, and these are the reasons not to ship it yet:
   review finds the bugs you thought of; it does not find the ones you did not.
   This needs someone else, ideally a cryptographer, before it guards anything
   real. This is the blocking item, and the only one that code cannot close.
-- **The service around this is unwritten.** TLS, CSRF, session transport, secret
-  storage and the HTTP routes. The library refuses to mint authority from a
-  name, so a route that forgets to authenticate fails closed, but that is only
-  as good as the integration.
+- **The service around this is unwritten.** TLS, CSRF, session transport and the
+  HTTP routes. DB encryption at rest is available via `SealedStore` /
+  `SIGIL_SEAL_PASS` (AES-256-GCM file seal; not SQLCipher page-level). The
+  plaintext working copy is mode `0600` in a private temp dir while open;
+  SQLCipher remains deferred for hostile co-resident hosts. The library
+  refuses to mint authority from a name, so a route that forgets to
+  authenticate fails closed, but that is only as good as the integration.
 - **Off-box anchor is opt-in.** Set `SIGIL_ANCHOR` (or `Store(..., anchor_path=)`)
   and every checkpoint also writes a signed tip outside the SQLite file.
   Rewriting the DB alone can no longer erase the tip. Without that path
@@ -154,3 +159,8 @@ The tests assert the logic is correct. They do not make it safe.
 
 `DESIGN.md` — the design page: rationale, flows, session stamps, the invariant
 and its reasoning, and what is deferred.
+
+`docs/PROD_RUNBOOK.md` — production checklist: require `SIGIL_SEAL_PASS` and
+`SIGIL_ANCHOR` on separate storage, verify-chain alerts, integration sharp edges.
+
+`requirements.txt` — `cryptography>=44.0.1,<45`; see `requirements.lock.txt` / `constraints.txt`.

constraints.txt

@@ -0,0 +1,3 @@
+# Exact pin matching requirements.lock.txt floor. Use with:
+#   pip install -r requirements.txt -c constraints.txt
+cryptography==44.0.1

docs/PROD_RUNBOOK.md

@@ -0,0 +1,46 @@
+# Sigil production runbook
+
+Short checklist before Sigil guards anything real. Library-only today — do **not**
+enable platform glue until an integration review covers session minting.
+
+## Required environment
+
+| Variable | Purpose | Notes |
+|----------|---------|--------|
+| `SIGIL_DB` | Path to the store file | Prefer a dedicated volume; backups are sensitive (roster + audit). |
+| `SIGIL_SEAL_PASS` | AES-256-GCM seal passphrase | **Required in prod.** ≥12 chars. Without it the DB file is cleartext on disk. Seal-on still uses a plaintext `0600` working copy while the process is open — not SQLCipher. |
+| `SIGIL_ANCHOR` | Off-box signed tip path | **Required in prod.** Must be on storage **separate** from `SIGIL_DB` (different volume/host). Every checkpoint publishes here; `verify_external_anchors` / `verify-chain` must alert on failure. |
+
+Optional: agent private keys stay **off** the platform host whenever possible; checkpoint signer key preferably offline.
+
+## Dedicated checkpoint-signer identity
+
+Enrol a separate agent id used **only** for `checkpoint` / `publish_anchor`
+(for example `sigil-checkpoint`). Do not use a working agent as the tip signer:
+compromising and permanently revoking that agent would otherwise retire its key
+(history still verifies, but ops alerting on "key retired" gets noisy). Never
+store the checkpoint signer's private key in the Sigil DB — public history only.
+
+## Ops loop
+
+1. Open with seal + anchor configured (CLI respects the env vars above).
+2. Periodically run chain verify **and** external-anchor verify; page on failure.
+3. After permanent revoke (compromise): bootstrap + `enrol_rekey`, then `reinstate`. Reinstate alone does not restore a wiped key.
+4. Single writer process until a shared rate-limit store exists (in-process quotas multiply across processes).
+
+## Integration sharp edges (do not ignore)
+
+- Never mint `issue_agent_session` from a client-supplied name or header. Only after a verified challenge-response. Prefer keeping that helper internal to verified paths.
+- Human view codes are ~40 bits; keep TTL short; never log raw codes (already hashed in audit).
+- Challenge signatures today sign the raw nonce (no domain separation). Changing the signed payload needs a coordinated version bump — tracked as a follow-up, not a silent break.
+
+## Deferred
+
+- SQLCipher / page-level encryption for multi-tenant co-resident hosts.
+- Domain-separated challenge payloads (`sigil-challenge-v1` prefix) — backward-incompatible without a protocol version.
+- Shared cross-process rate limiter.
+
+## Dependencies
+
+Install with `pip install -r requirements.txt -c constraints.txt`
+(or `--require-hashes -r requirements.lock.txt`). Floor is `cryptography>=44.0.1`.

requirements.lock.txt

@@ -0,0 +1,13 @@
+# Hash-pinned lock for Sigil runtime crypto deps.
+# Install: pip install --require-hashes -r requirements.lock.txt
+# Floor cryptography==44.0.1 addresses CVE-2024-12797 in older wheels.
+
+cryptography==44.0.1 \
+    --hash=sha256:53f23339864b617a3dfc2b0ac8d5c432625c80014c25caac9082314e9de56f41
+
+cffi==2.1.1 \
+    --hash=sha256:a931079504ecc49efed7744c476a5c343a92fabf66dec2db95edb1b2fdc770e2
+
+pycparser==3.0 \
+    --hash=sha256:b727414169a36b7d524c1c3e31839a521725078d7b2ff038656844266160a992
+

requirements.txt

@@ -0,0 +1,5 @@
+# Runtime dependency for Ed25519 + AES-GCM (SealedStore / optional key files).
+# Lower bound 44.0.1: older 43.x/44.0.0 wheels ship OpenSSL affected by
+# CVE-2024-12797 (GHSA-79v4-65xg-pq4g). Upper bound <45 keeps upgrades deliberate.
+# For hash-pinned install see requirements.lock.txt.
+cryptography>=44.0.1,<45

sigil.py

@@ -19,8 +19,9 @@
     sigil.py whoami <token>                   describe a session
     sigil.py events [actor]                   audit trail
 
-Uses SIGIL_DB (default ./sigil.db), SIGIL_KEY (default ./agent.key), and
-optional SIGIL_ANCHOR (filesystem path for the off-box signed tip).
+Uses SIGIL_DB (default ./sigil.db), SIGIL_KEY (default ./agent.key),
+optional SIGIL_ANCHOR (filesystem path for the off-box signed tip), and
+optional SIGIL_SEAL_PASS (when set, SIGIL_DB is an AES-GCM sealed file).
 """
 
 from __future__ import annotations
@@ -31,11 +32,13 @@ import os
 import sys
 
 from sigil import errors, identity, keys, sessions, viewcodes
+from sigil.sealed import SealedStore, is_sealed_file
 from sigil.store import Store
 
 DB = os.environ.get("SIGIL_DB", "./sigil.db")
 KEY = os.environ.get("SIGIL_KEY", "./agent.key")
 ANCHOR = os.environ.get("SIGIL_ANCHOR")  # off-box tip path; unset = in-DB only
+SEAL_PASS = os.environ.get("SIGIL_SEAL_PASS")  # seal passphrase; unset = cleartext DB
 
 
 def out(obj) -> None:
@@ -47,7 +50,14 @@ def main(argv) -> int:
         print(__doc__)
         return 2
     cmd = argv[1]
-    store = Store(DB, anchor_path=ANCHOR)
+    store = None
+    if SEAL_PASS or (os.path.exists(DB) and is_sealed_file(DB)):
+        if not SEAL_PASS:
+            print('{"error":"ValueError","message":"SIGIL_DB is sealed; set SIGIL_SEAL_PASS"}')
+            return 1
+        store = SealedStore(DB, SEAL_PASS, anchor_path=ANCHOR)
+    else:
+        store = Store(DB, anchor_path=ANCHOR)
 
     try:
         if cmd == "seed":
@@ -185,15 +195,26 @@ def main(argv) -> int:
 
     except errors.SigilError as exc:
         print(json.dumps({"error": type(exc).__name__, "message": str(exc)}))
+        _flush_if_sealed(store)
         return 1
     except (ValueError, OSError) as exc:
         # Wrong passphrase, unreadable key file. Report it as a plain result
         # rather than a traceback: it is an ordinary operator mistake.
         print(json.dumps({"error": type(exc).__name__, "message": str(exc)[:200]}))
+        _flush_if_sealed(store)
         return 1
 
+    _flush_if_sealed(store)
     return 0
 
 
+def _flush_if_sealed(store) -> None:
+    if store is not None and isinstance(store, SealedStore):
+        try:
+            store.flush()
+        except Exception:
+            pass
+
+
 if __name__ == "__main__":
     raise SystemExit(main(sys.argv))

sigil/__init__.py

@@ -31,17 +31,20 @@ without the private key, which never leaves the agent.
 
 import time
 
-from . import anchor, errors, identity, keys, sessions, store, viewcodes
+from . import anchor, errors, identity, keys, sealed, sessions, store, viewcodes
 from .store import Store
+from .sealed import SealedStore
 
-__version__ = "0.4.0"
+__version__ = "0.4.1"
 
 __all__ = [
     "Store",
+    "SealedStore",
     "anchor",
     "errors",
     "identity",
     "keys",
+    "sealed",
     "sessions",
     "store",
     "viewcodes",

sigil/identity.py

@@ -56,7 +56,9 @@ def seed_bootstrap(store: Store, agent: str, secret: Optional[str] = None) -> st
     """
     _validate_agent_id(agent)
     row = store.get_agent(agent)
-    if row is not None and is_revoked(row):
+    # Timed suspension: refuse. Permanent revoke (no live key): allow so ops
+    # can stage a rekey before or after reinstate.
+    if row is not None and is_revoked(row) and has_live_key(row):
         raise errors.UnknownAgent(f"{agent} is revoked")
     secret = secret or secrets.token_urlsafe(24)
     store.put_bootstrap(_hash_secret(secret), agent)
@@ -76,6 +78,10 @@ def enrol(
     _validate_agent_id(agent)
     if not store.allow(f"enrol:{agent}"):
         raise errors.RateLimited(f"too many enrolment attempts for {agent}")
+    # Already enrolled (e.g. after permanent revoke): refuse *before* spending
+    # the seed so ops following a wrong "re-enrol" path do not burn it.
+    if store.get_agent(agent) is not None:
+        raise errors.EnrolmentFailed(f"{agent} is already enrolled; use enrol_rekey")
     # Spend the secret first so a burned secret cannot be retried. The agent
     # row insert below is INSERT OR IGNORE: if two secrets race for the same
     # unenrolled agent, exactly one put_agent wins and the loser fails closed
@@ -88,6 +94,8 @@ def enrol(
     fingerprint = keys.fingerprint(public)
     if not store.put_agent(agent, keys.public_bytes(public), fingerprint):
         raise errors.EnrolmentFailed(f"{agent} is already enrolled")
+    # Spend any sibling seeds left over from double-seed / unused chat leaks.
+    store.clear_bootstrap_for_agent(agent)
     store.record(
         "identity.enrolled",
         actor=agent,
@@ -112,6 +120,18 @@ def is_revoked(row, now: Optional[float] = None) -> bool:
     return (now if now is not None else time.time()) < until
 
 
+def has_live_key(row) -> bool:
+    """Does this agent row still hold an enrolled public key?
+
+    Permanent revoke clears pubkey/fingerprint so reinstate alone cannot bring
+    a compromised key back. Callers that mint challenges or sessions must
+    refuse until bootstrap + enrol_rekey restores a real key.
+    """
+    fp = row["fingerprint"] if "fingerprint" in row.keys() else None
+    pubkey = row["pubkey"] if "pubkey" in row.keys() else None
+    return bool(fp) and isinstance(pubkey, (bytes, bytearray)) and len(pubkey) == 32
+
+
 def revoke(store: Store, agent: str, until: Optional[int] = None,
            reason: Optional[str] = None, actor: str = "system") -> int:
     """Revoke an agent, optionally until a time.
@@ -142,7 +162,14 @@ def revoke(store: Store, agent: str, until: Optional[int] = None,
 
 
 def reinstate(store: Store, agent: str, actor: str = "system") -> None:
-    """Lift a revocation. Audited, because it undoes a security decision."""
+    """Lift a revocation. Audited, because it undoes a security decision.
+
+    After a *permanent* revoke the enrolled key was cleared. Reinstate lifts
+    the ban flags only — the agent still cannot authenticate until ops seed a
+    bootstrap secret and call enrol_rekey with a fresh key. Timed suspensions
+    keep the key, so reinstate (or waiting out the window) restores the same
+    identity without rotation.
+    """
     _validate_agent_id(agent)
     if store.get_agent(agent) is None:
         raise errors.UnknownAgent(f"{agent} is not enrolled")
@@ -161,6 +188,10 @@ def challenge(store: Store, agent: str, ttl: int = DEFAULT_NONCE_TTL,
         raise errors.UnknownAgent(f"{agent} is not enrolled")
     if is_revoked(row):
         raise errors.UnknownAgent(f"{agent} is revoked")
+    if not has_live_key(row):
+        raise errors.UnknownAgent(
+            f"{agent} has no live key; bootstrap and enrol_rekey required"
+        )
     nonce = secrets.token_urlsafe(24)
     expires = int(time.time()) + ttl
     store.put_nonce(nonce, agent, expires)
@@ -219,6 +250,10 @@ def verify(
         raise errors.UnknownAgent(f"{agent} is not enrolled")
     if is_revoked(agent_row):
         raise errors.UnknownAgent(f"{agent} is revoked")
+    if not has_live_key(agent_row):
+        raise errors.UnknownAgent(
+            f"{agent} has no live key; bootstrap and enrol_rekey required"
+        )
 
     public = keys.from_public_bytes(agent_row["pubkey"])
     try:
@@ -257,10 +292,17 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
     row = store.get_agent(agent)
     if row is None:
         raise errors.UnknownAgent(f"{agent} is not enrolled")
-    if is_revoked(row):
+    # Permanent revoke clears the key. Allow rekey while still revoked so ops
+    # can rotate before reinstate; a revoked agent that still has a live key
+    # (timed suspend) must not rekey as a back door.
+    if is_revoked(row) and has_live_key(row):
         raise errors.UnknownAgent(f"{agent} is revoked")
     owner = store.spend_bootstrap(_hash_secret(secret))
     if owner != agent:
+        # Rate-limit only failed attempts so junk cannot burn the ops path, and
+        # use a dedicated rekey: bucket so exhausting enrol: cannot lock rekey.
+        if not store.allow(f"rekey:{agent}"):
+            raise errors.RateLimited(f"too many rekey attempts for {agent}")
         raise errors.EnrolmentFailed("bootstrap secret invalid or already spent")
     public = keys.import_public_b64(public_b64)
     fingerprint = keys.fingerprint(public)
@@ -270,6 +312,8 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
     # A new key must not leave old sessions alive, or rotating after a
     # compromise buys nothing.
     killed = store.revoke_sessions_for_agent(agent, "key rotated")
+    # Leftover bootstrap seeds must not remain a standing rekey capability.
+    store.clear_bootstrap_for_agent(agent)
     store.record(
         "identity.rekeyed",
         actor=agent,

sigil/sealed.py

@@ -0,0 +1,227 @@
+"""Sealed store: encryption at rest for the Sigil database file.
+
+Tokens, codes and bootstrap secrets are already stored as digests, so a stolen
+cleartext SQLite file does not yield live credentials. It does still yield the
+full agent roster, revocation state, and audit history. A sealed store wraps
+the on-disk bytes in AES-256-GCM so that material is unreadable without the
+passphrase.
+
+This is an incremental path short of SQLCipher (no page-level encryption, the
+working copy is plaintext in memory / a temp file while open). It is a real
+working seal: the file on disk is ciphertext, and tests assert a stolen seal
+has no SQLite header and no recoverable agent rows without the passphrase.
+
+The plaintext working copy is created mode 0600 inside a private temp dir.
+SQLCipher (or equivalent page-level encryption) remains a deferred prod
+follow-up for multi-tenant / hostile co-resident hosts — never treat seal-on
+as "encrypted DB" while the process is running.
+
+Format on disk (version 1):
+
+    b"SIGILSEAL1" + 16-byte salt + 12-byte nonce + AESGCM(ciphertext)
+    key = PBKDF2-HMAC-SHA256(passphrase, salt, 480_000 iterations, 32 bytes)
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import tempfile
+from typing import Optional
+
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+from cryptography.hazmat.primitives import hashes
+
+from .store import Store
+
+MAGIC = b"SIGILSEAL1"
+SALT_LEN = 16
+NONCE_LEN = 12
+KEY_LEN = 32
+PBKDF2_ITERS = 480_000
+MIN_PASSPHRASE = 12
+
+
+def derive_key(passphrase: str, salt: bytes) -> bytes:
+    if len(passphrase) < MIN_PASSPHRASE:
+        raise ValueError(
+            f"seal passphrase must be at least {MIN_PASSPHRASE} characters"
+        )
+    kdf = PBKDF2HMAC(
+        algorithm=hashes.SHA256(),
+        length=KEY_LEN,
+        salt=salt,
+        iterations=PBKDF2_ITERS,
+    )
+    return kdf.derive(passphrase.encode("utf-8"))
+
+
+def seal_bytes(plaintext: bytes, passphrase: str, *, salt: Optional[bytes] = None) -> bytes:
+    """Encrypt plaintext DB bytes into a sealed blob."""
+    salt = salt or os.urandom(SALT_LEN)
+    key = derive_key(passphrase, salt)
+    nonce = os.urandom(NONCE_LEN)
+    ct = AESGCM(key).encrypt(nonce, plaintext, MAGIC)
+    return MAGIC + salt + nonce + ct
+
+
+def unseal_bytes(blob: bytes, passphrase: str) -> bytes:
+    """Decrypt a sealed blob back to plaintext DB bytes."""
+    if not blob.startswith(MAGIC):
+        raise ValueError("not a Sigil sealed store (bad magic)")
+    salt = blob[len(MAGIC):len(MAGIC) + SALT_LEN]
+    nonce_start = len(MAGIC) + SALT_LEN
+    nonce = blob[nonce_start:nonce_start + NONCE_LEN]
+    ct = blob[nonce_start + NONCE_LEN:]
+    key = derive_key(passphrase, salt)
+    try:
+        return AESGCM(key).decrypt(nonce, ct, MAGIC)
+    except Exception as exc:
+        raise ValueError("wrong passphrase or corrupted sealed store") from exc
+
+
+def is_sealed_file(path: str) -> bool:
+    if not os.path.exists(path):
+        return False
+    with open(path, "rb") as handle:
+        return handle.read(len(MAGIC)) == MAGIC
+
+
+def _write_private_file(path: str, data: bytes) -> None:
+    """Write `data` to `path` with mode 0600, fsync, unlink on failure."""
+    flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
+    fd = os.open(path, flags, 0o600)
+    try:
+        with os.fdopen(fd, "wb") as handle:
+            fd = -1  # ownership transferred
+            handle.write(data)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.chmod(path, 0o600)
+    except Exception:
+        if fd >= 0:
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+        raise
+
+
+class SealedStore(Store):
+    """Store whose on-disk file is AES-GCM sealed.
+
+    Open decrypts into a temp SQLite file (or creates an empty one). Call
+    `flush()` or `close()` to re-encrypt and write the seal back to `path`.
+    The passphrase never touches the SQLite file; only the seal blob does.
+    """
+
+    def __init__(
+        self,
+        path: str,
+        passphrase: str,
+        *,
+        anchor_path: Optional[str] = None,
+    ):
+        if path == ":memory:":
+            raise ValueError("SealedStore needs a filesystem path to seal")
+        if len(passphrase) < MIN_PASSPHRASE:
+            raise ValueError(
+                f"seal passphrase must be at least {MIN_PASSPHRASE} characters"
+            )
+        self.seal_path = path
+        self._passphrase = passphrase
+        self._tmpdir = tempfile.TemporaryDirectory(prefix="sigil-seal-")
+        # TemporaryDirectory is already 0700; pin the dir mode explicitly.
+        os.chmod(self._tmpdir.name, 0o700)
+        self._plain_path = os.path.join(self._tmpdir.name, "store.db")
+        if os.path.exists(path):
+            with open(path, "rb") as handle:
+                blob = handle.read()
+            if not blob.startswith(MAGIC):
+                raise ValueError(
+                    f"{path} exists but is not a Sigil sealed store; "
+                    "refusing to treat a cleartext DB as sealed"
+                )
+            plain = unseal_bytes(blob, passphrase)
+            _write_private_file(self._plain_path, plain)
+        else:
+            # Touch an empty 0600 file so SQLite does not create a world-readable DB.
+            _write_private_file(self._plain_path, b"")
+        super().__init__(self._plain_path, anchor_path=anchor_path)
+        # Re-pin after SQLite open (WAL/SHM sidecars inherit umask otherwise).
+        try:
+            os.chmod(self._plain_path, 0o600)
+        except OSError:
+            pass
+
+    def flush(self) -> str:
+        """Checkpoint WAL, read plaintext bytes, write sealed blob to seal_path."""
+        with self._write_lock:
+            self.conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+            self.conn.commit()
+            # Copy via SQLite backup so we don't race readers mid-page.
+            dest_path = os.path.join(self._tmpdir.name, "export.db")
+            if os.path.exists(dest_path):
+                os.unlink(dest_path)
+            # Pre-create 0600 so SQLite backup does not leave a world-readable export.
+            _write_private_file(dest_path, b"")
+            dest = sqlite3.connect(dest_path)
+            try:
+                self.conn.backup(dest)
+                dest.commit()
+            finally:
+                dest.close()
+            try:
+                os.chmod(dest_path, 0o600)
+            except OSError:
+                pass
+            with open(dest_path, "rb") as handle:
+                plain = handle.read()
+            try:
+                os.unlink(dest_path)
+            except OSError:
+                pass
+            blob = seal_bytes(plain, self._passphrase)
+            directory = os.path.dirname(os.path.abspath(self.seal_path))
+            if directory:
+                os.makedirs(directory, exist_ok=True)
+            fd, tmp = tempfile.mkstemp(
+                prefix=".sigil-seal-", suffix=".tmp", dir=directory or "."
+            )
+            try:
+                with os.fdopen(fd, "wb") as handle:
+                    handle.write(blob)
+                    handle.flush()
+                    os.fsync(handle.fileno())
+                os.replace(tmp, self.seal_path)
+            except Exception:
+                try:
+                    os.unlink(tmp)
+                except OSError:
+                    pass
+                raise
+            os.chmod(self.seal_path, 0o600)
+        return self.seal_path
+
+    def close(self) -> None:
+        """Flush the seal and release the temp working copy."""
+        try:
+            self.flush()
+        finally:
+            try:
+                self.conn.close()
+            except Exception:
+                pass
+            self._tmpdir.cleanup()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        self.close()
+        return False

sigil/sessions.py

@@ -105,9 +105,16 @@ class GrantedCode:
     agent session (issue_code) or a valid code (redeem_code).
     """
 
-    __slots__ = ("agent", "scopes", "ttl", "_sigil_internal")
+    __slots__ = ("agent", "scopes", "ttl", "fingerprint", "_sigil_internal")
 
-    def __init__(self, agent: str, scopes: List[str], ttl: int, _internal=None):
+    def __init__(
+        self,
+        agent: str,
+        scopes: List[str],
+        ttl: int,
+        _internal=None,
+        fingerprint: Optional[str] = None,
+    ):
         if _internal is not _ISSUER:
             raise errors.SigilError(
                 "GrantedCode cannot be constructed directly; it comes from"
@@ -116,6 +123,7 @@ class GrantedCode:
         self.agent = agent
         self.scopes = list(scopes)
         self.ttl = ttl
+        self.fingerprint = fingerprint
         self._sigil_internal = _internal
 
     def __repr__(self) -> str:
@@ -151,6 +159,7 @@ def issue_view_session(store: Store, grant: "GrantedCode") -> dict:
         granted,
         expires,
         granted_by=agent,
+        fingerprint=grant.fingerprint,
     )
     store.record(
         "session.issued",
@@ -176,7 +185,14 @@ def issue_view_session(store: Store, grant: "GrantedCode") -> dict:
 
 
 def resolve(store: Store, token: str) -> dict:
-    """Look up a session and reject it if revoked or expired."""
+    """Look up a session and reject it if revoked, expired, or the agent died.
+
+    The session row alone is not enough: an in-flight verify/issue_code can
+    insert a session or code after revoke/rekey has already run. Re-check the
+    agent on every resolve so a race-surviving bearer cannot keep acting.
+    """
+    from . import identity as _identity
+
     row = store.get_session(hash_token(token))
     if row is None:
         raise errors.SessionUnknown("no such session")
@@ -184,6 +200,24 @@ def resolve(store: Store, token: str) -> dict:
         raise errors.SessionExpired("session revoked")
     if time.time() >= row["expires_at"]:
         raise errors.SessionExpired("session expired")
+
+    agent = row["on_behalf_of"]
+    agent_row = store.get_agent(agent)
+    if agent_row is None or _identity.is_revoked(agent_row):
+        raise errors.SessionExpired("agent revoked")
+    if not _identity.has_live_key(agent_row):
+        raise errors.SessionExpired("agent has no live key")
+    # Every session that carries a fingerprint is bound to that key epoch.
+    # After rekey, race-surviving agent:act *and* human_via_agent bearers die
+    # even if revoke_sessions missed them (N7).
+    sess_fp = row["fingerprint"] or ""
+    live_fp = agent_row["fingerprint"] or ""
+    if sess_fp or row["kind"] == AGENT:
+        # Agent sessions always require a matching fingerprint; human sessions
+        # require a match whenever a fingerprint was recorded at mint time.
+        if not sess_fp or sess_fp != live_fp:
+            raise errors.SessionExpired("session key epoch no longer current")
+
     return {
         "token": token,
         "kind": row["kind"],

sigil/store.py

@@ -36,6 +36,20 @@ CREATE TABLE IF NOT EXISTS agents (
     revoked_reason TEXT
 );
 
+-- Append-only public-key history. Revoke and rekey *retire* a key instead of
+-- destroying it, so historical checkpoints and off-box anchors stay verifiable.
+-- Never stores private keys.
+CREATE TABLE IF NOT EXISTS agent_keys (
+    agent          TEXT NOT NULL,
+    fingerprint    TEXT NOT NULL,
+    pubkey         BLOB NOT NULL,
+    active_from    INTEGER NOT NULL,
+    retired_at     INTEGER,
+    retired_reason TEXT,
+    PRIMARY KEY (agent, fingerprint)
+);
+CREATE INDEX IF NOT EXISTS agent_keys_agent ON agent_keys(agent, active_from);
+
 -- Enrolment bootstrap secrets. Spent secrets are deleted, not marked, so a
 -- spent secret cannot be replayed even if the table is read.
 CREATE TABLE IF NOT EXISTS bootstrap (
@@ -77,7 +91,11 @@ CREATE TABLE IF NOT EXISTS codes (
     redeem_expires_at INTEGER NOT NULL,  -- short: the code is a key
     view_ttl         INTEGER NOT NULL,   -- the door opens a full hour, but only
                                         -- once the key is used
-    used_at         INTEGER
+    used_at         INTEGER,
+    -- Fingerprint of the agent key at issue time. Redeem refuses if the agent
+    -- has since rekeyed (or been wiped), so an in-flight issue_code cannot
+    -- outlive a revoke/rekey that lands before put_code commits.
+    fingerprint     TEXT
 );
 
 -- Append-only, hash-chained audit log. The chain is what makes this an
@@ -130,6 +148,9 @@ RATE_LIMITS = {
     "challenge": (30, 60),      # 30 per 60s
     "verify": (60, 60),
     "enrol": (10, 300),
+    # Separate from enrol: junk pre-auth enrol_rekey must not lock out ops
+    # rotating a compromised agent during incident response.
+    "rekey": (10, 300),
     "code_issue": (20, 300),
     "code_redeem": (20, 60),
     "code_redeem_fail": (10, 300),
@@ -217,6 +238,22 @@ class Store:
             self.conn.execute(
                 "UPDATE codes SET view_ttl = 3600 WHERE view_ttl IS NULL"
             )
+        code_cols = {
+            row["name"] for row in self.conn.execute("PRAGMA table_info(codes)")
+        }
+        if "fingerprint" not in code_cols:
+            self.conn.execute("ALTER TABLE codes ADD COLUMN fingerprint TEXT")
+        # Backfill agent_keys from current agents rows (live or wiped-empty skipped).
+        for row in self.conn.execute(
+            "SELECT agent, pubkey, fingerprint, enrolled_at FROM agents"
+            " WHERE fingerprint IS NOT NULL AND fingerprint != ''"
+            " AND length(pubkey) = 32"
+        ):
+            self.conn.execute(
+                "INSERT OR IGNORE INTO agent_keys"
+                " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)",
+                (row["agent"], row["fingerprint"], row["pubkey"], row["enrolled_at"]),
+            )
         self.conn.commit()
 
     def close(self) -> None:
@@ -253,12 +290,19 @@ class Store:
         which preserves revocation columns.
         """
         with self._write_lock:
+            now = int(time.time())
             cur = self.conn.execute(
                 "INSERT OR IGNORE INTO agents"
                 " (agent, pubkey, fingerprint, enrolled_at)"
                 " VALUES (?, ?, ?, ?)",
-                (agent, pubkey, fingerprint, int(time.time())),
+                (agent, pubkey, fingerprint, now),
             )
+            if cur.rowcount == 1:
+                self.conn.execute(
+                    "INSERT OR IGNORE INTO agent_keys"
+                    " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)",
+                    (agent, fingerprint, pubkey, now),
+                )
             self.conn.commit()
             return cur.rowcount == 1
 
@@ -270,10 +314,33 @@ class Store:
         used to un-revoke. An UPDATE of only the key columns keeps the ban intact.
         """
         with self._write_lock:
+            now = int(time.time())
+            prev = self.conn.execute(
+                "SELECT pubkey, fingerprint, enrolled_at FROM agents WHERE agent = ?",
+                (agent,),
+            ).fetchone()
             cur = self.conn.execute(
                 "UPDATE agents SET pubkey = ?, fingerprint = ? WHERE agent = ?",
                 (pubkey, fingerprint, agent),
             )
+            if cur.rowcount == 1 and prev is not None:
+                old_fp = prev["fingerprint"] or ""
+                if old_fp and len(prev["pubkey"] or b"") == 32:
+                    self.conn.execute(
+                        "INSERT OR IGNORE INTO agent_keys"
+                        " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)",
+                        (agent, old_fp, prev["pubkey"], prev["enrolled_at"] or now),
+                    )
+                    self.conn.execute(
+                        "UPDATE agent_keys SET retired_at = ?, retired_reason = ?"
+                        " WHERE agent = ? AND fingerprint = ? AND retired_at IS NULL",
+                        (now, "rekey", agent, old_fp),
+                    )
+                self.conn.execute(
+                    "INSERT OR IGNORE INTO agent_keys"
+                    " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)",
+                    (agent, fingerprint, pubkey, now),
+                )
             self.conn.commit()
             return cur.rowcount == 1
 
@@ -284,14 +351,47 @@ class Store:
                      reason: Optional[str] = None) -> int:
         """Suspend an agent. `until` of None is permanent.
 
+        A permanent revoke clears the enrolled public key so `reinstate` alone
+        cannot restore a compromised identity — ops must bootstrap + enrol_rekey
+        before the agent can authenticate again. Timed suspension keeps the
+        key: when the pause lapses, the same key works.
+
         Returns how many live sessions were killed.
         """
         with self._write_lock:
-            self.conn.execute(
-                "UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?"
-                " WHERE agent = ?",
-                (int(time.time()), until, reason, agent),
-            )
+            if until is None:
+                # Permanent: retire the public key into agent_keys (so historical
+                # checkpoints stay verifiable), then wipe the live slot.
+                now = int(time.time())
+                prev = self.conn.execute(
+                    "SELECT pubkey, fingerprint, enrolled_at FROM agents WHERE agent = ?",
+                    (agent,),
+                ).fetchone()
+                if prev is not None:
+                    old_fp = prev["fingerprint"] or ""
+                    if old_fp and len(prev["pubkey"] or b"") == 32:
+                        self.conn.execute(
+                            "INSERT OR IGNORE INTO agent_keys"
+                            " (agent, fingerprint, pubkey, active_from)"
+                            " VALUES (?, ?, ?, ?)",
+                            (agent, old_fp, prev["pubkey"], prev["enrolled_at"] or now),
+                        )
+                        self.conn.execute(
+                            "UPDATE agent_keys SET retired_at = ?, retired_reason = ?"
+                            " WHERE agent = ? AND fingerprint = ? AND retired_at IS NULL",
+                            (now, reason or "permanent revoke", agent, old_fp),
+                        )
+                self.conn.execute(
+                    "UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?,"
+                    " pubkey = ?, fingerprint = ? WHERE agent = ?",
+                    (now, until, reason, b"", "", agent),
+                )
+            else:
+                self.conn.execute(
+                    "UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?"
+                    " WHERE agent = ?",
+                    (int(time.time()), until, reason, agent),
+                )
             # Drop outstanding challenges so an in-flight exchange cannot
             # complete after the revocation lands.
             self.conn.execute(
@@ -312,31 +412,68 @@ class Store:
 
     # -- bootstrap ---------------------------------------------------------
     def put_bootstrap(self, secret_hash: str, agent: str) -> None:
-        self._exec(
-            "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
-            (secret_hash, agent, int(time.time())),
-        )
+        """Register one unspent bootstrap secret for an agent.
+
+        Any prior unspent secrets for the same agent are spent first so only
+        one live seed exists at a time (ops double-seed or a chat leak of an
+        unused seed must not leave a standing rekey capability).
+        """
+        with self._write_lock:
+            now = int(time.time())
+            self.conn.execute(
+                "DELETE FROM bootstrap WHERE agent = ? AND spent_at IS NULL",
+                (agent,),
+            )
+            self.conn.execute(
+                "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
+                (secret_hash, agent, now),
+            )
+            self.conn.commit()
+
+    def clear_bootstrap_for_agent(self, agent: str, *, keep_hash: Optional[str] = None) -> int:
+        """Delete every unspent bootstrap secret for an agent.
+
+        After a successful enrol / enrol_rekey, leftover seeds must not remain
+        usable. keep_hash, when set, is left alone (already consumed by CAS).
+        """
+        with self._write_lock:
+            if keep_hash is None:
+                cur = self.conn.execute(
+                    "DELETE FROM bootstrap WHERE agent = ? AND spent_at IS NULL",
+                    (agent,),
+                )
+            else:
+                cur = self.conn.execute(
+                    "DELETE FROM bootstrap WHERE agent = ? AND spent_at IS NULL"
+                    " AND secret_hash != ?",
+                    (agent, keep_hash),
+                )
+            self.conn.commit()
+            return cur.rowcount
 
     def spend_bootstrap(self, secret_hash: str) -> Optional[str]:
         """Atomically consume an enrolment secret. Returns the agent or None.
 
-        Compare-and-set: the UPDATE only matches an unspent row, so exactly one
-        concurrent caller can win. Check-then-update would let a secret enrol
-        more than one key.
+        Compare-and-set then delete: exactly one concurrent caller can win, and
+        the spent row is removed so a table dump cannot offline-crack a
+        low-entropy custom secret. Matches the schema comment.
         """
         with self._write_lock:
+            row = self.conn.execute(
+                "SELECT agent FROM bootstrap WHERE secret_hash = ? AND spent_at IS NULL",
+                (secret_hash,),
+            ).fetchone()
+            if row is None:
+                return None
+            agent = row["agent"]
             cur = self.conn.execute(
-                "UPDATE bootstrap SET spent_at = ? WHERE secret_hash = ?"
-                " AND spent_at IS NULL",
-                (int(time.time()), secret_hash),
+                "DELETE FROM bootstrap WHERE secret_hash = ? AND spent_at IS NULL",
+                (secret_hash,),
             )
             self.conn.commit()
             if cur.rowcount != 1:
                 return None
-            row = self.conn.execute(
-                "SELECT agent FROM bootstrap WHERE secret_hash = ?", (secret_hash,)
-            ).fetchone()
-            return row["agent"] if row else None
+            return agent
 
     # -- nonces ------------------------------------------------------------
     def put_nonce(self, nonce: str, agent: str, expires_at: int) -> None:
@@ -402,12 +539,30 @@ class Store:
             (int(time.time()), token_hash),
         )
 
+    def invalidate_codes_for_agent(self, agent: str) -> int:
+        """Mark every unused view code for an agent as spent.
+
+        Revoke, suspend, and rekey must close outstanding codes the same way they
+        kill sessions. Leaving a pre-issued code redeemable after compromise
+        revoke or key rotation would mint a fresh human_via_agent session while
+        the agent is supposed to be dead.
+        """
+        with self._write_lock:
+            cur = self.conn.execute(
+                "UPDATE codes SET used_at = ? WHERE agent = ? AND used_at IS NULL",
+                (int(time.time()), agent),
+            )
+            self.conn.commit()
+            return cur.rowcount
+
     def revoke_sessions_for_agent(self, agent: str, reason: str) -> int:
         """Kill every live session for an agent. Used on revoke and rekey.
 
         Rotating a key must not leave old sessions usable, otherwise the new key
-        buys nothing after a compromise.
+        buys nothing after a compromise. Outstanding view codes are invalidated
+        in the same breath so a pre-issued code cannot mint a fresh session.
         """
+        self.invalidate_codes_for_agent(agent)
         cur = self._exec(
             "UPDATE sessions SET revoked_at = ?, revoked_reason = ?"
             " WHERE on_behalf_of = ? AND revoked_at IS NULL",
@@ -426,10 +581,11 @@ class Store:
         scopes: List[str],
         redeem_expires_at: int,
         view_ttl: int,
+        fingerprint: Optional[str] = None,
     ) -> None:
         self._exec(
             "INSERT INTO codes (code_hash, agent, scopes, issued_at, redeem_expires_at,"
-            " view_ttl) VALUES (?, ?, ?, ?, ?, ?)",
+            " view_ttl, fingerprint) VALUES (?, ?, ?, ?, ?, ?, ?)",
             (
                 code_hash,
                 agent,
@@ -437,6 +593,7 @@ class Store:
                 int(time.time()),
                 redeem_expires_at,
                 view_ttl,
+                fingerprint,
             ),
         )
 
@@ -444,14 +601,17 @@ class Store:
         return self._one("SELECT * FROM codes WHERE code_hash = ?", (code_hash,))
 
     def claim_code(self, code_hash: str) -> bool:
-        """Consume a one-time code exactly once. False if already used.
+        """Consume a one-time code exactly once. False if already used or expired.
 
-        Atomic, so a code cannot mint two sessions under a race.
+        Atomic CAS includes the expiry predicate so a race at the TTL boundary
+        cannot mint a session slightly after redeem_expires_at.
         """
         with self._write_lock:
+            now = int(time.time())
             cur = self.conn.execute(
-                "UPDATE codes SET used_at = ? WHERE code_hash = ? AND used_at IS NULL",
-                (int(time.time()), code_hash),
+                "UPDATE codes SET used_at = ? WHERE code_hash = ? AND used_at IS NULL"
+                " AND redeem_expires_at > ?",
+                (now, code_hash, now),
             )
             self.conn.commit()
             return cur.rowcount == 1
@@ -580,7 +740,26 @@ class Store:
                 }
             )
         return out
+    def get_key_by_fingerprint(self, agent: str, fingerprint: str):
+        """Return a historical (or live) public key row for verify.
+
+        Prefers agent_keys history. Falls back to the live agents row when the
+        fingerprint still matches (covers DBs created before agent_keys).
+        Never returns private key material — pubkey only.
+        """
+        row = self._one(
+            "SELECT * FROM agent_keys WHERE agent = ? AND fingerprint = ?",
+            (agent, fingerprint),
+        )
+        if row is not None:
+            return row
+        live = self.get_agent(agent)
+        if live is not None and (live["fingerprint"] or "") == fingerprint:
+            return live
+        return None
+
     # -- checkpoints -------------------------------------------------------
+
     @staticmethod
     def _checkpoint_payload(at: int, label: str, head: str, events: int,
                             signer: str, fingerprint: str) -> bytes:
@@ -609,8 +788,19 @@ class Store:
 
         public = _keys.public_from_private(private)
         fingerprint = _keys.fingerprint(public)
-        if self.get_agent(signer) is None:
+        from . import identity as _identity
+
+        agent_row = self.get_agent(signer)
+        if agent_row is None:
             raise ValueError(f"{signer} is not enrolled, so cannot sign a checkpoint")
+        if _identity.is_revoked(agent_row):
+            raise ValueError(f"{signer} is revoked, so cannot sign a checkpoint")
+        enrolled_fp = agent_row["fingerprint"] or ""
+        if not enrolled_fp or fingerprint != enrolled_fp:
+            raise ValueError(
+                f"private key fingerprint does not match enrolled key for {signer}; "
+                "refusing to sign a checkpoint with a stranger key"
+            )
         with self._write_lock:
             last = self.conn.execute(
                 "SELECT id, hash FROM events ORDER BY id DESC LIMIT 1"
@@ -674,17 +864,20 @@ class Store:
         if not notices or notices[-1].get("head") != tip.get("head"):
             notices = list(notices) + [tip]
         results = []
+        warnings = []
         for notice in notices:
-            agent_row = self.get_agent(notice["signer"])
-            if agent_row is None:
+            if self.get_agent(notice["signer"]) is None:
                 results.append({"ok": False, "head": notice.get("head"),
                                 "reason": f"signer {notice['signer']} no longer enrolled"})
                 continue
-            if agent_row["fingerprint"] != notice.get("fingerprint"):
+            key_row = self.get_key_by_fingerprint(
+                notice["signer"], notice.get("fingerprint") or ""
+            )
+            if key_row is None:
                 results.append({"ok": False, "head": notice.get("head"),
-                                "reason": "signer key changed since the tip was published"})
+                                "reason": "signer key for this fingerprint is unknown"})
                 continue
-            public = _keys.from_public_bytes(agent_row["pubkey"])
+            public = _keys.from_public_bytes(key_row["pubkey"])
             checked = _anchor.verify_notice(notice, public)
             if not checked["ok"]:
                 results.append({"ok": False, "head": notice.get("head"),
@@ -704,11 +897,17 @@ class Store:
                               "(database rewrite cannot erase the off-box tip)",
                 })
                 continue
-            results.append({"ok": True, "head": notice["head"],
-                            "events": notice["events"], "signer": notice["signer"]})
+            entry = {"ok": True, "head": notice["head"],
+                     "events": notice["events"], "signer": notice["signer"]}
+            retired_at = key_row["retired_at"] if "retired_at" in key_row.keys() else None
+            if retired_at is not None and notice.get("at", 0) > retired_at:
+                entry["flagged"] = True
+                entry["reason"] = "notice timestamp is after the signing key was retired"
+                warnings.append(entry)
+            results.append(entry)
         bad = [r for r in results if not r["ok"]]
         return {"ok": not bad, "anchors": len(results), "failures": bad,
-                "tip": tip, "path": self.anchor_path}
+                "warnings": warnings, "tip": tip, "path": self.anchor_path}
 
     def checkpoints(self) -> List[Dict[str, Any]]:
         rows = self.conn.execute("SELECT * FROM checkpoints ORDER BY id ASC").fetchall()
@@ -724,22 +923,23 @@ class Store:
         from . import keys as _keys
 
         results = []
+        warnings = []
         for row in self.conn.execute("SELECT * FROM checkpoints ORDER BY id ASC"):
-            agent_row = self.get_agent(row["signer"])
-            if agent_row is None:
+            if self.get_agent(row["signer"]) is None:
                 results.append({"id": row["id"], "ok": False,
                                 "reason": f"signer {row['signer']} no longer enrolled"})
                 continue
-            if agent_row["fingerprint"] != row["fingerprint"]:
+            key_row = self.get_key_by_fingerprint(row["signer"], row["fingerprint"])
+            if key_row is None:
                 results.append({"id": row["id"], "ok": False,
-                                "reason": "signer key changed since the checkpoint was made"})
+                                "reason": "signer key for this fingerprint is unknown"})
                 continue
             payload = self._checkpoint_payload(
                 row["at"], row["label"], row["head"], row["events"],
                 row["signer"], row["fingerprint"],
             )
             try:
-                public = _keys.from_public_bytes(agent_row["pubkey"])
+                public = _keys.from_public_bytes(key_row["pubkey"])
                 # verify() returns None on success and raises on failure, so it
                 # must be called, not inspected.
                 public.verify(base64.b64decode(row["signature"]), payload)
@@ -757,9 +957,16 @@ class Store:
                 results.append({"id": row["id"], "ok": False,
                                 "reason": "chain head at that position no longer matches"})
                 continue
-            results.append({"id": row["id"], "ok": True})
+            entry = {"id": row["id"], "ok": True}
+            retired_at = key_row["retired_at"] if "retired_at" in key_row.keys() else None
+            if retired_at is not None and row["at"] > retired_at:
+                entry["flagged"] = True
+                entry["reason"] = "checkpoint timestamp is after the signing key was retired"
+                warnings.append(entry)
+            results.append(entry)
         bad = [r for r in results if not r["ok"]]
-        return {"ok": not bad, "checkpoints": len(results), "failures": bad}
+        return {"ok": not bad, "checkpoints": len(results),
+                "failures": bad, "warnings": warnings}
 
     # -- housekeeping ------------------------------------------------------
     def sweep(self) -> Dict[str, int]:

sigil/viewcodes.py

@@ -91,9 +91,18 @@ def issue_code(
     if not store.allow(f"code_issue:{agent}", limit=limit, window=window):
         raise errors.RateLimited(f"too many view codes for {agent}")
 
+    from . import identity as _identity
+
     agent_row = store.get_agent(agent)
     if agent_row is None:
         raise errors.UnknownAgent(f"{agent} is not enrolled")
+    # Re-check after resolve: revoke/rekey can land between resolve and put_code.
+    if _identity.is_revoked(agent_row) or not _identity.has_live_key(agent_row):
+        raise errors.UnknownAgent(f"{agent} is revoked or has no live key")
+    live_fp = agent_row["fingerprint"] or ""
+    sess_fp = principal.get("fingerprint") or ""
+    if not live_fp or sess_fp != live_fp:
+        raise errors.SessionExpired("session key epoch no longer current")
 
     granted = list(scopes if scopes is not None else sessions.HUMAN_VIEW_SCOPES)
     # Refuse to bake write or act scopes into a human code at issue time.
@@ -115,12 +124,13 @@ def issue_code(
         granted,
         redeem_expires_at=now + code_ttl,
         view_ttl=view_ttl,
+        fingerprint=live_fp,
     )
     store.record(
         "view_code.issued",
         actor=agent,
         subject=f"{agent}'s human",
-        fingerprint=agent_row["fingerprint"],
+        fingerprint=live_fp,
         detail={
             "code_ref": _code_ref(code),
             "scopes": granted,
@@ -134,7 +144,6 @@ def issue_code(
         "code_ttl": code_ttl,
         "view_ttl": view_ttl,
         "redeem_expires_at": now + code_ttl,
-        "view_ttl": view_ttl,
     }
 
 
@@ -155,6 +164,28 @@ def redeem_code(store: Store, code: str) -> dict:
 
     agent = row["agent"]
 
+    # Defense in depth: refuse if agent revoked, wiped, or rekeyed since issue.
+    from . import identity as _identity
+    agent_row = store.get_agent(agent)
+    code_fp = row["fingerprint"] if "fingerprint" in row.keys() else None
+    live_fp = (agent_row["fingerprint"] or "") if agent_row is not None else ""
+    bad = (
+        agent_row is None
+        or _identity.is_revoked(agent_row)
+        or not _identity.has_live_key(agent_row)
+        or (code_fp is not None and code_fp != live_fp)
+    )
+    if bad:
+        if not store.allow(f"code_redeem_fail:{agent}"):
+            raise errors.RateLimited("too many redemption attempts")
+        store.record(
+            "view_code.redeem_refused",
+            actor=agent,
+            subject=f"{agent}'s human",
+            detail={"reason": "agent revoked, wiped, or rekeyed"},
+        )
+        raise errors.CodeInvalid("that code is not redeemable")
+
     # Known-agent attempts use the success quota (code_redeem), not the fail
     # bucket. Failures for this agent (expired / already used) charge a
     # per-agent fail bucket so one noisy human cannot lock every agent out.
@@ -200,6 +231,7 @@ def redeem_code(store: Store, code: str) -> dict:
         _json.loads(row["scopes"]),
         row["view_ttl"],
         sessions._ISSUER,
+        fingerprint=live_fp,
     )
     return sessions.issue_view_session(store, grant)
 

tests/test_hardening.py

@@ -360,3 +360,215 @@ class TestAgentIdValidation(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestClaimCodeExpiryCAS(unittest.TestCase):
+    """R6: claim_code CAS must include redeem_expires_at predicate."""
+
+    def test_claim_code_after_expiry_returns_false(self):
+        store, private = helpers.fresh()
+        issued = helpers.issue(store, private, code_ttl=1, view_ttl=60)
+        digest = viewcodes._hash_code(issued["code"])
+        # Force expiry in the row.
+        store.conn.execute(
+            "UPDATE codes SET redeem_expires_at = ? WHERE code_hash = ?",
+            (int(time.time()) - 1, digest),
+        )
+        store.conn.commit()
+        self.assertFalse(store.claim_code(digest))
+        row = store.get_code(digest)
+        self.assertIsNone(row["used_at"])
+
+
+class TestEnrolRekeyRateLimit(unittest.TestCase):
+    """N3: enrol_rekey uses a dedicated rekey: bucket; junk cannot lock ops."""
+
+    def test_exhausting_enrol_bucket_does_not_block_rekey(self):
+        store, _ = helpers.fresh("limited")
+        for _ in range(20):
+            store.allow("enrol:limited")
+        secret = identity.seed_bootstrap(store, "limited")
+        # Valid rekey must succeed even if enrol: is exhausted.
+        identity.enrol_rekey(
+            store, "limited", keys.export_public_b64(keys.generate()[1]), secret
+        )
+
+    def test_junk_rekey_fills_rekey_bucket_not_enrol(self):
+        store, _ = helpers.fresh("victim")
+        for _ in range(15):
+            try:
+                identity.enrol_rekey(
+                    store, "victim", keys.export_public_b64(keys.generate()[1]), "junk"
+                )
+            except (errors.EnrolmentFailed, errors.RateLimited):
+                pass
+        # enrol: bucket still usable
+        self.assertTrue(store.allow("enrol:victim"))
+        # Valid ops rekey still works: failures charge rekey: only after bad secret,
+        # and a valid secret bypasses the limiter.
+        secret = identity.seed_bootstrap(store, "victim")
+        identity.enrol_rekey(
+            store, "victim", keys.export_public_b64(keys.generate()[1]), secret
+        )
+
+    def test_failed_rekey_eventually_rate_limited(self):
+        store, _ = helpers.fresh("hammer")
+        hit = False
+        for _ in range(30):
+            try:
+                identity.enrol_rekey(
+                    store, "hammer", keys.export_public_b64(keys.generate()[1]), "nope"
+                )
+            except errors.RateLimited:
+                hit = True
+                break
+            except errors.EnrolmentFailed:
+                pass
+        self.assertTrue(hit, "junk rekey attempts should eventually RateLimited")
+
+
+
+class TestBootstrapRowDeletedOnSpend(unittest.TestCase):
+    """R8: spent bootstrap secrets are deleted, not retained as hashes."""
+
+    def test_spend_deletes_the_row(self):
+        store = Store(":memory:")
+        secret = identity.seed_bootstrap(store, "dave")
+        digest = __import__("hashlib").sha256(secret.encode()).hexdigest()
+        self.assertIsNotNone(store.conn.execute(
+            "SELECT 1 FROM bootstrap WHERE secret_hash = ?", (digest,)
+        ).fetchone())
+        owner = store.spend_bootstrap(digest)
+        self.assertEqual(owner, "dave")
+        self.assertIsNone(store.conn.execute(
+            "SELECT 1 FROM bootstrap WHERE secret_hash = ?", (digest,)
+        ).fetchone())
+
+
+class TestResolveRechecksAgentState(unittest.TestCase):
+    """N1: resolve must refuse sessions that outran revoke/rekey."""
+
+    def _plant_agent_session(self, store, agent, fingerprint):
+        """Insert a live session as if put_session won a race against revoke."""
+        import time as _t
+        token = "race-survivor-token"
+        store.put_session(
+            sessions.hash_token(token),
+            sessions.AGENT,
+            agent,
+            agent,
+            list(sessions.AGENT_SCOPES),
+            int(_t.time()) + 900,
+            fingerprint=fingerprint,
+        )
+        return token
+
+    def test_resolve_refuses_race_survivor_after_permanent_revoke(self):
+        store, private = helpers.fresh()
+        fp = store.get_agent("bigpickle")["fingerprint"]
+        identity.revoke(store, "bigpickle", reason="compromise")
+        # Session inserted *after* revoke (race); not marked revoked_at.
+        token = self._plant_agent_session(store, "bigpickle", fp)
+        with self.assertRaises(errors.SessionExpired):
+            sessions.resolve(store, token)
+
+    def test_resolve_refuses_stale_fingerprint_after_rekey(self):
+        store, private = helpers.fresh()
+        old_fp = store.get_agent("bigpickle")["fingerprint"]
+        new_pub = keys.generate()[1]
+        secret = identity.seed_bootstrap(store, "bigpickle")
+        identity.enrol_rekey(
+            store, "bigpickle", keys.export_public_b64(new_pub), secret
+        )
+        token = self._plant_agent_session(store, "bigpickle", old_fp)
+        with self.assertRaises(errors.SessionExpired):
+            sessions.resolve(store, token)
+
+    def test_redeem_refuses_code_after_rekey_even_if_not_invalidated(self):
+        """Fingerprint binding: plant a code with a stale fingerprint."""
+        store, private = helpers.fresh()
+        agent_row = store.get_agent("bigpickle")
+        old_fp = agent_row["fingerprint"]
+        # Issue normally, then surgically restore used_at and keep old fp after rekey
+        issued = helpers.issue(store, private)
+        digest = viewcodes._hash_code(issued["code"])
+        new_pub = keys.generate()[1]
+        secret = identity.seed_bootstrap(store, "bigpickle")
+        identity.enrol_rekey(
+            store, "bigpickle", keys.export_public_b64(new_pub), secret
+        )
+        # Rekey invalidated the code; un-burn it to simulate put_code after invalidate.
+        store.conn.execute(
+            "UPDATE codes SET used_at = NULL, fingerprint = ? WHERE code_hash = ?",
+            (old_fp, digest),
+        )
+        store.conn.commit()
+        with self.assertRaises(errors.CodeInvalid):
+            viewcodes.redeem_code(store, issued["code"])
+
+    def test_redeem_refuses_without_live_key(self):
+        store, private = helpers.fresh()
+        issued = helpers.issue(store, private)
+        digest = viewcodes._hash_code(issued["code"])
+        identity.revoke(store, "bigpickle", reason="compromise")
+        identity.reinstate(store, "bigpickle")  # ban lifted, key still wiped
+        store.conn.execute(
+            "UPDATE codes SET used_at = NULL WHERE code_hash = ?", (digest,)
+        )
+        store.conn.commit()
+        with self.assertRaises(errors.CodeInvalid):
+            viewcodes.redeem_code(store, issued["code"])
+
+    def test_verify_then_revoke_session_unusable(self):
+        """Session minted just before revoke must not resolve."""
+        store, private = helpers.fresh()
+        chal = identity.challenge(store, "bigpickle")
+        sig = helpers.sign(private, chal["nonce"])
+        sess = identity.verify(store, chal["nonce"], sig)
+        identity.revoke(store, "bigpickle", reason="compromise")
+        with self.assertRaises(errors.SessionExpired):
+            sessions.resolve(store, sess["token"])
+
+
+class TestHumanSessionFingerprintEpoch(unittest.TestCase):
+    """N7: rekey must kill race-surviving human_via_agent sessions too."""
+
+    def test_resolve_refuses_stale_human_fingerprint_after_rekey(self):
+        store, private = helpers.fresh()
+        old_fp = store.get_agent("bigpickle")["fingerprint"]
+        # Plant a human session as if redeem's put_session won a race against rekey.
+        import time as _t
+        token = "human-race-survivor"
+        store.put_session(
+            sessions.hash_token(token),
+            sessions.HUMAN_VIA_AGENT,
+            "bigpickle's human",
+            "bigpickle",
+            list(sessions.HUMAN_VIEW_SCOPES),
+            int(_t.time()) + 3600,
+            granted_by="bigpickle",
+            fingerprint=old_fp,
+        )
+        new_pub = keys.generate()[1]
+        secret = identity.seed_bootstrap(store, "bigpickle")
+        identity.enrol_rekey(
+            store, "bigpickle", keys.export_public_b64(new_pub), secret
+        )
+        # Clear revoked_at that rekey set, to isolate the fingerprint check.
+        store.conn.execute(
+            "UPDATE sessions SET revoked_at = NULL, revoked_reason = NULL"
+            " WHERE token_hash = ?",
+            (sessions.hash_token(token),),
+        )
+        store.conn.commit()
+        with self.assertRaises(errors.SessionExpired) as ctx:
+            sessions.resolve(store, token)
+        self.assertIn("key epoch", str(ctx.exception))
+
+    def test_issue_code_return_has_single_view_ttl(self):
+        """N8: issue_code return dict must not duplicate view_ttl."""
+        import inspect
+        from sigil import viewcodes as vc
+        src = inspect.getsource(vc.issue_code)
+        # Count literal "view_ttl" keys in the return dict construction region.
+        self.assertEqual(src.count('"view_ttl":'), 1, src[src.rfind("return"):])

tests/test_identity.py

@@ -6,6 +6,8 @@ import unittest
 from sigil import errors, identity, keys
 from sigil.store import Store
 
+import helpers
+
 
 class Base(unittest.TestCase):
     def setUp(self):
@@ -144,5 +146,59 @@ class TestRekey(Base):
         )
 
 
+class TestBootstrapHygiene(unittest.TestCase):
+    """R2: only one live bootstrap secret; leftovers unusable after enrol."""
+
+    def test_leftover_seed_fails_after_enrol(self):
+        store = Store(":memory:")
+        private, public = keys.generate()
+        s1 = identity.seed_bootstrap(store, "alice")
+        # Direct insert of a second hash simulates a pre-hygiene double-seed
+        # (put_bootstrap now clears siblings; plant one after s1 is recorded).
+        import hashlib
+        s2 = "leftover-seed-that-must-die"
+        store.conn.execute(
+            "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
+            (hashlib.sha256(s2.encode()).hexdigest(), "alice", 1),
+        )
+        store.conn.commit()
+        identity.enrol(store, "alice", keys.export_public_b64(public), s1)
+        with self.assertRaises(errors.EnrolmentFailed):
+            identity.enrol_rekey(
+                store, "alice", keys.export_public_b64(public), s2
+            )
+
+    def test_put_bootstrap_keeps_only_one_live_secret(self):
+        store = Store(":memory:")
+        s1 = identity.seed_bootstrap(store, "bob")
+        s2 = identity.seed_bootstrap(store, "bob")
+        private, public = keys.generate()
+        with self.assertRaises(errors.EnrolmentFailed):
+            identity.enrol(store, "bob", keys.export_public_b64(public), s1)
+        # Only the latest seed works.
+        identity.enrol(store, "bob", keys.export_public_b64(public), s2)
+
+    def test_leftover_seed_fails_after_rekey(self):
+        store = Store(":memory:")
+        helpers.enrol(store, "carol")
+        import hashlib
+        leftover = "another-unused-seed"
+        new_pub = keys.generate()[1]
+        secret = identity.seed_bootstrap(store, "carol")
+        # Plant a sibling after the live seed (simulates a concurrent leak).
+        store.conn.execute(
+            "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
+            (hashlib.sha256(leftover.encode()).hexdigest(), "carol", 2),
+        )
+        store.conn.commit()
+        identity.enrol_rekey(
+            store, "carol", keys.export_public_b64(new_pub), secret
+        )
+        newer_pub = keys.generate()[1]
+        with self.assertRaises(errors.EnrolmentFailed):
+            identity.enrol_rekey(
+                store, "carol", keys.export_public_b64(newer_pub), leftover
+            )
+
 if __name__ == "__main__":
     unittest.main()

tests/test_production.py

@@ -133,15 +133,25 @@ class TestSignedCheckpoints(unittest.TestCase):
         with self.assertRaises(ValueError):
             stranger.checkpoint(self.private, "not-enrolled")
 
-    def test_checkpoint_after_rekey_is_flagged(self):
+    def test_checkpoint_refuses_enrolled_name_with_wrong_key(self):
+        # R4: signer name is enrolled, but private key is not the enrolled one.
+        before = len(self.store.checkpoints())
+        other_private, _ = keys.generate()
+        with self.assertRaises(ValueError) as ctx:
+            self.store.checkpoint(other_private, "bigpickle")
+        self.assertIn("does not match enrolled key", str(ctx.exception))
+        self.assertEqual(len(self.store.checkpoints()), before)
+
+    def test_checkpoint_after_rekey_still_verifies_via_history(self):
+        # N2: historical checkpoints verify against retired public keys.
         secret = identity.seed_bootstrap(self.store, "bigpickle")
         identity.enrol_rekey(
             self.store, "bigpickle",
             keys.export_public_b64(keys.generate()[1]), secret,
         )
         result = self.store.verify_checkpoints()
-        self.assertFalse(result["ok"])
-        self.assertIn("key changed", result["failures"][0]["reason"])
+        self.assertTrue(result["ok"], result)
+        self.assertEqual(result["failures"], [])
 
     def test_empty_log_checkpoints_cleanly(self):
         store, private = helpers.fresh()
@@ -193,10 +203,19 @@ class TestRevocationSemantics(unittest.TestCase):
     def test_reinstating_works_and_is_audited(self):
         identity.revoke(self.store, "bigpickle")
         identity.reinstate(self.store, "bigpickle")
+        # Permanent revoke cleared the key; reinstate alone is not enough.
+        with self.assertRaises(errors.UnknownAgent):
+            identity.challenge(self.store, "bigpickle")
+        new_pub = keys.generate()[1]
+        secret = identity.seed_bootstrap(self.store, "bigpickle")
+        identity.enrol_rekey(
+            self.store, "bigpickle", keys.export_public_b64(new_pub), secret
+        )
         identity.challenge(self.store, "bigpickle")
         kinds = [e["kind"] for e in self.store.events(limit=50)]
         self.assertIn("identity.revoked", kinds)
         self.assertIn("identity.reinstated", kinds)
+        self.assertIn("identity.rekeyed", kinds)
 
     def test_revoking_an_unknown_agent_refused(self):
         with self.assertRaises(errors.UnknownAgent):
@@ -230,6 +249,33 @@ class TestRevocationSemantics(unittest.TestCase):
                             helpers.sign(self.private, chal["nonce"]))
 
 
+class TestCompromiseReinstateRequiresRekey(unittest.TestCase):
+    """R3: reinstate after permanent revoke must not restore the same key."""
+
+    def setUp(self):
+        self.store, self.private = helpers.fresh()
+
+    def test_reinstate_alone_cannot_use_old_key(self):
+        old_token_path_chal = identity.challenge(self.store, "bigpickle")
+        self.assertTrue(old_token_path_chal["fingerprint"])
+        identity.revoke(self.store, "bigpickle", reason="key leak")
+        identity.reinstate(self.store, "bigpickle")
+        with self.assertRaises(errors.UnknownAgent):
+            identity.challenge(self.store, "bigpickle")
+        # Old private key must not verify even if a nonce were planted.
+        self.assertFalse(identity.has_live_key(self.store.get_agent("bigpickle")))
+
+    def test_timed_suspend_reinstate_keeps_same_key(self):
+        import time
+        fp = self.store.get_agent("bigpickle")["fingerprint"]
+        identity.revoke(
+            self.store, "bigpickle", until=int(time.time()) + 60, reason="pause"
+        )
+        identity.reinstate(self.store, "bigpickle")
+        chal = identity.challenge(self.store, "bigpickle")
+        self.assertEqual(chal["fingerprint"], fp)
+
+
 class TestViewSessionsCannotBeForged(unittest.TestCase):
     """Finding: human sessions were minted from an agent name.
 
@@ -300,12 +346,14 @@ class TestRekeyDoesNotUnrevoke(unittest.TestCase):
     def setUp(self):
         self.store, self.private = helpers.fresh()
 
-    def test_rekey_refuses_while_revoked(self):
-        identity.revoke(self.store, "bigpickle", reason="incident")
-        secret = None
+    def test_rekey_refuses_while_timed_suspended(self):
+        import time
+        # Timed suspend keeps the key — rekey must not be a back door.
+        identity.revoke(
+            self.store, "bigpickle", until=int(time.time()) + 3600, reason="pause"
+        )
         with self.assertRaises(errors.UnknownAgent):
-            secret = identity.seed_bootstrap(self.store, "bigpickle")
-        # Even with a secretly planted bootstrap, rekey must refuse.
+            identity.seed_bootstrap(self.store, "bigpickle")
         planted = secrets_module.token_urlsafe(24)
         self.store.put_bootstrap(
             __import__("hashlib").sha256(planted.encode()).hexdigest(),
@@ -320,6 +368,28 @@ class TestRekeyDoesNotUnrevoke(unittest.TestCase):
             )
         row = self.store.get_agent("bigpickle")
         self.assertTrue(identity.is_revoked(row))
+        self.assertTrue(identity.has_live_key(row))
+
+    def test_permanent_revoke_allows_rekey_but_stays_revoked(self):
+        """Compromise recovery: rotate key while banned, then reinstate."""
+        old_fp = self.store.get_agent("bigpickle")["fingerprint"]
+        identity.revoke(self.store, "bigpickle", reason="compromise")
+        row = self.store.get_agent("bigpickle")
+        self.assertTrue(identity.is_revoked(row))
+        self.assertFalse(identity.has_live_key(row))
+        new_pub = keys.generate()[1]
+        secret = identity.seed_bootstrap(self.store, "bigpickle")
+        new_fp = identity.enrol_rekey(
+            self.store, "bigpickle", keys.export_public_b64(new_pub), secret
+        )
+        row = self.store.get_agent("bigpickle")
+        self.assertTrue(identity.is_revoked(row))
+        self.assertTrue(identity.has_live_key(row))
+        self.assertNotEqual(new_fp, old_fp)
+        with self.assertRaises(errors.UnknownAgent):
+            identity.challenge(self.store, "bigpickle")
+        identity.reinstate(self.store, "bigpickle")
+        identity.challenge(self.store, "bigpickle")
 
     def test_rekey_preserves_revocation_columns(self):
         # Reinstate path: if somehow update ran under a revoked row, columns stay.
@@ -345,10 +415,17 @@ class TestDualBootstrapEnrolRace(unittest.TestCase):
         import threading
 
         store = Store(":memory:")
-        secrets = [
-            identity.seed_bootstrap(store, "racer"),
-            identity.seed_bootstrap(store, "racer"),
-        ]
+        # Plant two live secrets directly: put_bootstrap now keeps only one
+        # live seed, so the dual-enrol race must bypass that hygiene.
+        import hashlib, time as _time
+        secrets = ["race-secret-alpha", "race-secret-beta"]
+        now = int(_time.time())
+        for s in secrets:
+            store.conn.execute(
+                "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
+                (hashlib.sha256(s.encode()).hexdigest(), "racer", now),
+            )
+        store.conn.commit()
         pubs = [keys.generate()[1] for _ in range(2)]
         results = []
         lock = threading.Lock()
@@ -415,3 +492,75 @@ class TestRedeemRateLimitIsolation(unittest.TestCase):
             self.fail("successes must not share the fail bucket")
         except errors.SigilError:
             pass
+
+
+class TestKeyHistoryKeepsCheckpointsVerifiable(unittest.TestCase):
+    """N2: permanent revoke / rekey must not make honest checkpoints unverifiable."""
+
+    def test_checkpoints_verify_after_signer_permanent_revoke(self):
+        import tempfile, os
+        d = tempfile.mkdtemp()
+        anchor = os.path.join(d, "tip")
+        store = Store(":memory:", anchor_path=anchor)
+        private, _ = helpers.enrol(store, "signer")
+        store.record("identity.enrolled", actor="signer", subject="signer")
+        store.checkpoint(private, "signer", "pre-compromise")
+        before = store.verify_checkpoints()
+        self.assertTrue(before["ok"], before)
+        anchors_before = store.verify_external_anchors()
+        self.assertTrue(anchors_before["ok"], anchors_before)
+        identity.revoke(store, "signer", reason="compromise")
+        after = store.verify_checkpoints()
+        self.assertTrue(after["ok"], after)
+        self.assertEqual(after["failures"], [])
+        anchors_after = store.verify_external_anchors()
+        self.assertTrue(anchors_after["ok"], anchors_after)
+        # Live key wiped; history retained.
+        self.assertFalse(identity.has_live_key(store.get_agent("signer")))
+        hist = store.conn.execute(
+            "SELECT fingerprint, retired_at FROM agent_keys WHERE agent = ?",
+            ("signer",),
+        ).fetchall()
+        self.assertEqual(len(hist), 1)
+        self.assertIsNotNone(hist[0]["retired_at"])
+
+    def test_private_key_never_in_agent_keys(self):
+        store, private = helpers.fresh("carol")
+        cols = [r["name"] for r in store.conn.execute("PRAGMA table_info(agent_keys)")]
+        self.assertNotIn("private", cols)
+        self.assertNotIn("privkey", cols)
+        row = store.conn.execute("SELECT * FROM agent_keys WHERE agent = ?", ("carol",)).fetchone()
+        self.assertEqual(len(row["pubkey"]), 32)
+
+
+class TestEnrolAfterPermanentRevoke(unittest.TestCase):
+    """N4: enrol must not burn the seed when the agent row still exists."""
+
+    def test_enrol_after_permanent_revoke_does_not_burn_seed(self):
+        store, private = helpers.fresh()
+        identity.revoke(store, "bigpickle", reason="compromise")
+        identity.reinstate(store, "bigpickle")
+        secret = identity.seed_bootstrap(store, "bigpickle")
+        with self.assertRaises(errors.EnrolmentFailed) as ctx:
+            identity.enrol(
+                store, "bigpickle", keys.export_public_b64(keys.generate()[1]), secret
+            )
+        self.assertIn("enrol_rekey", str(ctx.exception))
+        # Seed still usable for the correct path.
+        identity.enrol_rekey(
+            store, "bigpickle", keys.export_public_b64(keys.generate()[1]), secret
+        )
+
+
+class TestSuspendedSignerCannotCheckpoint(unittest.TestCase):
+    """N5: timed-suspended signer must not checkpoint."""
+
+    def test_timed_suspend_refuses_checkpoint(self):
+        import time
+        store, private = helpers.fresh()
+        identity.revoke(
+            store, "bigpickle", until=int(time.time()) + 3600, reason="pause"
+        )
+        with self.assertRaises(ValueError) as ctx:
+            store.checkpoint(private, "bigpickle")
+        self.assertIn("revoked", str(ctx.exception).lower())

tests/test_sealed.py

@@ -0,0 +1,153 @@
+"""Sealed store: encryption at rest for the Sigil database file.
+
+Finding: digests stop a stolen DB from yielding live credentials, but the file
+still holds the agent roster, revocation state and audit history in the clear.
+SealedStore wraps on-disk bytes in AES-256-GCM so that material is unreadable
+without the passphrase. Short of SQLCipher (no page-level encryption while
+open); real working seal of the sensitive file.
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import tempfile
+import unittest
+
+from sigil import identity, keys
+from sigil.sealed import (
+    MAGIC,
+    SealedStore,
+    is_sealed_file,
+    seal_bytes,
+    unseal_bytes,
+)
+from sigil.store import Store
+from tests import helpers
+
+
+PASS = "test-seal-passphrase-ok"
+
+
+class TestSealBytes(unittest.TestCase):
+    def test_round_trip(self):
+        blob = seal_bytes(b"SQLite format 3\x00hello", PASS)
+        self.assertTrue(blob.startswith(MAGIC))
+        self.assertEqual(unseal_bytes(blob, PASS), b"SQLite format 3\x00hello")
+
+    def test_wrong_passphrase_refused(self):
+        blob = seal_bytes(b"secret-db-bytes", PASS)
+        with self.assertRaises(ValueError):
+            unseal_bytes(blob, "wrong-passphrase!!")
+
+    def test_short_passphrase_refused(self):
+        with self.assertRaises(ValueError):
+            seal_bytes(b"x", "short")
+
+    def test_corrupt_magic_refused(self):
+        with self.assertRaises(ValueError):
+            unseal_bytes(b"NOTSEALED" + b"\x00" * 40, PASS)
+
+
+class TestSealedStore(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.path = os.path.join(self.tmp.name, "sigil.seal")
+
+    def test_flush_writes_ciphertext_without_sqlite_header(self):
+        store = SealedStore(self.path, PASS)
+        helpers.enrol(store, "bigpickle")
+        store.flush()
+        with open(self.path, "rb") as handle:
+            raw = handle.read()
+        self.assertTrue(raw.startswith(MAGIC))
+        self.assertNotIn(b"SQLite format 3", raw)
+        self.assertTrue(is_sealed_file(self.path))
+        # Agent id must not appear in the clear on disk.
+        self.assertNotIn(b"bigpickle", raw)
+        store.close()
+
+    def test_reopen_with_passphrase_recovers_agents(self):
+        with SealedStore(self.path, PASS) as store:
+            private, _ = helpers.enrol(store, "bigpickle")
+            store.flush()
+        with SealedStore(self.path, PASS) as store:
+            row = store.get_agent("bigpickle")
+            self.assertIsNotNone(row)
+            self.assertEqual(row["fingerprint"], keys.fingerprint(
+                keys.public_from_private(private)
+            ))
+
+    def test_reopen_with_wrong_passphrase_fails(self):
+        with SealedStore(self.path, PASS) as store:
+            helpers.enrol(store, "bigpickle")
+            store.flush()
+        with self.assertRaises(ValueError):
+            SealedStore(self.path, "definitely-wrong-passphrase")
+
+    def test_stolen_seal_is_not_a_sqlite_database(self):
+        with SealedStore(self.path, PASS) as store:
+            helpers.enrol(store, "bigpickle")
+            identity.challenge(store, "bigpickle")
+            store.flush()
+        with self.assertRaises(sqlite3.DatabaseError):
+            sqlite3.connect(self.path).execute("SELECT name FROM sqlite_master")
+
+    def test_refuses_to_open_cleartext_db_as_sealed(self):
+        clear = os.path.join(self.tmp.name, "clear.db")
+        Store(clear)  # creates a normal sqlite file
+        with self.assertRaises(ValueError) as ctx:
+            SealedStore(clear, PASS)
+        self.assertIn("not a Sigil sealed store", str(ctx.exception))
+
+    def test_seal_permissions_are_owner_only(self):
+        with SealedStore(self.path, PASS) as store:
+            helpers.enrol(store, "bigpickle")
+            store.flush()
+        mode = os.stat(self.path).st_mode & 0o777
+        self.assertEqual(mode, 0o600)
+
+    def test_sealed_store_works_with_offbox_anchor(self):
+        tip = os.path.join(self.tmp.name, "chain.tip")
+        with SealedStore(self.path, PASS, anchor_path=tip) as store:
+            private, _ = helpers.enrol(store, "bigpickle")
+            identity.challenge(store, "bigpickle")
+            result = store.checkpoint(private, "bigpickle", "sealed-daily")
+            self.assertEqual(result.get("anchor_path"), tip)
+            self.assertTrue(os.path.exists(tip))
+            self.assertTrue(store.verify_external_anchors()["ok"])
+            store.flush()
+        self.assertTrue(is_sealed_file(self.path))
+        self.assertTrue(os.path.exists(tip))
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class TestPlaintextWorkingCopyHardening(unittest.TestCase):
+    """R5: plaintext tmp must be 0600; SQLCipher remains deferred."""
+
+    def test_plaintext_working_copy_is_owner_only(self):
+        import tempfile
+        from sigil.sealed import SealedStore
+        from sigil import identity, keys
+
+        d = tempfile.mkdtemp()
+        path = os.path.join(d, "sealed.db")
+        store = SealedStore(path, "test-passphrase-ok")
+        try:
+            mode = os.stat(store._plain_path).st_mode & 0o777
+            self.assertEqual(mode, 0o600, oct(mode))
+            dir_mode = os.stat(store._tmpdir.name).st_mode & 0o777
+            self.assertEqual(dir_mode, 0o700, oct(dir_mode))
+            secret = identity.seed_bootstrap(store, "alice")
+            _, pub = keys.generate()
+            identity.enrol(store, "alice", keys.export_public_b64(pub), secret)
+            store.flush()
+            # export.db must not linger world-readable after flush
+            export = os.path.join(store._tmpdir.name, "export.db")
+            self.assertFalse(os.path.exists(export))
+        finally:
+            store.close()

tests/test_viewcodes.py

@@ -143,3 +143,48 @@ class TestAuditTrail(Base):
 
 if __name__ == "__main__":
     unittest.main()
+
+class TestRevokeAndRekeyCloseViewCodes(Base):
+    """R1: revoke / rekey / suspend must close outstanding view codes."""
+
+    def test_redeem_refused_after_permanent_revoke(self):
+        issued = helpers.issue(self.store, self.private)
+        identity.revoke(self.store, "bigpickle", reason="compromise")
+        with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
+            viewcodes.redeem_code(self.store, issued["code"])
+
+    def test_redeem_refused_during_timed_suspension(self):
+        import time
+        issued = helpers.issue(self.store, self.private)
+        identity.revoke(
+            self.store, "bigpickle", until=int(time.time()) + 3600, reason="pause"
+        )
+        with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
+            viewcodes.redeem_code(self.store, issued["code"])
+
+    def test_redeem_refused_after_rekey(self):
+        issued = helpers.issue(self.store, self.private)
+        new_private, new_public = keys.generate()
+        secret = identity.seed_bootstrap(self.store, "bigpickle")
+        identity.enrol_rekey(
+            self.store,
+            "bigpickle",
+            keys.export_public_b64(new_public),
+            secret,
+        )
+        # Invalidate-on-rekey marks the code used; redeem must not mint a session.
+        with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
+            viewcodes.redeem_code(self.store, issued["code"])
+        # New agent can still mint and redeem a fresh code.
+        fresh = helpers.issue(self.store, new_private)
+        session = viewcodes.redeem_code(self.store, fresh["code"])
+        self.assertEqual(session["kind"], "human_via_agent")
+
+    def test_unused_codes_marked_used_on_revoke(self):
+        issued = helpers.issue(self.store, self.private)
+        digest = viewcodes._hash_code(issued["code"])
+        row = self.store.get_code(digest)
+        self.assertIsNone(row["used_at"])
+        identity.revoke(self.store, "bigpickle")
+        row = self.store.get_code(digest)
+        self.assertIsNotNone(row["used_at"])