S14: NPE fail-closed adapter; npe mode never falls back to lab-xor

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
+    elif mode == "npe":  # production — sidecar only
-        raise NotImplementedError("npe: attach NPE sidecar; do not HPKE here")
+        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

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

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()

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(