Merge remote-tracking branch 'origin/main' into feat/upgrade-safety

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.<user> 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,

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
+}

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

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