Harden Sigil: tests/test_hardening.py

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()