sigil/tests/test_hardening.py

"""Hardening tests: the findings from the self-review.

Each test here corresponds to a defect that was found and fixed. If any of
these regress, the corresponding vulnerability is back.
"""

import base64
import os
import tempfile
import threading
import time
import unittest

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

import helpers


def enrolled(store=None, agent="bigpickle"):
    store = store or Store(":memory:")
    private, _ = helpers.enrol(store, agent)
    return store, private, None


sign = helpers.sign


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


class TestSingleUseUnderRace(unittest.TestCase):
    """Finding: check-then-update let one code mint several sessions.

    Measured before the fix: a single-use code succeeded 753 times across 200
    trials. These tests hold the invariant under real concurrency.
    """

    TRIALS = 40
    THREADS = 8

    def _race(self, setup, threads=THREADS):
        """Set up one reusable credential per trial, then race for it.

        setup returns (store, shared, private). `shared` is the single-use
        credential the threads all compete for, created once outside the
        threads, so this genuinely tests the same-secret-twice path.
        """
        for _ in range(self.TRIALS):
            store, shared, private = setup()
            barrier = threading.Barrier(threads)
            outcomes = []

            def go():
                barrier.wait()
                try:
                    work(store, shared, private)
                    outcomes.append(1)
                except Exception:
                    outcomes.append(0)

            pool = [threading.Thread(target=go) for _ in range(threads)]
            for t in pool:
                t.start()
            for t in pool:
                t.join()
            yield sum(outcomes)

    def test_one_code_yields_at_most_one_session(self):
        def setup():
            store, private, _ = enrolled()
            return store, helpers.issue(store, private)["code"], private

        def work(store, code, private):
            viewcodes.redeem_code(store, code)

        for successes in self._race(setup):
            self.assertLessEqual(successes, 1, "one code minted multiple sessions")

    def test_one_bootstrap_secret_enrols_at_most_once(self):
        def setup():
            store, private, _ = enrolled()
            return store, identity.seed_bootstrap(store, "racer"), private

        def work(store, secret, private):
            identity.enrol(
                store, "racer", keys.export_public_b64(keys.generate()[1]), secret
            )

        for successes in self._race(setup, threads=6):
            self.assertLessEqual(successes, 1, "one secret enrolled multiple keys")

    def test_one_nonce_verifies_at_most_once(self):
        def setup():
            store, private, _ = enrolled()
            nonce = identity.challenge(store, "bigpickle")["nonce"]
            return store, nonce, private

        def work(store, nonce, private):
            identity.verify(store, nonce, sign(private, nonce))

        for successes in self._race(setup, threads=6):
            self.assertLessEqual(successes, 1, "one nonce produced multiple sessions")


class TestTokensNotStoredInClear(unittest.TestCase):
    """Finding: session tokens were stored verbatim, so a database leak
    handed an attacker live sessions."""

    def setUp(self):
        self.store, self.private, self.public = enrolled()

    def _agent_token(self):
        chal = identity.challenge(self.store, "bigpickle")
        return identity.verify(self.store, chal["nonce"], sign(self.private, chal["nonce"]))["token"]

    def test_agent_token_appears_nowhere_in_the_database_file(self):
        import sqlite3

        path = tempfile.mktemp(suffix=".db")
        store = Store(path)
        private, public = keys.generate()
        secret = identity.seed_bootstrap(store, "bigpickle")
        identity.enrol(store, "bigpickle", keys.export_public_b64(public), secret)
        chal = identity.challenge(store, "bigpickle")
        token = identity.verify(
            store, chal["nonce"], sign(private, chal["nonce"])
        )["token"]
        store.close()
        with open(path, "rb") as handle:
            blob = handle.read()
        os.unlink(path)
        self.assertNotIn(token.encode(), blob, "session token stored in the clear")

    def test_human_token_appears_nowhere_in_the_database_file(self):
        import sqlite3

        path = tempfile.mktemp(suffix=".db")
        store = Store(path)
        private, public = keys.generate()
        secret = identity.seed_bootstrap(store, "bigpickle")
        identity.enrol(store, "bigpickle", keys.export_public_b64(public), secret)
        code = helpers.issue(store, private)["code"]
        token = viewcodes.redeem_code(store, code)["token"]
        store.close()
        with open(path, "rb") as handle:
            blob = handle.read()
        os.unlink(path)
        self.assertNotIn(token.encode(), blob, "view session token stored in the clear")

    def test_hashed_token_still_resolves(self):
        token = self._agent_token()
        self.assertEqual(
            sessions.resolve(self.store, token)["on_behalf_of"], "bigpickle"
        )


class TestAuditChain(unittest.TestCase):
    """The log is an attestation substrate, so it must be tamper-evident."""

    def test_empty_log_verifies(self):
        self.assertTrue(Store(":memory:").verify_chain()["ok"])

    def test_intact_log_verifies(self):
        store, private, public = enrolled()
        for _ in range(5):
            identity.challenge(store, "bigpickle")
        result = store.verify_chain()
        self.assertTrue(result["ok"])
        self.assertEqual(result["events"], len(store.events(limit=50)))

    def test_edited_detail_is_detected(self):
        store, private, public = enrolled()
        identity.challenge(store, "bigpickle")
        store.conn.execute(
            "UPDATE events SET detail = '{\"tampered\":true}' WHERE id = 1"
        )
        store.conn.commit()
        result = store.verify_chain()
        self.assertFalse(result["ok"])
        self.assertEqual(result["broken_at"], 1)

    def test_deleted_event_is_detected(self):
        store, private, public = enrolled()
        identity.challenge(store, "bigpickle")
        store.conn.execute("DELETE FROM events WHERE id = 1")
        store.conn.commit()
        self.assertFalse(store.verify_chain()["ok"])

    def test_reordered_events_are_detected(self):
        store, private, public = enrolled()
        for _ in range(3):
            identity.challenge(store, "bigpickle")
        store.conn.execute("UPDATE events SET kind = 'tampered' WHERE id = 2")
        store.conn.commit()
        self.assertFalse(store.verify_chain()["ok"])


class TestKeyRotationKillsSessions(unittest.TestCase):
    """Finding: rekey replaced the key but left existing sessions alive, so
    rotating after a compromise bought nothing."""

    def setUp(self):
        self.store, self.private, self.public = enrolled()

    def _token(self):
        chal = identity.challenge(self.store, "bigpickle")
        return identity.verify(
            self.store, chal["nonce"], sign(self.private, chal["nonce"])
        )["token"]

    def test_rekey_revokes_existing_sessions(self):
        token = self._token()
        sessions.resolve(self.store, token)  # alive before
        secret = identity.seed_bootstrap(self.store, "bigpickle")
        identity.enrol_rekey(
            self.store, "bigpickle",
            keys.export_public_b64(keys.generate()[1]), secret,
        )
        with self.assertRaises(errors.SessionExpired):
            sessions.resolve(self.store, token)

    def test_old_key_cannot_authenticate_after_rekey(self):
        secret = identity.seed_bootstrap(self.store, "bigpickle")
        identity.enrol_rekey(
            self.store, "bigpickle",
            keys.export_public_b64(keys.generate()[1]), secret,
        )
        chal = identity.challenge(self.store, "bigpickle")
        with self.assertRaises(errors.BadSignature):
            identity.verify(self.store, chal["nonce"], sign(self.private, chal["nonce"]))

    def test_rotation_is_recorded(self):
        secret = identity.seed_bootstrap(self.store, "bigpickle")
        identity.enrol_rekey(
            self.store, "bigpickle",
            keys.export_public_b64(keys.generate()[1]), secret,
        )
        kinds = [e["kind"] for e in self.store.events(limit=50)]
        self.assertIn("identity.rekeyed", kinds)


class TestRateLimits(unittest.TestCase):
    """Finding: challenge() was unmetered, an open signing oracle."""

    def setUp(self):
        self.store, self.private, self.public = enrolled()

    def test_challenge_is_rate_limited(self):
        raised = False
        for _ in range(200):
            try:
                identity.challenge(self.store, "bigpickle")
            except errors.RateLimited:
                raised = True
                break
        self.assertTrue(raised, "challenge() was never rate limited")

    def test_code_issue_is_rate_limited(self):
        raised = False
        for _ in range(200):
            try:
                helpers.issue(self.store, self.private)
            except errors.RateLimited:
                raised = True
                break
        self.assertTrue(raised, "issue_code() was never rate limited")

    def test_redemption_guessing_is_rate_limited(self):
        refused = 0
        for i in range(500):
            try:
                viewcodes.redeem_code(self.store, f"ZZZZ-{i:04d}")
            except errors.RateLimited:
                break
            except errors.SigilError:
                refused += 1
        else:
            self.fail("code guessing was not rate limited")

    def test_limits_are_per_agent(self):
        for _ in range(40):
            try:
                identity.challenge(self.store, "bigpickle")
            except errors.RateLimited:
                break
        # a different agent is unaffected
        self.store2, _, _ = enrolled(self.store, agent="second")
        identity.challenge(self.store2, "second")


class TestKeyEncryptionAtRest(unittest.TestCase):
    """Finding: keys were written with NoEncryption by default."""

    def test_encrypted_key_needs_passphrase(self):
        private, _ = keys.generate()
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "k.pem")
            keys.save_private(private, path, "a-long-enough-passphrase")
            self.assertTrue(keys.is_encrypted(path))
            with self.assertRaises(ValueError):
                keys.load_private(path)

    def test_wrong_passphrase_refuses(self):
        private, _ = keys.generate()
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "k.pem")
            keys.save_private(private, path, "a-long-enough-passphrase")
            with self.assertRaises(Exception):
                keys.load_private(path, "the-wrong-passphrase")

    def test_correct_passphrase_roundtrips(self):
        private, _ = keys.generate()
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "k.pem")
            keys.save_private(private, path, "a-long-enough-passphrase")
            loaded = keys.load_private(path, "a-long-enough-passphrase")
            self.assertEqual(loaded.sign(b"m"), private.sign(b"m"))

    def test_short_passphrase_refused(self):
        private, _ = keys.generate()
        with tempfile.TemporaryDirectory() as tmp:
            with self.assertRaises(ValueError):
                keys.save_private(private, os.path.join(tmp, "k.pem"), "short")

    def test_plaintext_key_still_loads(self):
        private, _ = keys.generate()
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "k.pem")
            keys.save_private(private, path)
            self.assertFalse(keys.is_encrypted(path))
            keys.load_private(path)


class TestAgentIdValidation(unittest.TestCase):
    """Finding: agent ids were unvalidated, so an id carrying a newline
    could forge audit lines."""

    def test_bad_ids_refused(self):
        store = Store(":memory:")
        for bad in ["", "Big Pickle", "a" * 65, "with\nnewline", "with space",
                    "-leading", "with/slash", "with;drop"]:
            with self.assertRaises(errors.SigilError, msg=f"accepted {bad!r}"):
                identity.seed_bootstrap(store, bad)

    def test_good_ids_accepted(self):
        store = Store(":memory:")
        for good in ["bigpickle", "agent-2", "a.b_c", "x" * 64]:
            private, public = keys.generate()
            secret = identity.seed_bootstrap(store, good)
            identity.enrol(store, good, keys.export_public_b64(public), secret)
            self.assertIn(good, store.agents())

    def test_newline_id_cannot_reach_the_audit_log(self):
        store = Store(":memory:")
        with self.assertRaises(errors.SigilError):
            identity.seed_bootstrap(store, "evil\nfake-event")


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