George Lambert
43cbf51d3e
Python router + Go sm-leaf reject missing to, empty ct, plaintext body. Summary logs never include ciphertext. MODULE.md.
131 lines
5.2 KiB
Python
131 lines
5.2 KiB
Python
"""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
|
|
|
|
self.assertTrue(hasattr(__import__("secure_messaging"), "NpeRequired"))
|
|
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 RouterTests(unittest.TestCase):
|
|
def test_good_passthrough(self):
|
|
from secure_messaging.envelope import seal
|
|
from secure_messaging.router import handle_send
|
|
|
|
env = seal(to="npe.inbox.abc", sender="alice", body={"note": "hello"}, mode="lab-xor", lab_key=b"lab")
|
|
r = handle_send(json.dumps(env.wire()).encode())
|
|
self.assertTrue(r.accepted)
|
|
self.assertEqual(r.ack()["lookup_id"], env.from_lookup_id)
|
|
self.assertNotIn("ct", r.ack())
|
|
self.assertEqual(r.events, [])
|
|
|
|
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":"lab-xor","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":"lab-xor","ct":""}')
|
|
self.assertEqual(r.error_code, "SM-EMPTY-CT")
|
|
r2 = handle_send(b'{"to":"npe.inbox.x","alg":"lab-xor","ct":"ab","body":{"secret":1}}')
|
|
self.assertEqual(r2.error_code, "SM-PLAINTEXT-BODY")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|