177 lines
6.9 KiB
Python
177 lines
6.9 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
|
|
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()
|