"""Signed config, passthrough, error bundle, admin history.""" from __future__ import annotations import json import tempfile import unittest from pathlib import Path from secure_messaging.admin_history import AdminHistory from secure_messaging.envelope import open_lab, seal from secure_messaging.error_bundle import build_bundle, log_summary, open_sender, open_system from secure_messaging.signed_config import UnsignedConfig, generate_signing_key, load_signed, save_signed, sign, verify class SignedConfigTests(unittest.TestCase): def test_roundtrip_and_tamper(self): priv = generate_signing_key() pub = priv.public_key() payload = {"crypto": {"mode": "lab-xor"}, "routing": {"mode": "passthrough"}} signed = sign(payload, priv) self.assertEqual(verify(signed.wrapper(), pub)["crypto"]["mode"], "lab-xor") bad = signed.wrapper() bad["payload"]["crypto"]["mode"] = "plain-lab" with self.assertRaises(UnsignedConfig): verify(bad, pub) with tempfile.TemporaryDirectory() as td: p = Path(td) / "c.json" save_signed(p, signed) self.assertEqual(load_signed(p, pub)["routing"]["mode"], "passthrough") p.write_text(json.dumps({"payload": payload})) with self.assertRaises(UnsignedConfig): load_signed(p, pub) class NpeAdapterTests(unittest.TestCase): def test_npe_mode_fail_closed(self): from secure_messaging import NpeRequired from secure_messaging.envelope import seal from secure_messaging.npe_adapter import probe self.assertTrue(hasattr(__import__("secure_messaging"), "NpeRequired")) st = probe() self.assertIn("usable", st) # mode=npe never falls back to lab-xor (missing seed or missing binary). with self.assertRaises(NpeRequired): seal(to="npe.inbox.x", sender="alice", body={"a": 1}, mode="npe") class EnvelopeTests(unittest.TestCase): def test_lab_xor_roundtrip_lookup(self): env = seal( to="npe.inbox.abc", sender="alice", body={"note": "hello"}, mode="lab-xor", lab_key=b"lab", ) self.assertEqual(env.to, "npe.inbox.abc") self.assertNotIn("alice", env.from_lookup_id) self.assertEqual(open_lab(env, b"lab")["note"], "hello") class ErrorBundleTests(unittest.TestCase): def test_sender_and_system_separate(self): b = build_bundle( lookup_id="lid1", error_code="SM-DEAD-1", dest_class="mailbox", sender_pub=b"sender-pub", system_pub=b"system-pub", detail="undeliverable", ) s = open_sender(b, b"sender-pub") sysb = open_system(b, b"system-pub") self.assertEqual(s["detail"], "undeliverable") self.assertNotIn("detail", sysb) self.assertTrue(sysb["respond"]) self.assertEqual(log_summary(b)["error_code"], "SM-DEAD-1") with self.assertRaises(Exception): open_sender(b, b"wrong") class HistoryTests(unittest.TestCase): def test_prev_new_diff(self): with tempfile.TemporaryDirectory() as td: h = AdminHistory(Path(td) / "admin-history") row = h.append_change(actor="alice", prev_text="a=1\n", new_text="a=2\n") self.assertEqual(row["payload"]["type"], "admin-config") self.assertIn("-a=1", row["payload"]["diff"]) self.assertIn("+a=2", row["payload"]["diff"]) with self.assertRaises(ValueError): h.append_change(actor="", prev_text="", new_text="x") class HpkeDirTests(unittest.TestCase): def test_roundtrip_and_router_cannot_see_body(self): from secure_messaging.content import routing_view, unwrap_content, wrap_content from secure_messaging.pubkey_dir import PubKeyDir with tempfile.TemporaryDirectory() as td: d = PubKeyDir(Path(td)) d.create_endpoint("alice") d.create_endpoint("pfc-repl") avail = d.available() self.assertIn("pfc-repl", avail["handles"]) self.assertTrue(all("sk" not in json.dumps(k) for k in avail["keys"])) env = wrap_content( to_handle="pfc-repl", sender="alice", body={"bytes_hex": "ab", "note": "secret-phi"}, directory=d, ) self.assertEqual(env["alg"], "npe") rv = routing_view(env) self.assertNotIn("ct", rv) blob = json.dumps(env) self.assertNotIn("secret-phi", blob) self.assertNotIn("bytes_hex", routing_view(env)) pt = unwrap_content(env, handle="pfc-repl", directory=d) self.assertEqual(pt["note"], "secret-phi") with self.assertRaises(Exception): unwrap_content(env, handle="alice", directory=d) class RouterTests(unittest.TestCase): def test_good_passthrough(self): from secure_messaging.router import handle_send env = { "to": "npe.inbox.abc", "to_handle": "pfc-repl", "alg": "npe", "ct": "abcd", "enc": "00", "from_lookup_id": "lid-npe", } r = handle_send(json.dumps(env).encode()) self.assertTrue(r.accepted) self.assertEqual(r.ack()["lookup_id"], "lid-npe") self.assertNotIn("ct", r.ack()) self.assertEqual(r.events, []) def test_xor_rejected_for_content(self): from secure_messaging.router import handle_send r = handle_send(b'{"to":"npe.inbox.x","alg":"lab-xor","ct":"ab"}') self.assertEqual(r.error_code, "SM-BAD-ALG") def test_missing_to_failure_path(self): from secure_messaging.router import handle_send from secure_messaging.error_bundle import open_sender, open_system r = handle_send(b'{"alg":"npe","ct":"ab","from_lookup_id":"lid-x"}') self.assertFalse(r.accepted) self.assertEqual(r.error_code, "SM-MISSING-TO") subjects = [e["subject"] for e in r.events] self.assertEqual(subjects, ["verae.sm.log.summary", "verae.sm.error", "verae.sm.dead"]) summary = r.events[0]["body"] self.assertNotIn("ct", summary) self.assertNotIn("detail", summary) s = open_sender(r.bundle, b"sender-pub") sysb = open_system(r.bundle, b"system-pub") self.assertEqual(s["error_code"], "SM-MISSING-TO") self.assertNotIn("note", sysb) def test_empty_ct_and_plaintext_body(self): from secure_messaging.router import handle_send r = handle_send(b'{"to":"npe.inbox.x","alg":"npe","ct":""}') self.assertEqual(r.error_code, "SM-EMPTY-CT") r2 = handle_send(b'{"to":"npe.inbox.x","alg":"npe","ct":"ab","body":{"secret":1}}') self.assertEqual(r2.error_code, "SM-PLAINTEXT-BODY") if __name__ == "__main__": unittest.main()