From 2a4fe8f5a702bc4d79915a15d05bfa247043704b Mon Sep 17 00:00:00 2001 From: Dan Draper Date: Sun, 18 Oct 2026 21:31:03 +1100 Subject: [PATCH 1/2] fix(golang): a call whose context ends fails on its own, and leaves the client open One encrypt.Client serves every request in a service, but a call whose context was cancelled or passed its deadline closed it for all of them, and so did the auth profile a client reads its tokens from. Both runtimes are built with wazero's WithCloseOnContextDone, and every guest call ran under the caller's context, so the runtime closed the module when that context ended. Worse, the wait for the client's lock ignored the context: a request that timed out in the queue behind a slow ZeroKMS round trip reached the guest with a context already done, and wazero closed the module before any guest code ran. The fix keeps the caller's cancellation and takes it away from the runtime. internal/guest gains three pieces, used by encrypt.Client and auth.ProfileStore alike: - Lock, a one-slot lock whose wait ends with the caller's context and which refuses a context that has already ended, so a request that timed out in the queue never starts. auth.Strategy's lock uses it too. - Detach, the context a guest call runs under: the caller's values but not its cancellation, so the runtime no longer closes the module when a caller gives up. - Caller, which the host functions use to reach the caller's context: a ZeroKMS request, a CTS exchange or an OIDC provider call in flight is cancelled, fails as a transport failure, and the guest returns through its ordinary error path with its keys intact. The error matches both the context's error and ErrTransport. A backstop remains for a guest or host function that ignores cancellation: Detach's context ends guest.StuckAfter (30s) after the caller's, and the runtime then closes the module, as before. That close now means a bug, never a closed browser tab. TestInterruptedCallClosesTheClient pinned the old behaviour and is replaced by tests of the new contract: a call cancelled during a ZeroKMS request, during a token fetch, and during an auth strategy's exchange leaves the client and the profile open; a call whose deadline passes in the queue leaves without reaching the guest; a stuck call still closes the client and the profile; and many callers cancelling at random share one client without it closing. Closes #1169 --- languages/golang/CHANGELOG.md | 10 + languages/golang/auth/cancel_test.go | 184 +++++++++ languages/golang/auth/doc.go | 7 + languages/golang/auth/guest.go | 7 +- languages/golang/auth/store.go | 52 ++- languages/golang/auth/strategy.go | 9 +- languages/golang/auth/transport.go | 5 +- languages/golang/encrypt/README.md | 21 +- languages/golang/encrypt/cancel_test.go | 391 +++++++++++++++++++ languages/golang/encrypt/checker.go | 6 +- languages/golang/encrypt/client.go | 87 +++-- languages/golang/encrypt/diagnostic_test.go | 6 +- languages/golang/encrypt/doc.go | 20 +- languages/golang/encrypt/export_test.go | 2 +- languages/golang/encrypt/guest.go | 15 +- languages/golang/encrypt/guest_test.go | 49 --- languages/golang/encrypt/memory_test.go | 6 +- languages/golang/encrypt/query.go | 6 +- languages/golang/encrypt/records.go | 12 +- languages/golang/encrypt/transport.go | 11 +- languages/golang/internal/guest/errors.go | 2 +- languages/golang/internal/guest/lock.go | 117 ++++++ languages/golang/internal/guest/lock_test.go | 82 ++++ 23 files changed, 986 insertions(+), 121 deletions(-) create mode 100644 languages/golang/auth/cancel_test.go create mode 100644 languages/golang/encrypt/cancel_test.go create mode 100644 languages/golang/internal/guest/lock.go create mode 100644 languages/golang/internal/guest/lock_test.go diff --git a/languages/golang/CHANGELOG.md b/languages/golang/CHANGELOG.md index 9d656a1aa..5dc6c0f3e 100644 --- a/languages/golang/CHANGELOG.md +++ b/languages/golang/CHANGELOG.md @@ -77,6 +77,16 @@ listed under **Breaking**. Releases are tagged `languages/golang/vX.Y.Z`. ### Changed +- **A call whose context ends fails on its own and leaves the client + open.** A cancelled or timed-out call used to close the `encrypt.Client`, + or the `auth` profile behind its strategy, for every later caller, and a + request whose deadline passed while it waited for the client closed it + before running. Now a call waiting for the client leaves the queue with + the context's error; a ZeroKMS request or a token fetch in flight is + cancelled, and the error matches both the context's error and + `ErrTransport`; and the client stays open. A client or profile still + closes itself (`ErrState`) when the guest traps, or when a call runs on + for 30 seconds after its context ended. - **stashgen refuses a match index on an index-only field** (`stash:"name,index=match"` with no `encrypt`). A match query's rows are checked by decrypting them, and such a field stores no ciphertext. diff --git a/languages/golang/auth/cancel_test.go b/languages/golang/auth/cancel_test.go new file mode 100644 index 000000000..dca54c7e1 --- /dev/null +++ b/languages/golang/auth/cancel_test.go @@ -0,0 +1,184 @@ +package auth + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/cipherstash/stack/languages/golang/internal/guest" + "github.com/tetratelabs/wazero/api" +) + +// A profile serves every strategy, and through them every encrypt client, +// in a process: a request cancelled while its token is fetched must fail +// on its own and leave the profile open for the next. + +// tokenServer answers the access-key exchange, holding each request until +// the client gives up while hold is set. +func tokenServer(t *testing.T, hold *atomic.Bool) *httptest.Server { + t.Helper() + var jwt string + // A request the client never cancels is let go at the end, so a + // failure reports itself rather than hanging the server's Close. + done := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if hold.Load() { + // The server notices a client that hung up only once the body + // is read. + _, _ = io.Copy(io.Discard, r.Body) + select { + case <-r.Context().Done(): + case <-done: + t.Error("a held request was never cancelled") + } + return + } + fmt.Fprintf(w, `{"accessToken":%q,"expiry":%d}`, jwt, time.Now().Add(time.Hour).Unix()) + })) + t.Cleanup(server.Close) + t.Cleanup(func() { close(done) }) + jwt = testJWT(t, server.URL) + return server +} + +func TestATokenFetchCancelledFailsOnlyItself(t *testing.T) { + guestOrSkip(t) + var hold atomic.Bool + server := tokenServer(t, &hold) + profile, err := Open(context.Background(), t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + strategy, err := profile.AccessKey(context.Background(), testCRN, "CSAKtestKeyId.testKeySecret", WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + + hold.Store(true) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + if _, err := strategy.Token(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("cancelled Token: %v, want the deadline", err) + } + if profile.root.closed || profile.root.inst.module.IsClosed() { + t.Fatal("the profile was closed") + } + hold.Store(false) + if token, err := strategy.Token(context.Background()); err != nil || token == "" { + t.Fatalf("the next Token: %q, %v", token, err) + } +} + +// A refresh runs on after its caller gives up, so the guest call under it +// can succeed after the caller's context ended. It still fails with the +// context's error; what the refresh brought is kept for the next call. +func TestARefreshThatSucceedsAfterItsContextEndsFailsWithIt(t *testing.T) { + guestOrSkip(t) + dir, _ := expiredDeviceProfile(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + cancel() + fmt.Fprint(w, `{"access_token":"fresh","token_type":"Bearer","expires_in":3600,"refresh_token":"refresh-2"}`) + })) + t.Cleanup(server.Close) + profile, err := Open(context.Background(), dir) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + ws, err := profile.WorkspaceStore(context.Background(), wsA) + if err != nil { + t.Fatal(err) + } + strategy, err := ws.DeviceSession(context.Background(), WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + + out, err := strategy.store.callArgs(withRefresh(ctx), func(i *instance) api.Function { return i.authRefresh }, guest.BufArg([]byte(strategy.handle))) + if !errors.Is(err, context.Canceled) || out != nil { + t.Fatalf("refresh: %q, %v, want no token and context.Canceled", out, err) + } + if token, err := strategy.Token(context.Background()); err != nil || token != "fresh" { + t.Fatalf("the next Token: %q, %v, want the refreshed token", token, err) + } +} + +func TestATokenRequestWhoseContextEndsInTheQueueNeverReachesTheGuest(t *testing.T) { + guestOrSkip(t) + var hold atomic.Bool + server := tokenServer(t, &hold) + profile, err := Open(context.Background(), t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + strategy, err := profile.AccessKey(context.Background(), testCRN, "CSAKtestKeyId.testKeySecret", WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + + for name, lock := range map[string]interface { + Acquire(context.Context) error + Unlock() + }{"the strategy's": &strategy.mu, "the profile's": &profile.root.mu} { + t.Run(name, func(t *testing.T) { + if err := lock.Acquire(context.Background()); err != nil { + t.Fatal(err) + } + released := make(chan struct{}) + time.AfterFunc(200*time.Millisecond, func() { lock.Unlock(); close(released) }) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if _, err := strategy.Token(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("queued Token: %v, want the deadline", err) + } + <-released + }) + } + if token, err := strategy.Token(context.Background()); err != nil || token == "" { + t.Fatalf("the next Token: %q, %v", token, err) + } +} + +// The backstop: a call still running long after its caller gave up closes +// the profile, and says so. +func TestATokenFetchThatRunsOnAfterItsContextEndsClosesTheProfile(t *testing.T) { + guestOrSkip(t) + profile, err := Open(context.Background(), t.TempDir(), WithRoundTripper(roundTripFunc(func(*http.Request) (*http.Response, error) { + // A RoundTripper that ignores the request's context. + time.Sleep(300 * time.Millisecond) + return nil, errors.New("too late") + }))) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + strategy, err := profile.AccessKey(context.Background(), testCRN, "CSAKtestKeyId.testKeySecret", WithBaseURL("https://cts.example.com")) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + profile.root.stuck = 20 * time.Millisecond + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if _, err := strategy.Token(ctx); err == nil || !strings.Contains(err.Error(), "the profile is closed") { + t.Fatalf("stuck Token: %v, want the profile closed", err) + } + if _, err := strategy.Token(context.Background()); !errors.Is(err, ErrState) { + t.Fatalf("a Token after the close: %v, want ErrState", err) + } +} diff --git a/languages/golang/auth/doc.go b/languages/golang/auth/doc.go index b0bf4f16d..de9d1bec9 100644 --- a/languages/golang/auth/doc.go +++ b/languages/golang/auth/doc.go @@ -39,6 +39,13 @@ // [OAuth2TokenSource] adapts an existing golang.org/x/oauth2.TokenSource // into the OIDC provider interface. // +// A call whose context ends, [Strategy.Token] included, fails with the +// context's error and leaves the store and its strategies open: a request +// cancelled while its token is exchanged or refreshed does not stop the +// encrypt clients that read tokens from the same store. The store closes +// itself, and every later call is [ErrState], only when its guest traps or +// a call runs on for 30 seconds after its context ended. +// // # Why a second guest // // The crypto guest behind encrypt has no filesystem and no diff --git a/languages/golang/auth/guest.go b/languages/golang/auth/guest.go index 5c17ef261..69e395a3a 100644 --- a/languages/golang/auth/guest.go +++ b/languages/golang/auth/guest.go @@ -111,6 +111,9 @@ func newInstance(ctx context.Context, wasm []byte, hostDir string, policy guest. return nil, fmt.Errorf("%w: %s: %w", ErrNoProfile, hostDir, err) } } + // A call never runs under its caller's context (see guest.Detach), so + // this closes the module only for a call that ran on long after its + // caller gave up: ProfileStore.callArgs. config := wazero.NewRuntimeConfig(). WithCompilationCache(compilationCache()). WithCloseOnContextDone(true) @@ -183,8 +186,8 @@ func newInstance(ctx context.Context, wasm []byte, hostDir string, policy guest. // release runs the guest's shutdown — every buffer it still holds wiped — // and closes the runtime, which frees the linear memory through the -// allocator's wipe, then the directory handle the mount holds. A module an -// interrupted call or a trap already closed cannot run sa_shutdown; the +// allocator's wipe, then the directory handle the mount holds. A module a +// stuck call or a trap already closed cannot run sa_shutdown; the // runtime close still wipes and frees its memory. func (inst *instance) release() error { ctx := context.Background() diff --git a/languages/golang/auth/store.go b/languages/golang/auth/store.go index b5baea7a6..0421c0f1e 100644 --- a/languages/golang/auth/store.go +++ b/languages/golang/auth/store.go @@ -10,7 +10,7 @@ import ( "path/filepath" "runtime" "strings" - "sync" + "time" "github.com/cipherstash/stack/languages/golang/internal/guest" "github.com/cipherstash/vitaminc/bindings/go/vcffi" @@ -53,6 +53,13 @@ func RequireLockedMemory() Option { // stores from one root share one guest instance over the one mounted // directory; Close on any of them releases it. It is safe for concurrent // use; calls are serialised, because a wasm instance is single-threaded. +// +// A call whose context ends fails with the context's error, and only that +// call: a request cancelled while its token is fetched or refreshed never +// closes the profile, or the strategies and encrypt clients that read +// tokens from it. The profile closes itself, and every later call is +// [ErrState], only when the guest's state is unknown: it trapped, or it +// went on running long after its caller gave up, which is a bug. type ProfileStore struct { root *root // dir is the store's directory as the guest sees it: guestRoot, or a @@ -62,13 +69,17 @@ type ProfileStore struct { // root is the guest instance the stores of one profile share. type root struct { - mu sync.Mutex + mu guest.Lock inst *instance hostDir string - closed bool - // released is the runtime's own state, apart from closed: an - // interrupted call closes the module, and so the stores, while the - // runtime is still allocated. Close frees it even then. + // stuck is how long a call may run on after its context ends before + // the guest is treated as stuck and closed. guest.StuckAfter; tests + // shorten it. + stuck time.Duration + closed bool + // released is the runtime's own state, apart from closed: a stuck + // call closes the module, and so the stores, while the runtime is + // still allocated. Close frees it even then. released bool cleanup runtime.Cleanup } @@ -135,7 +146,7 @@ func open(ctx context.Context, hostDir string, opts []Option) (*ProfileStore, er if err != nil { return nil, err } - r := &root{inst: inst, hostDir: hostDir} + r := &root{inst: inst, hostDir: hostDir, stuck: guest.StuckAfter} // The cleanup takes the instance, not the root: a cleanup whose // argument reaches its object keeps that object alive forever. r.cleanup = runtime.AddCleanup(r, func(inst *instance) { _ = inst.release() }, inst) @@ -204,7 +215,7 @@ func (s *ProfileStore) Close() error { type export func(*instance) api.Function // call runs one export under the profile's lock, closing the profile if -// the guest trapped or an interrupted call took the module down. +// the guest trapped or a stuck call took the module down. func (s *ProfileStore) call(ctx context.Context, fn export, args ...string) ([]byte, error) { // Every profile export works on the mount; without one the guest would // report an I/O error that says less than this does. @@ -221,23 +232,31 @@ func (s *ProfileStore) call(ctx context.Context, fn export, args ...string) ([]b // callArgs is the shared checked call path for profile exports and auth // exports. The latter pass secret-bearing byte buffers and wipe host copies. +// +// The caller's context bounds the wait for the lock, as encrypt's +// Client.call does, and the guest runs under guest.Detach, so a caller +// that gives up cancels its token request without closing the profile. func (s *ProfileStore) callArgs(ctx context.Context, fn export, args ...guest.Arg) ([]byte, error) { r := s.root - r.mu.Lock() + if err := r.mu.Acquire(ctx); err != nil { + return nil, err + } defer r.mu.Unlock() if r.closed || r.inst.module.IsClosed() { r.closed = true return nil, ErrState } growth := r.inst.mem.GrowthRefusal() - out, err := guest.Call(ctx, r.inst.mem, r.inst.module, r.inst.exports, fn(r.inst), args...) + run, done := guest.Detach(ctx, r.stuck) + out, err := guest.Call(run, r.inst.mem, r.inst.module, r.inst.exports, fn(r.inst), args...) + done() switch { case r.inst.module.IsClosed(): r.closed = true if err == nil { err = ErrState } - err = fmt.Errorf("%w: interrupted call closed the profile", err) + err = fmt.Errorf("%w: the guest ran on after its caller's context ended, and the profile is closed", err) case errors.Is(err, guest.ErrTrap): r.closed = true _ = r.inst.module.Close(context.Background()) @@ -253,8 +272,19 @@ func (s *ProfileStore) callArgs(ctx context.Context, fn export, args ...guest.Ar err = fmt.Errorf("%w (growth refused under RequireLockedMemory): %w", guest.MemoryLockError(g.Reason), err) } if err != nil { + if ctxErr := ctx.Err(); ctxErr != nil && !errors.Is(err, ctxErr) { + err = fmt.Errorf("%w: %w", ctxErr, err) + } return nil, err } + // A call that succeeded after its context ended fails too, as in + // encrypt's Client.call, and its output (a token, say) is wiped. What + // the guest wrote stays written: a refresh that ran on keeps the + // rotated token, and the next call reads it. + if ctxErr := ctx.Err(); ctxErr != nil { + guest.Wipe(out) + return nil, ctxErr + } return out, nil } diff --git a/languages/golang/auth/strategy.go b/languages/golang/auth/strategy.go index 5fadc7bce..a5fcb924f 100644 --- a/languages/golang/auth/strategy.go +++ b/languages/golang/auth/strategy.go @@ -7,7 +7,6 @@ import ( "fmt" "os" "strconv" - "sync" "github.com/cipherstash/stack/languages/golang/internal/guest" "github.com/tetratelabs/wazero/api" @@ -58,7 +57,7 @@ type Strategy struct { handle string providerID uint32 device bool - mu sync.Mutex + mu guest.Lock closed bool } @@ -182,7 +181,11 @@ func (s *ProfileStore) hasToken(ctx context.Context) (bool, error) { // Token gets the current CTS bearer credential. Device sessions read a fresh // token without a file lock; only a refresh call takes the cross-process lock. func (s *Strategy) Token(ctx context.Context) (string, error) { - s.mu.Lock() + // A request queued behind another's refresh leaves when its context + // ends, as at the profile's own lock. + if err := s.mu.Acquire(ctx); err != nil { + return "", err + } defer s.mu.Unlock() if s.closed { return "", ErrState diff --git a/languages/golang/auth/transport.go b/languages/golang/auth/transport.go index 599f40dab..de30684d2 100644 --- a/languages/golang/auth/transport.go +++ b/languages/golang/auth/transport.go @@ -124,7 +124,8 @@ func (t *authTransport) send(ctx context.Context, m api.Module, return t.placeResponse(ctx, m, respHeadersPtrOut, respHeadersLenOut, respBodyPtrOut, respBodyLenOut, -1, nil, []byte("request buffer out of range")) } requestBody := newAuthRequestBody(body) - req, err := http.NewRequestWithContext(ctx, string(method), string(url), requestBody) + // The caller's context, not the call's: see guest.Detach. + req, err := http.NewRequestWithContext(guest.Caller(ctx), string(method), string(url), requestBody) if err != nil { _ = requestBody.Close() return t.placeResponse(ctx, m, respHeadersPtrOut, respHeadersLenOut, respBodyPtrOut, respBodyLenOut, -1, nil, []byte(err.Error())) @@ -174,7 +175,7 @@ func (t *authTransport) oidcTokenGet(ctx context.Context, m api.Module, provider if source == nil { return 1 } - token, err := source.Token(ctx) + token, err := source.Token(guest.Caller(ctx)) if err != nil || token == "" || strings.ContainsAny(token, "\r\n\x00") { return 1 } diff --git a/languages/golang/encrypt/README.md b/languages/golang/encrypt/README.md index fe6021930..4c3145b11 100644 --- a/languages/golang/encrypt/README.md +++ b/languages/golang/encrypt/README.md @@ -117,6 +117,21 @@ ZeroKMS round trip `NewClient` makes. It has nothing to do with an *encryption* context, which is what a field is sealed under; that is the `context=` tag. +Every call takes a `ctx` too, and a call whose `ctx` is cancelled or passes +its deadline fails with the context's error. Only that call fails: one +request giving up never closes the shared client for the others. + +- A call waiting for the client (it runs one call at a time) leaves the + queue when its `ctx` ends, and never starts. +- A call waiting on ZeroKMS, or on its auth strategy for a token, has that + request cancelled. Its error matches both `context.Canceled` or + `context.DeadlineExceeded` and `ErrTransport`. + +A client closes itself, and every later call returns `ErrState`, only when +the guest's state is unknown: the guest trapped, or a call ran on for 30 +seconds after its `ctx` ended, which means a bug. A service that sees +`ErrState` from a client it did not close builds a new one. + Every other setting is a functional option with a default: ```go @@ -372,7 +387,11 @@ or for stored data. Check an error two ways. its field's context (ZeroKMS refuses the key, or the AEAD fails). - `ErrEncoding`: a stored value or a call that does not fit the declaration. - `ErrUnauthorized`, `ErrNotFound`, `ErrTransport`, `ErrKMS`: ZeroKMS. -- `ErrState`: a call on a closed client. `ErrMemoryLock`: see below. +- `ErrState`: a call on a closed client: you closed it, or the guest + trapped or ran on long after a call's context ended (see Connect). + `ErrMemoryLock`: see below. +- `context.Canceled`, `context.DeadlineExceeded`: the call's `ctx` ended. + The client is still open. `errors.As` with a `*Diagnostic` gives you the detail behind a failure the engine reports: a stable `Code` (`stack_encrypt::foreign_keyset`, diff --git a/languages/golang/encrypt/cancel_test.go b/languages/golang/encrypt/cancel_test.go new file mode 100644 index 000000000..c94c916ca --- /dev/null +++ b/languages/golang/encrypt/cancel_test.go @@ -0,0 +1,391 @@ +package encrypt + +import ( + "context" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "math/rand/v2" + "net/http" + "net/http/httptest" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cipherstash/stack/languages/golang/auth" + "github.com/cipherstash/stack/languages/golang/internal/guest" + "github.com/cipherstash/stack/languages/golang/internal/record" +) + +// One encrypt.Client serves every request in a service, so a caller's +// context ending must fail that caller's call and nothing else. These tests +// pin that contract, and the one case that still closes the client: a guest +// that runs on long after its caller gave up. + +// unauthorized answers every ZeroKMS request with a 401: a full round trip +// through the guest's network path that ends in a status the client can +// tell apart from ErrState. +func unauthorized(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("nope")), Header: http.Header{}}, nil +} + +// switchable is a RoundTripper whose behaviour a test changes between calls. +type switchable struct { + mu sync.Mutex + rt roundTripFunc + sent atomic.Int32 +} + +func (s *switchable) set(rt roundTripFunc) { + s.mu.Lock() + defer s.mu.Unlock() + s.rt = rt +} + +func (s *switchable) RoundTrip(r *http.Request) (*http.Response, error) { + s.sent.Add(1) + s.mu.Lock() + rt := s.rt + s.mu.Unlock() + return rt(r) +} + +// cancelClient is a guest not yet given a key, whose ZeroKMS requests go to +// rt and whose token comes from token. Its se_cipher_init is the call the +// tests drive: it makes a load-keyset request, so it reaches the network. +func cancelClient(t *testing.T, rt http.RoundTripper, token tokenSource) *Client { + t.Helper() + tr := &transport{rt: rt, token: token} + inst, err := newInstance(context.Background(), guestOrSkip(t), tr, guest.BestEffort) + if err != nil { + t.Fatal(err) + } + c := newClient(inst, tr) + t.Cleanup(func() { _ = c.Close() }) + return c +} + +// cipherInit runs se_cipher_init on c under ctx. +func cipherInit(ctx context.Context, c *Client) error { + encoded, err := encodeConfig(testInit("http://zerokms.invalid")) + if err != nil { + return err + } + defer wipe(encoded) + _, err = c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { + return inst.call(ctx, inst.cipherInit, buf(encoded)) + }) + return err +} + +// wantOpen fails the test unless c is still open: its next call reaches +// ZeroKMS and comes back with ZeroKMS's answer, not ErrState. +func wantOpen(t *testing.T, c *Client, rt *switchable) { + t.Helper() + if c.closed || c.inst.module.IsClosed() { + t.Fatal("the client was closed") + } + rt.set(unauthorized) + if err := cipherInit(context.Background(), c); !errors.Is(err, ErrUnauthorized) { + t.Fatalf("the next call: %v, want ErrUnauthorized from ZeroKMS", err) + } +} + +func TestACallCancelledDuringARequestFailsOnlyItself(t *testing.T) { + rt := &switchable{rt: func(r *http.Request) (*http.Response, error) { + <-r.Context().Done() + return nil, r.Context().Err() + }} + c := cancelClient(t, rt, staticToken("t")) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + err := cipherInit(ctx, c) + if !errors.Is(err, context.DeadlineExceeded) || !errors.Is(err, ErrTransport) { + t.Fatalf("cancelled call: %v, want the deadline and ErrTransport", err) + } + wantOpen(t, c, rt) +} + +func TestACallCancelledDuringATokenFetchFailsOnlyItself(t *testing.T) { + rt := &switchable{rt: unauthorized} + var blocking atomic.Bool + blocking.Store(true) + token := tokenFunc(func(ctx context.Context) (string, error) { + if blocking.Load() { + <-ctx.Done() + return "", ctx.Err() + } + return "t", nil + }) + c := cancelClient(t, rt, token) + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(50*time.Millisecond, cancel) + err := cipherInit(ctx, c) + if !errors.Is(err, context.Canceled) || !errors.Is(err, ErrTransport) { + t.Fatalf("cancelled call: %v, want context.Canceled and ErrTransport", err) + } + if n := rt.sent.Load(); n != 0 { + t.Fatalf("%d request(s) sent without a token", n) + } + blocking.Store(false) + wantOpen(t, c, rt) +} + +func TestACallWhoseContextEndsInTheQueueNeverReachesTheGuest(t *testing.T) { + rt := &switchable{rt: unauthorized} + c := cancelClient(t, rt, staticToken("t")) + + t.Run("ends while waiting", func(t *testing.T) { + // Another call holds the client: the slow ZeroKMS round trip the + // queue is waiting behind. + if err := c.mu.Acquire(context.Background()); err != nil { + t.Fatal(err) + } + released := make(chan struct{}) + time.AfterFunc(200*time.Millisecond, func() { c.mu.Unlock(); close(released) }) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + start := time.Now() + err := cipherInit(ctx, c) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("queued call: %v, want the deadline", err) + } + if waited := time.Since(start); waited > 150*time.Millisecond { + t.Errorf("the queued call waited %v, past its deadline, for the lock", waited) + } + <-released + }) + t.Run("already ended", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := cipherInit(ctx, c); !errors.Is(err, context.Canceled) { + t.Fatalf("call with an ended context: %v, want context.Canceled", err) + } + }) + if n := rt.sent.Load(); n != 0 { + t.Fatalf("%d request(s) reached ZeroKMS from calls whose context had ended", n) + } + wantOpen(t, c, rt) +} + +// The backstop: a call still running long after its caller gave up is a +// guest or host function that ignores cancellation. Its state is unknown, +// so the client closes, and says so. +func TestACallThatRunsOnAfterItsContextEndsClosesTheClient(t *testing.T) { + rt := &switchable{rt: func(*http.Request) (*http.Response, error) { + // A RoundTripper that ignores the request's context. + time.Sleep(300 * time.Millisecond) + return unauthorized(nil) + }} + c := cancelClient(t, rt, staticToken("t")) + c.stuck = 20 * time.Millisecond + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + err := cipherInit(ctx, c) + if err == nil || !strings.Contains(err.Error(), "the client is closed") { + t.Fatalf("stuck call: %v, want the client closed", err) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("stuck call: %v, want it to match the deadline", err) + } + if err := cipherInit(context.Background(), c); !errors.Is(err, ErrState) { + t.Fatalf("a call after the close: %v, want ErrState", err) + } +} + +// A call that finishes in time stops the backstop: the client stays open +// however long after the call its context ends. +func TestAFinishedCallStopsTheBackstop(t *testing.T) { + rt := &switchable{rt: unauthorized} + c := cancelClient(t, rt, staticToken("t")) + c.stuck = time.Millisecond + ctx, cancel := context.WithCancel(context.Background()) + if err := cipherInit(ctx, c); !errors.Is(err, ErrUnauthorized) { + t.Fatalf("call: %v, want ErrUnauthorized", err) + } + cancel() + time.Sleep(20 * time.Millisecond) + wantOpen(t, c, rt) +} + +// Work the guest does without the network runs on through a cancellation, +// so a call can succeed after its caller gave up. It still fails with the +// context's error, and the output nobody will read is wiped. +func TestACallThatSucceedsAfterItsContextEndsFailsWithIt(t *testing.T) { + rt := &switchable{rt: unauthorized} + c := cancelClient(t, rt, staticToken("t")) + ctx, cancel := context.WithCancel(context.Background()) + plaintext := []byte("plaintext") + out, err := c.call(ctx, func(context.Context, *instance) ([]byte, error) { + cancel() + return plaintext, nil + }) + if !errors.Is(err, context.Canceled) || out != nil { + t.Fatalf("call: %q, %v, want no output and context.Canceled", out, err) + } + if string(plaintext) != string(make([]byte, len(plaintext))) { + t.Error("the discarded output was not wiped") + } + wantOpen(t, c, rt) +} + +// A closed module is ErrState, and Close still releases the runtime. +func TestAClosedModuleIsState(t *testing.T) { + ctx := context.Background() + c := rawInstance(t) + if err := c.inst.module.Close(ctx); err != nil { + t.Fatal(err) + } + if _, err := c.Keyset(KeysetName("k")).KeysetID(ctx); !errors.Is(err, ErrState) { + t.Fatalf("Keyset on a closed module: %v, want ErrState", err) + } + if err := c.Close(); err != nil { + t.Fatalf("Close after the module closed: %v", err) + } + // The close must reach the runtime. A stuck call closes the module and + // marks the client closed; a Close that treated that as "already done" + // would leave the runtime and the host module it carries allocated for + // the life of the process. + if !c.released { + t.Error("Close after the module closed left the runtime unreleased") + } + if m := c.inst.runtime.Module(transportModule); m != nil { + t.Errorf("host module %s is still registered after Close", transportModule) + } + // Still idempotent. + if err := c.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } +} + +// Many callers share one client, as a service's requests do, and each call +// gets a deadline: none, already passed, or one that may pass in the queue +// or in the guest. The client never closes, every call whose context did +// not end succeeds, and nothing is left running afterwards. +func TestManyCallersCancellingAtRandomShareOneClient(t *testing.T) { + c, err := NewDeterministicClient(context.Background(), [32]byte{7}) + if errors.Is(err, ErrDeterministicGuestNotBuilt) { + t.Skip(err) + } + if err != nil { + t.Fatal(err) + } + defer c.Close() + cipher := c.DefaultKeyset() + plan := usersPlan() + before := runtime.NumGoroutine() + + const callers, calls = 16, 40 + var succeeded atomic.Int32 + var wg sync.WaitGroup + for g := range callers { + wg.Add(1) + go func() { + defer wg.Done() + r := rand.New(rand.NewPCG(uint64(g), 1)) + for i := range calls { + ctx, cancel := context.Background(), context.CancelFunc(func() {}) + switch r.IntN(4) { + case 0: // no deadline + case 1: + ctx, cancel = context.WithCancel(ctx) + cancel() + default: + ctx, cancel = context.WithTimeout(ctx, time.Duration(r.IntN(2000))*time.Microsecond) + } + rows := []record.Source{usersRow(uint32(i), "user@example.com")} + _, err := cipher.Seal(ctx, plan, rows) + ended := ctx.Err() != nil + cancel() + switch { + case errors.Is(err, ErrState): + t.Errorf("the client closed: %v", err) + return + case err == nil: + succeeded.Add(1) + case !ended: + t.Errorf("a call whose context did not end failed: %v", err) + case !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded): + t.Errorf("a cancelled call: %v, want the context's error", err) + } + } + }() + } + wg.Wait() + if succeeded.Load() == 0 { + t.Fatal("no call succeeded") + } + if _, err := cipher.Seal(context.Background(), plan, []record.Source{usersRow(1, "a@example.com")}); err != nil { + t.Fatalf("a call after the load: %v", err) + } + // Every backstop timer and context watcher has stopped. + deadline := time.Now().Add(2 * time.Second) + for runtime.NumGoroutine() > before && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if n := runtime.NumGoroutine(); n > before { + t.Errorf("%d goroutines after the load, %d before", n, before) + } +} + +// The encrypt client and the auth profile fail together or not at all: a +// call cancelled while the auth strategy exchanges its access key for a +// token leaves both open, and the next call gets a token and reaches +// ZeroKMS. +func TestACallCancelledDuringATokenExchangeLeavesTheProfileOpen(t *testing.T) { + var hold atomic.Bool + done := make(chan struct{}) + var jwt string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + if hold.Load() { + select { + case <-r.Context().Done(): + case <-done: + t.Error("a held token request was never cancelled") + } + return + } + fmt.Fprintf(w, `{"accessToken":%q,"expiry":%d}`, jwt, time.Now().Add(time.Hour).Unix()) + })) + t.Cleanup(server.Close) + t.Cleanup(func() { close(done) }) + claims, err := json.Marshal(map[string]any{"iss": server.URL, "sub": "CS|test", "workspace": "ZVATKW3VHMFG27DY", "exp": time.Now().Add(time.Hour).Unix()}) + if err != nil { + t.Fatal(err) + } + jwt = "e30." + base64.RawURLEncoding.EncodeToString(claims) + ".c2ln" + + store, err := auth.OpenWithoutProfile(context.Background()) + if errors.Is(err, auth.ErrGuestNotBuilt) { + t.Skip(err) + } + if err != nil { + t.Fatal(err) + } + defer store.Close() + strategy, err := store.AccessKey(context.Background(), testCRN, testAccessKey, auth.WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + + rt := &switchable{rt: unauthorized} + c := cancelClient(t, rt, strategy) + hold.Store(true) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + if err := cipherInit(ctx, c); !errors.Is(err, context.DeadlineExceeded) || !errors.Is(err, ErrTransport) { + t.Fatalf("call cancelled during the exchange: %v, want the deadline and ErrTransport", err) + } + hold.Store(false) + wantOpen(t, c, rt) + if n := rt.sent.Load(); n != 1 { + t.Fatalf("%d ZeroKMS request(s), want the one after the token arrived", n) + } +} diff --git a/languages/golang/encrypt/checker.go b/languages/golang/encrypt/checker.go index 38533b0c2..a4ce1228d 100644 --- a/languages/golang/encrypt/checker.go +++ b/languages/golang/encrypt/checker.go @@ -57,7 +57,7 @@ func (k *Checker) engineCheck(ctx context.Context, plan *record.Plan) error { if err != nil { return fmt.Errorf("%w: %v", ErrEncoding, err) } - _, err = k.c.call(ctx, func(inst *instance) ([]byte, error) { + _, err = k.c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.planCheck, buf(encoded)) }) return err @@ -66,7 +66,7 @@ func (k *Checker) engineCheck(ctx context.Context, plan *record.Plan) error { // Targets lists the EQL types this build of the engine produces: none until // the EQL target dispatch lands. func (k *Checker) Targets(ctx context.Context) ([]record.Target, error) { - out, err := k.c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := k.c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.targets) }) if err != nil { @@ -93,7 +93,7 @@ func (k *Checker) Operators(ctx context.Context, plan *record.Plan, field string if err != nil { return nil, fmt.Errorf("%w: %v", ErrEncoding, err) } - out, err := k.c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := k.c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.operators, buf(encoded), buf([]byte(field))) }) if err != nil { diff --git a/languages/golang/encrypt/client.go b/languages/golang/encrypt/client.go index 4f89bd4ff..b0156425a 100644 --- a/languages/golang/encrypt/client.go +++ b/languages/golang/encrypt/client.go @@ -8,7 +8,7 @@ import ( "net/http" "runtime" "strconv" - "sync" + "time" "github.com/cipherstash/stack/languages/golang/internal/guest" "github.com/cipherstash/vitaminc/bindings/go/vcffi" @@ -20,6 +20,15 @@ import ( // concurrent use; calls are serialised internally, because a wasm instance // is single-threaded. Close it when done, as with any resource. // +// A call whose context ends fails with the context's error, and only that +// call: one request's cancellation or deadline never closes the client for +// the others sharing it. A call waiting its turn leaves the queue when its +// context ends; a call waiting on ZeroKMS has its request cancelled, and +// the error matches both the context's error and [ErrTransport]. The +// client closes itself, and every later call is [ErrState], only when the +// guest's state is unknown: it trapped, or it went on running long after +// its caller gave up, which is a bug. +// // Its key material lives in the guest's linear memory, which this package // supplies: reserved once so it never moves, locked in RAM and excluded // from core dumps where the platform allows, and wiped before it is @@ -29,13 +38,17 @@ import ( // where the protection lives. [Client.MemoryLocked] reports whether the // lock was granted. type Client struct { - mu sync.Mutex + mu guest.Lock inst *instance transport *transport - // closed refuses further calls: either Close ran, or an interrupted - // call took the module down under us (see Client.call). released is - // the runtime's own state, tracked apart from it because those two - // things come apart: an interrupted call closes the module — and so + // stuck is how long a call may run on after its context ends before + // the guest is treated as stuck and closed. guest.StuckAfter; tests + // shorten it. + stuck time.Duration + // closed refuses further calls: either Close ran, or a stuck or + // trapped call took the module down under us (see Client.call). + // released is the runtime's own state, tracked apart from it because + // those two things come apart: a stuck call closes the module — and so // the client — while the runtime and the host modules beside it are // still allocated. Close is what frees those, so it must do its work // even on a client that is already closed. @@ -175,7 +188,7 @@ func NewClient(ctx context.Context, opts ...ClientOption) (_ *Client, err error) c.releaseCredentials = resolved.Close c.credentialsLockErr = resolved.MemoryLockError owned = true - out, err := c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.cipherInit, buf(encoded)) }) if err != nil { @@ -217,7 +230,7 @@ func consumeUnresolved(creds Credentials) { // instance, not the client: a cleanup whose argument reaches its object // keeps that object alive forever. func newClient(inst *instance, t *transport) *Client { - c := &Client{inst: inst, transport: t} + c := &Client{inst: inst, transport: t, stuck: guest.StuckAfter} c.cleanup = runtime.AddCleanup(c, func(inst *instance) { _ = inst.release() }, inst) return c } @@ -323,8 +336,8 @@ func encodeConfig(cfg initConfig) ([]byte, error) { func (c *Client) Close() error { c.mu.Lock() defer c.mu.Unlock() - // Idempotency turns on the runtime, not on the client: a client an - // interrupted call already closed has never released its runtime. + // Idempotency turns on the runtime, not on the client: a client a + // stuck call already closed has never released its runtime. if c.released { return nil } @@ -368,7 +381,7 @@ func (c *Client) resolveKeyset(ctx context.Context, sel KeysetSelector) (KeysetI if err != nil { return KeysetID{}, err } - out, err := c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.keyset, buf(encoded)) }) if err != nil { @@ -382,18 +395,27 @@ func (c *Client) resolveKeyset(ctx context.Context, sel KeysetSelector) (KeysetI return id, nil } -// call runs f on the instance under the client's lock. +// call runs f on the instance under the client's lock. f must run its +// guest call under the context it is given, never the caller's. // -// A call interrupted by its context (the runtime closes the module on a -// deadline or cancellation, see newInstance) leaves the instance closed: -// its key material is gone with its memory and no further call can run. -// The client is then closed, so later calls are ErrState rather than a -// runtime error, and Close releases the runtime without a shutdown call. -// A guest that trapped is closed the same way, by this method: the guest -// builds with panic-as-abort, so a trap is an abort mid-export, after -// which its state is unknown and its keys are better wiped than reused. -func (c *Client) call(ctx context.Context, f func(*instance) ([]byte, error)) ([]byte, error) { - c.mu.Lock() +// The caller's context bounds the wait for the lock, and is refused if it +// has already ended, so a request that timed out in the queue never +// reaches the guest. The guest then runs under guest.Detach: the caller's +// context reaches the host functions, where it cancels a ZeroKMS request +// or a token fetch, and the guest returns through its ordinary error path +// with the client intact. The error matches the context's error too. +// +// The client is closed, so later calls are ErrState rather than a runtime +// error, when the guest's state is unknown. A guest that trapped is one: +// it builds with panic-as-abort, so a trap is an abort mid-export, after +// which its keys are better wiped than reused. A guest still running +// c.stuck after its caller's context ended is the other: Detach's context +// ends, the runtime closes the module (WithCloseOnContextDone, see +// newInstance), and its key material goes with its memory. +func (c *Client) call(ctx context.Context, f func(context.Context, *instance) ([]byte, error)) ([]byte, error) { + if err := c.mu.Acquire(ctx); err != nil { + return nil, err + } defer c.mu.Unlock() if c.closed || c.inst.module.IsClosed() { c.closed = true @@ -403,14 +425,16 @@ func (c *Client) call(ctx context.Context, f func(*instance) ([]byte, error)) ([ if c.transport != nil { c.transport.tokenErr = nil } - out, err := f(c.inst) + run, done := guest.Detach(ctx, c.stuck) + out, err := f(run, c.inst) + done() switch { case c.inst.module.IsClosed(): c.closed = true if err == nil { err = ErrState } - err = fmt.Errorf("%w: interrupted call closed the client", err) + err = fmt.Errorf("%w: the guest ran on after its caller's context ended, and the client is closed", err) case errors.Is(err, errGuestTrap): // No guest code is running (f has returned), so the close is // immediate: the module's memory is wiped and freed here. @@ -438,8 +462,23 @@ func (c *Client) call(ctx context.Context, f func(*instance) ([]byte, error)) ([ err = fmt.Errorf("%w (token source: %w)", err, c.transport.tokenErr) } } + // A call that failed after its context ended failed because of it, or + // as good as: the caller asks errors.Is(err, context.Canceled) or + // DeadlineExceeded, and the cause (ErrTransport for a cancelled + // request) stays matchable too. if err != nil { + if ctxErr := ctx.Err(); ctxErr != nil && !errors.Is(err, ctxErr) { + err = fmt.Errorf("%w: %w", ctxErr, err) + } return nil, err } + // A call that succeeded after its context ended fails too: work the + // guest does without the network (a query term, a batch under a + // cached key) runs on through a cancellation, and the caller was + // promised the context's error. The output nobody will read is wiped. + if ctxErr := ctx.Err(); ctxErr != nil { + wipe(out) + return nil, ctxErr + } return out, nil } diff --git a/languages/golang/encrypt/diagnostic_test.go b/languages/golang/encrypt/diagnostic_test.go index 811496d44..c632bde94 100644 --- a/languages/golang/encrypt/diagnostic_test.go +++ b/languages/golang/encrypt/diagnostic_test.go @@ -36,7 +36,7 @@ func wantCode(t *testing.T, err error, code string) { func TestGuestRefusalsCarryTheirDiagnostic(t *testing.T) { ctx := context.Background() c := rawInstance(t) - _, err := c.call(ctx, func(inst *instance) ([]byte, error) { + _, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.planCheck, buf([]byte{0xff})) }) if !errors.Is(err, ErrEncoding) { @@ -48,7 +48,7 @@ func TestGuestRefusalsCarryTheirDiagnostic(t *testing.T) { if err != nil { t.Fatal(err) } - _, err = c.call(ctx, func(inst *instance) ([]byte, error) { + _, err = c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.keyset, buf(selector)) }) if !errors.Is(err, ErrState) { @@ -65,7 +65,7 @@ func TestAGuestWithoutLastErrorFailsWithTheBareSentinel(t *testing.T) { ctx := context.Background() c := rawInstance(t) c.inst.exports.LastError = nil - _, err := c.call(ctx, func(inst *instance) ([]byte, error) { + _, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.planCheck, buf([]byte{0xff})) }) if err != ErrEncoding { diff --git a/languages/golang/encrypt/doc.go b/languages/golang/encrypt/doc.go index 6fc832d00..8254579ff 100644 --- a/languages/golang/encrypt/doc.go +++ b/languages/golang/encrypt/doc.go @@ -184,6 +184,23 @@ // memory until the garbage collector reclaims it; a MatchQuery never prints // it. // +// # Cancellation +// +// Every call takes a [context.Context], and a call whose context is +// cancelled or passes its deadline fails with that context's error. Only +// that call fails: one Client is shared by every request in a service, and +// one request giving up never closes it for the others. A call waiting +// for the client, which runs one call at a time, leaves the queue when its +// context ends and never starts. A call waiting on ZeroKMS, or on its auth +// strategy for a token, has that request cancelled; its error matches both +// the context's error and [ErrTransport]. +// +// A Client closes itself, and every later call returns [ErrState], only +// when the guest's state is unknown: the guest trapped, or a call ran on +// for 30 seconds after its context ended, which means a bug, not a caller +// that gave up. A service that sees ErrState from a client it did not +// close builds a new one. +// // # Transport and auth // // The guest imports exactly two host functions: an HTTP send, served by any @@ -216,7 +233,8 @@ // // The generator and the compiler find a mistake in a declaration, so no call // returns an error for one. A call returns an error for a key, for the -// network, or for stored data: [ErrForeignKeyset] when a *Cipher is given a +// network, or for stored data, or because its context ended (see +// Cancellation): [ErrForeignKeyset] when a *Cipher is given a // record another keyset sealed, [ErrAuthentication] or [ErrForbidden] for a // ciphertext that does not open under its field's context, [ErrEncoding] for // a stored value that does not fit its declaration. Read the kind with diff --git a/languages/golang/encrypt/export_test.go b/languages/golang/encrypt/export_test.go index 8fd779c95..753d1a196 100644 --- a/languages/golang/encrypt/export_test.go +++ b/languages/golang/encrypt/export_test.go @@ -118,7 +118,7 @@ func deterministicClient(ctx context.Context, wasm []byte, seed [32]byte) (*Clie _ = c.Close() return nil, err } - out, err := c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.cipherInit, buf(encoded)) }) if err != nil { diff --git a/languages/golang/encrypt/guest.go b/languages/golang/encrypt/guest.go index def93995e..2cffd3be2 100644 --- a/languages/golang/encrypt/guest.go +++ b/languages/golang/encrypt/guest.go @@ -113,10 +113,13 @@ func guestModuleConfig() wazero.ModuleConfig { // strict policy, memory that cannot be locked fails instantiation with // ErrMemoryLock. func newInstance(ctx context.Context, wasm []byte, t *transport, policy guest.LockPolicy) (*instance, error) { - // WithCloseOnContextDone lets a caller's deadline or cancellation - // interrupt an in-flight guest call — which otherwise holds the Client's - // lock against every other user. An interrupted call closes the module, - // so the Client is done afterwards; the alternative is a wedged process. + // WithCloseOnContextDone closes the module when a call's context ends. + // A guest call never runs under the caller's context: Client.call runs + // it under guest.Detach, whose context ends only when the guest has run + // on long after its caller gave up. Closing the module then frees the + // Client's lock, which a stuck call would otherwise hold against every + // other user, at the cost of the Client: the alternative is a wedged + // process. config := wazero.NewRuntimeConfig(). WithCompilationCache(compilationCache()). WithCloseOnContextDone(true) @@ -196,8 +199,8 @@ func newInstance(ctx context.Context, wasm []byte, t *transport, policy guest.Lo // release runs the guest's shutdown — the client key and every loaded // index key wiped inside the instance — and closes the runtime, which // frees the linear memory through the allocator's wipe. It is what Close -// does, and what the cleanup on an unreachable Client does. A module an -// interrupted call or a trap already closed cannot run se_shutdown; the +// does, and what the cleanup on an unreachable Client does. A module a +// stuck call or a trap already closed cannot run se_shutdown; the // runtime close still wipes and frees its memory, so nothing is left // behind either way. func (inst *instance) release() error { diff --git a/languages/golang/encrypt/guest_test.go b/languages/golang/encrypt/guest_test.go index a98266f3b..9981d753e 100644 --- a/languages/golang/encrypt/guest_test.go +++ b/languages/golang/encrypt/guest_test.go @@ -15,13 +15,11 @@ import ( "strconv" "strings" "testing" - "time" "github.com/cipherstash/stack/languages/golang/auth" "github.com/cipherstash/stack/languages/golang/internal/guest" "github.com/cipherstash/stack/languages/golang/internal/record" "github.com/tetratelabs/wazero" - "github.com/tetratelabs/wazero/sys" ) // Tests that drive the embedded guest without a live ZeroKMS. What they @@ -278,53 +276,6 @@ type roundTripFunc func(*http.Request) (*http.Response, error) func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } -// An interrupted call closes the module (WithCloseOnContextDone); the -// client must then be closed rather than a wedge or a runtime error. -func TestInterruptedCallClosesTheClient(t *testing.T) { - t.Run("deadline during a request", func(t *testing.T) { - guestOrSkip(t) - cfg := testConfig("http://zerokms.invalid") - cfg = append(cfg, WithTransport(roundTripFunc(func(r *http.Request) (*http.Response, error) { - <-r.Context().Done() - return nil, r.Context().Err() - }))) - ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) - defer cancel() - _, err := NewClient(ctx, cfg...) - if !errors.Is(err, context.DeadlineExceeded) { - t.Fatalf("NewClient: %v, want the deadline", err) - } - }) - t.Run("closed module is ErrState", func(t *testing.T) { - ctx := context.Background() - c := rawInstance(t) - // What the runtime does to the module when a call's context ends. - if err := c.inst.module.CloseWithExitCode(ctx, sys.ExitCodeContextCanceled); err != nil { - t.Fatal(err) - } - if _, err := c.Keyset(KeysetName("k")).KeysetID(ctx); !errors.Is(err, ErrState) { - t.Fatalf("Keyset on a closed module: %v, want ErrState", err) - } - if err := c.Close(); err != nil { - t.Fatalf("Close after interruption: %v", err) - } - // The close must reach the runtime. An interrupted call closes the - // module and marks the client closed; a Close that treated that as - // "already done" would leave the runtime and the host module it - // carries allocated for the life of the process. - if !c.released { - t.Error("Close after interruption left the runtime unreleased") - } - if m := c.inst.runtime.Module(transportModule); m != nil { - t.Errorf("host module %s is still registered after Close", transportModule) - } - // Still idempotent. - if err := c.Close(); err != nil { - t.Fatalf("second Close: %v", err) - } - }) -} - // A response the host would have to buffer without bound is refused as a // transport failure, whether the size is announced or streamed. func TestOversizedResponseIsTransport(t *testing.T) { diff --git a/languages/golang/encrypt/memory_test.go b/languages/golang/encrypt/memory_test.go index 6be1561b5..2ab6becd4 100644 --- a/languages/golang/encrypt/memory_test.go +++ b/languages/golang/encrypt/memory_test.go @@ -202,7 +202,7 @@ func TestRequireLockedMemoryFailsTheCallThatCannotGrow(t *testing.T) { ctx := context.Background() c := strictClient(t) refusing := guest.RefuseGrowth(c.inst.mem, errors.New("refused for the test")) - stage := func(inst *instance) ([]byte, error) { return nil, stageLarge(ctx, inst) } + stage := func(ctx context.Context, inst *instance) ([]byte, error) { return nil, stageLarge(ctx, inst) } _, err := c.call(ctx, stage) if !errors.Is(err, ErrMemoryLock) || !strings.Contains(err.Error(), "growth refused") || !strings.Contains(err.Error(), refusing.Reason().Error()) { t.Fatalf("call needing a refused growth: %v; want ErrMemoryLock naming the refusal", err) @@ -247,7 +247,7 @@ func TestRequireLockedMemoryClosesTheClientOnARefusedInternalGrowth(t *testing.T t.Fatal(err) } var refusing *guest.Refusing - _, err = c.call(ctx, func(inst *instance) ([]byte, error) { + _, err = c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { staged, err := inst.exports.AllocWrite(ctx, inst.module, encoded) if err != nil { return nil, err @@ -268,7 +268,7 @@ func TestRequireLockedMemoryClosesTheClientOnARefusedInternalGrowth(t *testing.T if !c.inst.mem.IsFreed() { t.Fatal("the closed client's memory was not wiped and freed") } - if _, err := c.call(ctx, func(*instance) ([]byte, error) { return nil, nil }); !errors.Is(err, ErrState) { + if _, err := c.call(ctx, func(context.Context, *instance) ([]byte, error) { return nil, nil }); !errors.Is(err, ErrState) { t.Fatalf("a call after the abort: %v, want ErrState", err) } if !c.MemoryLocked() { diff --git a/languages/golang/encrypt/query.go b/languages/golang/encrypt/query.go index b98e76cb5..d4953b4cf 100644 --- a/languages/golang/encrypt/query.go +++ b/languages/golang/encrypt/query.go @@ -131,7 +131,7 @@ func (cph *Cipher) QueryTerm(ctx context.Context, plan *record.Plan, field, oper if err != nil { return QueryTerm{}, err } - out, err := cph.client.call(ctx, func(inst *instance) ([]byte, error) { + out, err := cph.client.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.queryTerm, buf(encodedInput), buf(encodedPlan), buf([]byte(field)), buf([]byte(operator)), buf(opts)) }) if err != nil { @@ -244,7 +244,7 @@ func (cph *Cipher) Verify(ctx context.Context, plan *record.Plan, field, operato if err != nil { return nil, fmt.Errorf("%w: %v", ErrEncoding, err) } - out, err := cph.client.call(ctx, func(inst *instance) ([]byte, error) { + out, err := cph.client.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.verify, buf(encodedInput), buf(encodedValues), buf(encodedPlan), buf([]byte(field)), buf([]byte(operator))) }) if err != nil { @@ -287,7 +287,7 @@ func (cph *Cipher) TargetLimit(ctx context.Context, plan *record.Plan, field, op if err != nil { return Limit{}, fmt.Errorf("%w: %v", ErrEncoding, err) } - out, err := cph.client.call(ctx, func(inst *instance) ([]byte, error) { + out, err := cph.client.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.limit, buf(term), buf(encodedStats), buf(encodedPlan), buf([]byte(field)), buf([]byte(operator)), scalar(want)) }) if err != nil { diff --git a/languages/golang/encrypt/records.go b/languages/golang/encrypt/records.go index 7a4748bbc..b2d0ea745 100644 --- a/languages/golang/encrypt/records.go +++ b/languages/golang/encrypt/records.go @@ -65,7 +65,7 @@ func (cph *Cipher) Seal(ctx context.Context, plan *record.Plan, rows []record.So if err != nil { return nil, err } - out, err := cph.client.call(ctx, func(inst *instance) ([]byte, error) { + out, err := cph.client.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.encryptRecord, buf(encodedSource), buf(encodedPlan), buf(opts)) }) if errors.Is(err, ErrTerm) { @@ -154,7 +154,7 @@ func (cph *Cipher) derive(ctx context.Context, plan *record.Plan, field string, if err != nil { return nil, err } - return cph.client.call(ctx, func(inst *instance) ([]byte, error) { + return cph.client.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.term, buf(encodedValue), buf(encodedContext), scalar(uint64(code)), buf(opts)) }) } @@ -359,9 +359,9 @@ var errNoContext = errors.New("encrypt: the record has no value for its context var errNoEQL = errors.New("encrypt: the record has no EQL value for a field that names an EQL type") // open decrypts records under the keyset the selector names, and, when -// context is not "", refuses a record whose stored context field names +// expected is not "", refuses a record whose stored context field names // another context before any key is retrieved. -func (c *Client) open(ctx context.Context, sel KeysetSelector, context string, p *record.Plan, records []record.Sealed) ([]record.Source, error) { +func (c *Client) open(ctx context.Context, sel KeysetSelector, expected string, p *record.Plan, records []record.Sealed) ([]record.Source, error) { trees := make([]any, len(records)) for i, rec := range records { tree := make(map[string]any, len(p.Fields)+1) @@ -404,11 +404,11 @@ func (c *Client) open(ctx context.Context, sel KeysetSelector, context string, p if err != nil { return nil, fmt.Errorf("%w: %v", ErrEncoding, err) } - opts, err := vcffi.Marshal(options(sel, context)) + opts, err := vcffi.Marshal(options(sel, expected)) if err != nil { return nil, err } - out, err := c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.decryptRecord, buf(encodedTree), buf(encodedPlan), buf(opts)) }) if err != nil { diff --git a/languages/golang/encrypt/transport.go b/languages/golang/encrypt/transport.go index 85e349555..2e1005b44 100644 --- a/languages/golang/encrypt/transport.go +++ b/languages/golang/encrypt/transport.go @@ -14,6 +14,8 @@ import ( "github.com/tetratelabs/wazero" "github.com/tetratelabs/wazero/api" + + "github.com/cipherstash/stack/languages/golang/internal/guest" ) // transportModule is the name of the guest's one import module. Its two @@ -117,7 +119,10 @@ func (t *transport) perform(ctx context.Context, mem api.Memory, // included, and a wipe racing that send would put a truncated or // zeroed request on the wire. reqBody := newRequestBody(body) - req, err := http.NewRequestWithContext(ctx, string(method), string(url), reqBody) + // The request runs under the caller's context, not the call's: the + // caller's cancellation ends the request, and the guest sees a + // transport failure it returns through its ordinary error path. + req, err := http.NewRequestWithContext(guest.Caller(ctx), string(method), string(url), reqBody) if err != nil { _ = reqBody.Close() return transportFailed, nil, []byte(err.Error()) @@ -212,7 +217,9 @@ func (b *requestBody) Close() error { // tokenGet is token_get: hand the guest the current bearer token. func (t *transport) tokenGet(ctx context.Context, m api.Module, tokenPtrOut, tokenLenOut uint32) int32 { - token, err := t.token.Token(ctx) + // The caller's context, as for a request: a token fetch or refresh the + // caller gave up on ends, and the client stays open. + token, err := t.token.Token(guest.Caller(ctx)) if err != nil { t.tokenErr = err return hostFailed diff --git a/languages/golang/internal/guest/errors.go b/languages/golang/internal/guest/errors.go index 347ec4609..aec29155f 100644 --- a/languages/golang/internal/guest/errors.go +++ b/languages/golang/internal/guest/errors.go @@ -19,7 +19,7 @@ var ( // it. ErrEncoding = errors.New("cipherstash: malformed input") // ErrState is a call on something that has been closed: a client, a - // store, or an instance an interrupted call took down. + // store, or an instance a trap or a stuck call took down. ErrState = errors.New("cipherstash: closed") // ErrInternal is a guest panic or any other unexpected guest failure. ErrInternal = errors.New("cipherstash: internal guest failure") diff --git a/languages/golang/internal/guest/lock.go b/languages/golang/internal/guest/lock.go new file mode 100644 index 000000000..5452837c0 --- /dev/null +++ b/languages/golang/internal/guest/lock.go @@ -0,0 +1,117 @@ +package guest + +import ( + "context" + "sync" + "time" +) + +// What a caller's context does to a guest call. A guest instance is shared: +// an encrypt.Client or an auth profile serves every request in a process, +// so one caller's cancellation must end that caller's call and nothing +// else. Three pieces make that so, and every guest package uses all three. +// +// - The wait for the instance ends with the caller's context ([Lock]). A +// request whose deadline passes in the queue leaves with the context's +// error and never reaches the guest. +// - The guest runs under a context the caller cannot end ([Detach]). The +// runtime closes a module whose call context ends (the runtimes are +// built with WithCloseOnContextDone), and a closed module is a client +// gone for everyone. The caller's context still reaches the host +// functions ([Caller]), so a ZeroKMS request or a token fetch in flight +// is cancelled, fails as a transport failure, and the guest unwinds +// through its ordinary error path with its keys intact. +// - A guest that has not returned [StuckAfter] after its caller's context +// ended is closed: the call context Detach returns ends then. That is +// the backstop for a guest or host function that ignores cancellation, +// which is a bug, never a closed browser tab. + +// StuckAfter is how long a guest call may go on after its caller's context +// ends before the guest is treated as stuck and closed. Once the network +// wait is cancelled, what remains of a call is the guest's own work, which +// takes milliseconds; the margin is wide so that a large batch cancelled +// mid-way still finishes and leaves the instance open. +const StuckAfter = 30 * time.Second + +// Lock is a guest instance's call lock: one call at a time, as a wasm +// instance is single-threaded, with a wait that ends with the caller's +// context. The zero value is unlocked. +type Lock struct { + once sync.Once + ch chan struct{} +} + +func (l *Lock) slot() chan struct{} { + l.once.Do(func() { l.ch = make(chan struct{}, 1) }) + return l.ch +} + +// Acquire takes the lock, or returns ctx's error if ctx ends first. A +// context that has already ended is refused even when the lock is free, +// so a call that would start with nothing to wait for never starts. +func (l *Lock) Acquire(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + select { + case l.slot() <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + } + // Both cases can be ready at once, and select picks either: a context + // that ended while the lock was free must not run the call. + if err := ctx.Err(); err != nil { + l.Unlock() + return err + } + return nil +} + +// Lock takes the lock with no way out, for Close, which must not be +// skippable. +func (l *Lock) Lock() { l.slot() <- struct{}{} } + +// Unlock releases the lock. +func (l *Lock) Unlock() { <-l.slot() } + +// callerKey carries the caller's context inside the one Detach returns. +type callerKey struct{} + +// Detach returns the context a guest call runs under: ctx's values, but not +// its cancellation, so the runtime does not close the module when the +// caller gives up. It ends stuck after ctx does, if the call has not +// finished by then, and the runtime then closes the module. done must be +// called when the call returns; it stops the backstop. +func Detach(ctx context.Context, stuck time.Duration) (run context.Context, done func()) { + run, cancel := context.WithCancel(context.WithValue(context.WithoutCancel(ctx), callerKey{}, ctx)) + var mu sync.Mutex + var timer *time.Timer + finished := false + stop := context.AfterFunc(ctx, func() { + mu.Lock() + defer mu.Unlock() + if !finished { + timer = time.AfterFunc(stuck, cancel) + } + }) + return run, func() { + stop() + mu.Lock() + finished = true + if timer != nil { + timer.Stop() + } + mu.Unlock() + cancel() + } +} + +// Caller is the caller's context inside a host function: what a request to +// the network or a token source must honour, so the caller's cancellation +// reaches it. Outside a call Detach began, it is ctx itself. +func Caller(ctx context.Context) context.Context { + if caller, ok := ctx.Value(callerKey{}).(context.Context); ok { + return caller + } + return ctx +} diff --git a/languages/golang/internal/guest/lock_test.go b/languages/golang/internal/guest/lock_test.go new file mode 100644 index 000000000..d4ab83ec7 --- /dev/null +++ b/languages/golang/internal/guest/lock_test.go @@ -0,0 +1,82 @@ +package guest + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestLockRefusesAnEndedContextEvenWhenFree(t *testing.T) { + var l Lock + ctx, cancel := context.WithCancel(context.Background()) + cancel() + for range 100 { + if err := l.Acquire(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("Acquire with an ended context: %v, want context.Canceled", err) + } + } + // And the refusals left it free. + if err := l.Acquire(context.Background()); err != nil { + t.Fatal(err) + } + l.Unlock() +} + +func TestLockWaitEndsWithTheContext(t *testing.T) { + var l Lock + l.Lock() + defer l.Unlock() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + if err := l.Acquire(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Acquire on a held lock: %v, want the deadline", err) + } +} + +type valueKey struct{} + +func TestDetachKeepsValuesNotCancellation(t *testing.T) { + caller, cancel := context.WithCancel(context.WithValue(context.Background(), valueKey{}, "v")) + run, done := Detach(caller, time.Hour) + defer done() + cancel() + if run.Err() != nil { + t.Fatal("the caller's cancellation reached the call's context") + } + if run.Value(valueKey{}) != "v" { + t.Error("the caller's values did not reach the call's context") + } + if Caller(run) != caller { + t.Error("Caller does not give back the caller's context") + } + if Caller(caller) != caller { + t.Error("Caller outside a detached call is not the context itself") + } +} + +func TestDetachEndsStuckAfterTheCaller(t *testing.T) { + caller, cancel := context.WithCancel(context.Background()) + run, done := Detach(caller, 10*time.Millisecond) + defer done() + cancel() + select { + case <-run.Done(): + case <-time.After(5 * time.Second): + t.Fatal("the call's context did not end after the caller's") + } +} + +// A call that finished before its caller's context ended keeps nothing +// running: the context the call ran under has ended by done, not by a timer. +func TestDetachDoneEndsTheCall(t *testing.T) { + caller, cancel := context.WithCancel(context.Background()) + defer cancel() + run, done := Detach(caller, time.Hour) + done() + select { + case <-run.Done(): + default: + t.Fatal("done left the call's context running") + } +} From 94f995d18cd57f8f6ab3973e69dd592dd048f65e Mon Sep 17 00:00:00 2001 From: Dan Draper Date: Sun, 18 Oct 2026 21:41:15 +1100 Subject: [PATCH 2/2] fix(golang): finish a device session's refresh in the background after its caller gives up, so a rotated refresh token is kept The token service rotates the refresh token when it answers a device session's refresh. A refresh cancelled with its caller after the service answered dropped the new token, and the next refresh sent one the service had retired, which it rejects: the strategy could not refresh again without a new login. This is the failure stack-auth's native refresh now avoids by finishing the refresh in a task after its caller leaves (#1172); the Go auth guest runs that refresh inline in WebAssembly, so the Go host has to do the same. Strategy.Token now runs its work on a goroutine that holds the strategy's lock until it is done, and returns to a caller whose context ends at once, without making a copy of a token nobody will read. The refresh call's context is marked, and only its requests run under guest.Detach of the caller's context with a grace of 10 seconds: the refresh finishes, the guest saves what came back, and the next Token reads it. Past the grace it fails as a transport failure, well inside the 30-second backstop, so the profile stays open. Close waits for such a refresh, so its rotated token is saved. Every other request to the token service, an access-key or OIDC exchange, rotates nothing and is still cancelled with its caller, as is a ZeroKMS request, so a cancelled NewClient or a shutdown does not wait on them. Refs #1169. --- languages/golang/CHANGELOG.md | 12 +- languages/golang/auth/cancel_test.go | 139 ++++++++++++++++++++++-- languages/golang/auth/doc.go | 11 +- languages/golang/auth/store.go | 7 +- languages/golang/auth/strategy.go | 50 +++++++-- languages/golang/auth/transport.go | 47 +++++++- languages/golang/encrypt/README.md | 10 +- languages/golang/encrypt/cancel_test.go | 8 +- languages/golang/encrypt/doc.go | 10 +- 9 files changed, 254 insertions(+), 40 deletions(-) diff --git a/languages/golang/CHANGELOG.md b/languages/golang/CHANGELOG.md index 5dc6c0f3e..edf47afb5 100644 --- a/languages/golang/CHANGELOG.md +++ b/languages/golang/CHANGELOG.md @@ -82,11 +82,13 @@ listed under **Breaking**. Releases are tagged `languages/golang/vX.Y.Z`. or the `auth` profile behind its strategy, for every later caller, and a request whose deadline passed while it waited for the client closed it before running. Now a call waiting for the client leaves the queue with - the context's error; a ZeroKMS request or a token fetch in flight is - cancelled, and the error matches both the context's error and - `ErrTransport`; and the client stays open. A client or profile still - closes itself (`ErrState`) when the guest traps, or when a call runs on - for 30 seconds after its context ended. + the context's error; a ZeroKMS request in flight is cancelled, and the + error matches both the context's error and `ErrTransport`; a device + session's refresh in flight finishes in the background, for up to 10 + seconds, so a rotated refresh token is never lost; and the client stays + open. A client or profile still closes itself (`ErrState`) when the + guest traps, or when a call runs on for 30 seconds after its context + ended. - **stashgen refuses a match index on an index-only field** (`stash:"name,index=match"` with no `encrypt`). A match query's rows are checked by decrypting them, and such a field stores no ciphertext. diff --git a/languages/golang/auth/cancel_test.go b/languages/golang/auth/cancel_test.go index dca54c7e1..1dda26166 100644 --- a/languages/golang/auth/cancel_test.go +++ b/languages/golang/auth/cancel_test.go @@ -2,12 +2,15 @@ package auth import ( "context" + "encoding/json" "errors" "fmt" "io" "net/http" "net/http/httptest" - "strings" + "os" + "path/filepath" + "sync" "sync/atomic" "testing" "time" @@ -25,17 +28,29 @@ import ( func tokenServer(t *testing.T, hold *atomic.Bool) *httptest.Server { t.Helper() var jwt string - // A request the client never cancels is let go at the end, so a - // failure reports itself rather than hanging the server's Close. + // The server notices a client that hung up a moment after the client + // has gone: it reads the closed connection on a goroutine of its own. + // So the test's end waits for each held request to see its + // cancellation, for a bounded time, before the server is closed (which + // would end the request's context itself and hide a request the client + // never cancelled). + var held sync.WaitGroup done := make(chan struct{}) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if hold.Load() { + held.Add(1) + defer held.Done() // The server notices a client that hung up only once the body // is read. _, _ = io.Copy(io.Discard, r.Body) select { case <-r.Context().Done(): + return case <-done: + } + select { + case <-r.Context().Done(): + case <-time.After(5 * time.Second): t.Error("a held request was never cancelled") } return @@ -43,7 +58,10 @@ func tokenServer(t *testing.T, hold *atomic.Bool) *httptest.Server { fmt.Fprintf(w, `{"accessToken":%q,"expiry":%d}`, jwt, time.Now().Add(time.Hour).Unix()) })) t.Cleanup(server.Close) - t.Cleanup(func() { close(done) }) + t.Cleanup(func() { + close(done) + held.Wait() + }) jwt = testJWT(t, server.URL) return server } @@ -69,13 +87,114 @@ func TestATokenFetchCancelledFailsOnlyItself(t *testing.T) { if _, err := strategy.Token(ctx); !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("cancelled Token: %v, want the deadline", err) } - if profile.root.closed || profile.root.inst.module.IsClosed() { - t.Fatal("the profile was closed") - } hold.Store(false) + // The next Token waits for the abandoned one to finish. if token, err := strategy.Token(context.Background()); err != nil || token == "" { t.Fatalf("the next Token: %q, %v", token, err) } + if profile.root.closed || profile.root.inst.module.IsClosed() { + t.Fatal("the profile was closed") + } +} + +// An access-key exchange rotates nothing, so it ends with its caller: the +// held request is cancelled at the caller's deadline, not run on for the +// refresh grace, and Close does not wait for it. +func TestAnAccessKeyExchangeEndsWithItsCaller(t *testing.T) { + guestOrSkip(t) + var hold atomic.Bool + server := tokenServer(t, &hold) + profile, err := Open(context.Background(), t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + strategy, err := profile.AccessKey(context.Background(), testCRN, "CSAKtestKeyId.testKeySecret", WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + + hold.Store(true) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + if _, err := strategy.Token(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("cancelled Token: %v, want the deadline", err) + } + // The grace is 10 seconds; a Close that waited for it would take that. + start := time.Now() + if err := strategy.Close(); err != nil { + t.Fatal(err) + } + if waited := time.Since(start); waited > 2*time.Second { + t.Fatalf("Close took %v: the exchange ran on past its caller", waited) + } +} + +// A device session's refresh is the request that rotates a credential: the +// service answers with a new refresh token and retires the old one. A +// caller that gives up returns at once, but the refresh runs on, and what +// it brings is saved. The next Token is served from it with no second +// request, and the profile on disk holds the rotated refresh token. +func TestARefreshAnsweredAfterItsCallerGaveUpIsKept(t *testing.T) { + guestOrSkip(t) + dir, _ := expiredDeviceProfile(t) + var requests atomic.Int32 + answered := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if requests.Add(1) > 1 { + // The old refresh token is retired once the service answers. + w.WriteHeader(http.StatusBadRequest) + fmt.Fprint(w, `{"error":"invalid_grant"}`) + return + } + _, _ = io.Copy(io.Discard, r.Body) + time.Sleep(150 * time.Millisecond) + fmt.Fprint(w, `{"access_token":"fresh","token_type":"Bearer","expires_in":3600,"refresh_token":"refresh-2"}`) + close(answered) + })) + t.Cleanup(server.Close) + profile, err := Open(context.Background(), dir) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + ws, err := profile.WorkspaceStore(context.Background(), wsA) + if err != nil { + t.Fatal(err) + } + strategy, err := ws.DeviceSession(context.Background(), WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + if _, err := strategy.Token(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("cancelled Token: %v, want the deadline", err) + } + select { + case <-answered: + t.Fatal("the cancelled Token waited for the service's answer") + default: + } + if token, err := strategy.Token(context.Background()); err != nil || token != "fresh" { + t.Fatalf("the next Token: %q, %v, want the refreshed token", token, err) + } + if n := requests.Load(); n != 1 { + t.Fatalf("%d refresh requests, want 1: the late answer was dropped", n) + } + data, err := os.ReadFile(filepath.Join(dir, "workspaces", wsA, "auth.json")) + if err != nil { + t.Fatal(err) + } + var stored map[string]any + if err := json.Unmarshal(data, &stored); err != nil { + t.Fatal(err) + } + if stored["refresh_token"] != "refresh-2" { + t.Fatalf("stored refresh token = %v, want the rotated one", stored["refresh_token"]) + } } // A refresh runs on after its caller gives up, so the guest call under it @@ -155,7 +274,7 @@ func TestATokenRequestWhoseContextEndsInTheQueueNeverReachesTheGuest(t *testing. } // The backstop: a call still running long after its caller gave up closes -// the profile, and says so. +// the profile. The caller has returned by then; the next call is ErrState. func TestATokenFetchThatRunsOnAfterItsContextEndsClosesTheProfile(t *testing.T) { guestOrSkip(t) profile, err := Open(context.Background(), t.TempDir(), WithRoundTripper(roundTripFunc(func(*http.Request) (*http.Response, error) { @@ -175,8 +294,8 @@ func TestATokenFetchThatRunsOnAfterItsContextEndsClosesTheProfile(t *testing.T) profile.root.stuck = 20 * time.Millisecond ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) defer cancel() - if _, err := strategy.Token(ctx); err == nil || !strings.Contains(err.Error(), "the profile is closed") { - t.Fatalf("stuck Token: %v, want the profile closed", err) + if _, err := strategy.Token(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("stuck Token: %v, want the deadline", err) } if _, err := strategy.Token(context.Background()); !errors.Is(err, ErrState) { t.Fatalf("a Token after the close: %v, want ErrState", err) diff --git a/languages/golang/auth/doc.go b/languages/golang/auth/doc.go index de9d1bec9..d61e99371 100644 --- a/languages/golang/auth/doc.go +++ b/languages/golang/auth/doc.go @@ -42,9 +42,14 @@ // A call whose context ends, [Strategy.Token] included, fails with the // context's error and leaves the store and its strategies open: a request // cancelled while its token is exchanged or refreshed does not stop the -// encrypt clients that read tokens from the same store. The store closes -// itself, and every later call is [ErrState], only when its guest traps or -// a call runs on for 30 seconds after its context ended. +// encrypt clients that read tokens from the same store. A request to the +// token service is cancelled with its caller, except a device session's +// refresh: the service rotates the refresh token when it answers one, so +// the caller returns at once while the refresh runs on, for up to 10 +// seconds, and the store keeps what it sends back for the next call. +// [Strategy.Close] waits for such a refresh. The store closes itself, and +// every later call is [ErrState], only when its guest traps or a call runs +// on for 30 seconds after its context ended. // // # Why a second guest // diff --git a/languages/golang/auth/store.go b/languages/golang/auth/store.go index 0421c0f1e..23b9b5915 100644 --- a/languages/golang/auth/store.go +++ b/languages/golang/auth/store.go @@ -57,7 +57,9 @@ func RequireLockedMemory() Option { // A call whose context ends fails with the context's error, and only that // call: a request cancelled while its token is fetched or refreshed never // closes the profile, or the strategies and encrypt clients that read -// tokens from it. The profile closes itself, and every later call is +// tokens from it. A device session's refresh already sent finishes in the +// background, for up to 10 seconds, so a rotated refresh token is never +// lost (see refreshGrace). The profile closes itself, and every later call is // [ErrState], only when the guest's state is unknown: it trapped, or it // went on running long after its caller gave up, which is a bug. type ProfileStore struct { @@ -235,7 +237,8 @@ func (s *ProfileStore) call(ctx context.Context, fn export, args ...string) ([]b // // The caller's context bounds the wait for the lock, as encrypt's // Client.call does, and the guest runs under guest.Detach, so a caller -// that gives up cancels its token request without closing the profile. +// that gives up never closes the profile. A refresh request runs on for +// refreshGrace, so the guest keeps the rotated token the service sends. func (s *ProfileStore) callArgs(ctx context.Context, fn export, args ...guest.Arg) ([]byte, error) { r := s.root if err := r.mu.Acquire(ctx); err != nil { diff --git a/languages/golang/auth/strategy.go b/languages/golang/auth/strategy.go index a5fcb924f..51379cbdf 100644 --- a/languages/golang/auth/strategy.go +++ b/languages/golang/auth/strategy.go @@ -186,27 +186,58 @@ func (s *Strategy) Token(ctx context.Context) (string, error) { if err := s.mu.Acquire(ctx); err != nil { return "", err } - defer s.mu.Unlock() if s.closed { + s.mu.Unlock() return "", ErrState } - ctx, status := withAuthHTTPStatus(ctx) - token, err := s.token(ctx) - return token, status.wrap(err) + // The work runs on its own goroutine, which holds the lock until it is + // done, so a caller that gives up returns at once. A device session's + // refresh already sent finishes there, and the guest keeps what the + // service sends (see refreshGrace): a rotated refresh token is never + // lost. The token it brings is dropped; the next Token reads it. Any + // other request ends with the caller. + type result struct { + token string + err error + } + done := make(chan result, 1) + go func() { + defer s.mu.Unlock() + ctx, status := withAuthHTTPStatus(ctx) + token, err := s.token(ctx) + done <- result{token, status.wrap(err)} + }() + select { + case r := <-done: + return r.token, r.err + case <-ctx.Done(): + // A call that finished as the context ended still answers. + select { + case r := <-done: + return r.token, r.err + default: + return "", ctx.Err() + } + } } // token is Token under the strategy's lock: a read, then a locked refresh // for a device session that needs one. func (s *Strategy) token(ctx context.Context) (string, error) { - call := func(export func(*instance) api.Function) (string, error) { + call := func(ctx context.Context, export func(*instance) api.Function) (string, error) { out, err := s.store.callArgs(ctx, export, guest.BufArg([]byte(s.handle))) if err != nil { return "", err } defer guest.Wipe(out) + // A caller that has left gets no copy: a string cannot be wiped, + // and the next Token reads the guest's. + if err := ctx.Err(); err != nil { + return "", err + } return string(out), nil } - token, err := call(func(i *instance) api.Function { return i.authToken }) + token, err := call(ctx, func(i *instance) api.Function { return i.authToken }) if !s.device || !errors.Is(err, guest.ErrAuthRefreshRequired) { return token, err } @@ -216,7 +247,9 @@ func (s *Strategy) token(ctx context.Context) (string, error) { } err = withRefreshLock(ctx, path, func() error { var err error - token, err = call(func(i *instance) api.Function { return i.authRefresh }) + // The one request that rotates a credential: it runs on after its + // caller leaves (see refreshGrace). + token, err = call(withRefresh(ctx), func(i *instance) api.Function { return i.authRefresh }) return err }) return token, err @@ -232,6 +265,9 @@ func (s *Strategy) token(ctx context.Context) (string, error) { func (s *Strategy) MemoryLockError() error { return s.store.MemoryLockError() } // Close drops the strategy's cached token and unregisters its provider. +// It waits for a call in flight, including a device session's refresh that +// a cancelled Token left running: that refresh finishes, for up to 10 +// seconds, so the rotated refresh token it brings is saved rather than lost. func (s *Strategy) Close() error { s.mu.Lock() defer s.mu.Unlock() diff --git a/languages/golang/auth/transport.go b/languages/golang/auth/transport.go index de30684d2..b2e18d89c 100644 --- a/languages/golang/auth/transport.go +++ b/languages/golang/auth/transport.go @@ -10,6 +10,7 @@ import ( "sort" "strings" "sync" + "time" "github.com/cipherstash/stack/languages/golang/internal/guest" "github.com/tetratelabs/wazero" @@ -70,7 +71,10 @@ func (s *authHTTPStatus) wrap(err error) error { } type authTransport struct { - rt http.RoundTripper + rt http.RoundTripper + // grace is how long a refresh request runs on after its caller gives + // up: refreshGrace, shortened by tests. + grace time.Duration mu sync.Mutex providers map[uint32]OIDCProvider nextProvider uint32 @@ -80,7 +84,7 @@ func newAuthTransport(rt http.RoundTripper) *authTransport { if rt == nil { rt = http.DefaultTransport } - return &authTransport{rt: rt, providers: make(map[uint32]OIDCProvider)} + return &authTransport{rt: rt, grace: refreshGrace, providers: make(map[uint32]OIDCProvider)} } func (t *authTransport) register(provider OIDCProvider) uint32 { @@ -101,6 +105,40 @@ func (t *authTransport) unregister(id uint32) { t.mu.Unlock() } +// refreshGrace is how long a device session's refresh request runs on after +// its caller's context ends. The token service rotates the refresh token +// when it answers a refresh, so a refresh abandoned after the service +// answered would lose the new token, and the next refresh would send one +// the service has retired. So a refresh is not cancelled with its caller: +// the caller returns at once (Strategy.Token), the request finishes, and +// the guest saves what came back. A refresh still running this long after +// its caller gave up fails as a transport failure, well inside +// guest.StuckAfter, so the profile stays open. +// +// Every other request to the token service (an access-key exchange, an +// OIDC exchange) rotates nothing, and is cancelled with its caller. +const refreshGrace = 10 * time.Second + +// refreshKey marks the context of a device session's refresh call +// (withRefresh), whose requests get refreshGrace. +type refreshKey struct{} + +// withRefresh marks ctx as a refresh call's. +func withRefresh(ctx context.Context) context.Context { + return context.WithValue(ctx, refreshKey{}, true) +} + +// requestContext is the context a request to the token service runs under: +// the caller's, so a caller that gives up cancels it, except for a refresh, +// which runs on for the grace (see refreshGrace). done must be called when +// the request is finished with. +func (t *authTransport) requestContext(ctx context.Context) (context.Context, func()) { + if refresh, _ := ctx.Value(refreshKey{}).(bool); refresh { + return guest.Detach(guest.Caller(ctx), t.grace) + } + return guest.Caller(ctx), func() {} +} + func (t *authTransport) instantiate(ctx context.Context, r wazero.Runtime) error { _, err := r.NewHostModuleBuilder("cipherstash_transport"). NewFunctionBuilder().WithFunc(t.send).Export("transport_send"). @@ -124,8 +162,9 @@ func (t *authTransport) send(ctx context.Context, m api.Module, return t.placeResponse(ctx, m, respHeadersPtrOut, respHeadersLenOut, respBodyPtrOut, respBodyLenOut, -1, nil, []byte("request buffer out of range")) } requestBody := newAuthRequestBody(body) - // The caller's context, not the call's: see guest.Detach. - req, err := http.NewRequestWithContext(guest.Caller(ctx), string(method), string(url), requestBody) + reqCtx, done := t.requestContext(ctx) + defer done() + req, err := http.NewRequestWithContext(reqCtx, string(method), string(url), requestBody) if err != nil { _ = requestBody.Close() return t.placeResponse(ctx, m, respHeadersPtrOut, respHeadersLenOut, respBodyPtrOut, respBodyLenOut, -1, nil, []byte(err.Error())) diff --git a/languages/golang/encrypt/README.md b/languages/golang/encrypt/README.md index 4c3145b11..79ff551ba 100644 --- a/languages/golang/encrypt/README.md +++ b/languages/golang/encrypt/README.md @@ -123,9 +123,13 @@ request giving up never closes the shared client for the others. - A call waiting for the client (it runs one call at a time) leaves the queue when its `ctx` ends, and never starts. -- A call waiting on ZeroKMS, or on its auth strategy for a token, has that - request cancelled. Its error matches both `context.Canceled` or - `context.DeadlineExceeded` and `ErrTransport`. +- A call waiting on ZeroKMS has that request cancelled. Its error matches + both `context.Canceled` or `context.DeadlineExceeded` and `ErrTransport`. +- A call waiting on its auth strategy for a token returns too, with the + context's error. If the strategy is refreshing a device session, the + refresh finishes in the background, for up to 10 seconds. The service + rotates the refresh token when it answers, and dropping that answer + would leave the strategy unable to refresh. A client closes itself, and every later call returns `ErrState`, only when the guest's state is unknown: the guest trapped, or a call ran on for 30 diff --git a/languages/golang/encrypt/cancel_test.go b/languages/golang/encrypt/cancel_test.go index c94c916ca..5acaae70e 100644 --- a/languages/golang/encrypt/cancel_test.go +++ b/languages/golang/encrypt/cancel_test.go @@ -150,13 +150,15 @@ func TestACallWhoseContextEndsInTheQueueNeverReachesTheGuest(t *testing.T) { time.AfterFunc(200*time.Millisecond, func() { c.mu.Unlock(); close(released) }) ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) defer cancel() - start := time.Now() err := cipherInit(ctx, c) if !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("queued call: %v, want the deadline", err) } - if waited := time.Since(start); waited > 150*time.Millisecond { - t.Errorf("the queued call waited %v, past its deadline, for the lock", waited) + // It left at its deadline, not when the lock came free. + select { + case <-released: + t.Error("the queued call waited past its deadline for the lock") + default: } <-released }) diff --git a/languages/golang/encrypt/doc.go b/languages/golang/encrypt/doc.go index 8254579ff..3650673b9 100644 --- a/languages/golang/encrypt/doc.go +++ b/languages/golang/encrypt/doc.go @@ -191,9 +191,13 @@ // that call fails: one Client is shared by every request in a service, and // one request giving up never closes it for the others. A call waiting // for the client, which runs one call at a time, leaves the queue when its -// context ends and never starts. A call waiting on ZeroKMS, or on its auth -// strategy for a token, has that request cancelled; its error matches both -// the context's error and [ErrTransport]. +// context ends and never starts. A call waiting on ZeroKMS has that request +// cancelled; its error matches both the context's error and +// [ErrTransport]. A call waiting on its auth strategy for a token returns +// too, with the context's error. If the strategy is refreshing a device +// session, the refresh finishes in the background, for up to 10 seconds: +// the service rotates the refresh token when it answers, and dropping that +// answer would leave the strategy unable to refresh. // // A Client closes itself, and every later call returns [ErrState], only // when the guest's state is unknown: the guest trapped, or a call ran on