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 secrets as secrets_module
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()

class TestRekeyDoesNotUnrevoke(unittest.TestCase):
    """Finding: enrol_rekey used INSERT OR REPLACE and cleared revoked_at."""

    def setUp(self):
        self.store, self.private = helpers.fresh()

    def test_rekey_refuses_while_revoked(self):
        identity.revoke(self.store, "bigpickle", reason="incident")
        secret = None
        with self.assertRaises(errors.UnknownAgent):
            secret = identity.seed_bootstrap(self.store, "bigpickle")
        # Even with a secretly planted bootstrap, rekey must refuse.
        planted = secrets_module.token_urlsafe(24)
        self.store.put_bootstrap(
            __import__("hashlib").sha256(planted.encode()).hexdigest(),
            "bigpickle",
        )
        with self.assertRaises(errors.UnknownAgent):
            identity.enrol_rekey(
                self.store,
                "bigpickle",
                keys.export_public_b64(keys.generate()[1]),
                planted,
            )
        row = self.store.get_agent("bigpickle")
        self.assertTrue(identity.is_revoked(row))

    def test_rekey_preserves_revocation_columns(self):
        # Reinstate path: if somehow update ran under a revoked row, columns stay.
        identity.revoke(self.store, "bigpickle", reason="incident")
        row = self.store.get_agent("bigpickle")
        self.assertIsNotNone(row["revoked_at"])
        new_pub = keys.generate()[1]
        self.store.update_agent_key(
            "bigpickle",
            keys.public_bytes(new_pub),
            keys.fingerprint(new_pub),
        )
        row2 = self.store.get_agent("bigpickle")
        self.assertEqual(row2["revoked_at"], row["revoked_at"])
        self.assertEqual(row2["revoked_reason"], "incident")
        self.assertTrue(identity.is_revoked(row2))


class TestDualBootstrapEnrolRace(unittest.TestCase):
    """Finding: two secrets for one unenrolled agent could both enrol."""

    def test_exactly_one_enrol_wins_under_race(self):
        import threading

        store = Store(":memory:")
        # Plant two live secrets directly: put_bootstrap now keeps only one
        # live seed, so the dual-enrol race must bypass that hygiene.
        import hashlib, time as _time
        secrets = ["race-secret-alpha", "race-secret-beta"]
        now = int(_time.time())
        for s in secrets:
            store.conn.execute(
                "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
                (hashlib.sha256(s.encode()).hexdigest(), "racer", now),
            )
        store.conn.commit()
        pubs = [keys.generate()[1] for _ in range(2)]
        results = []
        lock = threading.Lock()

        def attempt(i):
            try:
                fp = identity.enrol(
                    store,
                    "racer",
                    keys.export_public_b64(pubs[i]),
                    secrets[i],
                )
                with lock:
                    results.append(("ok", fp, keys.fingerprint(pubs[i])))
            except errors.EnrolmentFailed as e:
                with lock:
                    results.append(("fail", str(e)))

        threads = [threading.Thread(target=attempt, args=(i,)) for i in range(2)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        wins = [r for r in results if r[0] == "ok"]
        fails = [r for r in results if r[0] == "fail"]
        self.assertEqual(len(wins), 1, results)
        self.assertEqual(len(fails), 1, results)
        row = store.get_agent("racer")
        self.assertEqual(row["fingerprint"], wins[0][2])


class TestRedeemRateLimitIsolation(unittest.TestCase):
    """Finding: one global code_redeem_fail bucket locked every agent."""

    def test_failed_guesses_do_not_lock_other_agents_valid_code(self):
        store_a, private_a = helpers.fresh("alice")
        # Exhaust anonymous fail bucket
        hit_limit = False
        for i in range(50):
            try:
                viewcodes.redeem_code(store_a, f"ZZZZ-{i:04d}")
            except errors.RateLimited:
                hit_limit = True
                break
            except errors.SigilError:
                pass
        self.assertTrue(hit_limit, "anonymous guessing should still rate-limit")

        # A valid code for alice must still redeem — success uses code_redeem:alice
        issued = helpers.issue(store_a, private_a, agent="alice", limit=5000, window=60)
        view = viewcodes.redeem_code(store_a, issued["code"])
        self.assertEqual(view["on_behalf_of"], "alice")

    def test_successful_redeems_do_not_consume_fail_bucket(self):
        store, private = helpers.fresh()
        for _ in range(5):
            issued = helpers.issue(store, private, limit=5000, window=60)
            viewcodes.redeem_code(store, issued["code"])
        # Fail bucket should still accept unknown guesses (not exhausted by successes)
        try:
            viewcodes.redeem_code(store, "AAAA-AAAA")
        except errors.RateLimited:
            self.fail("successes must not share the fail bucket")
        except errors.SigilError:
            pass