Harden Sigil: tests/test_hardening.py
This commit is contained in:
parent
47abc7d717
commit
ea827c8e42
362
tests/test_hardening.py
Normal file
362
tests/test_hardening.py
Normal file
@ -0,0 +1,362 @@
|
||||
"""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()
|
||||
Loading…
x
Reference in New Issue
Block a user