Off-box audit chain tip: publish/verify outside the SQLite file
A signed checkpoint inside the same database is still only as strong as that file. Store(..., anchor_path=) / SIGIL_ANCHOR now writes the signed tip to a filesystem path outside the DB (atomic tip file + append-only .log). verify_external_anchors catches a rewrite even after in-DB checkpoints are deleted. 8 new tests in test_anchor.py; suite 119 green. Part of Quarry 18a5d762 (Sigil prod build). Do not merge to main yet: sealed store and independent crypto review still open.
This commit is contained in:
parent
1c2e9d16c2
commit
29fd9a1388
2
.gitignore
vendored
2
.gitignore
vendored
@ -3,3 +3,5 @@ __pycache__/
|
||||
*.db
|
||||
*.key
|
||||
*.pem
|
||||
*.tip
|
||||
*.tip.log
|
||||
|
||||
22
DESIGN.md
22
DESIGN.md
@ -195,10 +195,12 @@ 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. Real deployment anchors the
|
||||
head digest somewhere the database cannot reach. This narrows the attack from
|
||||
"rewrite anything" to "erase the record that you did", and leaves that second
|
||||
half open.
|
||||
checkpoints, because they live in the same database. Set `anchor_path` /
|
||||
`SIGIL_ANCHOR` and every checkpoint also publishes the signed tip to a file
|
||||
outside the SQLite path; `verify_external_anchors` then catches a rewrite even
|
||||
after the in-DB checkpoints table is wiped. Without that path the tip stays
|
||||
in-DB only, and deleting the checkpoints remains an option for whoever holds
|
||||
the file.
|
||||
|
||||
Secrets are not in the log and not in the database in the clear. Codes and
|
||||
session tokens are stored as digests, enrolment secrets are hashed and spent
|
||||
@ -222,15 +224,17 @@ 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). Independent review, off-box checkpoint anchor, 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. Independent review, sealed store, and platform glue remain.
|
||||
|
||||
**Still not production ready**, and the reason is not a to-do list:
|
||||
|
||||
- **No independent review.** Everything above is self-review, twice over.
|
||||
Self-review finds the bugs you already suspected. This needs someone else,
|
||||
ideally a cryptographer, and it is the one blocker code cannot close.
|
||||
- **Checkpoints are not anchored off-box.** Signed, yes; deleteable by the same
|
||||
attacker along with everything else, yes. See *Audit trail*.
|
||||
- **Off-box tip is opt-in.** With `anchor_path` set, every checkpoint publishes a
|
||||
signed tip outside the SQLite file and `verify_external_anchors` catches a DB
|
||||
rewrite that in-DB checkpoints alone would miss after deletion. Leave it unset
|
||||
and the tip stays in-DB only.
|
||||
- **The service is unwritten.** TLS, CSRF, transport, secret storage, routes.
|
||||
- **Rate limits are per-process**, so N processes means N times the quota.
|
||||
- **No key recovery.** Rotate and old sessions die. Deliberate.
|
||||
@ -243,8 +247,8 @@ The tests show the logic is correct. They do not make it safe.
|
||||
|
||||
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, so deleting the
|
||||
checkpoints stops being an option for an attacker who already has the file.
|
||||
2. ~~Anchor the checkpoint head digest outside the database~~ — done via
|
||||
`Store(..., anchor_path=)` / `SIGIL_ANCHOR` (signed tip file + `.log`).
|
||||
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.
|
||||
|
||||
21
README.md
21
README.md
@ -41,8 +41,9 @@ python3 sigil.py code bigpickle /tmp/demo.key # hand a human a code
|
||||
python3 sigil.py redeem X3ZB-NTLG # -> human_via_agent token
|
||||
python3 sigil.py whoami "$TOKEN" # kind, scopes, may_act_as_agent
|
||||
python3 sigil.py events bigpickle # audit trail
|
||||
python3 sigil.py checkpoint /tmp/demo.key bigpickle # sign the audit head
|
||||
python3 sigil.py verify-chain # chain + checkpoints
|
||||
export SIGIL_ANCHOR=/tmp/demo.chain.tip # off-box tip path
|
||||
python3 sigil.py checkpoint /tmp/demo.key bigpickle # sign head + publish tip
|
||||
python3 sigil.py verify-chain # chain + checkpoints + tip
|
||||
python3 sigil.py revoke bigpickle --reason "incident" # or --until UNIX
|
||||
python3 sigil.py sweep # prune stale rows
|
||||
```
|
||||
@ -53,10 +54,11 @@ python3 sigil.py sweep # prune stale rows
|
||||
python3 -m unittest discover -s tests -v
|
||||
```
|
||||
|
||||
111 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` and `test_production.py` have one test
|
||||
per defect found in review, so any of them regressing brings that defect back.
|
||||
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
|
||||
|
||||
@ -122,6 +124,7 @@ A hardening pass closed the defects below. Each is now covered by a test.
|
||||
| Rekey cleared `revoked_at` (INSERT OR REPLACE) | Rekey refuses while revoked; `update_agent_key` preserves ban columns |
|
||||
| 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`) |
|
||||
|
||||
Still open, and these are the reasons not to ship it yet:
|
||||
|
||||
@ -133,10 +136,10 @@ Still open, and these are the reasons not to ship it yet:
|
||||
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.
|
||||
- **Checkpoints need somewhere to live.** A signed checkpoint inside the same
|
||||
database is still only as good as the database. Real deployment wants the head
|
||||
digest anchored off-box — published, or countersigned elsewhere — and this
|
||||
does not do that for you.
|
||||
- **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
|
||||
configured, checkpoints remain in-DB only — fine for tests, not for prod.
|
||||
- **No key recovery.** Rotate and the old sessions die, which is the point. There
|
||||
is no escrow. Deliberate, not an oversight.
|
||||
- **Rate limits are in-process.** They are per-`Store`, so several processes
|
||||
|
||||
18
sigil.py
18
sigil.py
@ -6,7 +6,8 @@
|
||||
sigil.py enrol <agent> <pubkey.b64> <secret>
|
||||
sigil.py challenge <agent> nonce to sign
|
||||
sigil.py checkpoint <key> <agent> [label] sign the current audit head
|
||||
sigil.py verify-chain check the chain and its checkpoints
|
||||
(also publishes off-box when SIGIL_ANCHOR is set)
|
||||
sigil.py verify-chain check chain, checkpoints, and off-box tip
|
||||
sigil.py revoke <agent> [--until UNIX] [--reason R] [--actor A]
|
||||
--until omitted means permanent
|
||||
sigil.py reinstate <agent> [actor] lift a revocation
|
||||
@ -18,7 +19,8 @@
|
||||
sigil.py whoami <token> describe a session
|
||||
sigil.py events [actor] audit trail
|
||||
|
||||
Uses SIGIL_DB (default ./sigil.db) and SIGIL_KEY (default ./agent.key).
|
||||
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
|
||||
@ -33,6 +35,7 @@ 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
|
||||
|
||||
|
||||
def out(obj) -> None:
|
||||
@ -44,7 +47,7 @@ def main(argv) -> int:
|
||||
print(__doc__)
|
||||
return 2
|
||||
cmd = argv[1]
|
||||
store = Store(DB)
|
||||
store = Store(DB, anchor_path=ANCHOR)
|
||||
|
||||
try:
|
||||
if cmd == "seed":
|
||||
@ -115,9 +118,12 @@ def main(argv) -> int:
|
||||
|
||||
elif cmd == "verify-chain":
|
||||
chain = store.verify_chain()
|
||||
anchors = store.verify_checkpoints()
|
||||
out({"chain": chain, "checkpoints": anchors})
|
||||
return 0 if chain["ok"] and anchors["ok"] else 1
|
||||
checkpoints = store.verify_checkpoints()
|
||||
external = store.verify_external_anchors()
|
||||
out({"chain": chain, "checkpoints": checkpoints,
|
||||
"external_anchors": external})
|
||||
ok = chain["ok"] and checkpoints["ok"] and external["ok"]
|
||||
return 0 if ok else 1
|
||||
|
||||
elif cmd == "revoke":
|
||||
# Positional optional args are a trap here: "revoke alice please"
|
||||
|
||||
@ -31,13 +31,14 @@ without the private key, which never leaves the agent.
|
||||
|
||||
import time
|
||||
|
||||
from . import errors, identity, keys, sessions, store, viewcodes
|
||||
from . import anchor, errors, identity, keys, sessions, store, viewcodes
|
||||
from .store import Store
|
||||
|
||||
__version__ = "0.3.0"
|
||||
__version__ = "0.4.0"
|
||||
|
||||
__all__ = [
|
||||
"Store",
|
||||
"anchor",
|
||||
"errors",
|
||||
"identity",
|
||||
"keys",
|
||||
|
||||
122
sigil/anchor.py
Normal file
122
sigil/anchor.py
Normal file
@ -0,0 +1,122 @@
|
||||
"""Off-box anchors for the audit chain tip.
|
||||
|
||||
A signed checkpoint inside the same SQLite file as the events is still only as
|
||||
strong as that file: an attacker who can rewrite the database can delete the
|
||||
checkpoints table along with everything else. Publishing the signed tip to a
|
||||
path *outside* the database closes that. Rewriting the DB alone cannot erase
|
||||
the tip the operator already published.
|
||||
|
||||
The notice is the same canonical payload the in-DB checkpoint signs, plus the
|
||||
signature. verify_anchor() checks the signature against the enrolled public
|
||||
key and reports whether the live chain still ends at that head.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
||||
def notice_bytes(notice: Dict[str, Any]) -> bytes:
|
||||
"""Canonical encoding of a published tip, excluding the signature itself."""
|
||||
body = {
|
||||
"at": int(notice["at"]),
|
||||
"label": notice.get("label") or "",
|
||||
"head": notice["head"],
|
||||
"events": int(notice["events"]),
|
||||
"signer": notice["signer"],
|
||||
"fingerprint": notice["fingerprint"],
|
||||
}
|
||||
return json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
||||
|
||||
|
||||
def write_tip(path: str, notice: Dict[str, Any]) -> str:
|
||||
"""Atomically write the latest tip notice to `path`.
|
||||
|
||||
Also appends one JSON line to `path + ".log"` so earlier tips remain
|
||||
auditable after a later checkpoint overwrites the tip file.
|
||||
"""
|
||||
if not path or path == ":memory:":
|
||||
raise ValueError("anchor path must be a real filesystem path")
|
||||
payload = {
|
||||
"at": int(notice["at"]),
|
||||
"label": notice.get("label") or "",
|
||||
"head": notice["head"],
|
||||
"events": int(notice["events"]),
|
||||
"signer": notice["signer"],
|
||||
"fingerprint": notice["fingerprint"],
|
||||
"signature": notice["signature"],
|
||||
}
|
||||
text = json.dumps(payload, sort_keys=True, indent=2) + "\n"
|
||||
directory = os.path.dirname(os.path.abspath(path))
|
||||
if directory:
|
||||
os.makedirs(directory, exist_ok=True)
|
||||
fd, tmp = _tmp_beside(path)
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
handle.write(text)
|
||||
handle.flush()
|
||||
os.fsync(handle.fileno())
|
||||
os.replace(tmp, path)
|
||||
except Exception:
|
||||
try:
|
||||
os.unlink(tmp)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
# Append-only history beside the tip. Best-effort: tip write already landed.
|
||||
try:
|
||||
with open(path + ".log", "a", encoding="utf-8") as log:
|
||||
log.write(json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n")
|
||||
except OSError:
|
||||
pass
|
||||
return path
|
||||
|
||||
|
||||
def read_tip(path: str) -> Optional[Dict[str, Any]]:
|
||||
"""Load the tip notice, or None if the file is missing."""
|
||||
if not path or not os.path.exists(path):
|
||||
return None
|
||||
with open(path, "r", encoding="utf-8") as handle:
|
||||
return json.loads(handle.read())
|
||||
|
||||
|
||||
def read_log(path: str) -> List[Dict[str, Any]]:
|
||||
"""Load every historically published tip from the sibling .log file."""
|
||||
log_path = path + ".log"
|
||||
if not os.path.exists(log_path):
|
||||
tip = read_tip(path)
|
||||
return [tip] if tip else []
|
||||
out: List[Dict[str, Any]] = []
|
||||
with open(log_path, "r", encoding="utf-8") as handle:
|
||||
for line in handle:
|
||||
line = line.strip()
|
||||
if line:
|
||||
out.append(json.loads(line))
|
||||
return out
|
||||
|
||||
|
||||
def verify_notice(notice: Dict[str, Any], public_key) -> Dict[str, Any]:
|
||||
"""Check one notice's signature. `public_key` is an Ed25519 public key."""
|
||||
from . import keys as _keys
|
||||
|
||||
try:
|
||||
payload = notice_bytes(notice)
|
||||
sig = base64.b64decode(notice["signature"])
|
||||
public_key.verify(sig, payload)
|
||||
except Exception as exc:
|
||||
return {"ok": False, "reason": f"signature check failed: {exc}"}
|
||||
expected_fp = _keys.fingerprint(public_key)
|
||||
if notice.get("fingerprint") != expected_fp:
|
||||
return {"ok": False, "reason": "fingerprint does not match signer key"}
|
||||
return {"ok": True, "head": notice["head"], "events": notice["events"],
|
||||
"signer": notice["signer"]}
|
||||
|
||||
|
||||
def _tmp_beside(path: str):
|
||||
directory = os.path.dirname(os.path.abspath(path)) or "."
|
||||
return __import__("tempfile").mkstemp(
|
||||
prefix=".sigil-anchor-", suffix=".tmp", dir=directory
|
||||
)
|
||||
@ -137,7 +137,15 @@ RATE_LIMITS = {
|
||||
|
||||
|
||||
class Store:
|
||||
def __init__(self, path: str = ":memory:"):
|
||||
def __init__(self, path: str = ":memory:", anchor_path: Optional[str] = None):
|
||||
"""Open a Sigil store.
|
||||
|
||||
`anchor_path`, when set, is a filesystem path *outside* this database
|
||||
where every checkpoint also publishes its signed tip. Rewriting or
|
||||
deleting rows in the SQLite file cannot erase that published tip.
|
||||
"""
|
||||
self.path = path
|
||||
self.anchor_path = anchor_path
|
||||
self.conn = sqlite3.connect(path, check_same_thread=False, timeout=10.0)
|
||||
self.conn.row_factory = sqlite3.Row
|
||||
# Writers wait rather than failing instantly. Without this a second
|
||||
@ -591,8 +599,13 @@ class Store:
|
||||
digest, so later verifying that the chain still ends at the same
|
||||
committed head is a check against anyone who cannot forge a signature
|
||||
with this key.
|
||||
|
||||
When this Store was opened with `anchor_path`, the same signed tip is
|
||||
also published to that path. The published tip survives a rewrite of
|
||||
this database file.
|
||||
"""
|
||||
from . import keys as _keys
|
||||
from . import anchor as _anchor
|
||||
|
||||
public = _keys.public_from_private(private)
|
||||
fingerprint = _keys.fingerprint(public)
|
||||
@ -614,8 +627,88 @@ class Store:
|
||||
(at, label or "", head, events, signer, fingerprint, signature),
|
||||
)
|
||||
self.conn.commit()
|
||||
return {"at": at, "label": label, "head": head, "events": events,
|
||||
"signer": signer, "fingerprint": fingerprint}
|
||||
result = {"at": at, "label": label, "head": head, "events": events,
|
||||
"signer": signer, "fingerprint": fingerprint,
|
||||
"signature": signature}
|
||||
if self.anchor_path:
|
||||
published = _anchor.write_tip(self.anchor_path, result)
|
||||
result["anchor_path"] = published
|
||||
return result
|
||||
|
||||
def publish_anchor(self, private, signer: str, label: Optional[str] = None) -> dict:
|
||||
"""Checkpoint and require an off-box publish.
|
||||
|
||||
Same as checkpoint(), but refuses if this Store has no anchor_path, so
|
||||
a deployment that intends to keep the tip off-box cannot silently fall
|
||||
back to in-DB-only.
|
||||
"""
|
||||
if not self.anchor_path:
|
||||
raise ValueError(
|
||||
"publish_anchor requires Store(..., anchor_path=...); "
|
||||
"a checkpoint inside the database alone is not off-box"
|
||||
)
|
||||
return self.checkpoint(private, signer, label)
|
||||
|
||||
def verify_external_anchors(self) -> Dict[str, Any]:
|
||||
"""Verify the published tip (and history) against enrolled keys + chain.
|
||||
|
||||
Returns ok=False when:
|
||||
- the tip file is missing but an anchor_path was configured
|
||||
- a notice signature fails
|
||||
- the tip head is no longer present at the recorded event count
|
||||
(the DB was rewritten or truncated past the published tip)
|
||||
"""
|
||||
from . import keys as _keys
|
||||
from . import anchor as _anchor
|
||||
|
||||
if not self.anchor_path:
|
||||
return {"ok": True, "anchors": 0, "skipped": True,
|
||||
"reason": "no anchor_path configured"}
|
||||
notices = _anchor.read_log(self.anchor_path)
|
||||
tip = _anchor.read_tip(self.anchor_path)
|
||||
if tip is None:
|
||||
return {"ok": False, "anchors": 0, "failures": [
|
||||
{"reason": f"anchor tip missing at {self.anchor_path}"}
|
||||
]}
|
||||
# Ensure the tip itself is represented even if the log write failed.
|
||||
if not notices or notices[-1].get("head") != tip.get("head"):
|
||||
notices = list(notices) + [tip]
|
||||
results = []
|
||||
for notice in notices:
|
||||
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
|
||||
if agent_row["fingerprint"] != notice.get("fingerprint"):
|
||||
results.append({"ok": False, "head": notice.get("head"),
|
||||
"reason": "signer key changed since the tip was published"})
|
||||
continue
|
||||
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"),
|
||||
"reason": checked["reason"]})
|
||||
continue
|
||||
found = self.conn.execute(
|
||||
"SELECT hash FROM events WHERE id = ?", (notice["events"],)
|
||||
).fetchone()
|
||||
actual = GENESIS if notice["events"] == 0 else (
|
||||
found["hash"] if found else None
|
||||
)
|
||||
if actual != notice["head"]:
|
||||
results.append({
|
||||
"ok": False,
|
||||
"head": notice["head"],
|
||||
"reason": "published tip head no longer matches the chain "
|
||||
"(database rewrite cannot erase the off-box tip)",
|
||||
})
|
||||
continue
|
||||
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,
|
||||
"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()
|
||||
|
||||
137
tests/test_anchor.py
Normal file
137
tests/test_anchor.py
Normal file
@ -0,0 +1,137 @@
|
||||
"""Off-box audit chain anchor.
|
||||
|
||||
Finding: a signed checkpoint inside the same SQLite file is still only as
|
||||
strong as that file. An attacker who can rewrite the database can delete the
|
||||
checkpoints table and recompute the chain. Publishing the signed tip to a path
|
||||
outside the database means rewriting the DB alone cannot erase the tip.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from sigil import identity, keys
|
||||
from sigil.store import Store, GENESIS
|
||||
from sigil import anchor as anchor_mod
|
||||
from tests import helpers
|
||||
|
||||
|
||||
class TestOffBoxAnchor(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.db = os.path.join(self.tmp.name, "sigil.db")
|
||||
self.tip = os.path.join(self.tmp.name, "offbox", "chain.tip")
|
||||
self.store = Store(self.db, anchor_path=self.tip)
|
||||
self.private, _ = helpers.enrol(self.store, "bigpickle")
|
||||
identity.challenge(self.store, "bigpickle") # leaves an event
|
||||
|
||||
def test_checkpoint_publishes_signed_tip_outside_the_db(self):
|
||||
result = self.store.checkpoint(self.private, "bigpickle", "daily")
|
||||
self.assertEqual(result["anchor_path"], self.tip)
|
||||
self.assertTrue(os.path.exists(self.tip))
|
||||
# Tip must not live inside the SQLite file.
|
||||
with open(self.db, "rb") as handle:
|
||||
raw = handle.read()
|
||||
tip_text = open(self.tip, "r", encoding="utf-8").read()
|
||||
notice = json.loads(tip_text)
|
||||
self.assertNotIn(notice["signature"].encode(), raw)
|
||||
self.assertEqual(notice["head"], result["head"])
|
||||
self.assertEqual(notice["signer"], "bigpickle")
|
||||
self.assertTrue(self.store.verify_external_anchors()["ok"])
|
||||
|
||||
def test_publish_anchor_refuses_without_path(self):
|
||||
bare, private = helpers.fresh()
|
||||
with self.assertRaises(ValueError):
|
||||
bare.publish_anchor(private, "bigpickle")
|
||||
|
||||
def test_deleting_in_db_checkpoints_does_not_erase_the_tip(self):
|
||||
self.store.checkpoint(self.private, "bigpickle", "kept-offbox")
|
||||
self.store.conn.execute("DELETE FROM checkpoints")
|
||||
self.store.conn.commit()
|
||||
# In-DB check has nothing left to verify.
|
||||
self.assertTrue(self.store.verify_checkpoints()["ok"])
|
||||
self.assertEqual(self.store.verify_checkpoints()["checkpoints"], 0)
|
||||
# Off-box tip still present and valid against the live chain.
|
||||
external = self.store.verify_external_anchors()
|
||||
self.assertTrue(external["ok"], external)
|
||||
self.assertGreaterEqual(external["anchors"], 1)
|
||||
self.assertTrue(os.path.exists(self.tip))
|
||||
|
||||
def test_rewriting_the_db_is_caught_by_the_offbox_tip(self):
|
||||
self.store.checkpoint(self.private, "bigpickle", "baseline")
|
||||
tip_before = json.loads(open(self.tip).read())
|
||||
# Attacker deletes history, forges an event, rebuilds hashes, and
|
||||
# clears the in-DB checkpoints so verify_checkpoints is vacuously ok.
|
||||
self.store.conn.execute("DELETE FROM events")
|
||||
self.store.conn.execute("DELETE FROM checkpoints")
|
||||
self.store.conn.commit()
|
||||
self.store.record("forged", "attacker", subject="bigpickle",
|
||||
detail={"note": "never happened"})
|
||||
rows = self.store.conn.execute("SELECT * FROM events ORDER BY id").fetchall()
|
||||
prev = GENESIS
|
||||
for row in rows:
|
||||
digest = Store._event_hash(
|
||||
prev, row["at"], row["kind"], row["actor"],
|
||||
row["subject"], row["fingerprint"], row["detail"],
|
||||
)
|
||||
self.store.conn.execute(
|
||||
"UPDATE events SET prev_hash = ?, hash = ? WHERE id = ?",
|
||||
(prev, digest, row["id"]),
|
||||
)
|
||||
prev = digest
|
||||
self.store.conn.commit()
|
||||
self.assertTrue(self.store.verify_chain()["ok"])
|
||||
self.assertTrue(self.store.verify_checkpoints()["ok"])
|
||||
# Tip file still holds the pre-attack head.
|
||||
tip_after = json.loads(open(self.tip).read())
|
||||
self.assertEqual(tip_after["head"], tip_before["head"])
|
||||
external = self.store.verify_external_anchors()
|
||||
self.assertFalse(external["ok"], external)
|
||||
self.assertIn("off-box tip", external["failures"][0]["reason"])
|
||||
|
||||
def test_forged_offbox_signature_is_rejected(self):
|
||||
self.store.checkpoint(self.private, "bigpickle", "honest")
|
||||
notice = json.loads(open(self.tip).read())
|
||||
notice["signature"] = __import__("base64").b64encode(b"\x00" * 64).decode()
|
||||
open(self.tip, "w").write(json.dumps(notice))
|
||||
# Truncate the log so verify only sees the forged tip.
|
||||
log = self.tip + ".log"
|
||||
if os.path.exists(log):
|
||||
os.unlink(log)
|
||||
external = self.store.verify_external_anchors()
|
||||
self.assertFalse(external["ok"])
|
||||
self.assertIn("signature", external["failures"][0]["reason"])
|
||||
|
||||
def test_missing_configured_tip_fails_closed(self):
|
||||
# Configure a path that was never written.
|
||||
lonely = Store(self.db, anchor_path=os.path.join(self.tmp.name, "missing.tip"))
|
||||
external = lonely.verify_external_anchors()
|
||||
self.assertFalse(external["ok"])
|
||||
self.assertIn("missing", external["failures"][0]["reason"])
|
||||
|
||||
def test_no_anchor_path_skips_cleanly(self):
|
||||
store, private = helpers.fresh()
|
||||
store.checkpoint(private, "bigpickle", "in-db-only")
|
||||
external = store.verify_external_anchors()
|
||||
self.assertTrue(external["ok"])
|
||||
self.assertTrue(external.get("skipped"))
|
||||
|
||||
def test_anchor_log_keeps_prior_tips(self):
|
||||
first = self.store.checkpoint(self.private, "bigpickle", "one")
|
||||
identity.challenge(self.store, "bigpickle")
|
||||
second = self.store.checkpoint(self.private, "bigpickle", "two")
|
||||
self.assertNotEqual(first["head"], second["head"])
|
||||
history = anchor_mod.read_log(self.tip)
|
||||
self.assertGreaterEqual(len(history), 2)
|
||||
self.assertEqual(history[-1]["head"], second["head"])
|
||||
tip = anchor_mod.read_tip(self.tip)
|
||||
self.assertEqual(tip["head"], second["head"])
|
||||
self.assertTrue(self.store.verify_external_anchors()["ok"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
x
Reference in New Issue
Block a user