From c61212eba77c4359694af33f44b35abec92bf71c Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 09:51:01 +0200 Subject: [PATCH] Make the membership, its subject, stream, bare address and the pin public, so the route proxy leaves the controller's module (issue 1633) --- go/membership/membership.go | 165 ++++++++++++++++++++++++++ go/membership/membership_test.go | 76 ++++++++++++ go/process/conn.go | 10 +- go/process/follow.go | 53 +++++++++ go/process/follow_integration_test.go | 45 +++++++ go/process/pin_test.go | 23 ++++ 6 files changed, 368 insertions(+), 4 deletions(-) create mode 100644 go/membership/membership.go create mode 100644 go/membership/membership_test.go create mode 100644 go/process/follow.go create mode 100644 go/process/follow_integration_test.go create mode 100644 go/process/pin_test.go diff --git a/go/membership/membership.go b/go/membership/membership.go new file mode 100644 index 0000000..c942ccc --- /dev/null +++ b/go/membership/membership.go @@ -0,0 +1,165 @@ +// Package membership is what the mesh issues an assignment to serve and to reach (novox/hq ADR 0160), as a +// process reads it from the bus: the Membership shape, the one subject it is published on, the stream that +// holds it, and the bus address helpers a reader needs. The controller composes it and the route proxy reads +// it; both use this one definition, so there is no second copy to drift (novox/hq issue 1633). +package membership + +import ( + "encoding/json" + "strings" +) + +// AssignmentsStream holds every assignment's membership, the newest per subject. +const AssignmentsStream = "ASSIGNMENTS" + +// Subject is the one address a runtime derives for itself: where its own membership is published, from +// the two names its credential carries. Everything else is in the membership. +func Subject(node, module string) string { + return "mesh.assignment." + node + "." + module +} + +// DirectGetSubject is the subject-addressed direct get of a membership from its stream: the one request an +// account may make of the stream. +func DirectGetSubject(node, module string) string { + return "$JS.API.DIRECT.GET." + AssignmentsStream + "." + Subject(node, module) +} + +// Membership is one assignment's subjects: what this instance of a module on this machine serves, +// and what it may reach. +type Membership struct { + Node string `json:"node"` + Module string `json:"module"` + // Serves is every address a tool of this instance answers on. `{tool}` stands for the tool's + // own name. An address with a queue is shared with the module's other instances. + Serves []Served `json:"serves"` + // Seats is every verb of a seat this instance holds, at the subject the seat's callers use. + Seats []SeatServed `json:"seats,omitempty"` + // Emits is where an event of this module lands; `{event}` stands for the event's name. + Emits string `json:"emits"` + // Reaches is each tool this module may call, `.`, to the subjects that reach it. + Reaches map[string][]string `json:"reaches,omitempty"` + // Tools is where this instance answers what it serves. + Tools string `json:"tools"` + // Receives is what this assignment is given for each requirement it receives, by requirement (novox/hq + // ADR 0167): kept as JSON because the catalogue owns the shape of a contribution. + Receives map[string]json.RawMessage `json:"receives,omitempty"` + // Mesh is every machine's address on the private network (novox/hq ADR 0167). + Mesh []string `json:"mesh,omitempty"` + // State is every bucket this module's code may reach, and whether it may write it (novox/hq ADR 0201). + State []StateIssued `json:"state,omitempty"` + // Logs is every log this module's code may reach (novox/hq ADR 0297). + Logs []LogIssued `json:"logs,omitempty"` + // SeatTraffic is what this module's code may submit, say, hear, take, ask, answer and read on seats + // that name their caller or their kind (novox/hq ADR 0259 §3). + SeatTraffic *SeatTraffic `json:"seat-traffic,omitempty"` + // Streams are this module's own streams, Hears the module streams it hears (novox/hq ADR 0363). + Streams []StreamIssued `json:"streams,omitempty"` + Hears []HeardIssued `json:"hears,omitempty"` + // Waits are the event families a session may wait on through the mesh MCP server (novox/hq ADR 0367 §2). + Waits []WaitIssued `json:"waits,omitempty"` +} + +// Served is one address a tool is answered on. +type Served struct { + Subject string `json:"subject"` + Queue string `json:"queue,omitempty"` +} + +// SeatServed is one verb of a held seat, where its callers ask. +type SeatServed struct { + Seat string `json:"seat"` + Verb string `json:"verb"` + Subject string `json:"subject"` +} + +// StateIssued is one bucket an assignment may reach, by the name its module uses for it (novox/hq ADR 0201). +// Key, when set, is the one key it may reach (novox/hq ADR 0260): this machine's. +type StateIssued struct { + Name string `json:"name"` + Bucket string `json:"bucket"` + Writes bool `json:"writes,omitempty"` + Key string `json:"key,omitempty"` +} + +// LogIssued is one log an assignment may reach, by the name its module uses for it (novox/hq ADR 0297). +type LogIssued struct { + Name string `json:"name"` + Stream string `json:"stream"` + Subject string `json:"subject"` + Writes bool `json:"writes"` +} + +// Worker is one durable consumer a holder pulls a seat's work from. +type Worker struct { + Stream string + Consumer string + Filter string +} + +// SeatTraffic is one module's seat traffic beyond tools. Publish and Subscribe are subject patterns, in the +// server's wildcards. +type SeatTraffic struct { + Publish []string `json:"publish,omitempty"` + Subscribe []string `json:"subscribe,omitempty"` + Answers []string `json:"answers,omitempty"` + Workers []Worker `json:"workers,omitempty"` + Records []string `json:"records,omitempty"` + Kinds []KindHeld `json:"kinds,omitempty"` +} + +// KindHeld is one kind of a kinded bench and who holds it. +type KindHeld struct { + Seat string `json:"seat"` + Kind string `json:"kind"` + Module string `json:"module"` + Node string `json:"node"` + Capabilities []string `json:"capabilities,omitempty"` +} + +// StreamIssued is one of a module's own streams as its membership names it (ADR 0363 §2). +type StreamIssued struct { + Stream string `json:"stream"` + Events []string `json:"events"` + Subject string `json:"subject"` +} + +// HeardIssued is one module stream an assignment hears (ADR 0363 §8). +type HeardIssued struct { + Stream string `json:"stream"` + Consumer string `json:"consumer"` +} + +// WaitIssued is one family as a membership issues it: the name a session asks for, and its subject. +type WaitIssued struct { + Event string `json:"event"` + Subject string `json:"subject"` +} + +// CredentialIn reads the credential out of a bus address, `[scheme://]user:password@host:port`, splitting on +// the last at-sign so a password containing one stays whole, and returns the address without it. +func CredentialIn(address string) (user, password, bare string) { + at := strings.LastIndex(address, "@") + if at < 0 { + return "", "", address + } + scheme := "" + rest := address[:at] + if i := strings.Index(rest, "://"); i >= 0 { + scheme, rest = rest[:i+3], rest[i+3:] + } + user, password, _ = strings.Cut(rest, ":") + return user, password, scheme + address[at+1:] +} + +// BareAddress is a bus address with any credential stripped, for something that only needs to know +// whether a server is answering there. +func BareAddress(address string) string { + _, _, bare := CredentialIn(address) + if bare == "" { + return address + } + if strings.Contains(bare, "://") { + return bare + } + return "nats://" + bare +} diff --git a/go/membership/membership_test.go b/go/membership/membership_test.go new file mode 100644 index 0000000..88b742c --- /dev/null +++ b/go/membership/membership_test.go @@ -0,0 +1,76 @@ +package membership + +import ( + "encoding/json" + "reflect" + "testing" +) + +func TestACredentialIsReadOutOfABusAddress(t *testing.T) { + for _, c := range []struct{ in, user, password, bare string }{ + {"nats://controller:secret@127.0.0.1:4222", "controller", "secret", "nats://127.0.0.1:4222"}, + {"controller:secret@127.0.0.1:4222", "controller", "secret", "127.0.0.1:4222"}, + {"nats://127.0.0.1:4222", "", "", "nats://127.0.0.1:4222"}, + {"127.0.0.1:4222", "", "", "127.0.0.1:4222"}, + // A password containing an at-sign: split on the last one. + {"nats://controller:a@b@127.0.0.1:4222", "controller", "a@b", "nats://127.0.0.1:4222"}, + } { + user, password, bare := CredentialIn(c.in) + if user != c.user || password != c.password || bare != c.bare { + t.Errorf("%q read as %q/%q at %q; wanted %q/%q at %q", c.in, user, password, bare, c.user, c.password, c.bare) + } + } +} + +func TestABareAddressCarriesNoCredentialAndAScheme(t *testing.T) { + for in, want := range map[string]string{ + "nats://controller:secret@10.0.0.1:4222": "nats://10.0.0.1:4222", + "controller:secret@10.0.0.1:4222": "nats://10.0.0.1:4222", + "nats://10.0.0.1:4222": "nats://10.0.0.1:4222", + "10.0.0.1:4222": "nats://10.0.0.1:4222", + } { + if got := BareAddress(in); got != want { + t.Errorf("BareAddress(%q) = %q, want %q", in, got, want) + } + } +} + +func TestAMembershipIsPublishedAndReadAtTheSubjectDerivedFromTheTwoNames(t *testing.T) { + if got := Subject("ace", "route-proxy"); got != "mesh.assignment.ace.route-proxy" { + t.Errorf("subject: %q", got) + } + if got := DirectGetSubject("ace", "route-proxy"); got != "$JS.API.DIRECT.GET.ASSIGNMENTS.mesh.assignment.ace.route-proxy" { + t.Errorf("direct get: %q", got) + } + if AssignmentsStream != "ASSIGNMENTS" { + t.Errorf("stream: %q", AssignmentsStream) + } +} + +// The shape on the wire is the controller's: a membership as issued reads back whole, with the routes kept raw. +func TestAMembershipRoundTripsAsIssued(t *testing.T) { + in := Membership{Node: "ace", Module: "route-proxy", Emits: "e", Tools: "t", + Serves: []Served{{Subject: "s", Queue: "q"}}, + Receives: map[string]json.RawMessage{"route": json.RawMessage(`[{"name":"a"}]`)}, + Mesh: []string{"10.10.0.7"}, + State: []StateIssued{{Name: "n", Bucket: "b", Writes: true, Key: "ace"}}, + SeatTraffic: &SeatTraffic{Publish: []string{"p"}, Kinds: []KindHeld{{Seat: "channel", Kind: "tg"}}}} + raw, err := json.Marshal(in) + if err != nil { + t.Fatal(err) + } + var out Membership + if err := json.Unmarshal(raw, &out); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(in, out) { + t.Fatalf("round trip changed it:\n%+v\n%+v", in, out) + } + var wire map[string]json.RawMessage + _ = json.Unmarshal(raw, &wire) + for _, k := range []string{"receives", "mesh", "seat-traffic"} { + if _, ok := wire[k]; !ok { + t.Errorf("field %q is not on the wire under that name", k) + } + } +} diff --git a/go/process/conn.go b/go/process/conn.go index 50addf5..58efb40 100644 --- a/go/process/conn.go +++ b/go/process/conn.go @@ -40,7 +40,7 @@ var refusedPublish = regexp.MustCompile(`(?i)Publish to "(\S+)"`) // other certificate), authenticated as its user and password, on the inbox prefix _INBOX. and no // other, reconnecting without limit as the tool runner does (novox/hq issue 1192: a process outlives a // bus restart, never giving up the way the client library's own default of 60 attempts would). Ported -// from mesh-controller internal/broker/jetstream.go:106-132 (DialPinned, PinnedToFingerprint) since this +// from mesh-controller internal/broker/jetstream.go (DialPinned, PinnedToFingerprint), which now uses this one since this // package may not import the controller. // // An ErrorHandler among opts is chained behind this package's own, which watches every async error for a @@ -78,7 +78,7 @@ func (c *Conn) options(opts []nats.Option) []nats.Option { nats.MaxReconnects(-1), } if c.account.Fingerprint != "" { - all = append(all, nats.Secure(pinnedToFingerprint(c.account.Fingerprint))) + all = append(all, nats.Secure(PinnedToFingerprint(c.account.Fingerprint))) } all = append(all, opts...) return append(all, nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) { @@ -140,9 +140,11 @@ func (c *Conn) publishRefused(err error) bool { return len(cancels) > 0 } -// pinnedToFingerprint accepts exactly the certificate with this SHA-256 and no other — the leaf alone, its +// PinnedToFingerprint accepts exactly the certificate with this SHA-256 and no other — the leaf alone, its // name and chain deliberately not consulted, as mesh-controller's own pin does (jetstream.go:116-132). -func pinnedToFingerprint(want string) *tls.Config { +// PinnedToFingerprint is public so that a process dialling the bus on its own terms, and the controller, +// share this one pin (novox/hq issue 1633). +func PinnedToFingerprint(want string) *tls.Config { return &tls.Config{ InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter MinVersion: tls.VersionTLS12, diff --git a/go/process/follow.go b/go/process/follow.go new file mode 100644 index 0000000..07678bc --- /dev/null +++ b/go/process/follow.go @@ -0,0 +1,53 @@ +package process + +import ( + "encoding/json" + "fmt" + "time" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/membership" +) + +// FollowMembership reads this account's own membership and follows it (novox/hq ADR 0160): apply is called +// with every membership the mesh issues this account's node and module, the one already issued first and +// each later one as it is published, so a change reaches a running process without a restart. A body that is +// not a membership is handed to onBad (nil to ignore it) and skipped. +// +// The subject is followed before it is read, so an issue landing between the two is applied, not missed. The +// read is the subject-addressed direct get, the one request this account may make of the stream; no +// membership issued yet, or the read failing, is not an error: the process follows the subject and is told +// when one arrives. The returned error is a refusal to follow at all. +func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(error)) error { + if c.account.Node == "" || c.account.Module == "" { + return fmt.Errorf("the account names no node or module, so it has no membership") + } + bad := func(err error) { + if onBad != nil { + onBad(err) + } + } + take := func(body []byte) { + var issued membership.Membership + if err := json.Unmarshal(body, &issued); err != nil { + bad(fmt.Errorf("a membership arrived that is not one: %w", err)) + return + } + apply(issued) + } + subject := membership.Subject(c.account.Node, c.account.Module) + if _, err := c.nc.Subscribe(subject, func(m *nats.Msg) { take(m.Data) }); err != nil { + return fmt.Errorf("cannot follow %s: %w", subject, err) + } + got, err := c.nc.Request(membership.DirectGetSubject(c.account.Node, c.account.Module), nil, 5*time.Second) + switch { + case err != nil: + bad(fmt.Errorf("cannot read the membership issued on %s yet (%v); following it", subject, err)) + case got.Header.Get("Status") != "" || len(got.Data) == 0: + // None issued yet: the follow above delivers it. + default: + take(got.Data) + } + return nil +} diff --git a/go/process/follow_integration_test.go b/go/process/follow_integration_test.go new file mode 100644 index 0000000..cdb420b --- /dev/null +++ b/go/process/follow_integration_test.go @@ -0,0 +1,45 @@ +//go:build integration + +package process + +import ( + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/membership" +) + +// A process connected pinned to a test bus hears the membership published for its node and module, and +// ignores what is not one (novox/hq issue 1633, AC2). +func TestFollowMembershipHearsWhatIsIssuedToTheAccount(t *testing.T) { + url, fingerprint := selfSignedServer(t) + account := mailSenderAccount(t, url, fingerprint) + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + got := make(chan membership.Membership, 1) + bad := make(chan error, 4) + if err := conn.FollowMembership(func(m membership.Membership) { got <- m }, func(e error) { bad <- e }); err != nil { + t.Fatal(err) + } + if err := conn.nc.Publish(membership.Subject("novox", "mesh-mail"), []byte("not json")); err != nil { + t.Fatal(err) + } + if err := conn.nc.Publish(membership.Subject("novox", "mesh-mail"), []byte(`{"node":"novox","module":"mesh-mail","mesh":["10.0.0.1"]}`)); err != nil { + t.Fatal(err) + } + select { + case m := <-got: + if m.Module != "mesh-mail" || len(m.Mesh) != 1 { + t.Fatalf("membership: %+v", m) + } + case <-time.After(5 * time.Second): + t.Fatal("no membership heard") + } + if len(bad) == 0 { + t.Error("the body that is not a membership was not reported") + } +} diff --git a/go/process/pin_test.go b/go/process/pin_test.go new file mode 100644 index 0000000..f07b2b4 --- /dev/null +++ b/go/process/pin_test.go @@ -0,0 +1,23 @@ +package process + +import ( + "crypto/sha256" + "encoding/hex" + "testing" +) + +// The pin accepts exactly the leaf certificate with the named SHA-256, and no other (novox/hq issue 1633). +func TestThePinAcceptsOnlyTheNamedCertificate(t *testing.T) { + leaf := []byte("a certificate") + sum := sha256.Sum256(leaf) + cfg := PinnedToFingerprint("sha256:" + hex.EncodeToString(sum[:])) + if err := cfg.VerifyPeerCertificate([][]byte{leaf}, nil); err != nil { + t.Errorf("the pinned certificate was refused: %v", err) + } + if err := cfg.VerifyPeerCertificate([][]byte{[]byte("another")}, nil); err == nil { + t.Error("another certificate was accepted") + } + if err := cfg.VerifyPeerCertificate(nil, nil); err == nil { + t.Error("no certificate was accepted") + } +}