From 910ddfde4256cd5e6935876d3c5fae732e9434fc Mon Sep 17 00:00:00 2001 From: bigpickle Date: Mon, 5 Oct 2026 21:26:17 +0800 Subject: [PATCH] Harden Sigil: tests/test_hardening.py --- tests/test_hardening.py | 362 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 362 insertions(+) create mode 100644 tests/test_hardening.py diff --git a/tests/test_hardening.py b/tests/test_hardening.py new file mode 100644 index 0000000..34ba7cf --- /dev/null +++ b/tests/test_hardening.py @@ -0,0 +1,362 @@ +"""Hardening tests: the findings from the self-review. + +Each test here corresponds to a defect that was found and fixed. If any of +these regress, the corresponding vulnerability is back. +""" + +import base64 +import os +import tempfile +import threading +import time +import unittest + +from sigil import errors, identity, keys, sessions, viewcodes +from sigil.store import Store + +import helpers + + +def enrolled(store=None, agent="bigpickle"): + store = store or Store(":memory:") + private, _ = helpers.enrol(store, agent) + return store, private, None + + +sign = helpers.sign + + +def issue(store, private, agent="bigpickle", **kw): + return helpers.issue(store, private, agent, **kw) + + +class TestSingleUseUnderRace(unittest.TestCase): + """Finding: check-then-update let one code mint several sessions. + + Measured before the fix: a single-use code succeeded 753 times across 200 + trials. These tests hold the invariant under real concurrency. + """ + + TRIALS = 40 + THREADS = 8 + + def _race(self, setup, threads=THREADS): + """Set up one reusable credential per trial, then race for it. + + setup returns (store, shared, private). `shared` is the single-use + credential the threads all compete for, created once outside the + threads, so this genuinely tests the same-secret-twice path. + """ + for _ in range(self.TRIALS): + store, shared, private = setup() + barrier = threading.Barrier(threads) + outcomes = [] + + def go(): + barrier.wait() + try: + work(store, shared, private) + outcomes.append(1) + except Exception: + outcomes.append(0) + + pool = [threading.Thread(target=go) for _ in range(threads)] + for t in pool: + t.start() + for t in pool: + t.join() + yield sum(outcomes) + + def test_one_code_yields_at_most_one_session(self): + def setup(): + store, private, _ = enrolled() + return store, helpers.issue(store, private)["code"], private + + def work(store, code, private): + viewcodes.redeem_code(store, code) + + for successes in self._race(setup): + self.assertLessEqual(successes, 1, "one code minted multiple sessions") + + def test_one_bootstrap_secret_enrols_at_most_once(self): + def setup(): + store, private, _ = enrolled() + return store, identity.seed_bootstrap(store, "racer"), private + + def work(store, secret, private): + identity.enrol( + store, "racer", keys.export_public_b64(keys.generate()[1]), secret + ) + + for successes in self._race(setup, threads=6): + self.assertLessEqual(successes, 1, "one secret enrolled multiple keys") + + def test_one_nonce_verifies_at_most_once(self): + def setup(): + store, private, _ = enrolled() + nonce = identity.challenge(store, "bigpickle")["nonce"] + return store, nonce, private + + def work(store, nonce, private): + identity.verify(store, nonce, sign(private, nonce)) + + for successes in self._race(setup, threads=6): + self.assertLessEqual(successes, 1, "one nonce produced multiple sessions") + + +class TestTokensNotStoredInClear(unittest.TestCase): + """Finding: session tokens were stored verbatim, so a database leak + handed an attacker live sessions.""" + + def setUp(self): + self.store, self.private, self.public = enrolled() + + def _agent_token(self): + chal = identity.challenge(self.store, "bigpickle") + return identity.verify(self.store, chal["nonce"], sign(self.private, chal["nonce"]))["token"] + + def test_agent_token_appears_nowhere_in_the_database_file(self): + import sqlite3 + + path = tempfile.mktemp(suffix=".db") + store = Store(path) + private, public = keys.generate() + secret = identity.seed_bootstrap(store, "bigpickle") + identity.enrol(store, "bigpickle", keys.export_public_b64(public), secret) + chal = identity.challenge(store, "bigpickle") + token = identity.verify( + store, chal["nonce"], sign(private, chal["nonce"]) + )["token"] + store.close() + with open(path, "rb") as handle: + blob = handle.read() + os.unlink(path) + self.assertNotIn(token.encode(), blob, "session token stored in the clear") + + def test_human_token_appears_nowhere_in_the_database_file(self): + import sqlite3 + + path = tempfile.mktemp(suffix=".db") + store = Store(path) + private, public = keys.generate() + secret = identity.seed_bootstrap(store, "bigpickle") + identity.enrol(store, "bigpickle", keys.export_public_b64(public), secret) + code = helpers.issue(store, private)["code"] + token = viewcodes.redeem_code(store, code)["token"] + store.close() + with open(path, "rb") as handle: + blob = handle.read() + os.unlink(path) + self.assertNotIn(token.encode(), blob, "view session token stored in the clear") + + def test_hashed_token_still_resolves(self): + token = self._agent_token() + self.assertEqual( + sessions.resolve(self.store, token)["on_behalf_of"], "bigpickle" + ) + + +class TestAuditChain(unittest.TestCase): + """The log is an attestation substrate, so it must be tamper-evident.""" + + def test_empty_log_verifies(self): + self.assertTrue(Store(":memory:").verify_chain()["ok"]) + + def test_intact_log_verifies(self): + store, private, public = enrolled() + for _ in range(5): + identity.challenge(store, "bigpickle") + result = store.verify_chain() + self.assertTrue(result["ok"]) + self.assertEqual(result["events"], len(store.events(limit=50))) + + def test_edited_detail_is_detected(self): + store, private, public = enrolled() + identity.challenge(store, "bigpickle") + store.conn.execute( + "UPDATE events SET detail = '{\"tampered\":true}' WHERE id = 1" + ) + store.conn.commit() + result = store.verify_chain() + self.assertFalse(result["ok"]) + self.assertEqual(result["broken_at"], 1) + + def test_deleted_event_is_detected(self): + store, private, public = enrolled() + identity.challenge(store, "bigpickle") + store.conn.execute("DELETE FROM events WHERE id = 1") + store.conn.commit() + self.assertFalse(store.verify_chain()["ok"]) + + def test_reordered_events_are_detected(self): + store, private, public = enrolled() + for _ in range(3): + identity.challenge(store, "bigpickle") + store.conn.execute("UPDATE events SET kind = 'tampered' WHERE id = 2") + store.conn.commit() + self.assertFalse(store.verify_chain()["ok"]) + + +class TestKeyRotationKillsSessions(unittest.TestCase): + """Finding: rekey replaced the key but left existing sessions alive, so + rotating after a compromise bought nothing.""" + + def setUp(self): + self.store, self.private, self.public = enrolled() + + def _token(self): + chal = identity.challenge(self.store, "bigpickle") + return identity.verify( + self.store, chal["nonce"], sign(self.private, chal["nonce"]) + )["token"] + + def test_rekey_revokes_existing_sessions(self): + token = self._token() + sessions.resolve(self.store, token) # alive before + secret = identity.seed_bootstrap(self.store, "bigpickle") + identity.enrol_rekey( + self.store, "bigpickle", + keys.export_public_b64(keys.generate()[1]), secret, + ) + with self.assertRaises(errors.SessionExpired): + sessions.resolve(self.store, token) + + def test_old_key_cannot_authenticate_after_rekey(self): + secret = identity.seed_bootstrap(self.store, "bigpickle") + identity.enrol_rekey( + self.store, "bigpickle", + keys.export_public_b64(keys.generate()[1]), secret, + ) + chal = identity.challenge(self.store, "bigpickle") + with self.assertRaises(errors.BadSignature): + identity.verify(self.store, chal["nonce"], sign(self.private, chal["nonce"])) + + def test_rotation_is_recorded(self): + secret = identity.seed_bootstrap(self.store, "bigpickle") + identity.enrol_rekey( + self.store, "bigpickle", + keys.export_public_b64(keys.generate()[1]), secret, + ) + kinds = [e["kind"] for e in self.store.events(limit=50)] + self.assertIn("identity.rekeyed", kinds) + + +class TestRateLimits(unittest.TestCase): + """Finding: challenge() was unmetered, an open signing oracle.""" + + def setUp(self): + self.store, self.private, self.public = enrolled() + + def test_challenge_is_rate_limited(self): + raised = False + for _ in range(200): + try: + identity.challenge(self.store, "bigpickle") + except errors.RateLimited: + raised = True + break + self.assertTrue(raised, "challenge() was never rate limited") + + def test_code_issue_is_rate_limited(self): + raised = False + for _ in range(200): + try: + helpers.issue(self.store, self.private) + except errors.RateLimited: + raised = True + break + self.assertTrue(raised, "issue_code() was never rate limited") + + def test_redemption_guessing_is_rate_limited(self): + refused = 0 + for i in range(500): + try: + viewcodes.redeem_code(self.store, f"ZZZZ-{i:04d}") + except errors.RateLimited: + break + except errors.SigilError: + refused += 1 + else: + self.fail("code guessing was not rate limited") + + def test_limits_are_per_agent(self): + for _ in range(40): + try: + identity.challenge(self.store, "bigpickle") + except errors.RateLimited: + break + # a different agent is unaffected + self.store2, _, _ = enrolled(self.store, agent="second") + identity.challenge(self.store2, "second") + + +class TestKeyEncryptionAtRest(unittest.TestCase): + """Finding: keys were written with NoEncryption by default.""" + + def test_encrypted_key_needs_passphrase(self): + private, _ = keys.generate() + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "k.pem") + keys.save_private(private, path, "a-long-enough-passphrase") + self.assertTrue(keys.is_encrypted(path)) + with self.assertRaises(ValueError): + keys.load_private(path) + + def test_wrong_passphrase_refuses(self): + private, _ = keys.generate() + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "k.pem") + keys.save_private(private, path, "a-long-enough-passphrase") + with self.assertRaises(Exception): + keys.load_private(path, "the-wrong-passphrase") + + def test_correct_passphrase_roundtrips(self): + private, _ = keys.generate() + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "k.pem") + keys.save_private(private, path, "a-long-enough-passphrase") + loaded = keys.load_private(path, "a-long-enough-passphrase") + self.assertEqual(loaded.sign(b"m"), private.sign(b"m")) + + def test_short_passphrase_refused(self): + private, _ = keys.generate() + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(ValueError): + keys.save_private(private, os.path.join(tmp, "k.pem"), "short") + + def test_plaintext_key_still_loads(self): + private, _ = keys.generate() + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "k.pem") + keys.save_private(private, path) + self.assertFalse(keys.is_encrypted(path)) + keys.load_private(path) + + +class TestAgentIdValidation(unittest.TestCase): + """Finding: agent ids were unvalidated, so an id carrying a newline + could forge audit lines.""" + + def test_bad_ids_refused(self): + store = Store(":memory:") + for bad in ["", "Big Pickle", "a" * 65, "with\nnewline", "with space", + "-leading", "with/slash", "with;drop"]: + with self.assertRaises(errors.SigilError, msg=f"accepted {bad!r}"): + identity.seed_bootstrap(store, bad) + + def test_good_ids_accepted(self): + store = Store(":memory:") + for good in ["bigpickle", "agent-2", "a.b_c", "x" * 64]: + private, public = keys.generate() + secret = identity.seed_bootstrap(store, good) + identity.enrol(store, good, keys.export_public_b64(public), secret) + self.assertIn(good, store.agents()) + + def test_newline_id_cannot_reach_the_audit_log(self): + store = Store(":memory:") + with self.assertRaises(errors.SigilError): + identity.seed_bootstrap(store, "evil\nfake-event") + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file