S25: npe adapter matches real send/keygen CLI; config key rotate script

Probe reports usable without enabling PFC_REQUIRE_NPE. Fail-closed if
sender .seed or recipient .npeid is missing.

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.

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.
+The live binary is ``npe <keygen|id|send|listen|reply|decode|...>``.
-If the ``npe`` binary is missing, raise NpeRequired.
+There is **no** ``npe seal --to`` verb. Production HPKE is ``npe send``
-Live ns1 must not set this until CI review (see UserReview.MD).
+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."""
+    """Return path to npe or raise. Checks NPE_BIN, PATH, then /opt/pfc/bin/npe."""
-    path = shutil.which("npe")  # PATH lookup
+    cands = [os.environ.get("NPE_BIN") or "", shutil.which("npe") or "", "/opt/pfc/bin/npe"]
-    if not path:  # missing
+    for path in cands:  # first executable wins
-        raise NpeRequired("npe binary not on PATH; attach sidecar")  # fail closed
+        if path and Path(path).is_file() and os.access(path, os.X_OK):
-    return path  # found
+            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 <keygen|id|send|listen|reply|decode>",
+        "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
+    proc = subprocess.run(  # real CLI
-        [npe_bin(), "seal", "--to", to],
+        [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])
+        raise NpeRequired("npe send failed: " + proc.stderr.decode(errors="replace")[:200])
-    return proc.stdout.strip().decode()  # hex or token from sidecar
+    return proc.stdout.strip().decode() or "npe-send-ok"  # token from sidecar

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

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