diff --git a/go/process/asks.go b/go/process/asks.go index 0aae29b..baa431a 100644 --- a/go/process/asks.go +++ b/go/process/asks.go @@ -24,7 +24,7 @@ func (c *Conn) Ask(bucket, id string) (*stdio.StateEntry, error) { return nil, err } subject := "$JS.API.DIRECT.GET.KV_" + bucket + ".$KV." + bucket + "." + c.account.Module + "." + id - ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() reply, err := c.nc.RequestWithContext(ctx, subject, nil) if err != nil { diff --git a/go/process/conn.go b/go/process/conn.go index b9d48cb..0c31c4c 100644 --- a/go/process/conn.go +++ b/go/process/conn.go @@ -1,12 +1,15 @@ package process import ( + "context" "crypto/sha256" "crypto/tls" "crypto/x509" "encoding/hex" "errors" "fmt" + "regexp" + "sync" "github.com/nats-io/nats.go" ) @@ -17,13 +20,43 @@ import ( 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. 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{}} + + // 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(account.InboxPrefix()), nats.UserInfo(account.User, account.Password), @@ -32,16 +65,72 @@ func Connect(account Account, opts ...nats.Option) (*Conn, error) { all = append(all, nats.Secure(pinnedToFingerprint(account.Fingerprint))) } all = append(all, opts...) + // Last, so it is the one AsyncErrorCB nats.Connect keeps however the opts above ordered theirs. + all = append(all, nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) { + c.publishRefused(err) + if chained != nil { + chained(nc, sub, err) + } + })) 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 + c.nc = nc + return c, nil } // 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 { diff --git a/go/process/emit.go b/go/process/emit.go index c905ceb..f674665 100644 --- a/go/process/emit.go +++ b/go/process/emit.go @@ -51,7 +51,7 @@ func (c *Conn) Emit(eventType string, body any, opts ...stdio.EmitOptions) error if err != nil { return err } - ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() _, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id)) return err diff --git a/go/process/state.go b/go/process/state.go index 1f8b6b8..43221fa 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -7,16 +7,16 @@ import ( "fmt" "time" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" ) -// StateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a var, not a const, so a -// test whose account is not granted a subject — refused only as an async protocol error with no reply, -// never a reply naming the refusal — need not wait out the full bound (mesh-tools' own ProofTimeout, -// node-tools/internal/bus/seat.go:65, is a var for the same reason). -var StateTimeout = 30 * time.Second +// stateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a denied publish is +// caught at once by Conn's own ErrorHandler (conn.go withRefusal/publishRefused), never waited out, so +// this stays the ordinary request bound and is not exposed for a test to shrink. +const stateTimeout = 30 * time.Second // State is one of this process's module's states, read and written on this connection (novox/hq ADR 0436 // clause 4), in the bucket _ (mesh-controller internal/broker/state.go:49) — only as the @@ -33,14 +33,27 @@ type State struct { func (s State) bucket() string { return s.Module + "_" + s.Name } // State is the named state this connection's module keeps, bound lazily to its bucket on each call so a -// bucket the account is not granted fails at the call that needed it, naming the bucket. +// bucket the account is not granted fails at the call that needed it, naming the bucket. The bind itself +// (js.KeyValue) is what asks $JS.API.STREAM.INFO.KV_, so it is what withRefusal watches: a bucket +// not granted is refused there, promptly, as nats.ErrPermissionViolation. func (c *Conn) State(name string) State { + bucket := c.account.Module + "_" + name return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) { js, err := jetstream.New(c.nc) if err != nil { return nil, err } - return js.KeyValue(ctx, c.account.Module+"_"+name) + subject := "$JS.API.STREAM.INFO.KV_" + bucket + ctx, stop := c.withRefusal(ctx, subject) + defer stop() + kv, err := js.KeyValue(ctx, bucket) + if err != nil { + if errors.Is(context.Cause(ctx), errPermissionRefused) { + return nil, fmt.Errorf("%w: publishing %s", nats.ErrPermissionViolation, subject) + } + return nil, err + } + return kv, nil }} } @@ -54,7 +67,7 @@ func (s State) store(ctx context.Context) (jetstream.KeyValue, error) { // Get is a key's current value, or nil when it has none. func (s State) Get(key string) (*stdio.StateEntry, error) { - ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) if err != nil { @@ -73,7 +86,7 @@ func (s State) Get(key string) (*stdio.StateEntry, error) { // Create writes a key only when it has none, and returns the revision; a key that has one is refused as // stdio.ErrChanged. func (s State) Create(key string, value any) (uint64, error) { - ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) if err != nil { @@ -96,7 +109,7 @@ func (s State) Create(key string, value any) (uint64, error) { // Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile // is refused as stdio.ErrChanged. func (s State) Update(key string, value any, revision uint64) (uint64, error) { - ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) if err != nil { diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index 1cb63e9..cc01c15 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -6,8 +6,8 @@ import ( "errors" "strings" "testing" - "time" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" @@ -64,15 +64,18 @@ func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) { } } -// 1188/AC4: given an account not granted a bucket, when it writes there, then the bus's refusal comes back -// as an error naming the bucket. A denied publish is refused only as an async protocol error with no -// reply (never a reply naming the refusal), so StateTimeout is shortened for this test alone. +// 1188/AC4: given an account not granted a bucket, when it writes there, then the bus's refusal comes +// back as an error naming the bucket. The bucket exists (made by the admin) so the only reason Create +// can fail is the missing grant, never "bucket not found"; and Conn's own ErrorHandler (conn.go +// withRefusal/publishRefused) catches the denied publish at once, so this returns promptly rather than +// after a timeout. func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) { - old := StateTimeout - StateTimeout = 2 * time.Second - t.Cleanup(func() { StateTimeout = old }) + url, account := grantedSendProcess(t) + _, adminJetStream := adminJS(t, url) + if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_unrelated"}); err != nil { + t.Fatalf("the admin could not make the bucket: %v", err) + } - _, account := grantedSendProcess(t) conn, err := Connect(account) if err != nil { t.Fatal(err) @@ -80,9 +83,14 @@ func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) { defer conn.Close() s := conn.State("unrelated") - if _, err := s.Create("key", map[string]any{"v": 1}); err == nil { + _, err = s.Create("key", map[string]any{"v": 1}) + if err == nil { t.Fatal("writing to an ungranted bucket was not refused") - } else if !strings.Contains(err.Error(), "mesh-mail_unrelated") { + } + if !errors.Is(err, nats.ErrPermissionViolation) { + t.Errorf("the refusal %v does not wrap nats.ErrPermissionViolation", err) + } + if !strings.Contains(err.Error(), "mesh-mail_unrelated") { t.Errorf("the refusal %q does not name the bucket mesh-mail_unrelated", err.Error()) } }