bigpickle/sigil
0
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Compare commits

merge into: bigpickle:main
Branches Tags
bigpickle:main
bigpickle:leora/readme-links-license
bigpickle:sigil-initial
bigpickle:leora/fix-n7-n8-human-fp-view-ttl
bigpickle:leora/fix-r11-cryptography-pin
bigpickle:leora/fix-n4-n5-n6-docs-checkpoint-suspend
bigpickle:leora/fix-n3-rekey-rate-bucket
bigpickle:leora/fix-n2-key-history
bigpickle:leora/fix-n1-resolve-recheck
bigpickle:leora/fix-r11-runbook-deps
bigpickle:leora/fix-r6-r7-r8-cas-rekey-bootstrap
bigpickle:leora/fix-r5-sealed-tmp-hardening
bigpickle:leora/fix-r4-checkpoint-key-binding
bigpickle:leora/fix-r3-reinstate-requires-rekey
bigpickle:leora/fix-r2-bootstrap-hygiene
bigpickle:leora/fix-r1-invalidate-view-codes
bigpickle:leora/prod-sealed-store
bigpickle:leora/prod-offbox-anchor
bigpickle:leora/drop-probe
bigpickle:leora/close-review-blockers
...
pull from: bigpickle:leora/prod-sealed-store
Branches Tags
bigpickle:leora/readme-links-license
bigpickle:main
bigpickle:sigil-initial
bigpickle:leora/fix-n7-n8-human-fp-view-ttl
bigpickle:leora/fix-r11-cryptography-pin
bigpickle:leora/fix-n4-n5-n6-docs-checkpoint-suspend
bigpickle:leora/fix-n3-rekey-rate-bucket
bigpickle:leora/fix-n2-key-history
bigpickle:leora/fix-n1-resolve-recheck
bigpickle:leora/fix-r11-runbook-deps
bigpickle:leora/fix-r6-r7-r8-cas-rekey-bootstrap
bigpickle:leora/fix-r5-sealed-tmp-hardening
bigpickle:leora/fix-r4-checkpoint-key-binding
bigpickle:leora/fix-r3-reinstate-requires-rekey
bigpickle:leora/fix-r2-bootstrap-hygiene
bigpickle:leora/fix-r1-invalidate-view-codes
bigpickle:leora/prod-sealed-store
bigpickle:leora/prod-offbox-anchor
bigpickle:leora/drop-probe
bigpickle:leora/close-review-blockers

No commits in common. "main" and "leora/prod-sealed-store" have entirely different histories.
main ... leora/prod

16 changed files with 85 additions and 1050 deletions
Show Stats Expand all files Collapse all files

63
DESIGN.md
View File

@ -121,13 +121,9 @@ 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 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.
`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.
The distinction matters because the two failure modes are different. A key
compromise wants death — a lapsing revocation would hand the attacker back in
@ -135,18 +131,8 @@ 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, 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.
Revoking kills live sessions and drops outstanding challenges, so a revocation
that races an in-flight authentication cannot be overtaken by it.
### Existing controls stay
@ -204,15 +190,9 @@ 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, 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.
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.
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` /
@ -231,7 +211,7 @@ because the review looked for it.
## Status
158 passing tests, and two hardening passes that closed real defects. The worst were
106 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
@ -244,7 +224,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. 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.
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 and platform glue remain.
**Still not production ready**, and the reason is not a to-do list:
@ -256,8 +236,7 @@ Leora's contracted prod pass (Quarry 18a5d762) closed three remaining review blo
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, routes. DB file seal is
available (`SealedStore`, plaintext working copy `0600`); SQLCipher
page-level encryption is deferred (not yet available).
available (`SealedStore`); SQLCipher page-level encryption is not.
- **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
@ -267,27 +246,11 @@ The tests show the logic is correct. They do not make it safe.
## Next
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`.
1. Independent security review before this guards anything real. Two hardening
passes made it reviewable; neither substituted for one.
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.

11
README.md
View File

@ -55,7 +55,7 @@ python3 sigil.py sweep # prune stale rows
python3 -m unittest discover -s tests -v
```
158 tests. They cover the refusals, not just the happy paths: replayed
130 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`, `test_anchor.py` and
`test_sealed.py` have one test per defect found in review, so any of them
@ -137,9 +137,7 @@ Still open, and these are the reasons not to ship it yet:
- **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
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.
@ -159,8 +157,3 @@ 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`.

3
constraints.txt
View File

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

46
docs/PROD_RUNBOOK.md
View File

@ -1,46 +0,0 @@
# 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`.

13
requirements.lock.txt
View File

@ -1,13 +0,0 @@
# 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

5
requirements.txt
View File

@ -1,5 +0,0 @@
# 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

50
sigil/identity.py
View File

@ -56,9 +56,7 @@ def seed_bootstrap(store: Store, agent: str, secret: Optional[str] = None) -> st
"""
_validate_agent_id(agent)
row = store.get_agent(agent)
# 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):
if row is not None and is_revoked(row):
raise errors.UnknownAgent(f"{agent} is revoked")
secret = secret or secrets.token_urlsafe(24)
store.put_bootstrap(_hash_secret(secret), agent)
@ -78,10 +76,6 @@ 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
@ -94,8 +88,6 @@ 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,
@ -120,18 +112,6 @@ 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.
@ -162,14 +142,7 @@ 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.
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.
"""
"""Lift a revocation. Audited, because it undoes a security decision."""
_validate_agent_id(agent)
if store.get_agent(agent) is None:
raise errors.UnknownAgent(f"{agent} is not enrolled")
@ -188,10 +161,6 @@ 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)
@ -250,10 +219,6 @@ 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:
@ -292,17 +257,10 @@ 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")
# 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):
if is_revoked(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)
@ -312,8 +270,6 @@ 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,

52
sigil/sealed.py
View File

@ -11,11 +11,6 @@ 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)
@ -88,30 +83,6 @@ def is_sealed_file(path: str) -> bool:
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.
@ -136,8 +107,6 @@ class SealedStore(Store):
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:
@ -148,16 +117,9 @@ class SealedStore(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"")
with open(self._plain_path, "wb") as handle:
handle.write(plain)
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."""
@ -168,24 +130,14 @@ class SealedStore(Store):
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:

40
sigil/sessions.py
View File

@ -105,16 +105,9 @@ class GrantedCode:
agent session (issue_code) or a valid code (redeem_code).
"""
__slots__ = ("agent", "scopes", "ttl", "fingerprint", "_sigil_internal")
__slots__ = ("agent", "scopes", "ttl", "_sigil_internal")
def __init__(
self,
agent: str,
scopes: List[str],
ttl: int,
_internal=None,
fingerprint: Optional[str] = None,
):
def __init__(self, agent: str, scopes: List[str], ttl: int, _internal=None):
if _internal is not _ISSUER:
raise errors.SigilError(
"GrantedCode cannot be constructed directly; it comes from"
@ -123,7 +116,6 @@ class GrantedCode:
self.agent = agent
self.scopes = list(scopes)
self.ttl = ttl
self.fingerprint = fingerprint
self._sigil_internal = _internal
def __repr__(self) -> str:
@ -159,7 +151,6 @@ def issue_view_session(store: Store, grant: "GrantedCode") -> dict:
granted,
expires,
granted_by=agent,
fingerprint=grant.fingerprint,
)
store.record(
"session.issued",
@ -185,14 +176,7 @@ 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, 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
"""Look up a session and reject it if revoked or expired."""
row = store.get_session(hash_token(token))
if row is None:
raise errors.SessionUnknown("no such session")
@ -200,24 +184,6 @@ 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"],

279
sigil/store.py
View File

@ -36,20 +36,6 @@ 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 (
@ -91,11 +77,7 @@ 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,
-- 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
used_at INTEGER
);
-- Append-only, hash-chained audit log. The chain is what makes this an
@ -148,9 +130,6 @@ 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),
@ -238,22 +217,6 @@ 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:
@ -290,18 +253,11 @@ 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, now),
)
if cur.rowcount == 1:
self.conn.execute(
"INSERT OR IGNORE INTO agent_keys"
" (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)",
(agent, fingerprint, pubkey, now),
(agent, pubkey, fingerprint, int(time.time())),
)
self.conn.commit()
return cur.rowcount == 1
@ -314,33 +270,10 @@ 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
@ -351,42 +284,9 @@ 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:
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 = ?",
@ -412,68 +312,31 @@ class Store:
# -- bootstrap ---------------------------------------------------------
def put_bootstrap(self, secret_hash: str, agent: str) -> None:
"""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(
self._exec(
"INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
(secret_hash, agent, now),
(secret_hash, agent, int(time.time())),
)
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 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.
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.
"""
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(
"DELETE FROM bootstrap WHERE secret_hash = ? AND spent_at IS NULL",
(secret_hash,),
"UPDATE bootstrap SET spent_at = ? WHERE secret_hash = ?"
" AND spent_at IS NULL",
(int(time.time()), secret_hash),
)
self.conn.commit()
if cur.rowcount != 1:
return None
return agent
row = self.conn.execute(
"SELECT agent FROM bootstrap WHERE secret_hash = ?", (secret_hash,)
).fetchone()
return row["agent"] if row else None
# -- nonces ------------------------------------------------------------
def put_nonce(self, nonce: str, agent: str, expires_at: int) -> None:
@ -539,30 +402,12 @@ 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. Outstanding view codes are invalidated
in the same breath so a pre-issued code cannot mint a fresh session.
buys nothing after a compromise.
"""
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",
@ -581,11 +426,10 @@ 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, fingerprint) VALUES (?, ?, ?, ?, ?, ?, ?)",
" view_ttl) VALUES (?, ?, ?, ?, ?, ?)",
(
code_hash,
agent,
@ -593,7 +437,6 @@ class Store:
int(time.time()),
redeem_expires_at,
view_ttl,
fingerprint,
),
)
@ -601,17 +444,14 @@ 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 or expired.
"""Consume a one-time code exactly once. False if already used.
Atomic CAS includes the expiry predicate so a race at the TTL boundary
cannot mint a session slightly after redeem_expires_at.
Atomic, so a code cannot mint two sessions under a race.
"""
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"
" AND redeem_expires_at > ?",
(now, code_hash, now),
"UPDATE codes SET used_at = ? WHERE code_hash = ? AND used_at IS NULL",
(int(time.time()), code_hash),
)
self.conn.commit()
return cur.rowcount == 1
@ -740,26 +580,7 @@ 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:
@ -788,19 +609,8 @@ class Store:
public = _keys.public_from_private(private)
fingerprint = _keys.fingerprint(public)
from . import identity as _identity
agent_row = self.get_agent(signer)
if agent_row is None:
if self.get_agent(signer) 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"
@ -864,20 +674,17 @@ class Store:
if not notices or notices[-1].get("head") != tip.get("head"):
notices = list(notices) + [tip]
results = []
warnings = []
for notice in notices:
if self.get_agent(notice["signer"]) is None:
agent_row = self.get_agent(notice["signer"])
if agent_row is None:
results.append({"ok": False, "head": notice.get("head"),
"reason": f"signer {notice['signer']} no longer enrolled"})
continue
key_row = self.get_key_by_fingerprint(
notice["signer"], notice.get("fingerprint") or ""
)
if key_row is None:
if agent_row["fingerprint"] != notice.get("fingerprint"):
results.append({"ok": False, "head": notice.get("head"),
"reason": "signer key for this fingerprint is unknown"})
"reason": "signer key changed since the tip was published"})
continue
public = _keys.from_public_bytes(key_row["pubkey"])
public = _keys.from_public_bytes(agent_row["pubkey"])
checked = _anchor.verify_notice(notice, public)
if not checked["ok"]:
results.append({"ok": False, "head": notice.get("head"),
@ -897,17 +704,11 @@ class Store:
"(database rewrite cannot erase the off-box tip)",
})
continue
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)
results.append({"ok": True, "head": notice["head"],
"events": notice["events"], "signer": notice["signer"]})
bad = [r for r in results if not r["ok"]]
return {"ok": not bad, "anchors": len(results), "failures": bad,
"warnings": warnings, "tip": tip, "path": self.anchor_path}
"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()
@ -923,23 +724,22 @@ class Store:
from . import keys as _keys
results = []
warnings = []
for row in self.conn.execute("SELECT * FROM checkpoints ORDER BY id ASC"):
if self.get_agent(row["signer"]) is None:
agent_row = self.get_agent(row["signer"])
if agent_row is None:
results.append({"id": row["id"], "ok": False,
"reason": f"signer {row['signer']} no longer enrolled"})
continue
key_row = self.get_key_by_fingerprint(row["signer"], row["fingerprint"])
if key_row is None:
if agent_row["fingerprint"] != row["fingerprint"]:
results.append({"id": row["id"], "ok": False,
"reason": "signer key for this fingerprint is unknown"})
"reason": "signer key changed since the checkpoint was made"})
continue
payload = self._checkpoint_payload(
row["at"], row["label"], row["head"], row["events"],
row["signer"], row["fingerprint"],
)
try:
public = _keys.from_public_bytes(key_row["pubkey"])
public = _keys.from_public_bytes(agent_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)
@ -957,16 +757,9 @@ class Store:
results.append({"id": row["id"], "ok": False,
"reason": "chain head at that position no longer matches"})
continue
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)
results.append({"id": row["id"], "ok": True})
bad = [r for r in results if not r["ok"]]
return {"ok": not bad, "checkpoints": len(results),
"failures": bad, "warnings": warnings}
return {"ok": not bad, "checkpoints": len(results), "failures": bad}
# -- housekeeping ------------------------------------------------------
def sweep(self) -> Dict[str, int]:

36
sigil/viewcodes.py
View File

@ -91,18 +91,9 @@ 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.
@ -124,13 +115,12 @@ 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=live_fp,
fingerprint=agent_row["fingerprint"],
detail={
"code_ref": _code_ref(code),
"scopes": granted,
@ -144,6 +134,7 @@ def issue_code(
"code_ttl": code_ttl,
"view_ttl": view_ttl,
"redeem_expires_at": now + code_ttl,
"view_ttl": view_ttl,
}
@ -164,28 +155,6 @@ 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.
@ -231,7 +200,6 @@ 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)

212
tests/test_hardening.py
View File

@ -360,215 +360,3 @@ 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"):])

56
tests/test_identity.py
View File

@ -6,8 +6,6 @@ import unittest
from sigil import errors, identity, keys
from sigil.store import Store
import helpers
class Base(unittest.TestCase):
def setUp(self):
@ -146,59 +144,5 @@ 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()

173
tests/test_production.py
View File

@ -133,25 +133,15 @@ class TestSignedCheckpoints(unittest.TestCase):
with self.assertRaises(ValueError):
stranger.checkpoint(self.private, "not-enrolled")
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.
def test_checkpoint_after_rekey_is_flagged(self):
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.assertTrue(result["ok"], result)
self.assertEqual(result["failures"], [])
self.assertFalse(result["ok"])
self.assertIn("key changed", result["failures"][0]["reason"])
def test_empty_log_checkpoints_cleanly(self):
store, private = helpers.fresh()
@ -203,19 +193,10 @@ 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):
@ -249,33 +230,6 @@ 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.
@ -346,14 +300,12 @@ class TestRekeyDoesNotUnrevoke(unittest.TestCase):
def setUp(self):
self.store, self.private = helpers.fresh()
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"
)
def test_rekey_refuses_while_revoked(self):
identity.revoke(self.store, "bigpickle", reason="incident")
secret = None
with self.assertRaises(errors.UnknownAgent):
identity.seed_bootstrap(self.store, "bigpickle")
secret = identity.seed_bootstrap(self.store, "bigpickle")
# Even with a secretly planted bootstrap, rekey must refuse.
planted = secrets_module.token_urlsafe(24)
self.store.put_bootstrap(
__import__("hashlib").sha256(planted.encode()).hexdigest(),
@ -368,28 +320,6 @@ 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.
@ -415,17 +345,10 @@ class TestDualBootstrapEnrolRace(unittest.TestCase):
import threading
store = Store(":memory:")
# 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()
secrets = [
identity.seed_bootstrap(store, "racer"),
identity.seed_bootstrap(store, "racer"),
]
pubs = [keys.generate()[1] for _ in range(2)]
results = []
lock = threading.Lock()
@ -492,75 +415,3 @@ 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())

27
tests/test_sealed.py
View File

@ -124,30 +124,3 @@ class TestSealedStore(unittest.TestCase):
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()

45
tests/test_viewcodes.py
View File

@ -143,48 +143,3 @@ 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"])