Sealed store: AES-256-GCM encryption at rest for the DB file
Tokens/codes/secrets were already digests; a stolen cleartext SQLite file still leaked the agent roster, revocation state and audit history. SealedStore / SIGIL_SEAL_PASS wraps on-disk bytes in AES-256-GCM (PBKDF2-HMAC-SHA256, 480k iters). Working copy is plaintext only while open (temp file); flush/close re-seals. Incremental path short of SQLCipher page-level encryption. 11 new tests in test_sealed.py (stolen seal has no SQLite header / no agent rows; wrong passphrase fails; works with off-box anchor). Suite 130 green. Part of Quarry 18a5d762. Independent crypto review + platform glue remain.
This commit is contained in:
parent
29fd9a1388
commit
bd7d62c21c
@ -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
|
unkeyed chain that a database-level attacker could rebuild, and SQLite without
|
||||||
a busy timeout.
|
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 and platform glue remain.
|
||||||
|
|
||||||
**Still not production ready**, and the reason is not a to-do list:
|
**Still not production ready**, and the reason is not a to-do list:
|
||||||
|
|
||||||
@ -235,7 +235,8 @@ 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
|
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
|
rewrite that in-DB checkpoints alone would miss after deletion. Leave it unset
|
||||||
and the tip stays in-DB only.
|
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`); SQLCipher page-level encryption is not.
|
||||||
- **Rate limits are per-process**, so N processes means N times the quota.
|
- **Rate limits are per-process**, so N processes means N times the quota.
|
||||||
- **No key recovery.** Rotate and old sessions die. Deliberate.
|
- **No key recovery.** Rotate and old sessions die. Deliberate.
|
||||||
- **Humans remain unattributable** to themselves, which is accepted and is a
|
- **Humans remain unattributable** to themselves, which is accepted and is a
|
||||||
|
|||||||
17
README.md
17
README.md
@ -28,6 +28,7 @@ is **not independently reviewed** — see *Status* below.
|
|||||||
|
|
||||||
```sh
|
```sh
|
||||||
export SIGIL_DB=/tmp/demo.db
|
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 seed bigpickle # -> one-time secret
|
||||||
python3 sigil.py keygen /tmp/demo.key # -> public_b64, fingerprint
|
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
|
python3 -m unittest discover -s tests -v
|
||||||
```
|
```
|
||||||
|
|
||||||
119 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
|
signatures, wrong keys, expired nonces, single-use codes, scope escalation, and
|
||||||
the invariant below. `test_hardening.py`, `test_production.py` and
|
the invariant below. `test_hardening.py`, `test_production.py`, `test_anchor.py` and
|
||||||
`test_anchor.py` have one test per defect found in review, so any of them
|
`test_sealed.py` have one test per defect found in review, so any of them
|
||||||
regressing brings that defect back.
|
regressing brings that defect back.
|
||||||
|
|
||||||
## The invariant
|
## 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 |
|
| 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 |
|
| 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`) |
|
| 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:
|
Still open, and these are the reasons not to ship it yet:
|
||||||
|
|
||||||
@ -132,10 +134,11 @@ 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.
|
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
|
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.
|
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
|
- **The service around this is unwritten.** TLS, CSRF, session transport and the
|
||||||
storage and the HTTP routes. The library refuses to mint authority from a
|
HTTP routes. DB encryption at rest is available via `SealedStore` /
|
||||||
name, so a route that forgets to authenticate fails closed, but that is only
|
`SIGIL_SEAL_PASS` (AES-256-GCM file seal; not SQLCipher page-level). The
|
||||||
as good as the integration.
|
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=)`)
|
- **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.
|
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
|
Rewriting the DB alone can no longer erase the tip. Without that path
|
||||||
|
|||||||
25
sigil.py
25
sigil.py
@ -19,8 +19,9 @@
|
|||||||
sigil.py whoami <token> describe a session
|
sigil.py whoami <token> describe a session
|
||||||
sigil.py events [actor] audit trail
|
sigil.py events [actor] audit trail
|
||||||
|
|
||||||
Uses SIGIL_DB (default ./sigil.db), SIGIL_KEY (default ./agent.key), and
|
Uses SIGIL_DB (default ./sigil.db), SIGIL_KEY (default ./agent.key),
|
||||||
optional SIGIL_ANCHOR (filesystem path for the off-box signed tip).
|
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
|
from __future__ import annotations
|
||||||
@ -31,11 +32,13 @@ import os
|
|||||||
import sys
|
import sys
|
||||||
|
|
||||||
from sigil import errors, identity, keys, sessions, viewcodes
|
from sigil import errors, identity, keys, sessions, viewcodes
|
||||||
|
from sigil.sealed import SealedStore, is_sealed_file
|
||||||
from sigil.store import Store
|
from sigil.store import Store
|
||||||
|
|
||||||
DB = os.environ.get("SIGIL_DB", "./sigil.db")
|
DB = os.environ.get("SIGIL_DB", "./sigil.db")
|
||||||
KEY = os.environ.get("SIGIL_KEY", "./agent.key")
|
KEY = os.environ.get("SIGIL_KEY", "./agent.key")
|
||||||
ANCHOR = os.environ.get("SIGIL_ANCHOR") # off-box tip path; unset = in-DB only
|
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:
|
def out(obj) -> None:
|
||||||
@ -47,6 +50,13 @@ def main(argv) -> int:
|
|||||||
print(__doc__)
|
print(__doc__)
|
||||||
return 2
|
return 2
|
||||||
cmd = argv[1]
|
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)
|
store = Store(DB, anchor_path=ANCHOR)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@ -185,15 +195,26 @@ def main(argv) -> int:
|
|||||||
|
|
||||||
except errors.SigilError as exc:
|
except errors.SigilError as exc:
|
||||||
print(json.dumps({"error": type(exc).__name__, "message": str(exc)}))
|
print(json.dumps({"error": type(exc).__name__, "message": str(exc)}))
|
||||||
|
_flush_if_sealed(store)
|
||||||
return 1
|
return 1
|
||||||
except (ValueError, OSError) as exc:
|
except (ValueError, OSError) as exc:
|
||||||
# Wrong passphrase, unreadable key file. Report it as a plain result
|
# Wrong passphrase, unreadable key file. Report it as a plain result
|
||||||
# rather than a traceback: it is an ordinary operator mistake.
|
# rather than a traceback: it is an ordinary operator mistake.
|
||||||
print(json.dumps({"error": type(exc).__name__, "message": str(exc)[:200]}))
|
print(json.dumps({"error": type(exc).__name__, "message": str(exc)[:200]}))
|
||||||
|
_flush_if_sealed(store)
|
||||||
return 1
|
return 1
|
||||||
|
|
||||||
|
_flush_if_sealed(store)
|
||||||
return 0
|
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__":
|
if __name__ == "__main__":
|
||||||
raise SystemExit(main(sys.argv))
|
raise SystemExit(main(sys.argv))
|
||||||
@ -31,17 +31,20 @@ without the private key, which never leaves the agent.
|
|||||||
|
|
||||||
import time
|
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 .store import Store
|
||||||
|
from .sealed import SealedStore
|
||||||
|
|
||||||
__version__ = "0.4.0"
|
__version__ = "0.4.1"
|
||||||
|
|
||||||
__all__ = [
|
__all__ = [
|
||||||
"Store",
|
"Store",
|
||||||
|
"SealedStore",
|
||||||
"anchor",
|
"anchor",
|
||||||
"errors",
|
"errors",
|
||||||
"identity",
|
"identity",
|
||||||
"keys",
|
"keys",
|
||||||
|
"sealed",
|
||||||
"sessions",
|
"sessions",
|
||||||
"store",
|
"store",
|
||||||
"viewcodes",
|
"viewcodes",
|
||||||
|
|||||||
179
sigil/sealed.py
Normal file
179
sigil/sealed.py
Normal file
@ -0,0 +1,179 @@
|
|||||||
|
"""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
|
||||||
126
tests/test_sealed.py
Normal file
126
tests/test_sealed.py
Normal file
@ -0,0 +1,126 @@
|
|||||||
|
"""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()
|
||||||
Loading…
x
Reference in New Issue
Block a user