diff --git a/README.md b/README.md index db74a86..a35b8c1 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,8 @@ Config must be a signed wrapper. Unsigned files are rejected. Admin changes append prev + new + unified diff to a `kind=admin-history` JSONL chain. `crypto.mode`: `npe` | `lab-xor` | `plain-lab` (see signed payload). -Live ns1 lab uses `lab-xor` until `PFC_REQUIRE_NPE=1` is explicitly cut over. +Live ns1 lab uses `lab-xor`. The real NPE CLI is `npe send|keygen|id` +(not `npe seal`). Probe `GET /v1/npe`. Do not set `PFC_REQUIRE_NPE=1` on +pfc-py-admin until every bus client uses `npe send`. Not a HIPAA/SOC 2/ISO certificate. diff --git a/python/secure_messaging/npe_adapter.py b/python/secure_messaging/npe_adapter.py index ce400b5..0a39b42 100644 --- a/python/secure_messaging/npe_adapter.py +++ b/python/secure_messaging/npe_adapter.py @@ -1,15 +1,21 @@ """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). +The live binary is ``npe ``. +There is **no** ``npe seal --to`` verb. Production HPKE is ``npe send`` +with a sender ``.seed`` and recipient ``.npeid``. + +``crypto.mode=npe`` must not fall back to lab-xor. +``PFC_REQUIRE_NPE=1`` on pfc-py-admin is a bus-wide fail-close; do not +set it on ns1 until every NATS client uses ``npe send``. """ from __future__ import annotations # annotations as strings import json # encode sidecar request +import os # NPE_BIN / NPE_SENDER_SEED import shutil # look up npe on PATH import subprocess # run sidecar +from pathlib import Path # /opt/pfc/bin/npe from typing import Any, Dict, Mapping # types @@ -18,23 +24,54 @@ class NpeRequired(RuntimeError): 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 + """Return path to npe or raise. Checks NPE_BIN, PATH, then /opt/pfc/bin/npe.""" + cands = [os.environ.get("NPE_BIN") or "", shutil.which("npe") or "", "/opt/pfc/bin/npe"] + for path in cands: # first executable wins + if path and Path(path).is_file() and os.access(path, os.X_OK): + return path # found + raise NpeRequired("npe binary not on PATH or /opt/pfc/bin/npe; attach sidecar") # fail closed + + +def probe() -> Dict[str, Any]: + """Does the real CLI exist and speak send/keygen? Never enables fail-close.""" + try: # binary lookup + path = npe_bin() # may raise + except NpeRequired as exc: # missing + return {"ok": False, "usable": False, "error": str(exc)} + proc = subprocess.run([path], capture_output=True, timeout=5, check=False) # usage on stderr + text = (proc.stderr or proc.stdout).decode(errors="replace") # usage line + usable = "send" in text and "keygen" in text # real npe CLI + return { + "ok": usable, + "usable": usable, + "bin": path, + "cli": "npe ", + "usage": text.strip()[:240], + "require_npe_env": os.environ.get("PFC_REQUIRE_NPE", ""), + } def seal_npe(to: str, body: Mapping[str, Any]) -> str: - """Ask sidecar to HPKE-seal body for mailbox ``to``. Returns hex ct.""" + """HPKE-seal via ``npe send``. Returns a handle, never lab-xor. + + Requires ``NPE_SENDER_SEED`` (sender .seed) and ``to`` as a path to a + recipient ``.npeid``. This publishes on NATS; it is not a hex-only + sidecar. Missing keys → NpeRequired (fail closed). + """ + sender = os.environ.get("NPE_SENDER_SEED", "") # host-only seed + if not sender or not Path(sender).is_file(): # no identity + raise NpeRequired("NPE_SENDER_SEED missing; npe send needs sender .seed + recipient .npeid") + to_path = to # dest is .npeid path for the real CLI + if not Path(to_path).is_file(): # mailbox id is not a file + raise NpeRequired("npe send --to expects a .npeid file, not a mailbox string") 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], + proc = subprocess.run( # real CLI + [npe_bin(), "send", "--id", sender, "--to", to_path, "--data", "-"], 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 + raise NpeRequired("npe send failed: " + proc.stderr.decode(errors="replace")[:200]) + return proc.stdout.strip().decode() or "npe-send-ok" # token from sidecar diff --git a/scripts/rotate_config_key.py b/scripts/rotate_config_key.py new file mode 100755 index 0000000..9c3a02e --- /dev/null +++ b/scripts/rotate_config_key.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +"""Rotate lab Ed25519 config key; re-sign payload; append admin-history.""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "python")) + +from cryptography.hazmat.primitives.serialization import load_pem_private_key + +from secure_messaging.admin_history import AdminHistory +from secure_messaging.signed_config import ( + generate_signing_key, + load_signed, + pem_private, + pem_public, + save_signed, + sign, +) + + +def main() -> None: + p = argparse.ArgumentParser() + p.add_argument("--pem", default="/opt/pfc/etc/sm-keys/config.ed25519.pem") + p.add_argument("--signed", default="/opt/pfc/etc/secure-messaging.signed.json") + p.add_argument("--history", default="/opt/pfc/data/admin/admin-history") + p.add_argument("--actor", default="rotate-lab") + args = p.parse_args() + pem = Path(args.pem) + signed_path = Path(args.signed) + old = load_pem_private_key(pem.read_bytes(), password=None) + payload = load_signed(signed_path, old.public_key()) + new = generate_signing_key() + pem.rename(pem.with_suffix(pem.suffix + ".prev")) + pem.write_bytes(pem_private(new)) + pem.chmod(0o600) + pub = pem.with_name("config.ed25519.pub.pem") + pub.write_bytes(pem_public(new.public_key())) + prev_text = signed_path.read_text() + signed = sign(payload, new) + save_signed(signed_path, signed) + row = AdminHistory(Path(args.history)).append_change( + actor=args.actor, prev_text=prev_text, new_text=signed.dumps() + ) + print("rotated key_id", signed.key_id, "history_seq", row.get("seq")) + + +if __name__ == "__main__": + main() diff --git a/tests/test_sm.py b/tests/test_sm.py index 37e9948..edb8a88 100644 --- a/tests/test_sm.py +++ b/tests/test_sm.py @@ -37,8 +37,12 @@ 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")