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_refuses_enrolled_name_with_wrong_key(self):
        # R4: signer name is enrolled, but private key is not the enrolled one.
        before = len(self.store.checkpoints())
        other_private, _ = keys.generate()
        with self.assertRaises(ValueError) as ctx:
            self.store.checkpoint(other_private, "bigpickle")
        self.assertIn("does not match enrolled key", str(ctx.exception))
        self.assertEqual(len(self.store.checkpoints()), before)

    def test_checkpoint_after_rekey_still_verifies_via_history(self):
        # N2: historical checkpoints verify against retired public keys.
        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.assertTrue(result["ok"], result)
        self.assertEqual(result["failures"], [])

    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")
        # Permanent revoke cleared the key; reinstate alone is not enough.
        with self.assertRaises(errors.UnknownAgent):
            identity.challenge(self.store, "bigpickle")
        new_pub = keys.generate()[1]
        secret = identity.seed_bootstrap(self.store, "bigpickle")
        identity.enrol_rekey(
            self.store, "bigpickle", keys.export_public_b64(new_pub), secret
        )
        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)
        self.assertIn("identity.rekeyed", 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 TestCompromiseReinstateRequiresRekey(unittest.TestCase):
    """R3: reinstate after permanent revoke must not restore the same key."""

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

    def test_reinstate_alone_cannot_use_old_key(self):
        old_token_path_chal = identity.challenge(self.store, "bigpickle")
        self.assertTrue(old_token_path_chal["fingerprint"])
        identity.revoke(self.store, "bigpickle", reason="key leak")
        identity.reinstate(self.store, "bigpickle")
        with self.assertRaises(errors.UnknownAgent):
            identity.challenge(self.store, "bigpickle")
        # Old private key must not verify even if a nonce were planted.
        self.assertFalse(identity.has_live_key(self.store.get_agent("bigpickle")))

    def test_timed_suspend_reinstate_keeps_same_key(self):
        import time
        fp = self.store.get_agent("bigpickle")["fingerprint"]
        identity.revoke(
            self.store, "bigpickle", until=int(time.time()) + 60, reason="pause"
        )
        identity.reinstate(self.store, "bigpickle")
        chal = identity.challenge(self.store, "bigpickle")
        self.assertEqual(chal["fingerprint"], fp)


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_timed_suspended(self):
        import time
        # Timed suspend keeps the key — rekey must not be a back door.
        identity.revoke(
            self.store, "bigpickle", until=int(time.time()) + 3600, reason="pause"
        )
        with self.assertRaises(errors.UnknownAgent):
            identity.seed_bootstrap(self.store, "bigpickle")
        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))
        self.assertTrue(identity.has_live_key(row))

    def test_permanent_revoke_allows_rekey_but_stays_revoked(self):
        """Compromise recovery: rotate key while banned, then reinstate."""
        old_fp = self.store.get_agent("bigpickle")["fingerprint"]
        identity.revoke(self.store, "bigpickle", reason="compromise")
        row = self.store.get_agent("bigpickle")
        self.assertTrue(identity.is_revoked(row))
        self.assertFalse(identity.has_live_key(row))
        new_pub = keys.generate()[1]
        secret = identity.seed_bootstrap(self.store, "bigpickle")
        new_fp = identity.enrol_rekey(
            self.store, "bigpickle", keys.export_public_b64(new_pub), secret
        )
        row = self.store.get_agent("bigpickle")
        self.assertTrue(identity.is_revoked(row))
        self.assertTrue(identity.has_live_key(row))
        self.assertNotEqual(new_fp, old_fp)
        with self.assertRaises(errors.UnknownAgent):
            identity.challenge(self.store, "bigpickle")
        identity.reinstate(self.store, "bigpickle")
        identity.challenge(self.store, "bigpickle")

    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


class TestKeyHistoryKeepsCheckpointsVerifiable(unittest.TestCase):
    """N2: permanent revoke / rekey must not make honest checkpoints unverifiable."""

    def test_checkpoints_verify_after_signer_permanent_revoke(self):
        import tempfile, os
        d = tempfile.mkdtemp()
        anchor = os.path.join(d, "tip")
        store = Store(":memory:", anchor_path=anchor)
        private, _ = helpers.enrol(store, "signer")
        store.record("identity.enrolled", actor="signer", subject="signer")
        store.checkpoint(private, "signer", "pre-compromise")
        before = store.verify_checkpoints()
        self.assertTrue(before["ok"], before)
        anchors_before = store.verify_external_anchors()
        self.assertTrue(anchors_before["ok"], anchors_before)
        identity.revoke(store, "signer", reason="compromise")
        after = store.verify_checkpoints()
        self.assertTrue(after["ok"], after)
        self.assertEqual(after["failures"], [])
        anchors_after = store.verify_external_anchors()
        self.assertTrue(anchors_after["ok"], anchors_after)
        # Live key wiped; history retained.
        self.assertFalse(identity.has_live_key(store.get_agent("signer")))
        hist = store.conn.execute(
            "SELECT fingerprint, retired_at FROM agent_keys WHERE agent = ?",
            ("signer",),
        ).fetchall()
        self.assertEqual(len(hist), 1)
        self.assertIsNotNone(hist[0]["retired_at"])

    def test_private_key_never_in_agent_keys(self):
        store, private = helpers.fresh("carol")
        cols = [r["name"] for r in store.conn.execute("PRAGMA table_info(agent_keys)")]
        self.assertNotIn("private", cols)
        self.assertNotIn("privkey", cols)
        row = store.conn.execute("SELECT * FROM agent_keys WHERE agent = ?", ("carol",)).fetchone()
        self.assertEqual(len(row["pubkey"]), 32)