From 9ce8da75730dbd4647598b0be44ef69074d55bb0 Mon Sep 17 00:00:00 2001 From: bigpickle Date: Mon, 5 Oct 2026 21:51:18 +0800 Subject: [PATCH] One test per finding in the second review pass --- tests/test_production.py | 294 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 294 insertions(+) create mode 100644 tests/test_production.py diff --git a/tests/test_production.py b/tests/test_production.py new file mode 100644 index 0000000..3c09102 --- /dev/null +++ b/tests/test_production.py @@ -0,0 +1,294 @@ +"""Production-readiness tests: the second hardening pass. + +Same rule as test_hardening.py: one test per defect, so a regression brings the +defect back rather than quietly passing. +""" + +import base64 +import os +import tempfile +import time +import unittest + +from sigil import errors, identity, keys, sessions, viewcodes +from sigil import store as store_mod +from sigil.store import Store + +import helpers + + +class TestAuditLogDoesNotLeakCodes(unittest.TestCase): + """Finding: the audit log recorded `code[:5]`. + + A code is 8 characters from a 32-character alphabet, so 40 bits. The first + four are 20 of those bits, which leaves about a million candidates - and the + audit log is the artifact most likely to be exported or shared, so it is the + worst place to leave 20 bits of a live credential. + """ + + def setUp(self): + self.store, self.private = helpers.fresh() + self.code = helpers.issue(self.store, self.private)["code"] + + def _log_text(self): + return str(self.store.events(limit=100)) + + def test_no_part_of_the_code_appears_in_the_log(self): + bare = self.code.replace("-", "") + text = self._log_text() + for size in range(1, len(bare) + 1): + for start in range(len(bare) - size + 1): + fragment = bare[start:start + size] + if len(fragment) >= 3: # 1-2 chars are coincidence, not a leak + self.assertNotIn( + fragment, text, f"log leaked {size} code characters" + ) + + def test_log_records_a_digest_reference_instead(self): + import hashlib + + events = self.store.events(limit=100) + issued = [e for e in events if e["kind"] == "view_code.issued"][0] + expected = hashlib.sha256( + self.code.replace("-", "").upper().encode() + ).hexdigest()[:12] + self.assertEqual(issued["detail"]["code_ref"], f"code:{expected}") + self.assertNotIn("code_prefix", issued["detail"]) + + def test_revoking_a_session_does_not_log_the_token(self): + token = helpers.agent_token(self.store, self.private) + sessions.revoke(self.store, token) + events = [e for e in self.store.events(limit=100) + if e["kind"] == "session.revoked"] + self.assertTrue(events) + self.assertNotIn(token[:8], str(events[0]["detail"])) + self.assertIn("token_ref", events[0]["detail"]) + + +class TestSignedCheckpoints(unittest.TestCase): + """Finding: the hash chain is unkeyed. + + Anyone with write access to the database can rewrite events and recompute + every digest, and verify_chain agrees. A checkpoint is signed by a key the + database does not contain, which makes that detectable. + """ + + def setUp(self): + self.store, self.private = helpers.fresh() + identity.challenge(self.store, "bigpickle") + self.store.checkpoint(self.private, "bigpickle", "baseline") + + def _rechain(self): + """Recompute every digest, as a database-level attacker would.""" + rows = self.store.conn.execute("SELECT * FROM events ORDER BY id").fetchall() + prev = store_mod.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() + + def test_honest_log_passes_both_checks(self): + self.assertTrue(self.store.verify_chain()["ok"]) + self.assertTrue(self.store.verify_checkpoints()["ok"]) + + def test_edited_event_breaks_the_chain(self): + self.store.conn.execute( + "UPDATE events SET detail = '{\"lies\":1}' WHERE id = 1" + ) + self.store.conn.commit() + self.assertFalse(self.store.verify_chain()["ok"]) + + def test_full_rewrite_with_recomputed_hashes_is_caught_by_the_checkpoint(self): + # Delete real history, forge something, then rebuild the chain so + # verify_chain is happy again. Only the signature catches this. + self.store.conn.execute("DELETE FROM events WHERE id IN (2, 3)") + self.store.conn.commit() + self.store.record("forged", "attacker", subject="bigpickle", + detail={"note": "never happened"}) + self._rechain() + self.assertTrue(self.store.verify_chain()["ok"], "chain was recomputed") + result = self.store.verify_checkpoints() + self.assertFalse(result["ok"], "checkpoint should catch the rewrite") + self.assertIn("no longer matches", result["failures"][0]["reason"]) + + def test_forged_checkpoint_signature_is_rejected(self): + self.store.conn.execute( + "UPDATE checkpoints SET signature = ? WHERE id = 1", + (base64.b64encode(b"\x00" * 64).decode(),), + ) + self.store.conn.commit() + result = self.store.verify_checkpoints() + self.assertFalse(result["ok"]) + + def test_checkpoint_cannot_be_signed_by_a_stranger(self): + stranger = Store(":memory:") + with self.assertRaises(ValueError): + stranger.checkpoint(self.private, "not-enrolled") + + 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.assertFalse(result["ok"]) + self.assertIn("key changed", result["failures"][0]["reason"]) + + def test_empty_log_checkpoints_cleanly(self): + store, private = helpers.fresh() + store.checkpoint(private, "bigpickle", "genesis") + self.assertTrue(store.verify_checkpoints()["ok"]) + + +class TestRevocationSemantics(unittest.TestCase): + """Finding: there was no revoke operation at all, only an unused column.""" + + def setUp(self): + self.store, self.private = helpers.fresh() + + def _token(self): + return helpers.agent_token(self.store, self.private) + + def test_revoking_stops_challenges(self): + identity.revoke(self.store, "bigpickle", reason="incident") + with self.assertRaises(errors.UnknownAgent): + identity.challenge(self.store, "bigpickle") + + def test_revoking_kills_live_sessions(self): + token = self._token() + identity.revoke(self.store, "bigpickle", reason="incident") + with self.assertRaises(errors.SessionExpired): + sessions.resolve(self.store, token) + + def test_revocation_is_permanent_by_default(self): + identity.revoke(self.store, "bigpickle") + row = self.store.get_agent("bigpickle") + self.assertTrue(identity.is_revoked(row, now=time.time() + 10 ** 9)) + + def test_temporary_revocation_lapses(self): + identity.revoke(self.store, "bigpickle", + until=int(time.time()) + 1, reason="pause") + row = self.store.get_agent("bigpickle") + self.assertTrue(identity.is_revoked(row)) + self.assertFalse( + identity.is_revoked(row, now=time.time() + 60), + "a time-boxed revocation should expire", + ) + + def test_agent_works_again_after_a_lapsed_revocation(self): + identity.revoke(self.store, "bigpickle", until=int(time.time()) - 1) + chal = identity.challenge(self.store, "bigpickle") + identity.verify(self.store, chal["nonce"], + helpers.sign(self.private, chal["nonce"])) + + def test_reinstating_works_and_is_audited(self): + identity.revoke(self.store, "bigpickle") + identity.reinstate(self.store, "bigpickle") + 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) + + def test_revoking_an_unknown_agent_refused(self): + with self.assertRaises(errors.UnknownAgent): + identity.revoke(self.store, "ghost") + + def test_revocation_records_the_actor_and_reason(self): + identity.revoke(self.store, "bigpickle", reason="key leak", + actor="leora") + event = [e for e in self.store.events(limit=50) + if e["kind"] == "identity.revoked"][0] + self.assertEqual(event["actor"], "leora") + self.assertEqual(event["detail"]["reason"], "key leak") + self.assertTrue(event["detail"]["permanent"]) + + def test_revocation_drops_outstanding_challenges(self): + chal = identity.challenge(self.store, "bigpickle") + identity.revoke(self.store, "bigpickle") + self.assertIsNone(self.store.get_nonce(chal["nonce"])) + + def test_a_revoked_agent_cannot_still_verify(self): + """An in-flight challenge must not complete after revocation. + + Revocation drops outstanding nonces, so this is refused at the nonce + rather than at the agent check. Either refusal is correct; what matters + is that it does not succeed. + """ + chal = identity.challenge(self.store, "bigpickle") + identity.revoke(self.store, "bigpickle") + with self.assertRaises(errors.SigilError): + identity.verify(self.store, chal["nonce"], + helpers.sign(self.private, chal["nonce"])) + + +class TestViewSessionsCannotBeForged(unittest.TestCase): + """Finding: human sessions were minted from an agent name. + + The only route to a human session should be a redeemed code. + """ + + def setUp(self): + self.store, self.private = helpers.fresh() + + def test_granted_code_cannot_be_constructed_directly(self): + with self.assertRaises(errors.SigilError): + sessions.GrantedCode("bigpickle", ["view:read"], 3600) + + def test_issue_view_session_rejects_a_bare_agent_name(self): + with self.assertRaises((AttributeError, errors.SigilError, TypeError)): + sessions.issue_view_session(self.store, "bigpickle") + + def test_the_real_path_still_works(self): + code = helpers.issue(self.store, self.private)["code"] + view = viewcodes.redeem_code(self.store, code) + session = sessions.resolve(self.store, view["token"]) + self.assertEqual(session["kind"], sessions.HUMAN_VIA_AGENT) + + +class TestDatabaseRobustness(unittest.TestCase): + """Concurrency and durability settings a production database needs.""" + + def test_failed_transaction_does_not_leave_state_open(self): + store, private = helpers.fresh() + before = store.get_agent("bigpickle") + with self.assertRaises(Exception): + store._exec("INSERT INTO agents (agent, pubkey, fingerprint, enrolled_at)" + " VALUES ('bigpickle', x'00', 'dup', 0)") + # The failed insert must not be committed by some later unrelated write. + store.record("after.failure", actor="test") + after = store.get_agent("bigpickle") + self.assertEqual(after["pubkey"], before["pubkey"]) + self.assertTrue(store.verify_chain()["ok"]) + + def test_busy_timeout_is_set(self): + store, private = helpers.fresh() + value = store.conn.execute("PRAGMA busy_timeout").fetchone()[0] + self.assertGreater(value, 0, "writers would fail instantly without this") + + def test_sweep_prunes_only_bookkeeping(self): + store, private = helpers.fresh() + chal = identity.challenge(store, "bigpickle") + identity.verify(store, chal["nonce"], helpers.sign(private, chal["nonce"])) + # backdate a spent nonce so it is genuinely past its window + store.conn.execute("UPDATE nonces SET expires_at = ? WHERE nonce = ?", + (int(time.time()) - 10, chal["nonce"])) + store.conn.commit() + sessions_before = len(store.sessions()) + swept = store.sweep() + self.assertGreaterEqual(swept["nonces"], 1) + # Sessions are retained even though sweep runs: the audit trail refers + # to them, so pruning them would break the story the log tells. + self.assertEqual(len(store.sessions()), sessions_before) + self.assertEqual(swept["sessions"], 0) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file