sigil/tests/test_identity.py

"""Enrolment and challenge-response: the properties that matter are the refusals."""

import base64
import unittest

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

import helpers


class Base(unittest.TestCase):
    def setUp(self):
        self.store = Store(":memory:")
        self.private, self.public = keys.generate()
        self.pub_b64 = keys.export_public_b64(self.public)
        self.secret = identity.seed_bootstrap(self.store, "bigpickle")

    def sign_nonce(self, nonce, private=None):
        return base64.b64encode(
            (private or self.private).sign(nonce.encode("utf-8"))
        ).decode("ascii")

    def enrol(self):
        return identity.enrol(self.store, "bigpickle", self.pub_b64, self.secret)


class TestEnrolment(Base):
    def test_enrol_returns_fingerprint(self):
        fingerprint = self.enrol()
        self.assertEqual(fingerprint, keys.fingerprint(self.public))

    def test_bootstrap_is_single_use(self):
        self.enrol()
        _, second_public = keys.generate()
        with self.assertRaises(errors.EnrolmentFailed):
            identity.enrol(
                self.store, "bigpickle",
                keys.export_public_b64(second_public), self.secret,
            )

    def test_wrong_bootstrap_secret_refused(self):
        with self.assertRaises(errors.EnrolmentFailed):
            identity.enrol(self.store, "bigpickle", self.pub_b64, "not-the-secret")

    def test_cannot_enrol_twice_with_different_secret(self):
        self.enrol()
        other = identity.seed_bootstrap(self.store, "bigpickle")
        with self.assertRaises(errors.EnrolmentFailed):
            identity.enrol(self.store, "bigpickle", self.pub_b64, other)

    def test_secret_for_one_agent_cannot_enrol_another(self):
        with self.assertRaises(errors.EnrolmentFailed):
            identity.enrol(self.store, "someone-else", self.pub_b64, self.secret)

    def test_challenge_unknown_agent(self):
        with self.assertRaises(errors.UnknownAgent):
            identity.challenge(self.store, "nobody")


class TestChallengeResponse(Base):
    def setUp(self):
        super().setUp()
        self.enrol()

    def test_happy_path_issues_agent_session(self):
        chal = identity.challenge(self.store, "bigpickle")
        session = identity.verify(self.store, chal["nonce"], self.sign_nonce(chal["nonce"]))
        self.assertEqual(session["kind"], "agent")
        self.assertEqual(session["subject"], "bigpickle")
        self.assertIn("agent:act", session["scopes"])

    def test_replay_of_a_valid_signature_is_refused(self):
        chal = identity.challenge(self.store, "bigpickle")
        sig = self.sign_nonce(chal["nonce"])
        identity.verify(self.store, chal["nonce"], sig)
        with self.assertRaises(errors.NonceUsed):
            identity.verify(self.store, chal["nonce"], sig)

    def test_signature_from_the_wrong_key_is_refused(self):
        other_private, _ = keys.generate()
        chal = identity.challenge(self.store, "bigpickle")
        with self.assertRaises(errors.BadSignature):
            identity.verify(
                self.store, chal["nonce"],
                self.sign_nonce(chal["nonce"], other_private),
            )

    def test_bad_signature_still_consumes_the_nonce(self):
        """A wrong attempt must not leave the nonce open for another try."""
        chal = identity.challenge(self.store, "bigpickle")
        with self.assertRaises(errors.BadSignature):
            identity.verify(self.store, chal["nonce"], base64.b64encode(b"junk").decode())
        with self.assertRaises(errors.NonceUsed):
            identity.verify(self.store, chal["nonce"], self.sign_nonce(chal["nonce"]))

    def test_expired_nonce_refused(self):
        chal = identity.challenge(self.store, "bigpickle", ttl=-1)
        with self.assertRaises(errors.NonceExpired):
            identity.verify(self.store, chal["nonce"], self.sign_nonce(chal["nonce"]))

    def test_unknown_nonce_refused(self):
        with self.assertRaises(errors.NonceUnknown):
            identity.verify(self.store, "no-such-nonce", self.sign_nonce("x"))

    def test_malformed_signature_b64_refused(self):
        chal = identity.challenge(self.store, "bigpickle")
        with self.assertRaises(errors.BadSignature):
            identity.verify(self.store, chal["nonce"], "!!!not-base64!!!")

    def test_signature_is_bound_to_that_nonce_only(self):
        """A signature captured from one challenge cannot answer another."""
        first = identity.challenge(self.store, "bigpickle")
        second = identity.challenge(self.store, "bigpickle")
        captured = self.sign_nonce(first["nonce"])
        with self.assertRaises(errors.BadSignature):
            identity.verify(self.store, second["nonce"], captured)


class TestRekey(Base):
    def setUp(self):
        super().setUp()
        self.enrol()

    def test_rekey_replaces_the_key(self):
        new_private, new_public = keys.generate()
        secret = identity.seed_bootstrap(self.store, "bigpickle")
        old_fingerprint = keys.fingerprint(self.public)
        new_fingerprint = identity.enrol_rekey(
            self.store, "bigpickle", keys.export_public_b64(new_public), secret
        )
        self.assertNotEqual(old_fingerprint, new_fingerprint)

        # old key no longer authenticates
        chal = identity.challenge(self.store, "bigpickle")
        with self.assertRaises(errors.BadSignature):
            identity.verify(self.store, chal["nonce"], self.sign_nonce(chal["nonce"]))

        # new key does
        chal = identity.challenge(self.store, "bigpickle")
        signed = base64.b64encode(
            new_private.sign(chal["nonce"].encode())
        ).decode()
        self.assertEqual(
            identity.verify(self.store, chal["nonce"], signed)["kind"], "agent"
        )


class TestBootstrapHygiene(unittest.TestCase):
    """R2: only one live bootstrap secret; leftovers unusable after enrol."""

    def test_leftover_seed_fails_after_enrol(self):
        store = Store(":memory:")
        private, public = keys.generate()
        s1 = identity.seed_bootstrap(store, "alice")
        # Direct insert of a second hash simulates a pre-hygiene double-seed
        # (put_bootstrap now clears siblings; plant one after s1 is recorded).
        import hashlib
        s2 = "leftover-seed-that-must-die"
        store.conn.execute(
            "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
            (hashlib.sha256(s2.encode()).hexdigest(), "alice", 1),
        )
        store.conn.commit()
        identity.enrol(store, "alice", keys.export_public_b64(public), s1)
        with self.assertRaises(errors.EnrolmentFailed):
            identity.enrol_rekey(
                store, "alice", keys.export_public_b64(public), s2
            )

    def test_put_bootstrap_keeps_only_one_live_secret(self):
        store = Store(":memory:")
        s1 = identity.seed_bootstrap(store, "bob")
        s2 = identity.seed_bootstrap(store, "bob")
        private, public = keys.generate()
        with self.assertRaises(errors.EnrolmentFailed):
            identity.enrol(store, "bob", keys.export_public_b64(public), s1)
        # Only the latest seed works.
        identity.enrol(store, "bob", keys.export_public_b64(public), s2)

    def test_leftover_seed_fails_after_rekey(self):
        store = Store(":memory:")
        helpers.enrol(store, "carol")
        import hashlib
        leftover = "another-unused-seed"
        new_pub = keys.generate()[1]
        secret = identity.seed_bootstrap(store, "carol")
        # Plant a sibling after the live seed (simulates a concurrent leak).
        store.conn.execute(
            "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
            (hashlib.sha256(leftover.encode()).hexdigest(), "carol", 2),
        )
        store.conn.commit()
        identity.enrol_rekey(
            store, "carol", keys.export_public_b64(new_pub), secret
        )
        newer_pub = keys.generate()[1]
        with self.assertRaises(errors.EnrolmentFailed):
            identity.enrol_rekey(
                store, "carol", keys.export_public_b64(newer_pub), leftover
            )

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