From f4da7ff44634973e1b32db93c694f6ee47c34e41 Mon Sep 17 00:00:00 2001 From: George Lambert Date: Tue, 15 Sep 2026 22:40:50 -0400 Subject: [PATCH] S15: sm-leaf health HTTP, verae.sm.* acks, systemd unit Loopback GET /health, nats-leaf:// rewrite, subject request-reply tests. Do not replace pfc-py-admin. NpeRequired stays on the package surface. --- README.md | 5 ++- deploy/pfc-sm-leaf.service | 14 ++++++ go/cmd/sm-leaf/main.go | 17 +++++++- go/internal/leaf/leaf.go | 66 ++++++++++++++++++++++++++--- go/internal/leaf/leaf_test.go | 62 ++++++++++++++++++++++++++- python/secure_messaging/__init__.py | 2 + tests/test_sm.py | 3 +- 7 files changed, 158 insertions(+), 11 deletions(-) create mode 100644 deploy/pfc-sm-leaf.service diff --git a/README.md b/README.md index 118206b..db74a86 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,8 @@ history, and Network Error Bundles. * Python spec: `python/secure_messaging/` (line comments) * Go leaf: `go/cmd/sm-leaf` (in-process NATS + optional `SM_LEAF_HUB`) +* Loopback health: `SM_HTTP` default `127.0.0.1:18783` (`GET /health`) +* systemd: `deploy/pfc-sm-leaf.service` (does not replace `pfc-py-admin`) * Catalog: https://git.georgelambert.org/marchon/nats-service-endpoints * Hub: https://git.georgelambert.org/marchon/system-git-sync @@ -12,5 +14,6 @@ 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. -Not a HIPAA/SOC 2/ISO certificate. ns1: no deploy until CI review. +Not a HIPAA/SOC 2/ISO certificate. diff --git a/deploy/pfc-sm-leaf.service b/deploy/pfc-sm-leaf.service new file mode 100644 index 0000000..037bdb7 --- /dev/null +++ b/deploy/pfc-sm-leaf.service @@ -0,0 +1,14 @@ +[Unit] +Description=PFC secure-messaging leaf (in-process NATS + hub leaf) +After=network.target + +[Service] +Environment=SM_LEAF_HUB=nats://10.10.10.21:7422 +Environment=SM_HTTP=127.0.0.1:18783 +ExecStart=/opt/pfc/bin/sm-leaf -hub nats://10.10.10.21:7422 -http 127.0.0.1:18783 +Restart=on-failure +User=root +LimitNOFILE=65535 + +[Install] +WantedBy=multi-user.target diff --git a/go/cmd/sm-leaf/main.go b/go/cmd/sm-leaf/main.go index 2ace866..24f9ee4 100644 --- a/go/cmd/sm-leaf/main.go +++ b/go/cmd/sm-leaf/main.go @@ -12,15 +12,28 @@ import ( ) func main() { - hub := flag.String("hub", os.Getenv("SM_LEAF_HUB"), "leaf hub URL, e.g. nats-leaf://10.10.10.21:7422") + hub := flag.String("hub", os.Getenv("SM_LEAF_HUB"), "leaf hub URL, e.g. nats://10.10.10.21:7422") + httpAddr := flag.String("http", getenv("SM_HTTP", "127.0.0.1:18783"), "loopback health HTTP") flag.Parse() n, err := leaf.Start(*hub) if err != nil { log.Fatal(err) } - log.Printf("sm-leaf in-process nats %s hub=%q", n.Addr(), *hub) + log.Printf("sm-leaf in-process nats %s hub=%q http=%s", n.Addr(), *hub, *httpAddr) + go func() { + if err := leaf.ServeHealth(*httpAddr); err != nil { + log.Printf("health: %v", err) + } + }() ch := make(chan os.Signal, 1) signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM) <-ch _ = n.Shutdown() } + +func getenv(k, d string) string { + if v := os.Getenv(k); v != "" { + return v + } + return d +} diff --git a/go/internal/leaf/leaf.go b/go/internal/leaf/leaf.go index d10ca28..f1d150a 100644 --- a/go/internal/leaf/leaf.go +++ b/go/internal/leaf/leaf.go @@ -2,8 +2,12 @@ package leaf import ( + "encoding/json" "errors" + "log" + "net/http" "net/url" + "strings" "time" natsserver "github.com/nats-io/nats-server/v2/server" @@ -16,11 +20,17 @@ type Node struct { nc *nats.Conn } +// ParseHub rewrites nats-leaf:// to nats:// then parses the URL. +func ParseHub(hub string) (*url.URL, error) { + raw := strings.Replace(hub, "nats-leaf://", "nats://", 1) + return url.Parse(raw) +} + // Start binds 127.0.0.1:0 and optionally leaf-connects to hub. func Start(hub string) (*Node, error) { opts := &natsserver.Options{Host: "127.0.0.1", Port: -1} if hub != "" { - u, err := url.Parse(hub) + u, err := ParseHub(hub) if err != nil { return nil, err } @@ -40,17 +50,61 @@ func Start(hub string) (*Node, error) { return nil, err } n := &Node{ns: ns, nc: nc} - if _, err := n.nc.Subscribe("verae.sm.send", func(msg *nats.Msg) { - if msg.Reply != "" { - _ = n.nc.Publish(msg.Reply, []byte(`{"accepted":true}`)) - } - }); err != nil { + if err := n.subscribe(); err != nil { n.Shutdown() return nil, err } return n, nil } +func (n *Node) subscribe() error { + ack := func(msg *nats.Msg, body []byte) { + if msg.Reply != "" { + _ = n.nc.Publish(msg.Reply, body) + } + } + if _, err := n.nc.Subscribe("verae.sm.send", func(msg *nats.Msg) { + // Passthrough: do not log ciphertext. + ack(msg, []byte(`{"accepted":true}`)) + }); err != nil { + return err + } + if _, err := n.nc.Subscribe("verae.sm.dead", func(msg *nats.Msg) { + ack(msg, []byte(`{"queued":true}`)) + }); err != nil { + return err + } + if _, err := n.nc.Subscribe("verae.sm.error", func(msg *nats.Msg) { + ack(msg, []byte(`{"emitted":true}`)) + }); err != nil { + return err + } + if _, err := n.nc.Subscribe("verae.sm.log.summary", func(msg *nats.Msg) { + var hdr map[string]any + _ = json.Unmarshal(msg.Data, &hdr) + log.Printf("sm-summary code=%v lookup=%v class=%v", hdr["error_code"], hdr["lookup_id"], hdr["dest_class"]) + ack(msg, []byte(`{"logged":true}`)) + }); err != nil { + return err + } + return nil +} + +// HealthHandler is the loopback JSON health mux (does not expose NATS). +func HealthHandler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"ok":true,"service":"sm-leaf"}`)) + }) + return mux +} + +// ServeHealth binds a loopback health listener (does not expose NATS). +func ServeHealth(addr string) error { + return http.ListenAndServe(addr, HealthHandler()) +} + // Addr is the in-process client URL. func (n *Node) Addr() string { return n.ns.ClientURL() } diff --git a/go/internal/leaf/leaf_test.go b/go/internal/leaf/leaf_test.go index e678eb6..f72be65 100644 --- a/go/internal/leaf/leaf_test.go +++ b/go/internal/leaf/leaf_test.go @@ -1,6 +1,12 @@ package leaf -import "testing" +import ( + "encoding/json" + "io" + "net/http/httptest" + "testing" + "time" +) func TestStartInProcess(t *testing.T) { n, err := Start("") @@ -12,3 +18,57 @@ func TestStartInProcess(t *testing.T) { t.Fatal("empty addr") } } + +func TestParseHubLeafScheme(t *testing.T) { + u, err := ParseHub("nats-leaf://10.10.10.21:7422") + if err != nil { + t.Fatal(err) + } + if u.Scheme != "nats" { + t.Fatalf("scheme %s", u.Scheme) + } + if u.Host != "10.10.10.21:7422" { + t.Fatalf("host %s", u.Host) + } +} + +func TestHealthHandler(t *testing.T) { + srv := httptest.NewServer(HealthHandler()) + defer srv.Close() + resp, err := srv.Client().Get(srv.URL + "/health") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + var m map[string]any + if err := json.Unmarshal(body, &m); err != nil { + t.Fatal(err) + } + if m["ok"] != true || m["service"] != "sm-leaf" { + t.Fatalf("health %s", body) + } +} + +func TestSMSubjectsAck(t *testing.T) { + n, err := Start("") + if err != nil { + t.Fatal(err) + } + defer n.Shutdown() + cases := map[string]string{ + "verae.sm.send": `{"accepted":true}`, + "verae.sm.dead": `{"queued":true}`, + "verae.sm.error": `{"emitted":true}`, + "verae.sm.log.summary": `{"logged":true}`, + } + for subj, want := range cases { + msg, err := n.nc.Request(subj, []byte(`{"lookup_id":"t"}`), time.Second) + if err != nil { + t.Fatalf("%s: %v", subj, err) + } + if string(msg.Data) != want { + t.Fatalf("%s got %s want %s", subj, msg.Data, want) + } + } +} diff --git a/python/secure_messaging/__init__.py b/python/secure_messaging/__init__.py index 7506e33..7b8a2a6 100644 --- a/python/secure_messaging/__init__.py +++ b/python/secure_messaging/__init__.py @@ -3,6 +3,7 @@ from .signed_config import SignedConfig, UnsignedConfig # signed wrapper load/s from .envelope import PassthroughEnvelope # dest-in-clear, body ciphertext from .error_bundle import NetworkErrorBundle # system-key + sender-only from .admin_history import AdminHistory # DataCube-shaped append-only config log +from .npe_adapter import NpeRequired # fail-closed NPE __all__ = [ "SignedConfig", @@ -10,4 +11,5 @@ __all__ = [ "PassthroughEnvelope", "NetworkErrorBundle", "AdminHistory", + "NpeRequired", ] diff --git a/tests/test_sm.py b/tests/test_sm.py index b5ff7a3..fbc885e 100644 --- a/tests/test_sm.py +++ b/tests/test_sm.py @@ -35,9 +35,10 @@ class SignedConfigTests(unittest.TestCase): class NpeAdapterTests(unittest.TestCase): def test_npe_mode_fail_closed(self): - from secure_messaging.npe_adapter import NpeRequired + from secure_messaging import NpeRequired from secure_messaging.envelope import seal + self.assertTrue(hasattr(__import__("secure_messaging"), "NpeRequired")) with self.assertRaises(NpeRequired): seal(to="npe.inbox.x", sender="alice", body={"a": 1}, mode="npe")