diff --git a/python/secure_messaging/envelope.py b/python/secure_messaging/envelope.py index 9adab43..0666d14 100644 --- a/python/secure_messaging/envelope.py +++ b/python/secure_messaging/envelope.py @@ -75,8 +75,10 @@ def seal( raise ValueError("lab_key required for lab-xor") key = hashlib.sha256(lab_key).digest() # 32-byte key ct = _xor(key, raw).hex() # hex ct - elif mode == "npe": # production — sidecar not in this module - raise NotImplementedError("npe: attach NPE sidecar; do not HPKE here") + elif mode == "npe": # production — sidecar only + from .npe_adapter import seal_npe # fail-closed import + + ct = seal_npe(to, body) # never lab-xor here else: # unknown raise ValueError("unknown crypto.mode " + mode) token = hashlib.sha256(nonce + b"error-token").hexdigest()[:32] # return-path token stub diff --git a/python/secure_messaging/npe_adapter.py b/python/secure_messaging/npe_adapter.py new file mode 100644 index 0000000..ce400b5 --- /dev/null +++ b/python/secure_messaging/npe_adapter.py @@ -0,0 +1,40 @@ +"""NPE sidecar adapter (fail-closed). + +Production ``crypto.mode=npe`` must not fall back to lab-xor. +If the ``npe`` binary is missing, raise NpeRequired. +Live ns1 must not set this until CI review (see UserReview.MD). +""" + +from __future__ import annotations # annotations as strings + +import json # encode sidecar request +import shutil # look up npe on PATH +import subprocess # run sidecar +from typing import Any, Dict, Mapping # types + + +class NpeRequired(RuntimeError): + """Raised when NPE is required but the sidecar is not usable.""" + + +def npe_bin() -> str: + """Return path to npe or raise.""" + path = shutil.which("npe") # PATH lookup + if not path: # missing + raise NpeRequired("npe binary not on PATH; attach sidecar") # fail closed + return path # found + + +def seal_npe(to: str, body: Mapping[str, Any]) -> str: + """Ask sidecar to HPKE-seal body for mailbox ``to``. Returns hex ct.""" + raw = json.dumps(body, sort_keys=True, separators=(",", ":")).encode() # canonical body + proc = subprocess.run( # npe CLI contract: stdin body, arg dest + [npe_bin(), "seal", "--to", to], + input=raw, + capture_output=True, + timeout=15, + check=False, + ) + if proc.returncode != 0: # sidecar failed + raise NpeRequired("npe seal failed: " + proc.stderr.decode(errors="replace")[:200]) + return proc.stdout.strip().decode() # hex or token from sidecar diff --git a/scripts/sign_config.py b/scripts/sign_config.py new file mode 100644 index 0000000..740ff41 --- /dev/null +++ b/scripts/sign_config.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +"""Generate a lab Ed25519 key and a signed secure-messaging.json. + +Usage: + python3 scripts/sign_config.py --out examples/secure-messaging.signed.json +Private key is written to keys/config.ed25519.pem (gitignored). +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "python")) + +from secure_messaging.admin_history import AdminHistory +from secure_messaging.signed_config import generate_signing_key, pem_private, pem_public, save_signed, sign + + +DEFAULT_PAYLOAD = { + "crypto": {"mode": "lab-xor", "system_key_id": "lab-system"}, + "routing": {"mode": "passthrough"}, + "admin": {"history_cube": "admin-history"}, + "logging": {"mode": "summary"}, +} + + +def main() -> None: + p = argparse.ArgumentParser() + p.add_argument("--out", default=str(ROOT / "examples" / "secure-messaging.signed.json")) + p.add_argument("--actor", default="lab-admin") + args = p.parse_args() + keydir = ROOT / "keys" + keydir.mkdir(mode=0o700, exist_ok=True) + priv_path = keydir / "config.ed25519.pem" + if priv_path.exists(): + from cryptography.hazmat.primitives.serialization import load_pem_private_key + + priv = load_pem_private_key(priv_path.read_bytes(), password=None) + else: + priv = generate_signing_key() + priv_path.write_bytes(pem_private(priv)) + priv_path.chmod(0o600) + (keydir / "config.ed25519.pub.pem").write_bytes(pem_public(priv.public_key())) + out = Path(args.out) + prev = out.read_text() if out.exists() else "" + signed = sign(DEFAULT_PAYLOAD, priv) + out.parent.mkdir(parents=True, exist_ok=True) + save_signed(out, signed) + hist = AdminHistory(ROOT / "examples" / "admin-history") + hist.append_change(actor=args.actor, prev_text=prev, new_text=signed.dumps()) + print("wrote", out) + print("history", hist.path) + + +if __name__ == "__main__": + main() diff --git a/tests/test_sm.py b/tests/test_sm.py index bb741ea..b5ff7a3 100644 --- a/tests/test_sm.py +++ b/tests/test_sm.py @@ -33,6 +33,15 @@ class SignedConfigTests(unittest.TestCase): load_signed(p, pub) +class NpeAdapterTests(unittest.TestCase): + def test_npe_mode_fail_closed(self): + from secure_messaging.npe_adapter import NpeRequired + from secure_messaging.envelope import seal + + 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(