sigil/tests/test_production.py

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