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