mesh-sdk/go/process/conn.go

package process

import (
	"context"
	"crypto/sha256"
	"crypto/tls"
	"crypto/x509"
	"encoding/hex"
	"errors"
	"fmt"
	"regexp"
	"sync"

	"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

	mu      sync.Mutex
	next    uint64
	pending map[string]map[uint64]context.CancelCauseFunc

	// What Keeps announced (keeps.go): the states' names (nil before Keeps), the announcement, and whether a
	// reconnect announces it again.
	kept        map[string]bool
	announced   []byte
	reannounces bool
}

// errPermissionRefused marks a context cancelled because the bus refused this account's publish of the
// subject a pending call was waiting an answer on: a denied publish is refused only as an async protocol
// error with no reply (mesh-tools node-tools/internal/bus/caller.go publishRefused), so without this a
// call would wait out its whole timeout instead of failing at once.
var errPermissionRefused = errors.New("the bus refused this account's publish")

// refusedPublish reads the subject out of the bus's "-ERR 'Permissions Violation for Publish to
// "<subject>"'" (mesh-tools node-tools/internal/bus/caller.go's own regexp).
var refusedPublish = regexp.MustCompile(`(?i)Publish to "(\S+)"`)

// 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, 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 (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
// permission violation refusing a publish this connection is waiting an answer to, and cancels that wait
// at once.
func Connect(account Account, opts ...nats.Option) (*Conn, error) {
	c := &Conn{account: account, pending: map[string]map[uint64]context.CancelCauseFunc{}}
	nc, err := nats.Connect(account.URL, c.options(opts)...)
	if err != nil {
		return nil, fmt.Errorf("connecting the process account %s: %w", account.User, err)
	}
	c.nc = nc
	return c, nil
}

// options is the full option list Connect dials with, apart from Connect so a unit test can check what
// it carries without a live bus: this package's own (the inbox prefix, the credentials, the pin,
// reconnecting without limit, and an ErrorHandler chaining any the caller passed), then opts, then this
// package's own ErrorHandler again last, so it is the one AsyncErrorCB nats.Connect keeps however opts
// ordered theirs.
func (c *Conn) options(opts []nats.Option) []nats.Option {
	// opts are functions over a *nats.Options; applied to a scratch value, this reads off a caller's own
	// ErrorHandler (if any) to chain, without needing nats.Connect's own URL/Servers parsing here.
	var scratch nats.Options
	for _, opt := range opts {
		if opt != nil {
			_ = opt(&scratch)
		}
	}
	chained := scratch.AsyncErrorCB

	all := []nats.Option{
		nats.CustomInboxPrefix(c.account.InboxPrefix()),
		nats.UserInfo(c.account.User, c.account.Password),
		nats.MaxReconnects(-1),
	}
	if 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) {
		c.publishRefused(err)
		if chained != nil {
			chained(nc, sub, err)
		}
	}))
}

// Close ends the connection.
func (c *Conn) Close() { c.nc.Close() }

// withRefusal derives a context from ctx that is also cancelled, with errPermissionRefused as its cause,
// the moment the bus refuses this account's publish of subject — before ctx's own deadline, which a
// denied publish would otherwise run out, refused only as an async error with no reply. Every call that
// requests on the bus registers the subject it is about to publish to here first.
func (c *Conn) withRefusal(ctx context.Context, subject string) (context.Context, func()) {
	derived, cancel := context.WithCancelCause(ctx)
	c.mu.Lock()
	c.next++
	id := c.next
	if c.pending[subject] == nil {
		c.pending[subject] = map[uint64]context.CancelCauseFunc{}
	}
	c.pending[subject][id] = cancel
	c.mu.Unlock()
	return derived, func() {
		c.mu.Lock()
		if list := c.pending[subject]; list != nil {
			delete(list, id)
			if len(list) == 0 {
				delete(c.pending, subject)
			}
		}
		c.mu.Unlock()
		cancel(nil)
	}
}

// publishRefused ends every call waiting on the subject the bus's error names, with errPermissionRefused
// as the cause (mesh-tools node-tools/internal/bus/caller.go publishRefused): true when there was one
// waiting.
func (c *Conn) publishRefused(err error) bool {
	if !errors.Is(err, nats.ErrPermissionViolation) {
		return false
	}
	m := refusedPublish.FindStringSubmatch(err.Error())
	if len(m) < 2 {
		return false
	}
	c.mu.Lock()
	cancels := c.pending[m[1]]
	delete(c.pending, m[1])
	c.mu.Unlock()
	for _, cancel := range cancels {
		cancel(errPermissionRefused)
	}
	return len(cancels) > 0
}

// 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).
// 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,
		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
		},
	}
}