One test per finding in the second review pass

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