sigil/tests/test_viewcodes.py

"""One-time codes: single use, short redemption window, capped capability."""

import time
import unittest

from sigil import errors, sessions, viewcodes
from sigil.store import Store
from sigil import identity, keys

import helpers


class Base(unittest.TestCase):
    def setUp(self):
        self.store = Store(":memory:")
        self.private, _ = helpers.enrol(self.store, "bigpickle")

    def issue(self, agent="bigpickle", **kw):
        return helpers.issue(self.store, self.private, agent, **kw)


class TestIssueCodes(Base):
    def test_issue_returns_code_and_two_lifetimes(self):
        issued = self.issue("bigpickle", limit=1000)
        self.assertIn("-", issued["code"])
        self.assertEqual(len(issued["code"].replace("-", "")), 8)
        # the code window is strictly shorter than the session window
        self.assertLess(issued["code_ttl"], issued["view_ttl"])
        self.assertLess(issued["redeem_expires_at"], issued["redeem_expires_at"] + issued["view_ttl"])

    def test_code_ttl_must_be_shorter_than_view_ttl(self):
        with self.assertRaises(errors.SigilError):
            self.issue("bigpickle", code_ttl=3600, view_ttl=60)

    def test_cannot_issue_code_with_write_scope(self):
        with self.assertRaises(errors.ScopeDenied):
            self.issue("bigpickle", scopes=["view:write"])

    def test_cannot_issue_code_for_unknown_agent(self):
        with self.assertRaises(errors.UnknownAgent):
            self.issue("ghost")

    def test_codes_are_distinct(self):
        codes = {self.issue("bigpickle", limit=1000)["code"]
                 for _ in range(25)}
        self.assertEqual(len(codes), 25)

    def test_code_alphabet_has_no_ambiguous_characters(self):
        for _ in range(200):
            code = self.issue("bigpickle", limit=1000)["code"]
            self.assertNotIn("0", code)
            self.assertNotIn("O", code)
            self.assertNotIn("1", code)
            self.assertNotIn("I", code)


class TestRedeemCodes(Base):
    def test_happy_path(self):
        issued = self.issue("bigpickle", limit=1000)
        view = viewcodes.redeem_code(self.store, issued["code"])
        self.assertEqual(view["kind"], "human_via_agent")
        self.assertEqual(view["on_behalf_of"], "bigpickle")

    def test_code_is_single_use(self):
        issued = self.issue("bigpickle", limit=1000)
        viewcodes.redeem_code(self.store, issued["code"])
        with self.assertRaises(errors.CodeUsed):
            viewcodes.redeem_code(self.store, issued["code"])

    def test_expired_code_refused(self):
        issued = self.issue("bigpickle", code_ttl=1, view_ttl=600)
        time.sleep(1.2)
        with self.assertRaises(errors.CodeExpired):
            viewcodes.redeem_code(self.store, issued["code"])

    def test_unknown_code_refused(self):
        with self.assertRaises(errors.CodeInvalid):
            viewcodes.redeem_code(self.store, "ZZZZ-ZZZZ")

    def test_code_is_case_and_space_insensitive(self):
        issued = self.issue("bigpickle", limit=1000)
        mangled = issued["code"].lower().replace("-", " ")
        view = viewcodes.redeem_code(self.store, mangled)
        self.assertEqual(view["on_behalf_of"], "bigpickle")

    def test_expired_code_grants_no_session(self):
        issued = self.issue("bigpickle", code_ttl=1, view_ttl=600)
        time.sleep(1.2)
        before = len(self.store.sessions())
        with self.assertRaises(errors.CodeExpired):
            viewcodes.redeem_code(self.store, issued["code"])
        self.assertEqual(len(self.store.sessions()), before)

    def test_session_outlives_the_code_window(self):
        """The point of splitting the lifetimes: after the code dies the
        session it opened is still good."""
        issued = self.issue("bigpickle", code_ttl=1, view_ttl=600)
        view = viewcodes.redeem_code(self.store, issued["code"])
        time.sleep(1.2)
        session = sessions.resolve(self.store, view["token"])
        self.assertEqual(session["kind"], "human_via_agent")


class TestAuditTrail(Base):
    def test_issue_and_redeem_are_recorded(self):
        issued = self.issue("bigpickle", limit=1000)
        viewcodes.redeem_code(self.store, issued["code"])
        kinds = [e["kind"] for e in self.store.events(limit=50)]
        self.assertIn("view_code.issued", kinds)
        self.assertIn("view_code.redeemed", kinds)

    def test_human_appears_as_a_human_in_the_log(self):
        issued = self.issue("bigpickle", limit=1000)
        viewcodes.redeem_code(self.store, issued["code"])
        redemption = [e for e in self.store.events(limit=50)
                      if e["kind"] == "view_code.redeemed"][0]
        self.assertEqual(redemption["subject"], "bigpickle's human")

    def test_replay_attempt_is_recorded(self):
        issued = self.issue("bigpickle", limit=1000)
        viewcodes.redeem_code(self.store, issued["code"])
        with self.assertRaises(errors.CodeUsed):
            viewcodes.redeem_code(self.store, issued["code"])
        kinds = [e["kind"] for e in self.store.events(limit=50)]
        self.assertIn("view_code.redeem_refused", kinds)

    def test_replay_attempt_never_names_the_code(self):
        issued = self.issue("bigpickle", limit=1000)
        viewcodes.redeem_code(self.store, issued["code"])
        with self.assertRaises(errors.CodeUsed):
            viewcodes.redeem_code(self.store, issued["code"])
        blob = repr(self.store.events(limit=50))
        self.assertNotIn(issued["code"], blob)

    def test_secret_is_never_stored_in_the_clear(self):
        fresh = Store(":memory:")
        secret = identity.seed_bootstrap(fresh, "bigpickle")
        _, public = keys.generate()
        identity.enrol(fresh, "bigpickle", keys.export_public_b64(public), secret)
        blob = repr(fresh.events(limit=100))
        self.assertNotIn(secret, blob)


if __name__ == "__main__":
    unittest.main()

class TestRevokeAndRekeyCloseViewCodes(Base):
    """R1: revoke / rekey / suspend must close outstanding view codes."""

    def test_redeem_refused_after_permanent_revoke(self):
        issued = helpers.issue(self.store, self.private)
        identity.revoke(self.store, "bigpickle", reason="compromise")
        with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
            viewcodes.redeem_code(self.store, issued["code"])

    def test_redeem_refused_during_timed_suspension(self):
        import time
        issued = helpers.issue(self.store, self.private)
        identity.revoke(
            self.store, "bigpickle", until=int(time.time()) + 3600, reason="pause"
        )
        with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
            viewcodes.redeem_code(self.store, issued["code"])

    def test_redeem_refused_after_rekey(self):
        issued = helpers.issue(self.store, self.private)
        new_private, new_public = keys.generate()
        secret = identity.seed_bootstrap(self.store, "bigpickle")
        identity.enrol_rekey(
            self.store,
            "bigpickle",
            keys.export_public_b64(new_public),
            secret,
        )
        # Invalidate-on-rekey marks the code used; redeem must not mint a session.
        with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
            viewcodes.redeem_code(self.store, issued["code"])
        # New agent can still mint and redeem a fresh code.
        fresh = helpers.issue(self.store, new_private)
        session = viewcodes.redeem_code(self.store, fresh["code"])
        self.assertEqual(session["kind"], "human_via_agent")

    def test_unused_codes_marked_used_on_revoke(self):
        issued = helpers.issue(self.store, self.private)
        digest = viewcodes._hash_code(issued["code"])
        row = self.store.get_code(digest)
        self.assertIsNone(row["used_at"])
        identity.revoke(self.store, "bigpickle")
        row = self.store.get_code(digest)
        self.assertIsNotNone(row["used_at"])