Give mesh-sdk a process's own pinned bus connection, reading its delivered account (issue 1171)

go/internal/testbus/testbus_integration_test.go

@@ -11,7 +11,7 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 )
 
-// 1175/AC3: a JetStream bus of this test alone answers a publish and a key-value create.
+// 1175/AC3, 1171/AC1: a JetStream bus of this test alone answers a publish and a key-value create.
 func TestBusAnswersPublishAndKeyValueCreate(t *testing.T) {
 	conn, err := nats.Connect(URL(t))
 	if err != nil {
@@ -52,7 +52,7 @@ func TestBusAnswersPublishAndKeyValueCreate(t *testing.T) {
 	}
 }
 
-// 1175/AC3: the bus is gone when the test that started it ends.
+// 1175/AC3, 1171/AC1: the bus is gone when the test that started it ends.
 func TestBusGoneWhenItsTestEnds(t *testing.T) {
 	var url string
 	t.Run("one", func(t *testing.T) {

go/live_before_serve_test.go

@@ -0,0 +1,36 @@
+package stdio
+
+import (
+	"os"
+	"os/exec"
+	"testing"
+)
+
+// 1171/AC4: before Serve, a bundle's bus functions still refuse exactly as on main (stdio.go:391-398,
+// unchanged for bundles). Run in a fresh process: once any other test in this package has called Serve,
+// current stays set for the rest of the test binary, so "before Serve" can only be checked honestly in a
+// process of its own (the self-exec pattern Go's own testing package uses for os.Exit).
+func TestBusFunctionsRefuseBeforeServeInAFreshProcess(t *testing.T) {
+	const marker = "MESH_SDK_TEST_FRESH_PROCESS"
+	if os.Getenv(marker) == "1" {
+		if err := Emit("x", map[string]any{}); err == nil ||
+			err.Error() != "emit before Serve: a bundle emits through the runtime that launched it" {
+			t.Fatalf("Emit before Serve: %v", err)
+		}
+		if _, err := Ask("m.t", nil); err == nil ||
+			err.Error() != "before Serve: a bundle reaches the bus through the runtime that launched it" {
+			t.Fatalf("Ask before Serve: %v", err)
+		}
+		if err := Subscribe("x", func(Envelope) error { return nil }); err == nil ||
+			err.Error() != "before Serve: a bundle reaches the bus through the runtime that launched it" {
+			t.Fatalf("Subscribe before Serve: %v", err)
+		}
+		return
+	}
+	cmd := exec.Command(os.Args[0], "-test.run=^TestBusFunctionsRefuseBeforeServeInAFreshProcess$", "-test.v")
+	cmd.Env = append(os.Environ(), marker+"=1")
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("a fresh process did not refuse every bus function before Serve: %v\n%s", err, out)
+	}
+}

go/process/account.go

@@ -0,0 +1,54 @@
+// Package process is a module's own process's connection to the bus (novox/hq ADR 0436 clause 4): it reads
+// the account the controller delivers, connects as it with its pinned fingerprint and its inbox prefix, and
+// offers emitting an event, the module's states with compare-and-set, consuming a work queue, reading the
+// router's records of the module's own asks, and calling a tool or verb — each only as the account's grants
+// allow. It offers no Serve and no way to serve a tool: a process never answers a call. A sibling of stdio,
+// which it imports only for stdio's pure types (ErrChanged, CodeChanged, StateEntry, EmitOptions); it never
+// touches stdio's current, channel or live(), which stay the bundle's own way to the bus.
+package process
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+)
+
+// Account is a process's own bus account, exactly as the controller delivers it as the module's own secret
+// broker-<process id> (mesh-controller cmd/mesh-controller/modules.go issueProcess, heldProcessAccount, and
+// issueWith for a module's own broker account, which also carries claims). A process account carries no
+// claims at all; this type reads either file without error.
+type Account struct {
+	URL         string `json:"url"`
+	Fingerprint string `json:"fingerprint,omitempty"`
+	Node        string `json:"node"`
+	Module      string `json:"module"`
+	User        string `json:"user"`
+	Password    string `json:"password"`
+	// Claims rides on a module's own broker account only; a process holds no seat, so its own account
+	// carries none (novox/hq ADR 0436 §2, §3). Kept raw: this package never reads it.
+	Claims json.RawMessage `json:"claims,omitempty"`
+}
+
+// InboxPrefix is this account's own inbox, and nothing wider: `_INBOX.<user>`.
+func (a Account) InboxPrefix() string { return "_INBOX." + a.User }
+
+// ReadAccount reads and validates the account file the controller delivers: it refuses a file missing url,
+// user or password, naming the field — never the password, which does not appear in an error.
+func ReadAccount(path string) (Account, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return Account{}, fmt.Errorf("reading the process account at %s: %w", path, err)
+	}
+	var a Account
+	if err := json.Unmarshal(data, &a); err != nil {
+		return Account{}, fmt.Errorf("the process account at %s is not JSON: %w", path, err)
+	}
+	for _, f := range []struct{ name, value string }{
+		{"url", a.URL}, {"user", a.User}, {"password", a.Password},
+	} {
+		if f.value == "" {
+			return Account{}, fmt.Errorf("the process account at %s has no %s", path, f.name)
+		}
+	}
+	return a, nil
+}

go/process/account_test.go

@@ -0,0 +1,84 @@
+package process
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// 1171/AC2: given an account file broker-<process id> in the controller's form {url, fingerprint, node,
+// module, user, password, claims}, when the process package reads it, then it yields those fields and
+// tolerates claims.
+func TestReadAccountYieldsEveryFieldAndToleratesClaims(t *testing.T) {
+	path := writeAccount(t, map[string]any{
+		"url": "nats://novox:4222", "fingerprint": "sha256:ab", "node": "novox", "module": "mesh-mail",
+		"user": "novox.mesh-mail.send", "password": "secret-pw",
+		"claims": []map[string]any{{"seat": "operator-channel", "verbs": []string{"tell"}}},
+	})
+	a, err := ReadAccount(path)
+	if err != nil {
+		t.Fatalf("a well-formed account file was refused: %v", err)
+	}
+	if a.URL != "nats://novox:4222" || a.Fingerprint != "sha256:ab" || a.Node != "novox" ||
+		a.Module != "mesh-mail" || a.User != "novox.mesh-mail.send" || a.Password != "secret-pw" {
+		t.Fatalf("the account read back: %+v", a)
+	}
+}
+
+// 1171/AC2: a process account carries no claims at all (heldProcessAccount); reading it is not refused for
+// the field being absent.
+func TestReadAccountToleratesNoClaimsAtAll(t *testing.T) {
+	path := writeAccount(t, map[string]any{
+		"url": "nats://novox:4222", "node": "novox", "module": "mesh-mail",
+		"user": "novox.mesh-mail.send", "password": "secret-pw",
+	})
+	a, err := ReadAccount(path)
+	if err != nil {
+		t.Fatalf("a process account with no claims was refused: %v", err)
+	}
+	if a.User != "novox.mesh-mail.send" {
+		t.Fatalf("the account read back: %+v", a)
+	}
+}
+
+// 1171/AC2: a file missing url, user or password is refused, naming the field; the password never appears
+// in the error even when it is the field present and another is missing.
+func TestReadAccountRefusesAFileMissingAField(t *testing.T) {
+	for _, tc := range []struct {
+		field  string
+		values map[string]any
+	}{
+		{"url", map[string]any{"node": "novox", "module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "secret-pw"}},
+		{"user", map[string]any{"url": "nats://novox:4222", "node": "novox", "module": "mesh-mail", "password": "secret-pw"}},
+		{"password", map[string]any{"url": "nats://novox:4222", "node": "novox", "module": "mesh-mail", "user": "novox.mesh-mail.send"}},
+	} {
+		t.Run(tc.field, func(t *testing.T) {
+			path := writeAccount(t, tc.values)
+			_, err := ReadAccount(path)
+			if err == nil {
+				t.Fatalf("an account file missing %s was accepted", tc.field)
+			}
+			if !strings.Contains(err.Error(), tc.field) {
+				t.Errorf("the error %q does not name the missing field %s", err.Error(), tc.field)
+			}
+			if strings.Contains(err.Error(), "secret-pw") {
+				t.Errorf("the password appears in the error: %q", err.Error())
+			}
+		})
+	}
+}
+
+func writeAccount(t *testing.T, fields map[string]any) string {
+	t.Helper()
+	b, err := json.Marshal(fields)
+	if err != nil {
+		t.Fatal(err)
+	}
+	path := filepath.Join(t.TempDir(), "broker-send")
+	if err := os.WriteFile(path, b, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return path
+}

go/process/conn.go

@@ -0,0 +1,64 @@
+package process
+
+import (
+	"crypto/sha256"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/hex"
+	"errors"
+	"fmt"
+
+	"github.com/nats-io/nats.go"
+)
+
+// Conn is one process's own connection to the bus (novox/hq ADR 0436 clause 4): dialled with the account
+// the controller delivered, pinned to its fingerprint, on its own inbox prefix. It offers no Serve and no
+// way to serve a tool — a process reaches the bus only as itself, and never answers a call.
+type Conn struct {
+	nc      *nats.Conn
+	account Account
+}
+
+// Connect dials the bus with account: pinned to its fingerprint when it carries one (refused with any
+// other certificate), authenticated as its user and password, on the inbox prefix _INBOX.<user> and no
+// other. Ported from mesh-controller internal/broker/jetstream.go:106-132 (DialPinned,
+// PinnedToFingerprint) since this package may not import the controller.
+func Connect(account Account, opts ...nats.Option) (*Conn, error) {
+	all := []nats.Option{
+		nats.CustomInboxPrefix(account.InboxPrefix()),
+		nats.UserInfo(account.User, account.Password),
+	}
+	if account.Fingerprint != "" {
+		all = append(all, nats.Secure(pinnedToFingerprint(account.Fingerprint)))
+	}
+	all = append(all, opts...)
+	nc, err := nats.Connect(account.URL, all...)
+	if err != nil {
+		return nil, fmt.Errorf("connecting the process account %s: %w", account.User, err)
+	}
+	return &Conn{nc: nc, account: account}, nil
+}
+
+// Close ends the connection.
+func (c *Conn) Close() { c.nc.Close() }
+
+// 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 {
+	return &tls.Config{
+		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
+		MinVersion:         tls.VersionTLS12,
+		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
+			if len(rawCerts) == 0 {
+				return errors.New("the bus presented no certificate")
+			}
+			sum := sha256.Sum256(rawCerts[0])
+			got := "sha256:" + hex.EncodeToString(sum[:])
+			if got != want {
+				return fmt.Errorf("the bus presented a certificate this mesh does not know (%s…), expected %s…",
+					got[:23], want[:23])
+			}
+			return nil
+		},
+	}
+}

go/process/conn_integration_test.go

@@ -0,0 +1,94 @@
+//go:build integration
+
+package process
+
+import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/sha256"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/hex"
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+)
+
+// selfSignedServer is a nats-server with JetStream and TLS, self-signed, for this test alone (mirroring
+// go/internal/testbus.Start, with a certificate added since testbus has none): a mesh's bus presents its own
+// certificate in no public trust store, which is exactly what a process pins to (ADR 0436 clause 4).
+func selfSignedServer(t *testing.T) (url, fingerprint string) {
+	t.Helper()
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	tmpl := &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject:      pkix.Name{CommonName: "mesh-sdk-test-bus"},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().Add(time.Hour),
+		KeyUsage:     x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+	}
+	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	sum := sha256.Sum256(der)
+	fingerprint = "sha256:" + hex.EncodeToString(sum[:])
+	cert := tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}
+	opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
+		NoLog: true, NoSigs: true, TLSConfig: &tls.Config{Certificates: []tls.Certificate{cert}},
+		Username: "novox.mesh-mail.send", Password: "the-pw"}
+	s, err := server.NewServer(opts)
+	if err != nil {
+		t.Fatalf("a TLS bus for this test could not be made: %v", err)
+	}
+	go s.Start()
+	if !s.ReadyForConnections(10 * time.Second) {
+		s.Shutdown()
+		t.Fatal("a TLS bus for this test did not come up within 10s")
+	}
+	t.Cleanup(func() {
+		s.Shutdown()
+		s.WaitForShutdown()
+	})
+	return s.ClientURL(), fingerprint
+}
+
+func mailSenderAccount(url, fingerprint string) Account {
+	return Account{URL: url, Fingerprint: fingerprint, Node: "novox", Module: "mesh-mail",
+		User: "novox.mesh-mail.send", Password: "the-pw"}
+}
+
+// 1171/AC3: given a test bus with TLS and the account composed for novox.mesh-mail.send, when the package
+// connects with the file, then it connects with the inbox prefix _INBOX.novox.mesh-mail.send.
+func TestConnectPinsTheFingerprintAndSetsTheInboxPrefix(t *testing.T) {
+	url, fingerprint := selfSignedServer(t)
+	account := mailSenderAccount(url, fingerprint)
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatalf("connecting with the right fingerprint was refused: %v", err)
+	}
+	defer conn.Close()
+	if !conn.nc.IsConnected() {
+		t.Fatal("the connection did not come up")
+	}
+	if got := conn.nc.Opts.InboxPrefix; got != "_INBOX.novox.mesh-mail.send" {
+		t.Errorf("inbox prefix: %q, want %q", got, "_INBOX.novox.mesh-mail.send")
+	}
+}
+
+// 1171/AC3: connecting with a fingerprint that is not the bus's own leaf certificate is refused.
+func TestConnectRefusesTheWrongFingerprint(t *testing.T) {
+	url, _ := selfSignedServer(t)
+	account := mailSenderAccount(url, "sha256:0000000000000000000000000000000000000000000000000000000000000000")
+	if _, err := Connect(account); err == nil {
+		t.Fatal("connecting with the wrong fingerprint was not refused")
+	}
+}