Compare commits

..

No commits in common. "c0027bb9f698a7bf5b179481cb3606574937faab" and "746e3e590ad197773820d52b42288055d7c0a69d" have entirely different histories.

6 changed files with 14 additions and 347 deletions

View File

@ -224,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 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. Independent review, sealed store, and platform glue remain.
**Still not production ready**, and the reason is not a to-do list:
@ -235,8 +235,7 @@ 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, routes. DB file seal is
available (`SealedStore`); SQLCipher page-level encryption is not.
- **The service is unwritten.** TLS, CSRF, transport, secret storage, routes.
- **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

View File

@ -28,7 +28,6 @@ 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
@ -55,10 +54,10 @@ python3 sigil.py sweep # prune stale rows
python3 -m unittest discover -s tests -v
```
130 tests. They cover the refusals, not just the happy paths: replayed
119 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
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
regressing brings that defect back.
## The invariant
@ -126,7 +125,6 @@ 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:
@ -134,11 +132,10 @@ 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 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
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, 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.
- **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

View File

@ -19,9 +19,8 @@
sigil.py whoami <token> describe a session
sigil.py events [actor] audit trail
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).
Uses SIGIL_DB (default ./sigil.db), SIGIL_KEY (default ./agent.key), and
optional SIGIL_ANCHOR (filesystem path for the off-box signed tip).
"""
from __future__ import annotations
@ -32,13 +31,11 @@ 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:
@ -50,13 +47,6 @@ def main(argv) -> int:
print(__doc__)
return 2
cmd = argv[1]
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:
@ -195,26 +185,15 @@ 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))

View File

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

View File

@ -1,179 +0,0 @@
"""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.
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
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-")
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)
with open(self._plain_path, "wb") as handle:
handle.write(plain)
super().__init__(self._plain_path, anchor_path=anchor_path)
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)
dest = sqlite3.connect(dest_path)
try:
self.conn.backup(dest)
dest.commit()
finally:
dest.close()
with open(dest_path, "rb") as handle:
plain = handle.read()
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

View File

@ -1,126 +0,0 @@
"""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()