diff --git a/Cargo.lock b/Cargo.lock index b8e275d66..6de780f42 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3325,6 +3325,7 @@ dependencies = [ name = "stack-guest-abi" version = "0.0.0" dependencies = [ + "futures", "miette", "proptest", "serde_json", diff --git a/languages/golang/CHANGELOG.md b/languages/golang/CHANGELOG.md index 9a20b3106..4dee4ef44 100644 --- a/languages/golang/CHANGELOG.md +++ b/languages/golang/CHANGELOG.md @@ -93,6 +93,23 @@ listed under **Breaking**. Releases are tagged `languages/golang/vX.Y.Z`. 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. +- **A call no longer holds the client across a ZeroKMS round trip.** The + guest returns each ZeroKMS request to the client instead of performing it + through a blocking host import (`transport_send` is gone; the guest's one + import is `token_get`). The client performs the request with its lock + free, so other callers run meanwhile, then resumes the call. A call whose + context ends between those steps returns the context's error, matching + `ErrTransport` too, and the guest drops the paused call, wiping what it + held; the client stays open. + The token is fetched once per call, before its first step and outside + the lock. That includes a call the guest answers from its cache (a query + term on a loaded keyset, a cached keyset lookup), which never asked for + a token before, so a token source sees one request per such call; one + that fails is still reported only when the guest needs the token. At + most 64 calls that can reach ZeroKMS are in progress at once (the + guest's cap on paused calls); one past that waits for another to finish, + as it waited for the client before. This is the step loop an + edge runtime's host needs, where `fetch` cannot block. - **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/encrypt/README.md b/languages/golang/encrypt/README.md index 9618d2801..6d5cf15d4 100644 --- a/languages/golang/encrypt/README.md +++ b/languages/golang/encrypt/README.md @@ -102,7 +102,11 @@ that stores separate byte columns (`index=`) against real ZeroKMS. A `Client` is one ZeroKMS client: its client key, its default keyset, and the keysets it has loaded since. Make one per process and share it; it is safe -for concurrent use. +for concurrent use. The guest runs one call at a time, but never holds the +client across a ZeroKMS round trip: a call that needs ZeroKMS hands its +request to the client, which performs it with the client free for other +callers, and then carries on. A call whose context ends while its request +is in flight returns the context's error and leaves the client open. ```go client, err := encrypt.NewClient(ctx) diff --git a/languages/golang/encrypt/cancel_test.go b/languages/golang/encrypt/cancel_test.go index 5acaae70e..0e7d7018a 100644 --- a/languages/golang/encrypt/cancel_test.go +++ b/languages/golang/encrypt/cancel_test.go @@ -57,7 +57,8 @@ func (s *switchable) RoundTrip(r *http.Request) (*http.Response, error) { // 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. +// tests drive: it pauses on a load-keyset request, so it reaches the +// network between two steps. func cancelClient(t *testing.T, rt http.RoundTripper, token tokenSource) *Client { t.Helper() tr := &transport{rt: rt, token: token} @@ -77,7 +78,7 @@ func cipherInit(ctx context.Context, c *Client) error { return err } defer wipe(encoded) - _, err = c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { + _, err = c.callKMS(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.cipherInit, buf(encoded)) }) return err @@ -96,21 +97,28 @@ func wantOpen(t *testing.T, c *Client, rt *switchable) { } } -func TestACallCancelledDuringARequestFailsOnlyItself(t *testing.T) { +// The request runs between two steps, under the caller's context, with the +// guest paused: cancelling it ends the call and aborts the paused +// operation, and the client stays open. +func TestACallCancelledBetweenStepsFailsOnlyItself(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() rt := &switchable{rt: func(r *http.Request) (*http.Response, error) { + // The caller gives up once its request is in flight. + cancel() <-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) + if !errors.Is(err, context.Canceled) || !errors.Is(err, ErrTransport) { + t.Fatalf("cancelled call: %v, want context.Canceled and ErrTransport", err) } wantOpen(t, c, rt) } +// The token is fetched before the operation's first step: a caller that +// gives up waiting on it never starts the operation. func TestACallCancelledDuringATokenFetchFailsOnlyItself(t *testing.T) { rt := &switchable{rt: unauthorized} var blocking atomic.Bool @@ -126,8 +134,8 @@ func TestACallCancelledDuringATokenFetchFailsOnlyItself(t *testing.T) { 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 !errors.Is(err, context.Canceled) || errors.Is(err, ErrState) { + t.Fatalf("cancelled call: %v, want context.Canceled", err) } if n := rt.sent.Load(); n != 0 { t.Fatalf("%d request(s) sent without a token", n) @@ -175,31 +183,53 @@ func TestACallWhoseContextEndsInTheQueueNeverReachesTheGuest(t *testing.T) { wantOpen(t, c, rt) } -// The backstop: a call still running long after its caller gave up is a +// The backstop: a step 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) - }} +// so the client closes, and says so. The step here waits out its caller's +// context and the backstop and only then enters the guest, as a guest +// stuck in a loop would still be running when the backstop fired. +func TestAStepThatRunsOnAfterItsContextEndsClosesTheClient(t *testing.T) { + rt := &switchable{rt: unauthorized} c := cancelClient(t, rt, staticToken("t")) c.stuck = 20 * time.Millisecond - ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + ctx, cancel := context.WithCancel(context.Background()) defer cancel() - err := cipherInit(ctx, c) + _, err := c.call(ctx, func(run context.Context, inst *instance) ([]byte, error) { + cancel() + <-run.Done() + return inst.call(run, inst.targets) + }) 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 !errors.Is(err, context.Canceled) { + t.Errorf("stuck call: %v, want it to match context.Canceled", err) } if err := cipherInit(context.Background(), c); !errors.Is(err, ErrState) { t.Fatalf("a call after the close: %v, want ErrState", err) } } +// A request that ignores its context holds up only its own call: the guest +// is paused while it runs, so nothing is stuck, and the client stays open. +func TestARequestThatIgnoresItsContextLeavesTheClientOpen(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + rt := &switchable{rt: func(*http.Request) (*http.Response, error) { + // The caller gives up once its request is in flight, and the + // request runs on past the backstop regardless. + cancel() + time.Sleep(100 * time.Millisecond) + return unauthorized(nil) + }} + c := cancelClient(t, rt, staticToken("t")) + c.stuck = 20 * time.Millisecond + if err := cipherInit(ctx, c); !errors.Is(err, context.Canceled) || !errors.Is(err, ErrTransport) { + t.Fatalf("call: %v, want context.Canceled and ErrTransport", err) + } + wantOpen(t, c, rt) +} + // 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) { @@ -236,6 +266,67 @@ func TestACallThatSucceedsAfterItsContextEndsFailsWithIt(t *testing.T) { wantOpen(t, c, rt) } +// A step that pauses after its caller gave up sends no request: the +// operation is aborted in the guest, so its handle is spent, and the call +// fails with the context's error. +func TestAStepThatPausesAfterItsContextEndsSendsNothing(t *testing.T) { + rt := &switchable{rt: unauthorized} + c := cancelClient(t, rt, staticToken("t")) + encoded, err := encodeConfig(testInit("http://zerokms.invalid")) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + _, err = c.callKMS(ctx, func(run context.Context, inst *instance) ([]byte, error) { + cancel() + return inst.call(run, inst.cipherInit, buf(encoded)) + }) + if !errors.Is(err, context.Canceled) || !errors.Is(err, ErrTransport) { + t.Fatalf("call: %v, want context.Canceled and ErrTransport", err) + } + if n := rt.sent.Load(); n != 0 { + t.Fatalf("%d request(s) sent for a caller that had given up", n) + } + if _, err := c.inst.call(context.Background(), c.inst.abort, scalar(1)); !errors.Is(err, ErrState) { + t.Fatalf("the paused operation's handle: %v, want ErrState (already dropped)", err) + } + wantOpen(t, c, rt) +} + +// A caller whose context ends while its resume waits for the client never +// resumes: the paused operation is still aborted in the guest once the +// client is free. +func TestAResumeWhoseContextEndsInTheQueueAbortsTheOperation(t *testing.T) { + stub := newHeldStub(t) + c := stepClient(t, staticToken("stub-token")) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + _, err := c.callKMS(ctx, initStep(t, stub.URL)) + done <- err + }() + waitFor(t, stub.arrived, "the init's request") + // Another call holds the client when the response comes back. + if err := c.mu.Acquire(context.Background()); err != nil { + t.Fatal(err) + } + close(stub.release) + time.Sleep(50 * time.Millisecond) + cancel() + time.Sleep(20 * time.Millisecond) + c.mu.Unlock() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("call: %v, want context.Canceled", err) + } + if _, err := c.inst.call(context.Background(), c.inst.abort, scalar(1)); !errors.Is(err, ErrState) { + t.Fatalf("the paused operation's handle: %v, want ErrState (already dropped)", err) + } + if c.closed || c.inst.module.IsClosed() { + t.Fatal("giving up closed the client") + } +} + // A closed module is ErrState, and Close still releases the runtime. func TestAClosedModuleIsState(t *testing.T) { ctx := context.Background() @@ -382,8 +473,8 @@ func TestACallCancelledDuringATokenExchangeLeavesTheProfileOpen(t *testing.T) { 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) + if err := cipherInit(ctx, c); !errors.Is(err, context.DeadlineExceeded) || errors.Is(err, ErrState) { + t.Fatalf("call cancelled during the exchange: %v, want the deadline", err) } hold.Store(false) wantOpen(t, c, rt) diff --git a/languages/golang/encrypt/client.go b/languages/golang/encrypt/client.go index b0156425a..042a7d09c 100644 --- a/languages/golang/encrypt/client.go +++ b/languages/golang/encrypt/client.go @@ -17,8 +17,10 @@ import ( // Client is one wasm instance holding one ZeroKMS client: its key, its // default keyset, and the keysets it has loaded since. It is safe for -// concurrent use; calls are serialised internally, because a wasm instance -// is single-threaded. Close it when done, as with any resource. +// concurrent use; the guest runs one call's step at a time, because a wasm +// instance is single-threaded, but never waits on the network inside one: +// a ZeroKMS request is performed between steps, with the client free for +// other callers (see step.go). 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 @@ -45,6 +47,12 @@ type Client struct { // the guest is treated as stuck and closed. guest.StuckAfter; tests // shorten it. stuck time.Duration + // paused bounds the operations that may be paused in the guest at once + // to its cap (maxPaused): an operation that can reach ZeroKMS takes a + // slot before its first step and gives it back when it ends, so a + // caller past the cap waits here for one rather than being refused by + // the guest. + paused chan struct{} // 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 @@ -188,7 +196,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(ctx context.Context, inst *instance) ([]byte, error) { + out, err := c.callKMS(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.cipherInit, buf(encoded)) }) if err != nil { @@ -230,7 +238,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, stuck: guest.StuckAfter} + c := &Client{inst: inst, transport: t, stuck: guest.StuckAfter, paused: make(chan struct{}, maxPaused)} c.cleanup = runtime.AddCleanup(c, func(inst *instance) { _ = inst.release() }, inst) return c } @@ -381,7 +389,7 @@ func (c *Client) resolveKeyset(ctx context.Context, sel KeysetSelector) (KeysetI if err != nil { return KeysetID{}, err } - out, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { + out, err := c.callKMS(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.keyset, buf(encoded)) }) if err != nil { @@ -395,15 +403,165 @@ func (c *Client) resolveKeyset(ctx context.Context, sel KeysetSelector) (KeysetI return id, nil } -// 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. +// maxPaused is the most operations the guest holds paused at once: MAX_OPEN +// in stack-guest-abi's step module. One more is refused there, so the +// client never starts more than this many operations that can pause. +const maxPaused = 64 + +// call runs an operation that never reaches ZeroKMS (a plan check, the +// catalog, a verify): the step loop with no token to fetch. +func (c *Client) call(ctx context.Context, f stepFunc) ([]byte, error) { + return c.drive(ctx, false, f) +} + +// callKMS runs an operation that can reach ZeroKMS: the step loop, with a +// token fetched before its first step. +func (c *Client) callKMS(ctx context.Context, f stepFunc) ([]byte, error) { + return c.drive(ctx, true, f) +} + +// drive runs an operation to its end, a step at a time (see step.go): its +// first step is f, and each later one resumes it with the responses to the +// requests the step before asked for. The client's lock is held for each +// step and released while the requests run, so callers share the client +// between steps rather than queue behind a round trip. +// +// When kms is set, the operation first takes one of the guest's paused +// slots (see Client.paused), and the token is fetched once, before its +// first step and outside the lock; every step of the operation hands it +// over, so the guest's token_get never waits. The token source is asked +// once per such operation, including one the guest answers from its cache +// without asking for the token (a query term on a loaded keyset, a cached +// keyset lookup): which operations reach ZeroKMS is known only once the +// guest has run. A source that fails is reported only if the guest asks. +// +// The caller's context governs everything the host waits on: the slot, the +// token fetch, the wait for the lock and the requests, which run under it. It +// never reaches the guest, which runs each step under guest.Detach (see +// step). An operation whose caller gives up between steps is aborted, +// which drops it in the guest with everything it held, and the call fails +// with the context's error; one abandoned while it waited on ZeroKMS +// matches ErrTransport too. +func (c *Client) drive(ctx context.Context, kms bool, f stepFunc) ([]byte, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + token, tokenErr := "", error(nil) + if kms { + select { + case c.paused <- struct{}{}: + defer func() { <-c.paused }() + case <-ctx.Done(): + return nil, ctx.Err() + } + } + if kms && c.transport != nil { + token, tokenErr = c.transport.fetchToken(ctx) + // A caller that gave up waiting on its token never starts the + // operation. + if err := ctx.Err(); err != nil { + return nil, err + } + } + step, resumed := f, false + var handle uint64 + var responses []byte + for { + out, io, err := c.step(ctx, token, tokenErr, step) + if resumed { + // The resume step wipes its responses when it runs; when it + // never ran (its caller gave up waiting for the client, or the + // client closed meanwhile) they are wiped here. + wipe(responses) + } + if errors.Is(err, errLockWait) { + // The step never ran: a paused operation is still the + // guest's to drop. + if resumed { + c.abort(handle) + } + return nil, ctx.Err() + } + if io == nil { + return settle(ctx, out, err) + } + // A caller that gave up while the guest ran sends no request. + if err := ctx.Err(); err != nil { + io.wipe() + c.abort(io.handle) + return nil, fmt.Errorf("%w: %w", err, ErrTransport) + } + responses, err = c.transport.performStep(ctx, io) + if err != nil { + c.abort(io.handle) + return nil, err + } + // The requests ran under the caller's context, so one it cancelled + // came back a transport failure: the guest is not resumed with it. + if err := ctx.Err(); err != nil { + wipe(responses) + c.abort(io.handle) + return nil, fmt.Errorf("%w: %w", err, ErrTransport) + } + h, pending := io.handle, responses + handle, resumed = h, true + step = func(ctx context.Context, inst *instance) ([]byte, error) { + defer wipe(pending) + return inst.call(ctx, inst.resume, scalar(h), buf(pending)) + } + } +} + +// settle is how an operation ends for its caller. 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 +// stays matchable too. A call that succeeded after its context ended fails +// as well: 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. +func settle(ctx context.Context, out []byte, err error) ([]byte, error) { + ctxErr := ctx.Err() + if err != nil { + if ctxErr != nil && !errors.Is(err, ctxErr) { + err = fmt.Errorf("%w: %w", ctxErr, err) + } + return nil, err + } + if ctxErr != nil { + wipe(out) + return nil, ctxErr + } + return out, nil +} + +// abort drops a paused operation in the guest, for a caller that gave up +// on it between steps. It waits for the lock however long that takes, and +// runs under no deadline: the operation's plaintext and keys are wiped +// only when it is dropped. +func (c *Client) abort(handle uint64) { + c.mu.Lock() + defer c.mu.Unlock() + if c.closed || c.inst.module.IsClosed() { + return + } + ctx := context.Background() + _, _ = c.inst.call(ctx, c.inst.abort, scalar(handle)) +} + +// errLockWait marks a step that never ran because its caller's context +// ended while it waited for the client. +var errLockWait = errors.New("encrypt: the caller's context ended waiting for the client") + +// step runs one step of an operation on the instance under the client's +// lock: the operation's output when it finishes, or its next step when it +// pauses, taken from the guest before the lock is released (a host reads a +// step before it runs any other). // // 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. +// has already ended, so a step whose caller timed out in the queue never +// reaches the guest. The guest then runs under guest.Detach, never the +// caller's context: the runtime closes a module whose call context ends, +// and one caller giving up must not close the client for the others. // // 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: @@ -412,21 +570,28 @@ func (c *Client) resolveKeyset(ctx context.Context, sel KeysetSelector) (KeysetI // 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) { +func (c *Client) step(ctx context.Context, token string, tokenErr error, f stepFunc) ([]byte, *ioStep, error) { if err := c.mu.Acquire(ctx); err != nil { - return nil, err + return nil, nil, fmt.Errorf("%w: %w", errLockWait, err) } defer c.mu.Unlock() if c.closed || c.inst.module.IsClosed() { c.closed = true - return nil, ErrState + return nil, nil, ErrState } growth := c.inst.mem.GrowthRefusal() if c.transport != nil { - c.transport.tokenErr = nil + c.transport.held, c.transport.fetchErr, c.transport.tokenErr = token, tokenErr, nil + defer func() { c.transport.held, c.transport.fetchErr = "", nil }() } run, done := guest.Detach(ctx, c.stuck) out, err := f(run, c.inst) + var io *ioStep + if errors.Is(err, guest.ErrNeedsIO) { + // A step that cannot be read cannot be resumed; the guest drops + // the operation at shutdown, and refuses new pauses past its cap. + io, err = c.inst.takeIO(run) + } done() switch { case c.inst.module.IsClosed(): @@ -462,23 +627,8 @@ func (c *Client) call(ctx context.Context, f func(context.Context, *instance) ([ 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 nil, nil, err } - return out, nil + return out, io, nil } diff --git a/languages/golang/encrypt/doc.go b/languages/golang/encrypt/doc.go index ae207856c..a87303c0c 100644 --- a/languages/golang/encrypt/doc.go +++ b/languages/golang/encrypt/doc.go @@ -208,9 +208,11 @@ // // # Transport and auth // -// The guest imports exactly two host functions: an HTTP send, served by any -// [net/http.RoundTripper], and a bearer-token fetch, served by the -// credentials' auth strategy. What crosses per ZeroKMS call is what would +// The guest imports exactly one host function, a bearer-token fetch, served +// by the credentials' auth strategy. It never performs a request itself: an +// operation that needs ZeroKMS pauses and hands the request to the client, +// which sends it with any [net/http.RoundTripper] and resumes the operation +// with the response. What crosses per ZeroKMS call is what would // cross TLS anyway; derived key material never leaves the guest. Under // [AutoCredentials] and [OIDCFederation] the same RoundTripper also carries // the authentication requests to CTS, so one scoped to the ZeroKMS host alone diff --git a/languages/golang/encrypt/guest.go b/languages/golang/encrypt/guest.go index 2cffd3be2..04d01730c 100644 --- a/languages/golang/encrypt/guest.go +++ b/languages/golang/encrypt/guest.go @@ -88,6 +88,9 @@ type instance struct { queryTerm, verify, limit api.Function encryptRecord, decryptRecord api.Function planCheck, targets, operators api.Function + // The step loop (step.go): take a paused export's step, resume it with + // the responses, or abort it. + io, resume, abort api.Function } // guestModuleConfig is the module configuration every guest instance runs @@ -183,6 +186,9 @@ func newInstance(ctx context.Context, wasm []byte, t *transport, policy guest.Lo "se_plan_check": &inst.planCheck, "se_targets": &inst.targets, "se_operators": &inst.operators, + "se_io": &inst.io, + "se_resume": &inst.resume, + "se_abort": &inst.abort, } for name, slot := range exports { if *slot = module.ExportedFunction(name); *slot == nil { diff --git a/languages/golang/encrypt/guest/Cargo.toml b/languages/golang/encrypt/guest/Cargo.toml index 561493d45..e39ab37b0 100644 --- a/languages/golang/encrypt/guest/Cargo.toml +++ b/languages/golang/encrypt/guest/Cargo.toml @@ -51,7 +51,7 @@ stack-auth = { path = "../../../../packages/stack-auth", default-features = fals stack-encrypt = { path = "../../../../packages/stack-encrypt", default-features = false, features = ["dynamic"] } # The guest ABI every guest under bindings/go shares: the allocator and # buffer registry (and with them the `se_alloc`/`se_dealloc` exports), the -# packed-result helpers, the status table and the `transport_send` import. +# packed-result helpers, the status table and the step loop. # This crate defines only the exports that are its own. stack-guest-abi = { path = "../../../../packages/stack-guest-abi" } stack-kms = { path = "../../../../packages/stack-kms", default-features = false } diff --git a/languages/golang/encrypt/guest/src/abi.rs b/languages/golang/encrypt/guest/src/abi.rs index 1cd4ce19a..122425494 100644 --- a/languages/golang/encrypt/guest/src/abi.rs +++ b/languages/golang/encrypt/guest/src/abi.rs @@ -16,17 +16,32 @@ //! AAD and plan buffers are not secret and are left untouched. //! - Output buffers from the decrypt exports contain plaintext; the host must //! copy them out and immediately `se_dealloc` (which zeroizes). +//! - **The guest never waits on the network: it steps.** An export that can +//! reach ZeroKMS ([`se_cipher_init`], [`se_keyset`], [`se_term`], +//! [`se_encrypt_record`], [`se_decrypt_record`], [`se_query_term`]) runs +//! its work as a future, polled once. Done, it returns its output as +//! always. Waiting on a request, it returns `STATUS_NEEDS_IO` instead: the +//! host takes the paused operation's handle and requests with [`se_io`], +//! performs them, and hands the responses to [`se_resume`], which returns +//! the same three ways; or it gives up with [`se_abort`] +//! (`stack_guest_abi::step`, and [`crate::io`] for the wire form). Several +//! operations may be paused at once, each under its own handle. A paused +//! operation owns its inputs — the plaintext among them, wiped when it +//! ends or is aborted — so a host's input buffers are free once the +//! export returns, paused or not. A host reads a step's requests (and a +//! failure's `se_last_error`) before it steps another operation. //! - **One instance is one client.** [`se_cipher_init`] runs once per //! instance: it builds the //! `StackCipher>` (one -//! `load-keyset` round trip through the host transport for the default +//! `load-keyset` round trip, through the step loop, for the default //! keyset) and returns that keyset's id. There is no cipher handle: the //! keysets a client uses are selected per call through the options object //! ([`crate::options`]), loaded on first use through the cipher's own //! cache. Nothing crosses the boundary that the host could allocate, //! alias or free. -//! - [`se_shutdown`] is the one lifetime call: it drops the cipher (client -//! key and every loaded index key wiped by `ZeroizeOnDrop`) and wipes +//! - [`se_shutdown`] is the one lifetime call: it drops every paused +//! operation, then the cipher (client +//! key and every loaded index key wiped by `ZeroizeOnDrop`), and wipes //! every buffer the registry still holds. It exists because closing a //! wasm instance frees linear memory without running Rust destructors — //! without it, key material would sit in freed host memory. After it, a @@ -37,10 +52,10 @@ //! [`se_shutdown`] is a no-op, and a *malformed* call is //! `STATUS_ENCODING` in any state, because validation runs first (see //! below). -//! - During an entry-point call the host's imported functions may re-enter -//! the guest **only** through `se_alloc` (to place the transport response -//! / token); calling any other export from inside a host import is -//! undefined behaviour of the embedding, not of this module. +//! - During an entry-point call the host's imported function may re-enter +//! the guest **only** through `se_alloc` (to place the token); calling any +//! other export from inside a host import is undefined behaviour of the +//! embedding, not of this module. //! - **Every export runs through `stack_guest_abi::abi::export`**, so a //! failure leaves its full error for `se_last_error`: the code, message, //! help and fields of the `stack_encrypt`, `stack_kms` or `stack_auth` @@ -69,22 +84,25 @@ //! inside the call, after every check. //! //! Wasm modules are single-threaded; the host must serialize calls into one -//! instance. +//! instance. Between calls it may interleave steps of different operations. use std::cell::{Cell, RefCell}; use std::panic::{catch_unwind, AssertUnwindSafe}; +use std::rc::Rc; -use futures::executor::block_on; -use stack_encrypt::{KeysetCipher, StackCipher}; +use stack_encrypt::StackCipher; use stack_guest_abi::abi::{export, input, take_plaintext, wipe_input}; use stack_guest_abi::buffers; +use stack_guest_abi::call; use stack_guest_abi::last_error::{self, malformed, out_of_order}; +use stack_guest_abi::status::STATUS_NEEDS_IO; +use stack_guest_abi::step::{self, Operation, Step}; +use crate::io; use crate::ops; use crate::options::{parse_options, parse_selector, scope_for, KeysetSelector, Side}; #[cfg(not(feature = "deterministic-kms"))] use crate::status::STATUS_INTERNAL; -use stack_encrypt::dynamic::Scope; #[cfg(not(feature = "deterministic-kms"))] use stack_encrypt::ErrorPayload; @@ -102,54 +120,48 @@ type GuestCipher = StackCipher; thread_local! { // Wasm is single-threaded, so a thread-local `RefCell` is a plain owner - // of the cipher — no `Send`/`Sync` bounds required. - static CIPHER: RefCell> = const { RefCell::new(None) }; + // of the cipher — no `Send`/`Sync` bounds required. Paused operations + // share it through the `Rc`, so their handles are dropped before it at + // shutdown, and the last owner wipes it. + static CIPHER: RefCell>> = const { RefCell::new(None) }; /// Set by `se_shutdown`: after it, `se_cipher_init` is refused too, so /// an instance the host has torn down cannot be quietly revived with /// stale buffers around. static SHUT_DOWN: Cell = const { Cell::new(false) }; + /// Set while an `se_cipher_init` is paused, so a second one is refused + /// before the first has installed its cipher. + static INITIALISING: Cell = const { Cell::new(false) }; } -/// Run `f` with the instance's cipher, or report `STATUS_STATE` when there -/// is none (never initialised, or shut down). -fn with_cipher(f: impl FnOnce(&GuestCipher) -> Result) -> Result { - CIPHER.with(|c| { - let c = c.borrow(); - let cipher = c.as_ref().ok_or_else(|| { - out_of_order( - "there is no cipher: call se_cipher_init first, and none after se_shutdown", - ) - })?; - f(cipher) +/// The instance's cipher, or `STATUS_STATE` when there is none (never +/// initialised, still initialising, or shut down). +fn cipher() -> Result, u32> { + CIPHER.with(|c| c.borrow().clone()).ok_or_else(|| { + out_of_order("there is no cipher: call se_cipher_init first, and none after se_shutdown") }) } -/// Run `f` with the keyset the mint-side options in `opts` select, -/// resolving it through the cipher (a first use is one `load-keyset` round -/// trip). The options are decoded and validated before the cipher is -/// consulted. -fn with_keyset( - opts: &[u8], - f: impl FnOnce(&KeysetCipher<'_, GuestKms>) -> Result, -) -> Result { - let options = parse_options(opts, Side::Mint)?; - with_cipher(|cipher| { - let keyset = block_on(options.keyset.resolve(cipher))?; - f(&keyset) - }) +/// Turn what came of an operation's step into an export's result: the +/// output when done, `STATUS_NEEDS_IO` with the step registered for +/// [`se_io`] when paused, the status when failed. +fn deliver(step: Result) -> Result, u32> { + match step? { + Step::Done(out) => Ok(out), + Step::NeedsIo { handle, requests } => { + let encoded = io::encode_step(handle, &requests); + drop(requests); + let encoded = encoded.inspect_err(|_| { + let _ = step::abort(handle); + })?; + step::set_pending(encoded); + Err(STATUS_NEEDS_IO) + } + } } -/// Run `f` with the scope the open-side options in `opts` select: the -/// client for `{"any"}`, one keyset's cipher otherwise. -fn with_scope( - opts: &[u8], - f: impl FnOnce(Scope<'_, GuestKms>) -> Result, -) -> Result { - let options = parse_options(opts, Side::Open)?; - with_cipher(|cipher| { - let scope = block_on(scope_for(cipher, &options.keyset))?; - f(scope) - }) +/// [`deliver`] an operation's first step. +fn run(work: Operation) -> Result, u32> { + deliver(step::start(work)) } /// Initialise the instance's cipher from a config frame @@ -205,12 +217,13 @@ fn cipher_init(config: InitConfig) -> Result, u32> { use crate::status::STATUS_KMS_TRANSPORT; use stack_kms::{ClientOpts, StackKms}; - if SHUT_DOWN.with(Cell::get) || CIPHER.with(|c| c.borrow().is_some()) { + if init_refused() { return Err(init_out_of_order()); } - // One request at a time: the host import is synchronous, so concurrency - // would only interleave nothing; keep the executor honest about it. + // One request at a time: a step hands the host one request, and the + // first version of the step loop keeps it that way. Raising it later + // changes no ABI: a step's request list can hold several. // // `max_keys_per_req` stays at the client default (500). That is what // bounds "one ZeroKMS call": a batch is assembled once, then @@ -240,7 +253,16 @@ fn cipher_init(config: InitConfig) -> Result, u32> { if let Some(size) = config.keyset_cache_size { builder = builder.keyset_cache_size(size); } - install(block_on(builder.init()).map_err(|e| crate::status::fail_error(&e))?) + let initialising = Initialising::begin(); + run(Box::pin(async move { + let _initialising = initialising; + install( + builder + .init() + .await + .map_err(|e| crate::status::fail_error(&e))?, + ) + })) } /// The `deterministic-kms` build's init: the config buffer is the @@ -249,23 +271,62 @@ fn cipher_init(config: InitConfig) -> Result, u32> { /// [`crate::deterministic`]. #[cfg(feature = "deterministic-kms")] fn cipher_init(seed: InitConfig) -> Result, u32> { - if SHUT_DOWN.with(Cell::get) || CIPHER.with(|c| c.borrow().is_some()) { + if init_refused() { return Err(init_out_of_order()); } let builder = StackCipher::builder().kms(seed.source()); - install(block_on(builder.init()).map_err(|e| crate::status::fail_error(&e))?) + let initialising = Initialising::begin(); + run(Box::pin(async move { + let _initialising = initialising; + install( + builder + .init() + .await + .map_err(|e| crate::status::fail_error(&e))?, + ) + })) +} + +/// Whether an `se_cipher_init` must be refused: after shutdown, with a +/// cipher installed, or with another init paused. +fn init_refused() -> bool { + SHUT_DOWN.with(Cell::get) + || INITIALISING.with(Cell::get) + || CIPHER.with(|c| c.borrow().is_some()) +} + +/// Marks an init in flight from the moment it starts until its work ends: +/// done, failed or aborted. +struct Initialising; + +impl Initialising { + fn begin() -> Self { + INITIALISING.with(|i| i.set(true)); + Initialising + } +} + +impl Drop for Initialising { + fn drop(&mut self) { + INITIALISING.with(|i| i.set(false)); + } } -/// `se_cipher_init` called a second time, or after `se_shutdown`. +/// `se_cipher_init` called a second time, while one is paused, or after +/// `se_shutdown`. fn init_out_of_order() -> u32 { - out_of_order("the cipher is already initialised, or the instance is shut down") + out_of_order("the cipher is already initialised or initialising, or the instance is shut down") } /// Install the built cipher as the instance's and return its default /// keyset's id. fn install(cipher: GuestCipher) -> Result, u32> { + // A shutdown that overtook a paused init installs nothing. + if SHUT_DOWN.with(Cell::get) { + return Err(init_out_of_order()); + } let default = cipher.default_keyset().keyset_id().as_bytes().to_vec(); - CIPHER.with(|c| *c.borrow_mut() = Some(cipher)); + CIPHER.with(|c| *c.borrow_mut() = Some(Rc::new(cipher))); Ok(default) } @@ -278,16 +339,15 @@ fn install(cipher: GuestCipher) -> Result, u32> { /// is `STATUS_STATE`, [`se_cipher_init`] included; a malformed one is /// `STATUS_ENCODING` first, as in any other state. /// -/// The index keys' wipe holds because each `HmacSha256Prf` clone a -/// derivation takes is created and dropped inside one `block_on`'d call, -/// and the guest is single-threaded, so no clone is alive when the host -/// calls this. That is the precondition, not a property of the drop: -/// anything that later parks a PRF clone beyond an ABI call turns this -/// wipe into a no-op for that key. +/// Every paused operation is dropped first: each holds the cipher, and may +/// hold a PRF clone a derivation took, plaintext, or keys it was handed. +/// Dropping them before the cipher is what makes the cipher's drop the last +/// owner's, and so its wipe real. #[no_mangle] pub extern "C" fn se_shutdown() { let _ = catch_unwind(AssertUnwindSafe(|| { SHUT_DOWN.with(|s| s.set(true)); + step::drop_all(); CIPHER.with(|c| { let _ = c.borrow_mut().take(); }); @@ -314,10 +374,11 @@ pub unsafe extern "C" fn se_keyset(sel_ptr: *const u8, sel_len: u32) -> u64 { if selector == KeysetSelector::Any { return Err(malformed(r#"{"any"} is not a keyset"#)); } - with_cipher(|cipher| { - let keyset = block_on(selector.resolve(cipher))?; + let cipher = cipher()?; + run(Box::pin(async move { + let keyset = selector.resolve(&cipher).await?; Ok(keyset.keyset_id().as_bytes().to_vec()) - }) + })) }) } @@ -355,15 +416,17 @@ pub unsafe extern "C" fn se_term( ) -> u64 { export(|| { let value = unsafe { take_plaintext(val_ptr, val_len)? }; - let value = value.as_slice(); // SAFETY: host-owned ranges the export was handed; the borrows end // before it returns and before any wipe of an overlapping range. - let context = unsafe { input(ctx_ptr, ctx_len)? }; + let context = unsafe { input(ctx_ptr, ctx_len)? }.to_vec(); let opts = unsafe { input(opt_ptr, opt_len)? }; - ops::validate::term(value, context, kind)?; - with_keyset(opts, |keyset| { - block_on(ops::term(keyset, value, context, kind)) - }) + ops::validate::term(&value, &context, kind)?; + let options = parse_options(opts, Side::Mint)?; + let cipher = cipher()?; + run(Box::pin(async move { + let keyset = options.keyset.resolve(&cipher).await?; + ops::term(&keyset, &value, &context, kind).await + })) }) } @@ -388,15 +451,17 @@ pub unsafe extern "C" fn se_encrypt_record( ) -> u64 { export(|| { let source = unsafe { take_plaintext(src_ptr, src_len)? }; - let source = source.as_slice(); // SAFETY: host-owned ranges the export was handed; the borrows end // before it returns and before any wipe of an overlapping range. - let plan = unsafe { input(plan_ptr, plan_len)? }; + let plan = unsafe { input(plan_ptr, plan_len)? }.to_vec(); let opts = unsafe { input(opt_ptr, opt_len)? }; - ops::validate::record(source, plan)?; - with_keyset(opts, |keyset| { - block_on(ops::encrypt_record(keyset, source, plan)) - }) + ops::validate::record(&source, &plan)?; + let options = parse_options(opts, Side::Mint)?; + let cipher = cipher()?; + run(Box::pin(async move { + let keyset = options.keyset.resolve(&cipher).await?; + ops::encrypt_record(&keyset, &source, &plan).await + })) }) } @@ -428,14 +493,16 @@ pub unsafe extern "C" fn se_decrypt_record( export(|| { // SAFETY: host-owned ranges the export was handed; the borrows end // before it returns and before any wipe of an overlapping range. - let record = unsafe { input(rec_ptr, rec_len)? }; - let plan = unsafe { input(plan_ptr, plan_len)? }; + let record = unsafe { input(rec_ptr, rec_len)? }.to_vec(); + let plan = unsafe { input(plan_ptr, plan_len)? }.to_vec(); let opts = unsafe { input(opt_ptr, opt_len)? }; - let expected = parse_options(opts, Side::Open)?.context; - ops::validate::record_tree(record, plan, expected.as_ref())?; - with_scope(opts, |scope| { - block_on(ops::decrypt_record(scope, record, plan, expected)) - }) + let options = parse_options(opts, Side::Open)?; + ops::validate::record_tree(&record, &plan, options.context.as_ref())?; + let cipher = cipher()?; + run(Box::pin(async move { + let scope = scope_for(&cipher, &options.keyset).await?; + ops::decrypt_record(scope, &record, &plan, options.context).await + })) }) } @@ -487,17 +554,19 @@ pub unsafe extern "C" fn se_query_term( ) -> u64 { export(|| { let value = unsafe { take_plaintext(val_ptr, val_len)? }; - let value = value.as_slice(); // SAFETY: host-owned ranges the export was handed; the borrows end // before it returns and before any wipe of an overlapping range. - let plan = unsafe { input(plan_ptr, plan_len)? }; - let field = unsafe { input(field_ptr, field_len)? }; - let operator = unsafe { input(op_ptr, op_len)? }; + let plan = unsafe { input(plan_ptr, plan_len)? }.to_vec(); + let field = unsafe { input(field_ptr, field_len)? }.to_vec(); + let operator = unsafe { input(op_ptr, op_len)? }.to_vec(); let opts = unsafe { input(opt_ptr, opt_len)? }; - ops::validate::query_term(value, plan, field, operator)?; - with_keyset(opts, |keyset| { - block_on(ops::query_term(keyset, value, plan, field, operator)) - }) + ops::validate::query_term(&value, &plan, &field, &operator)?; + let options = parse_options(opts, Side::Mint)?; + let cipher = cipher()?; + run(Box::pin(async move { + let keyset = options.keyset.resolve(&cipher).await?; + ops::query_term(&keyset, &value, &plan, &field, &operator).await + })) }) } @@ -609,3 +678,49 @@ pub unsafe extern "C" fn se_operators( ops::operators(plan, field) }) } + +/// The step a paused export left: its handle and the requests it waits on, +/// as a packed buffer ([`crate::io`]), handed over once — the slot is +/// emptied, so a second call returns zero, as does a call with no step +/// waiting. The host calls it after `STATUS_NEEDS_IO`, before it runs any +/// other export, copies the buffer out and releases it with `se_dealloc` +/// (which wipes it: it carries the bearer token). +#[no_mangle] +pub extern "C" fn se_io() -> u64 { + match step::take_pending() { + Some((ptr, len)) => call::pack_buffer(ptr as usize as u32, len as u32), + None => 0, + } +} + +/// Hand a paused operation the responses to the requests its last step +/// asked for ([`crate::io`] has the frame), and run its next step. Returns +/// as the export that started it does: the output when done, +/// `STATUS_NEEDS_IO` when paused again (under the same handle), or the +/// status of its failure. The responses buffer can carry wrapped keys and +/// is wiped in place. A handle that names no paused operation is +/// `STATUS_STATE`; responses that do not decode, or are not one per +/// request, are `STATUS_ENCODING`, and the operation is dropped. +/// +/// # Safety +/// +/// As for [`se_term`]. +#[no_mangle] +pub unsafe extern "C" fn se_resume(handle: u64, resp_ptr: *mut u8, resp_len: u32) -> u64 { + export(|| { + let bytes = unsafe { take_plaintext(resp_ptr, resp_len) }; + let responses = bytes.and_then(|bytes| io::decode_responses(&bytes)); + let responses = responses.inspect_err(|_| { + let _ = step::abort(handle); + })?; + deliver(step::resume(handle, responses)) + }) +} + +/// Give up on a paused operation: drop it, wiping everything it held. An +/// empty output; `STATUS_STATE` for a handle that names no paused +/// operation (one that finished, failed or was aborted). +#[no_mangle] +pub extern "C" fn se_abort(handle: u64) -> u64 { + export(|| step::abort(handle).map(|()| Vec::new())) +} diff --git a/languages/golang/encrypt/guest/src/host.rs b/languages/golang/encrypt/guest/src/host.rs index 7c7564f5f..605bf7d42 100644 --- a/languages/golang/encrypt/guest/src/host.rs +++ b/languages/golang/encrypt/guest/src/host.rs @@ -1,7 +1,16 @@ -//! The host imports and the types built over them: [`WasiHostConnection`] -//! (a [`ZeroKMSConnection`] whose transport is the host's HTTP client) and -//! [`HostTokenStrategy`] (an [`AuthStrategy`] that asks the host for the -//! bearer token). wasm32-only — everything here calls an imported function. +//! The guest's two connections to its host: [`WasiHostConnection`] (a +//! [`ZeroKMSConnection`] whose requests the host performs, through the step +//! loop) and [`HostTokenStrategy`] (an [`AuthStrategy`] that asks the host +//! for the bearer token). wasm32-only — the token is an imported function. +//! +//! # Requests +//! +//! The guest never waits on the network. A ZeroKMS request is recorded for +//! the step being run (`stack_guest_abi::step::request`) and the export +//! returns to the host with it, paused; the host performs it and resumes the +//! operation with the response ([`crate::io`] has the wire form). So a host +//! whose HTTP client cannot block — `fetch` on the edge — can run the guest, +//! and the guest is never running while a request is in flight. //! //! # Import contract (module `cipherstash_transport`) //! @@ -9,11 +18,9 @@ //! guest buffers with `se_alloc` and the guest reclaims them through the //! shared registry (`stack_guest_abi::buffers`). //! -//! - `transport_send(..)` — perform one HTTP request. The import and its -//! contract are `stack_guest_abi::transport`'s, shared with every guest; -//! this module only builds ZeroKMS requests over it. //! - `token_get(token_out) -> status` — hand over the current bearer token -//! (Phase-1 auth: minting and refresh stay on the host). `token_out` is a +//! (Phase-1 auth: minting and refresh stay on the host, which holds a +//! fresh token before each step, so this never waits). `token_out` is a //! `(ptr_out, len_out)` slot pair filled with an `se_alloc`'d buffer; //! status `0` is success, anything else a host-side failure. This //! guest's own: the credential guest supplies tokens rather than asking @@ -34,7 +41,7 @@ use std::sync::Mutex; use stack_auth::{AuthError, AuthStrategy, CustomError, SecretToken, ServiceToken}; use stack_guest_abi::buffers; use stack_guest_abi::headers::header_value; -use stack_guest_abi::transport; +use stack_guest_abi::step; use stack_kms::{BaseUrlUnresolved, ZeroKMSConnection, ZeroKMSConnectionInit, ZeroKmsEndpoint}; use zeroize::Zeroizing; use zerokms_protocol::{ViturRequest, ViturRequestError}; @@ -47,9 +54,9 @@ extern "C" { fn token_get(token_ptr_out: *mut u32, token_len_out: *mut u32) -> i32; } -/// A [`ZeroKMSConnection`] whose transport is the `transport_send` host -/// import. The host call is synchronous from the guest's perspective, so -/// `send` resolves immediately — `block_on` in the ABI layer never parks. +/// A [`ZeroKMSConnection`] whose requests the host performs: `send` records +/// its request for the step being run and resolves when the host resumes the +/// operation with the response (see the [module docs](self)). /// /// The endpoint is pinned at init when the cipher config named one; /// otherwise it is discovered by `StackKms` from the access token's @@ -126,11 +133,12 @@ impl ZeroKMSConnection for WasiHostConnection { let auth = Zeroizing::new(format!("Bearer {access_token}")); let headers = Zeroizing::new(request_headers(auth.as_str())); - // The shared import reclaims both response slots before judging - // either, and the body it hands back wipes on drop (it carries - // wrapped key material). - let response = transport::send("POST", url.as_str(), &headers, &body) - .map_err(|e| ViturRequestError::parse("Host response buffer failed validation", e))?; + // The request's headers and body wipe on drop, and so does the + // response's body: it carries wrapped key material. + let request = step::Request::new("POST", url.to_string(), headers, body); + let response = step::request(request) + .await + .map_err(|e| ViturRequestError::prepare("The host was not asked for the request", e))?; let content_type = header_value(&response.headers, "content-type"); map_response(response.status, content_type, &response.body) diff --git a/languages/golang/encrypt/guest/src/io.rs b/languages/golang/encrypt/guest/src/io.rs new file mode 100644 index 000000000..13618e8d2 --- /dev/null +++ b/languages/golang/encrypt/guest/src/io.rs @@ -0,0 +1,188 @@ +//! How a step of the step loop crosses the boundary +//! ([`stack_guest_abi::step`]): the requests a paused operation hands the +//! host, and the responses the host hands back. Both are transport frames, +//! read and written with serde. +//! +//! When an export pauses it returns `STATUS_NEEDS_IO` instead of an output, +//! and the host takes the step with `se_io`: +//! +//! ```text +//! { "handle": , +//! "requests": [ { "method": "POST", "url": , +//! "headers": , "body": }, ... ] } +//! ``` +//! +//! The host performs every request and calls `se_resume` with the handle and +//! their responses, in the same order: +//! +//! ```text +//! [ { "status": , "headers": , "body": }, ... ] +//! ``` +//! +//! `headers` is the `name: value` line format of +//! [`stack_guest_abi::headers`] both ways. A `status` is the HTTP status, or +//! negative for a request the host could not perform, with its error text as +//! the body: the contract `transport_send` had, field for field. A step +//! carries the bearer token and a request body that can hold key-material +//! contexts, and a response can carry wrapped keys, so the guest wipes its +//! copies, and the host wipes the buffers with `se_dealloc`. + +use serde::{Deserialize, Serialize, Serializer}; +use stack_guest_abi::last_error::malformed; +use stack_guest_abi::step::{Request, Response}; +use vitaminc_aead_frame::serde::{from_slice, to_vec}; +use zeroize::Zeroizing; + +use crate::status::STATUS_ENCODING; + +/// The step the host takes with `se_io`, as frame bytes. The bytes carry a +/// bearer token, so the caller registers them and the host's `se_dealloc` +/// wipes them; the staging buffer is wiped here. +pub fn encode_step(handle: u64, requests: &[Request]) -> Result, u32> { + #[derive(Serialize)] + struct Wire<'a> { + handle: u64, + requests: Vec>, + } + #[derive(Serialize)] + struct RequestWire<'a> { + method: &'a str, + url: &'a str, + headers: Bytes<'a>, + body: Bytes<'a>, + } + let wire = Wire { + handle, + requests: requests + .iter() + .map(|request| RequestWire { + method: request.method, + url: &request.url, + headers: Bytes(&request.headers), + body: Bytes(&request.body), + }) + .collect(), + }; + let frame = Zeroizing::new(to_vec(&wire).map_err(|_| STATUS_ENCODING)?); + // Sized exactly, as every output is: the registry would otherwise + // reallocate a buffer with spare capacity and leave the old one unwiped. + let mut out = Vec::new(); + out.try_reserve_exact(frame.len()) + .map_err(|_| stack_guest_abi::last_error::internal("no memory for a step's requests"))?; + out.extend_from_slice(&frame); + Ok(out) +} + +/// The responses the host handed `se_resume`, copied out of its buffer into +/// ones the guest wipes. +pub fn decode_responses(bytes: &[u8]) -> Result, u32> { + #[derive(Deserialize)] + #[serde(deny_unknown_fields)] + struct ResponseWire<'a> { + status: i32, + #[serde(borrow)] + headers: &'a [u8], + #[serde(borrow)] + body: &'a [u8], + } + let wire: Vec> = + from_slice(bytes).map_err(|e| malformed(format!("the responses: {e}")))?; + Ok(wire + .into_iter() + .map(|r| { + Response::new( + r.status, + r.headers.to_vec(), + Zeroizing::new(r.body.to_vec()), + ) + }) + .collect()) +} + +/// Bytes, written as a bytes leaf. +struct Bytes<'a>(&'a [u8]); + +impl Serialize for Bytes<'_> { + fn serialize(&self, s: S) -> Result { + s.serialize_bytes(self.0) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_frame::{frame, obj, V}; + + #[test] + fn a_step_is_its_handle_and_requests() { + let request = Request::new( + "POST", + "https://kms.example/keys".to_owned(), + Zeroizing::new(b"authorization: Bearer t\n".to_vec()), + Zeroizing::new(b"{\"k\":1}".to_vec()), + ); + let bytes = encode_step(7, &[request]).unwrap(); + let expected = frame(&obj(vec![ + ("handle", V::U64(7)), + ( + "requests", + V::Arr(vec![obj(vec![ + ("method", V::Str("POST".into())), + ("url", V::Str("https://kms.example/keys".into())), + ("headers", V::Bytes(b"authorization: Bearer t\n".to_vec())), + ("body", V::Bytes(b"{\"k\":1}".to_vec())), + ])]), + ), + ])); + assert_eq!(bytes, expected); + assert_eq!(bytes.capacity(), bytes.len()); + } + + #[test] + fn responses_read_in_order() { + let bytes = frame(&V::Arr(vec![ + obj(vec![ + ("status", V::I32(200)), + ( + "headers", + V::Bytes(b"content-type: application/json\n".to_vec()), + ), + ("body", V::Bytes(b"{}".to_vec())), + ]), + obj(vec![ + ("status", V::I32(-1)), + ("headers", V::Bytes(Vec::new())), + ("body", V::Bytes(b"dial tcp: refused".to_vec())), + ]), + ])); + let responses = decode_responses(&bytes).unwrap(); + assert_eq!(responses.len(), 2); + assert_eq!(responses[0].status, 200); + assert_eq!(&*responses[0].body, b"{}"); + assert_eq!(responses[1].status, -1); + assert_eq!(&*responses[1].body, b"dial tcp: refused"); + } + + #[test] + fn malformed_responses_are_refused_without_echoing_them() { + for bad in [ + V::Null, + obj(vec![("status", V::I32(200))]), + V::Arr(vec![obj(vec![ + ("status", V::I32(200)), + ("headers", V::Str("secret-looking".into())), + ("body", V::Bytes(Vec::new())), + ])]), + V::Arr(vec![obj(vec![ + ("status", V::I32(200)), + ("headers", V::Bytes(Vec::new())), + ("body", V::Bytes(Vec::new())), + ("extra", V::Bool(true)), + ])]), + ] { + assert_eq!(decode_responses(&frame(&bad)).err(), Some(STATUS_ENCODING)); + let error = stack_guest_abi::last_error::take().unwrap(); + assert!(!String::from_utf8_lossy(&error).contains("secret-looking")); + } + } +} diff --git a/languages/golang/encrypt/guest/src/lib.rs b/languages/golang/encrypt/guest/src/lib.rs index fefd48176..2d3c7acd3 100644 --- a/languages/golang/encrypt/guest/src/lib.rs +++ b/languages/golang/encrypt/guest/src/lib.rs @@ -76,7 +76,8 @@ //! - [`abi`], [`host`] (wasm32 only) — this guest's export surface and its //! token import. The conventions every guest shares — `se_alloc` / //! `se_dealloc`, the buffer registry, the packed result encoding, the -//! status table, the `transport_send` import — are `stack_guest_abi`'s; +//! status table, the step loop (`stack_guest_abi::step`) — are +//! `stack_guest_abi`'s, and [`io`] is how this guest's steps cross; //! [`abi`]'s module docs give this guest's contract on top of them. //! //! On wasm32 `vitaminc-encrypt` uses its pure-Rust (RustCrypto `aes-gcm`) @@ -91,6 +92,7 @@ pub mod config; #[cfg(feature = "deterministic-kms")] pub mod deterministic; pub mod headers; +pub mod io; pub mod ops; pub mod options; pub mod response; diff --git a/languages/golang/encrypt/guest/src/test_frame.rs b/languages/golang/encrypt/guest/src/test_frame.rs index f18ee4d5a..dc4e77943 100644 --- a/languages/golang/encrypt/guest/src/test_frame.rs +++ b/languages/golang/encrypt/guest/src/test_frame.rs @@ -10,6 +10,7 @@ pub(crate) enum V { Arr(Vec), Str(String), Bytes(Vec), + I32(i32), U32(u32), U64(u64), Bool(bool), @@ -51,6 +52,7 @@ pub(crate) fn frame(v: &V) -> Vec { } V::Str(text) => scalar(to_vec(text)), V::Bytes(bytes) => scalar(to_vec(&RawBytes(bytes))), + V::I32(n) => scalar(to_vec(n)), V::U32(n) => scalar(to_vec(n)), V::U64(n) => scalar(to_vec(n)), V::Bool(b) => scalar(to_vec(b)), diff --git a/languages/golang/encrypt/guest/tests/leak.rs b/languages/golang/encrypt/guest/tests/leak.rs index 6912ec3d8..d16656f38 100644 --- a/languages/golang/encrypt/guest/tests/leak.rs +++ b/languages/golang/encrypt/guest/tests/leak.rs @@ -13,9 +13,10 @@ //! back through `se_last_error`'s packing. Each scenario also pins the codes //! it reaches, so a path that stops being driven fails here, by name, //! instead of quietly dropping out of the check. -//! Unreachable natively, and so not here: the host imports (`token_get`, -//! `transport_send`), whose failures arrive as stack-auth and ZeroKMS -//! errors the key-source and token scenarios below already encode. +//! Unreachable natively, and so not here: the host's side of the step loop +//! (the `token_get` import, and the responses `se_resume` is handed), whose +//! failures arrive as stack-auth and ZeroKMS errors the key-source and token +//! scenarios below already encode. use std::borrow::Cow; use std::future::IntoFuture; diff --git a/languages/golang/encrypt/guest_test.go b/languages/golang/encrypt/guest_test.go index 9981d753e..e1cf3b7ec 100644 --- a/languages/golang/encrypt/guest_test.go +++ b/languages/golang/encrypt/guest_test.go @@ -113,7 +113,9 @@ func testInit(url string) initConfig { return initConfig{clientID: testClientID, clientKey: NewClientKey([]byte(testClientKey)), zerokmsURL: url} } -func TestImportSurfaceIsWASIPlusTransport(t *testing.T) { +// The guest reaches its host through one import, token_get: its ZeroKMS +// requests come back through the step loop, so it has no HTTP import. +func TestImportSurfaceIsWASIPlusTokenGet(t *testing.T) { ctx := context.Background() r := wazero.NewRuntime(ctx) defer r.Close(ctx) @@ -138,12 +140,12 @@ func TestImportSurfaceIsWASIPlusTransport(t *testing.T) { t.Errorf("guest imports %s::%s, outside the allowed surface", module, name) } } - want := []string{"token_get", "transport_send"} + want := []string{"token_get"} sort.Strings(transportImports) if !reflect.DeepEqual(transportImports, want) { t.Fatalf("transport imports = %v, want %v", transportImports, want) } - for name := range map[string]bool{"se_alloc": true, "se_dealloc": true, "se_cipher_init": true, "se_shutdown": true, "se_keyset": true, "se_term": true, "se_query_term": true, "se_verify": true, "se_target_limit": true, "se_encrypt_record": true, "se_decrypt_record": true, "se_plan_check": true, "se_targets": true, "se_operators": true} { + for name := range map[string]bool{"se_alloc": true, "se_dealloc": true, "se_cipher_init": true, "se_shutdown": true, "se_keyset": true, "se_term": true, "se_query_term": true, "se_verify": true, "se_target_limit": true, "se_encrypt_record": true, "se_decrypt_record": true, "se_plan_check": true, "se_targets": true, "se_operators": true, "se_io": true, "se_resume": true, "se_abort": true} { if _, ok := compiled.ExportedFunctions()[name]; !ok { t.Errorf("guest does not export %s", name) } @@ -912,7 +914,10 @@ func TestClientKeyDoesNotRemainInGuestMemory(t *testing.T) { t.Fatal(err) } defer func() { _ = inst.release() }() - _, err = inst.call(ctx, inst.cipherInit, buf(encoded)) + c := newClient(inst, tr) + _, err = c.callKMS(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { + return inst.call(ctx, inst.cipherInit, buf(encoded)) + }) if !errors.Is(err, ErrUnauthorized) { t.Fatalf("init: %v", err) } @@ -937,7 +942,9 @@ func TestClientKeyDoesNotRemainInGuestMemory(t *testing.T) { } } -func TestTransportSendCounterAndResponseHeaders(t *testing.T) { +// The init's one load-keyset request is one step: one request performed, +// and the operation's handle spent once it ends. +func TestEachRequestIsOneStep(t *testing.T) { guestOrSkip(t) stub := newStub(t, http.StatusUnauthorized, "text/plain", "nope") tr := &transport{rt: http.DefaultTransport, token: staticToken("stub-token")} @@ -947,13 +954,19 @@ func TestTransportSendCounterAndResponseHeaders(t *testing.T) { t.Fatal(err) } defer func() { _ = inst.release() }() + c := newClient(inst, tr) encoded, _ := encodeConfig(testInit(stub.URL)) - if _, err := inst.call(ctx, inst.cipherInit, buf(encoded)); !errors.Is(err, ErrUnauthorized) { + if _, err := c.callKMS(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { + return inst.call(ctx, inst.cipherInit, buf(encoded)) + }); !errors.Is(err, ErrUnauthorized) { t.Fatalf("init: %v", err) } if n := tr.sends.Load(); n != 1 { t.Fatalf("transport sends = %d, want 1", n) } + if _, err := inst.call(ctx, inst.abort, scalar(1)); !errors.Is(err, ErrState) { + t.Fatalf("aborting the finished operation's handle: %v, want ErrState", err) + } } func ExampleNewClient() { diff --git a/languages/golang/encrypt/query.go b/languages/golang/encrypt/query.go index d4953b4cf..540ec475e 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(ctx context.Context, inst *instance) ([]byte, error) { + out, err := cph.client.callKMS(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 { diff --git a/languages/golang/encrypt/records.go b/languages/golang/encrypt/records.go index b2d0ea745..6ff270867 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(ctx context.Context, inst *instance) ([]byte, error) { + out, err := cph.client.callKMS(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(ctx context.Context, inst *instance) ([]byte, error) { + return cph.client.callKMS(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.term, buf(encodedValue), buf(encodedContext), scalar(uint64(code)), buf(opts)) }) } @@ -408,7 +408,7 @@ func (c *Client) open(ctx context.Context, sel KeysetSelector, expected string, if err != nil { return nil, err } - out, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { + out, err := c.callKMS(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/step.go b/languages/golang/encrypt/step.go new file mode 100644 index 000000000..19d970a89 --- /dev/null +++ b/languages/golang/encrypt/step.go @@ -0,0 +1,178 @@ +package encrypt + +import ( + "context" + "errors" + "fmt" + + "github.com/cipherstash/stack/languages/golang/internal/guest" + "github.com/cipherstash/vitaminc/bindings/go/vcffi" + "github.com/cipherstash/vitaminc/bindings/go/vcvalue" +) + +// The step loop. The guest never waits on the network: an export that +// needs ZeroKMS pauses instead, returning the requests it is waiting on and +// a handle (guest.ErrNeedsIO, then se_io). The host performs the requests, +// outside the client's lock, and resumes the operation with the responses +// (se_resume), until it is done. So the lock is held only while the guest +// runs, never across a network round trip, and a caller that gives up +// between two steps aborts the operation (se_abort) while the guest is not +// running at all. + +// stepFunc runs one step of an operation in the guest, under ctx. +type stepFunc func(ctx context.Context, inst *instance) ([]byte, error) + +// ioStep is a paused operation's step, as the guest hands it over: the +// operation's handle and the requests it waits on. +type ioStep struct { + handle uint64 + requests []ioRequest +} + +// ioRequest is one HTTP request the guest needs performed. The headers +// carry the bearer token and the body key-material contexts, so both are +// wiped once performed. +type ioRequest struct { + method, url string + headers, body []byte +} + +func (s *ioStep) wipe() { + for _, r := range s.requests { + wipe(r.headers) + wipe(r.body) + } +} + +var errMalformedStep = errors.New("encrypt: the guest's step is malformed") + +// takeIO takes the step a paused export left (se_io), once. +func (inst *instance) takeIO(ctx context.Context) (*ioStep, error) { + // The same bracket guest.Call keeps: the memory stays mapped until the + // step is copied out and its buffer freed. + inst.mem.Enter() + defer inst.mem.Exit() + res, err := inst.io.Call(ctx) + if err != nil { + return nil, fmt.Errorf("%w: guest call: %w", errGuestTrap, err) + } + if len(res) == 0 || res[0]>>32 == 0 { + return nil, fmt.Errorf("%w: the guest paused without a step", guest.ErrInternal) + } + out := guest.Buf{Ptr: uint32(res[0] >> 32), Len: uint32(res[0])} //nolint:gosec // splits the packed u64 into its two u32 halves + defer inst.exports.Free(ctx, out) + view, ok := inst.module.Memory().Read(out.Ptr, out.Len) + if !ok { + return nil, errors.New("encrypt: guest returned an out-of-range buffer") + } + raw := make([]byte, len(view)) + copy(raw, view) + defer wipe(raw) + return decodeStep(raw) +} + +// decodeStep reads `{"handle": u64, "requests": [{"method", "url", +// "headers", "body"}, ...]}`. Every byte slice the decoder returns is its +// own copy. +func decodeStep(raw []byte) (*ioStep, error) { + decoded, err := vcffi.Unmarshal(raw) + if err != nil { + return nil, fmt.Errorf("%w: %w", errMalformedStep, err) + } + obj, ok := decoded.(vcvalue.Object) + if !ok { + return nil, errMalformedStep + } + step := &ioStep{} + var sawHandle bool + for _, f := range obj { + switch f.Key { + case "handle": + h, ok := f.Value.(uint64) + if !ok { + return nil, errMalformedStep + } + step.handle, sawHandle = h, true + case "requests": + items, ok := f.Value.([]any) + if !ok { + return nil, errMalformedStep + } + for _, item := range items { + r, ok := decodeRequest(item) + if !ok { + step.wipe() + return nil, errMalformedStep + } + step.requests = append(step.requests, r) + } + default: + step.wipe() + return nil, errMalformedStep + } + } + if !sawHandle || len(step.requests) == 0 { + step.wipe() + return nil, errMalformedStep + } + return step, nil +} + +func decodeRequest(item any) (ioRequest, bool) { + obj, ok := item.(vcvalue.Object) + if !ok { + return ioRequest{}, false + } + var r ioRequest + for _, f := range obj { + switch f.Key { + case "method": + r.method, ok = f.Value.(string) + case "url": + r.url, ok = f.Value.(string) + case "headers": + r.headers, ok = f.Value.([]byte) + case "body": + r.body, ok = f.Value.([]byte) + default: + ok = false + } + if !ok { + wipe(r.headers) + wipe(r.body) + return ioRequest{}, false + } + } + return r, r.method != "" && r.url != "" +} + +// performStep performs every request of a step, in order, and encodes the +// responses for se_resume: `[{"status": i32, "headers": bytes, "body": +// bytes}, ...]`. The responses can carry wrapped keys: the caller wipes the +// encoding once the guest has it, and every other copy is wiped here. +func (t *transport) performStep(ctx context.Context, step *ioStep) ([]byte, error) { + defer step.wipe() + responses := make([]any, 0, len(step.requests)) + var bodies [][]byte + defer func() { + for _, b := range bodies { + wipe(b) + } + }() + for _, r := range step.requests { + status, headers, body := t.perform(ctx, r.method, r.url, r.headers, r.body) + if headers == nil { + headers = []byte{} + } + if body == nil { + body = []byte{} + } + bodies = append(bodies, body) + responses = append(responses, vcvalue.Object{ + {Key: "status", Value: status}, + {Key: "headers", Value: headers}, + {Key: "body", Value: body}, + }) + } + return vcffi.Marshal(responses) +} diff --git a/languages/golang/encrypt/step_test.go b/languages/golang/encrypt/step_test.go new file mode 100644 index 000000000..f9977d04f --- /dev/null +++ b/languages/golang/encrypt/step_test.go @@ -0,0 +1,313 @@ +package encrypt + +import ( + "bytes" + "context" + "errors" + "io" + "io/fs" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "regexp" + "strconv" + "sync/atomic" + "testing" + "time" + + "github.com/cipherstash/stack/languages/golang/internal/guest" + "github.com/cipherstash/vitaminc/bindings/go/vcffi" +) + +// The step loop, driven hermetically: the guest's init pauses on its +// load-keyset request, which a stub ZeroKMS holds until the test releases +// it. + +// heldStub answers every request with 401 once released, and reports each +// request as it arrives. +type heldStub struct { + *httptest.Server + arrived chan struct{} + release chan struct{} +} + +func newHeldStub(t *testing.T) *heldStub { + t.Helper() + s := &heldStub{arrived: make(chan struct{}, 8), release: make(chan struct{})} + s.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + s.arrived <- struct{}{} + select { + case <-s.release: + case <-r.Context().Done(): + return + } + w.WriteHeader(http.StatusUnauthorized) + _, _ = io.WriteString(w, "nope") + })) + t.Cleanup(func() { + select { + case <-s.release: + default: + close(s.release) + } + s.Close() + }) + return s +} + +// stepClient is a client over a fresh instance whose init has not run. +func stepClient(t *testing.T, token tokenSource) *Client { + t.Helper() + tr := &transport{rt: http.DefaultTransport, 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 +} + +func initStep(t *testing.T, url string) stepFunc { + t.Helper() + encoded, err := encodeConfig(testInit(url)) + if err != nil { + t.Fatal(err) + } + return func(ctx context.Context, inst *instance) ([]byte, error) { + return inst.call(ctx, inst.cipherInit, buf(encoded)) + } +} + +func targetsStep(ctx context.Context, inst *instance) ([]byte, error) { + return inst.call(ctx, inst.targets) +} + +func waitFor(t *testing.T, ch <-chan struct{}, what string) { + t.Helper() + select { + case <-ch: + case <-time.After(10 * time.Second): + t.Fatalf("timed out waiting for %s", what) + } +} + +// While an operation's request is in flight the guest is not running, and +// the client's lock is free: another call runs to completion meanwhile. +func TestAStepDoesNotHoldTheLockWhileItsRequestRuns(t *testing.T) { + stub := newHeldStub(t) + c := stepClient(t, staticToken("stub-token")) + done := make(chan error, 1) + go func() { + _, err := c.callKMS(context.Background(), initStep(t, stub.URL)) + done <- err + }() + waitFor(t, stub.arrived, "the init's request") + if _, err := c.call(context.Background(), targetsStep); err != nil { + t.Fatalf("a call while the init's request is in flight: %v", err) + } + close(stub.release) + if err := <-done; !errors.Is(err, ErrUnauthorized) { + t.Fatalf("init: %v, want ErrUnauthorized", err) + } +} + +// The guest refuses an operation past its cap of paused operations, so the +// client never starts one: a caller past the cap waits for a slot, leaves +// the queue when its context ends, and runs once a slot is free. Here the +// cap is one, so the paused init holds the only slot. +func TestACallerPastThePausedCapWaitsForASlot(t *testing.T) { + stub := newHeldStub(t) + c := stepClient(t, staticToken("stub-token")) + if got := cap(c.paused); got != maxPaused { + t.Fatalf("paused slots: %d, want maxPaused (%d)", got, maxPaused) + } + c.paused = make(chan struct{}, 1) + first := make(chan error, 1) + go func() { + _, err := c.callKMS(context.Background(), initStep(t, stub.URL)) + first <- err + }() + waitFor(t, stub.arrived, "the first init's request") + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if _, err := c.callKMS(ctx, initStep(t, stub.URL)); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("a call past the cap: %v, want context.DeadlineExceeded (it waited)", err) + } + + second := make(chan error, 1) + go func() { + _, err := c.callKMS(context.Background(), initStep(t, stub.URL)) + second <- err + }() + select { + case <-stub.arrived: + t.Fatal("a call past the cap sent a request while the slot was held") + case <-time.After(100 * time.Millisecond): + } + close(stub.release) + if err := <-first; !errors.Is(err, ErrUnauthorized) { + t.Fatalf("first init: %v, want ErrUnauthorized", err) + } + if err := <-second; !errors.Is(err, ErrUnauthorized) { + t.Fatalf("the waiting init: %v, want ErrUnauthorized (it ran once the slot was free)", err) + } + if n := len(c.paused); n != 0 { + t.Fatalf("%d slots still held after every call ended", n) + } +} + +// maxPaused is the guest's MAX_OPEN, written twice: a Go bound above the +// guest's cap lets the guest refuse a caller the client should have +// queued. +func TestMaxPausedIsTheGuestsCap(t *testing.T) { + src, err := os.ReadFile(filepath.Join("..", "..", "..", "packages", "stack-guest-abi", "src", "step.rs")) + if errors.Is(err, fs.ErrNotExist) { + t.Skip("stack-guest-abi's source is not beside this module") + } + if err != nil { + t.Fatal(err) + } + m := regexp.MustCompile(`pub const MAX_OPEN: usize = (\d+);`).FindSubmatch(src) + if m == nil { + t.Fatal("MAX_OPEN not found in stack-guest-abi's step.rs") + } + if got := string(m[1]); got != strconv.Itoa(maxPaused) { + t.Fatalf("the guest's MAX_OPEN is %s, maxPaused is %d", got, maxPaused) + } +} + +// A caller that gives up while its request is in flight returns with its +// context's error. The operation is aborted in the guest, so its handle is +// spent and a new init may start; the client stays open. +func TestACallerThatGivesUpBetweenStepsAbortsTheOperation(t *testing.T) { + stub := newHeldStub(t) + c := stepClient(t, staticToken("stub-token")) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + _, err := c.callKMS(ctx, initStep(t, stub.URL)) + done <- err + }() + waitFor(t, stub.arrived, "the init's request") + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("cancelled init: %v, want context.Canceled", err) + } + if c.closed || c.inst.module.IsClosed() { + t.Fatal("giving up closed the client") + } + if _, err := c.inst.call(context.Background(), c.inst.abort, scalar(1)); !errors.Is(err, ErrState) { + t.Fatalf("the aborted operation's handle: %v, want ErrState (already dropped)", err) + } + close(stub.release) + if _, err := c.callKMS(context.Background(), initStep(t, stub.URL)); !errors.Is(err, ErrUnauthorized) { + t.Fatalf("an init after the aborted one: %v, want ErrUnauthorized (it runs)", err) + } +} + +// A context that has already ended never reaches the guest. +func TestACallWhoseContextHasEndedNeverStarts(t *testing.T) { + stub := newStub(t, http.StatusUnauthorized, "", "nope") + c := stepClient(t, staticToken("stub-token")) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := c.callKMS(ctx, initStep(t, stub.URL)); !errors.Is(err, context.Canceled) { + t.Fatalf("init under an ended context: %v, want context.Canceled", err) + } + if len(stub.requests) != 0 { + t.Fatalf("%d requests sent for a call that never started", len(stub.requests)) + } +} + +// lockProbe is a token source that records whether the client's lock was +// free when it was asked. +type lockProbe struct { + c **Client + asked, held atomic.Int32 +} + +func (p *lockProbe) Token(context.Context) (string, error) { + p.asked.Add(1) + if !(*p.c).mu.TryLock() { + p.held.Add(1) + } else { + (*p.c).mu.Unlock() + } + return "stub-token", nil +} + +// The token is fetched once per operation, before its first step and +// outside the client's lock, so a refresh that waits on the network holds +// up no other caller. +func TestTheTokenIsFetchedOutsideTheLock(t *testing.T) { + stub := newStub(t, http.StatusUnauthorized, "", "nope") + var c *Client + probe := &lockProbe{c: &c} + c = stepClient(t, probe) + if _, err := c.callKMS(context.Background(), initStep(t, stub.URL)); !errors.Is(err, ErrUnauthorized) { + t.Fatalf("init: %v", err) + } + if probe.asked.Load() == 0 { + t.Fatal("the token source was never asked") + } + if n := probe.held.Load(); n != 0 { + t.Fatalf("the token was fetched %d times under the client's lock", n) + } + if got := stub.requests[0].auth; got != "Bearer stub-token" { + t.Fatalf("the request carried %q", got) + } +} + +// An operation paused on its request and then aborted leaves neither the +// client key nor the bearer token in guest memory: dropping it wipes what +// it held. +func TestAnAbortedInitLeavesNoKeyInGuestMemory(t *testing.T) { + stub := newHeldStub(t) + c := stepClient(t, staticToken("stub-token")) + ctx := context.Background() + out, step, err := c.step(ctx, "stub-token", nil, initStep(t, stub.URL)) + if err != nil || step == nil { + t.Fatalf("the init's first step: %v, %v; want a pause", out, err) + } + if len(step.requests) != 1 || !bytes.Contains(step.requests[0].headers, []byte("Bearer stub-token")) { + t.Fatalf("the init's step: %+v", step.requests) + } + step.wipe() + c.abort(step.handle) + mem := c.inst.module.Memory() + view, ok := mem.Read(0, mem.Size()) + if !ok { + t.Fatal("cannot read guest memory") + } + for name, needle := range map[string][]byte{ + "key hex": []byte(testClientKey), + "key bytes": mustHex(testClientKey), + "bearer": []byte("stub-token"), + } { + if n := bytes.Count(view, needle); n != 0 { + t.Errorf("%s found %d times in guest memory after an aborted init", name, n) + } + } +} + +// A step the host cannot decode is refused, never performed. +func TestAMalformedStepIsRefused(t *testing.T) { + for _, raw := range [][]byte{nil, {0xff}, marshalStep(t, map[string]any{"handle": uint64(1)})} { + if _, err := decodeStep(raw); !errors.Is(err, errMalformedStep) { + t.Errorf("decodeStep(%x): %v, want errMalformedStep", raw, err) + } + } +} + +func marshalStep(t *testing.T, v any) []byte { + t.Helper() + b, err := vcffi.Marshal(v) + if err != nil { + t.Fatal(err) + } + return b +} diff --git a/languages/golang/encrypt/transport.go b/languages/golang/encrypt/transport.go index 2e1005b44..d463a6b22 100644 --- a/languages/golang/encrypt/transport.go +++ b/languages/golang/encrypt/transport.go @@ -14,52 +14,54 @@ 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 -// functions are the whole host surface the guest can reach. +// transportModule is the name of the guest's one import module. Its one +// function, token_get, is the whole host surface the guest can reach: its +// ZeroKMS requests come back to the host through the step loop (step.go). const transportModule = "cipherstash_transport" // tokenSource supplies the bearer token the guest presents to ZeroKMS. It -// is asked on every request, so a source that rotates tokens needs no -// re-initialisation of the client. Outside this package's tests it is -// always a *auth.Strategy: minting and refresh stay host-side, out of -// the crypto guest, in auth's credential guest. The interface is -// unexported so that no caller can hand the client a raw token, which could -// not be refreshed and would bypass the strategies' refresh lock. +// is asked once before each operation that can reach ZeroKMS, so a source +// that rotates tokens needs no re-initialisation of the client. +// Outside this package's tests it is always a *auth.Strategy: minting and +// refresh stay host-side, out of the crypto guest, in auth's credential +// guest. The interface is unexported so that no caller can hand the client +// a raw token, which could not be refreshed and would bypass the +// strategies' refresh lock. type tokenSource interface { Token(ctx context.Context) (string, error) } -// transport implements the guest's two host imports over a RoundTripper -// and a tokenSource. One per Client; it is bound to the module at -// instantiation and reaches the guest's allocator through the module the -// call arrives on. +// transport performs the guest's ZeroKMS requests over a RoundTripper and +// implements its token_get import over a tokenSource. One per Client. type transport struct { rt http.RoundTripper token tokenSource - // sends counts transport_send excursions, so tests can pin the batching + // sends counts the requests performed, so tests can pin the batching // contract (one ZeroKMS call per operation) instead of trusting it. sends atomic.Int64 - // tokenErr is why the token source last failed during the call in - // flight: the guest sees only that token_get failed, so Client.call - // attaches the cause — ErrNoCredentials, a refused refresh — to the - // error it returns. Only touched under the client's lock, which every - // guest call holds. + // held is the token fetched for the operation in flight, which + // token_get hands over, and fetchErr why the source gave none. tokenErr is why + // token_get failed, when the guest asked: the guest sees only that it + // failed, so Client.call attaches the cause — ErrNoCredentials, a + // refused refresh — to the error it returns. All three are set for each + // step and only touched under the client's lock, which every step + // holds. + held string + fetchErr error tokenErr error } -// transportFailed is the return value of transport_send when the request -// could not be performed at all; the body then carries the error text. +// transportFailed is the status of a request that could not be performed +// at all; the body then carries the error text. const transportFailed int32 = -1 // hostFailed is the return value of token_get when no token is available. const hostFailed int32 = 1 -// maxResponseBytes bounds what transport_send will buffer from ZeroKMS. The -// guest issues at most one 500-key batch per request, which is well under a +// maxResponseBytes bounds what perform will buffer from ZeroKMS. The guest +// issues at most one 500-key batch per request, which is well under a // megabyte either way; the bound exists so that an endpoint the transport // was pointed at cannot make the host allocate without limit. const maxResponseBytes = 16 << 20 @@ -67,7 +69,6 @@ const maxResponseBytes = 16 << 20 // instantiate registers the host module in r. func (t *transport) instantiate(ctx context.Context, r wazero.Runtime) error { _, err := r.NewHostModuleBuilder(transportModule). - NewFunctionBuilder().WithFunc(t.send).Export("transport_send"). NewFunctionBuilder().WithFunc(t.tokenGet).Export("token_get"). Instantiate(ctx) if err != nil { @@ -76,42 +77,29 @@ func (t *transport) instantiate(ctx context.Context, r wazero.Runtime) error { return nil } -// send is transport_send: one HTTP request on the guest's behalf. Inputs -// are (ptr, len) pairs borrowed for the call; the two outputs are slot -// pairs filled with buffers obtained from the guest's se_alloc. Returns -// the HTTP status, or transportFailed with the error text as the body. -func (t *transport) send(ctx context.Context, m api.Module, - methodPtr, methodLen, urlPtr, urlLen, headersPtr, headersLen, bodyPtr, bodyLen uint32, - respHeadersPtrOut, respHeadersLenOut, respBodyPtrOut, respBodyLenOut uint32, -) int32 { - t.sends.Add(1) - mem := m.Memory() - status, respHeaders, respBody := t.perform(ctx, mem, - methodPtr, methodLen, urlPtr, urlLen, headersPtr, headersLen, bodyPtr, bodyLen) - // The response carries wrapped key material; once it is in guest memory - // the host copy is wiped. - defer wipe(respBody) - if !place(ctx, m, respHeadersPtrOut, respHeadersLenOut, respHeaders) || - !place(ctx, m, respBodyPtrOut, respBodyLenOut, respBody) { - // The guest reclaims whatever was placed and refuses an unplaced - // slot as an unregistered buffer; nothing more this side can do. - return transportFailed +// fetchToken asks the token source for the token an operation's steps hand +// the guest. It runs before the first step, outside the client's lock, so a +// refresh that waits on the network holds up no other caller; the steps +// then never wait. A failure is kept for a step to report if the guest +// asks. +func (t *transport) fetchToken(ctx context.Context) (string, error) { + if t.token == nil { + return "", nil } - return status + token, err := t.token.Token(ctx) + if err == nil && token == "" { + err = errors.New("encrypt: the token source returned an empty token") + } + return token, err } -func (t *transport) perform(ctx context.Context, mem api.Memory, - methodPtr, methodLen, urlPtr, urlLen, headersPtr, headersLen, bodyPtr, bodyLen uint32, -) (int32, []byte, []byte) { - method, ok1 := mem.Read(methodPtr, methodLen) - url, ok2 := mem.Read(urlPtr, urlLen) - headers, ok3 := mem.Read(headersPtr, headersLen) - body, ok4 := mem.Read(bodyPtr, bodyLen) - if !ok1 || !ok2 || !ok3 || !ok4 { - return transportFailed, nil, []byte("guest request buffers out of range") - } +// perform is one HTTP request on the guest's behalf. It returns the HTTP +// status, the response headers in the guest's line format and the body, +// or transportFailed with the error text as the body. +func (t *transport) perform(ctx context.Context, method, url string, headers, body []byte) (int32, []byte, []byte) { + t.sends.Add(1) // The request body may carry key-material contexts; it is copied - // because the guest wipes its own buffer when the call returns, and the + // because the caller wipes its own copy when the step ends, and the // RoundTripper may read it after this function has. The copy is owned // by the request body handed to the RoundTripper and is wiped when that // body is closed — not here: RoundTrip may go on reading, and close, @@ -119,10 +107,7 @@ 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) - // 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) + req, err := http.NewRequestWithContext(ctx, method, url, reqBody) if err != nil { _ = reqBody.Close() return transportFailed, nil, []byte(err.Error()) @@ -215,17 +200,15 @@ func (b *requestBody) Close() error { return nil } -// tokenGet is token_get: hand the guest the current bearer token. +// tokenGet is token_get: hand the guest the token fetched for this step. +// It never waits: the fetch happened before the step (fetchToken). func (t *transport) tokenGet(ctx context.Context, m api.Module, tokenPtrOut, tokenLenOut uint32) int32 { - // 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 - } + token := t.held if token == "" { - t.tokenErr = errors.New("encrypt: the token source returned an empty token") + t.tokenErr = t.fetchErr + if t.tokenErr == nil { + t.tokenErr = errors.New("encrypt: no token was fetched for this step") + } return hostFailed } // The credential's transport copy is wiped once it is in guest memory; diff --git a/languages/golang/internal/guest/call.go b/languages/golang/internal/guest/call.go index 0c0c48d4a..6c2619000 100644 --- a/languages/golang/internal/guest/call.go +++ b/languages/golang/internal/guest/call.go @@ -115,6 +115,10 @@ func Call(ctx context.Context, mem *Allocator, m api.Module, e Exports, fn api.F return nil, fmt.Errorf("%w: guest call: %w", ErrTrap, err) } ptr, n, err := PackedResult(res[0]) + if errors.Is(err, ErrNeedsIO) { + // A pause records no error: there is nothing to diagnose. + return nil, err + } if err != nil { return nil, e.diagnose(ctx, m, err) } diff --git a/languages/golang/internal/guest/errors.go b/languages/golang/internal/guest/errors.go index aec29155f..9020d2a2a 100644 --- a/languages/golang/internal/guest/errors.go +++ b/languages/golang/internal/guest/errors.go @@ -86,6 +86,11 @@ var ( // ErrContextMismatch is a row whose stored context field is not the // context the caller named, refused before any key is retrieved. ErrContextMismatch = errors.New("cipherstash: the stored context is not the one expected") + // ErrNeedsIO is not a failure: the export paused, waiting for the host + // to perform HTTP requests on its behalf and resume it (the step loop). + // A guest package's call loop takes the step, performs it and resumes; + // it never reaches a caller. + ErrNeedsIO = errors.New("cipherstash: the guest is waiting on the host for I/O") // ErrMemoryLock is guest memory that could not be locked in RAM (or, // on Linux, excluded from core dumps). A constructor returns it when // asked for locked memory and refused, and so does any later call under diff --git a/languages/golang/internal/guest/lock.go b/languages/golang/internal/guest/lock.go index 5452837c0..febdd1718 100644 --- a/languages/golang/internal/guest/lock.go +++ b/languages/golang/internal/guest/lock.go @@ -71,6 +71,17 @@ func (l *Lock) Acquire(ctx context.Context) error { // skippable. func (l *Lock) Lock() { l.slot() <- struct{}{} } +// TryLock takes the lock if it is free, without waiting, and reports +// whether it did. +func (l *Lock) TryLock() bool { + select { + case l.slot() <- struct{}{}: + return true + default: + return false + } +} + // Unlock releases the lock. func (l *Lock) Unlock() { <-l.slot() } diff --git a/languages/golang/internal/guest/lock_test.go b/languages/golang/internal/guest/lock_test.go index d4ab83ec7..bdd0c9fc7 100644 --- a/languages/golang/internal/guest/lock_test.go +++ b/languages/golang/internal/guest/lock_test.go @@ -34,6 +34,21 @@ func TestLockWaitEndsWithTheContext(t *testing.T) { } } +func TestTryLockTakesOnlyAFreeLock(t *testing.T) { + var l Lock + if !l.TryLock() { + t.Fatal("TryLock on a free lock did not take it") + } + if l.TryLock() { + t.Fatal("TryLock on a held lock took it") + } + l.Unlock() + if !l.TryLock() { + t.Fatal("TryLock after Unlock did not take it") + } + l.Unlock() +} + type valueKey struct{} func TestDetachKeepsValuesNotCancellation(t *testing.T) { diff --git a/languages/golang/internal/guest/status.go b/languages/golang/internal/guest/status.go index 130834b05..f93e6a1d9 100644 --- a/languages/golang/internal/guest/status.go +++ b/languages/golang/internal/guest/status.go @@ -37,6 +37,9 @@ const ( StatusAuthRefreshRequired = 27 // Appended by the crypto guest after the credential guest's codes. StatusContextMismatch = 28 + // Not a failure: the operation is paused, waiting on HTTP requests the + // host performs for it. See ErrNeedsIO. + StatusNeedsIO = 29 ) // StatusError is the sentinel a guest status decodes to. A status this host @@ -100,6 +103,8 @@ func StatusError(status uint32) error { return ErrAuthRefreshRequired case StatusContextMismatch: return ErrContextMismatch + case StatusNeedsIO: + return ErrNeedsIO default: return unrecognizedStatus(status) } diff --git a/mise.toml b/mise.toml index 99c81a215..e9c094ce9 100644 --- a/mise.toml +++ b/mise.toml @@ -159,7 +159,9 @@ module=target/wasm32-wasip1/release/stack_encrypt_guest.wasm echo "guest module: languages/golang/encrypt/guest/$module" # The security contract is about the *linked* module, which a successful # build says nothing about: the guest may reach the outside world only -# through the two host functions the Go embedder provides. Fail-closed — +# through the one host function the Go embedder provides, token_get (its +# ZeroKMS requests return to the host through the step loop, not an +# import). Fail-closed — # any other import, or a missing one, fails here rather than widening the # surface silently. WASI is allowed as a module but denied the # capability-granting half of its namespace (ambient filesystem and @@ -174,7 +176,6 @@ python3 "$root/scripts/check-wasm-imports.py" "$module" \\ --deny-prefix wasi_snapshot_preview1:sock_ \\ --deny-prefix wasi_snapshot_preview1:fd_prestat \\ --require wasi_snapshot_preview1:random_get \\ - --require cipherstash_transport:transport_send \\ --require cipherstash_transport:token_get # The Go module embeds the checked artefact (languages/golang/encrypt/wasm, # gitignored): copying it here is what makes `go test` in the binding run @@ -239,7 +240,6 @@ python3 "$root/scripts/check-wasm-imports.py" "$module" \\ --deny-prefix wasi_snapshot_preview1:sock_ \\ --deny-prefix wasi_snapshot_preview1:fd_prestat \\ --require wasi_snapshot_preview1:random_get \\ - --require cipherstash_transport:transport_send \\ --require cipherstash_transport:token_get mkdir -p ../eql/wasm cp "$module" ../eql/wasm/stack_encrypt_guest_eql.wasm diff --git a/packages/stack-guest-abi/Cargo.toml b/packages/stack-guest-abi/Cargo.toml index a7ed2db77..d125c2d48 100644 --- a/packages/stack-guest-abi/Cargo.toml +++ b/packages/stack-guest-abi/Cargo.toml @@ -36,3 +36,6 @@ vitaminc-protected = { workspace = true } # host holds, run natively and under Miri. `std` only: the `fork` and # `timeout` defaults spawn processes and threads Miri cannot interpret. proptest = { version = "1.7", default-features = false, features = ["std"] } +# `step`'s tests drive the loop through the combinators the ZeroKMS client +# runs its requests with (`buffered`, `join`). +futures = { version = "0.3", default-features = false, features = ["std", "async-await"] } diff --git a/packages/stack-guest-abi/src/buffers.rs b/packages/stack-guest-abi/src/buffers.rs index f5cc9207b..109b56e9f 100644 --- a/packages/stack-guest-abi/src/buffers.rs +++ b/packages/stack-guest-abi/src/buffers.rs @@ -135,6 +135,7 @@ pub fn wipe_all() { // The recorded last error is one of these buffers; its slot must not // outlive the wipe and name freed memory. crate::last_error::forget(); + crate::step::forget_pending(); let live = BUFFERS.with(|b| core::mem::take(&mut *b.borrow_mut())); for (ptr, len) in live { // SAFETY: every entry was registered by `register`, which leaked a diff --git a/packages/stack-guest-abi/src/call.rs b/packages/stack-guest-abi/src/call.rs index c1b5bdc49..9ed9f9e78 100644 --- a/packages/stack-guest-abi/src/call.rs +++ b/packages/stack-guest-abi/src/call.rs @@ -16,6 +16,7 @@ use std::panic::{catch_unwind, AssertUnwindSafe}; use crate::buffers; use crate::last_error; +use crate::status::STATUS_NEEDS_IO; /// Run an export's body: clear the last error, run `f` (a panic is an /// internal failure), and then clear the last error again if `f` succeeded, @@ -34,6 +35,11 @@ pub fn run(f: impl FnOnce() -> Result) -> Result { last_error::clear(); Ok(out) } + // A paused operation has not failed: nothing is recorded. + Err(STATUS_NEEDS_IO) => { + last_error::clear(); + Err(STATUS_NEEDS_IO) + } Err(status) => { last_error::ensure(status); Err(status) @@ -93,6 +99,18 @@ pub fn take_last_error() -> Option> { mod tests { use super::*; use crate::status::{STATUS_ENCODING, STATUS_INTERNAL}; + + /// A paused operation is not a failure: it records no error, and clears + /// one recorded on the way to the pause. + #[test] + fn a_pause_records_no_error() { + let paused = run::<()>(|| { + let _ = last_error::malformed("on the way"); + Err(STATUS_NEEDS_IO) + }); + assert_eq!(paused, Err(STATUS_NEEDS_IO)); + assert!(!last_error::is_set()); + } use crate::value::HostValue; /// A packed result, read the way the Go host reads it. diff --git a/packages/stack-guest-abi/src/lib.rs b/packages/stack-guest-abi/src/lib.rs index 6b4db1a3f..37a246a52 100644 --- a/packages/stack-guest-abi/src/lib.rs +++ b/packages/stack-guest-abi/src/lib.rs @@ -104,6 +104,7 @@ pub mod call; pub mod headers; pub mod last_error; pub mod status; +pub mod step; pub mod value; #[cfg(target_arch = "wasm32")] diff --git a/packages/stack-guest-abi/src/status.rs b/packages/stack-guest-abi/src/status.rs index a62ba51c3..7e32fdc4d 100644 --- a/packages/stack-guest-abi/src/status.rs +++ b/packages/stack-guest-abi/src/status.rs @@ -15,8 +15,8 @@ //! run this call against. Codes 5–10 are the outcomes of a request to //! ZeroKMS, so a host can distinguish a refused credential from a tampered //! ciphertext without parsing strings; 11 and 12 were appended by the -//! crypto guest, 13–27 by the credential guest, and 28 by the crypto guest -//! again. +//! crypto guest, 13–27 by the credential guest, and 28 and 29 by the +//! crypto guest again. 29 is not a failure: see [`STATUS_NEEDS_IO`]. //! //! Each code is documented here as the *verdict* it carries to a host — the //! thing the host can act on. Which of a library's errors reach which code, @@ -122,10 +122,16 @@ pub const STATUS_AUTH_REFRESH_REQUIRED: u32 = 27; /// and only that — never a verdict on the record's integrity. pub const STATUS_CONTEXT_MISMATCH: u32 = 28; +/// Not a failure: the operation is paused, waiting for the host to perform +/// HTTP requests on its behalf (the step loop, [`crate::step`]). The host +/// takes the requests and the operation's handle from the guest, performs +/// them, and resumes it with the responses. No error is recorded. +pub const STATUS_NEEDS_IO: u32 = 29; + /// The last code in the table. A guest appending a code of its own starts /// at `LAST_STATUS + 1` and moves this constant with it, so two guests can /// never claim one number. -pub const LAST_STATUS: u32 = STATUS_CONTEXT_MISMATCH; +pub const LAST_STATUS: u32 = STATUS_NEEDS_IO; #[cfg(test)] mod tests { @@ -164,6 +170,7 @@ mod tests { STATUS_AUTH_OTHER, STATUS_AUTH_REFRESH_REQUIRED, STATUS_CONTEXT_MISMATCH, + STATUS_NEEDS_IO, ]; for (i, code) in codes.iter().enumerate() { assert_eq!(*code, i as u32 + 1, "code {i} is out of sequence"); diff --git a/packages/stack-guest-abi/src/step.rs b/packages/stack-guest-abi/src/step.rs new file mode 100644 index 000000000..e99047dbc --- /dev/null +++ b/packages/stack-guest-abi/src/step.rs @@ -0,0 +1,659 @@ +//! The step loop: how a guest export that needs the network returns to its +//! host instead of blocking on it. +//! +//! An export that may reach ZeroKMS runs its work as a future, polled once +//! ([`start`]). Where the work sends an HTTP request ([`request`]), the +//! request is recorded and the future stays pending. The export then ends: +//! the future is kept in a table under a handle, and the host is handed the +//! handle and the requests it recorded ([`Step::NeedsIo`]). The host performs +//! them however it likes (a goroutine, `fetch`) and hands the responses back +//! ([`resume`]), in the order of the requests. The future is polled again, +//! and the loop repeats until it is done ([`Step::Done`]) or fails. A host +//! that gives up drops the operation ([`abort`]). +//! +//! The guest is never running while a request is in flight, so a host's +//! cancellation never interrupts it: an aborted operation is a future +//! dropped between two polls, and everything it held is wiped on drop. +//! +//! # Handles +//! +//! - A handle names one paused operation. It is issued when an operation +//! first pauses, never before, so an export that completes in one poll +//! costs no handle. +//! - Handles are issued in increasing order and never reused, so a stale +//! [`resume`] can never reach another operation: it is an unknown handle. +//! They are 64-bit so a long-lived instance never runs out: at a million +//! pauses a second, that takes over half a million years. +//! - At most [`MAX_OPEN`] operations may be paused at once. One more is +//! refused, and dropped, so a host that forgets to abort cannot grow the +//! table without bound. The cap guards against a faulty host; a host +//! with more concurrent callers than that bounds itself to it, so they +//! wait rather than fail (the Go client does, in `Client.drive`). +//! - An unknown, finished or aborted handle is `STATUS_STATE`. It never +//! panics. +//! +//! # Wakers +//! +//! The work may run requests through combinators that poll an inner future +//! again only once its waker has been woken (`FuturesOrdered`, `buffered`). +//! So each request stores the waker it was polled with, and [`resume`] wakes +//! it when its response arrives, before polling the operation again. The +//! operation itself is polled with a waker that does nothing: the loop polls +//! it on every step regardless. +//! +//! This module holds no encoding: the guest decides how a step's requests +//! and the host's responses cross the boundary. + +use std::cell::{Cell, RefCell}; +use std::collections::BTreeMap; +use std::future::Future; +use std::pin::Pin; +use std::rc::Rc; +use std::task::{Context, Poll, Waker}; + +use vitaminc_protected::OpaqueDebug; +use zeroize::Zeroizing; + +use crate::buffers; +use crate::last_error; + +/// An operation's work: what an export would have run with `block_on`, +/// owning its inputs, with the export's output or status as its result. +pub type Operation = Pin, u32>>>>; + +/// The most operations that may be paused at once. +pub const MAX_OPEN: usize = 64; + +/// One HTTP request an operation needs the host to perform. +/// +/// The headers (in the [`crate::headers`] line format, a bearer token +/// among them) and the body (which can carry key-material contexts) are +/// wiped on drop. +#[derive(OpaqueDebug)] +#[non_exhaustive] +pub struct Request { + /// The HTTP method. + #[non_sensitive] + pub method: &'static str, + /// The URL. + #[non_sensitive] + pub url: String, + /// The request headers. + pub headers: Zeroizing>, + /// The request body. + pub body: Zeroizing>, +} + +impl Request { + /// A request. + pub fn new( + method: &'static str, + url: String, + headers: Zeroizing>, + body: Zeroizing>, + ) -> Self { + Self { + method, + url, + headers, + body, + } + } +} + +/// The host's answer to one [`Request`]. +/// +/// `Debug` shows the status alone: the body can carry wrapped key material +/// or a credential. +#[derive(OpaqueDebug)] +#[non_exhaustive] +pub struct Response { + /// The HTTP status code, or negative for a transport-level failure — + /// then `body` is the host's error text. + #[non_sensitive] + pub status: i32, + /// The response headers, in the [`crate::headers`] line format. + pub headers: Vec, + /// The response body. Wiped on drop. + pub body: Zeroizing>, +} + +impl Response { + /// A response. + pub fn new(status: i32, headers: Vec, body: Zeroizing>) -> Self { + Self { + status, + headers, + body, + } + } +} + +/// What a step of an operation came to, when it did not fail. +#[derive(Debug)] +pub enum Step { + /// The operation finished, with this output. + Done(Vec), + /// The operation is paused under `handle`, waiting for the host to + /// perform `requests` and [`resume`] it with their responses, in order. + NeedsIo { + /// The paused operation. + handle: u64, + /// What it waits on. + requests: Vec, + }, +} + +/// A request made outside a step: the work was polled by something other +/// than [`start`] or [`resume`], so no host can be asked to perform it. +#[derive(Debug, thiserror::Error)] +#[error("an HTTP request was made outside a guest step")] +pub struct NotInStep; + +/// One request of a paused operation, from the moment the work makes it to +/// the moment it reads the response. +#[derive(Default)] +struct Exchange { + /// The request, until the step that made it hands it to the host. + request: Option, + /// The response, from [`resume`] until the work reads it. + response: Option, + /// The waker the request was last polled with. + waker: Option, +} + +/// The requests of one operation. +#[derive(Default)] +struct Io { + exchanges: RefCell>, +} + +/// A paused operation. +struct Open { + work: Operation, + io: Rc, + /// The exchanges whose requests the host was handed at the last step, + /// in the order it was handed them: where [`resume`]'s responses go. + awaiting: Vec, +} + +thread_local! { + /// The requests of the operation being polled, so [`request`] can record + /// into them without the work knowing which operation it belongs to. + static CURRENT: RefCell>> = const { RefCell::new(None) }; + static OPEN: RefCell> = const { RefCell::new(BTreeMap::new()) }; + /// The next handle to issue. Zero is never issued. + static NEXT: Cell = const { Cell::new(1) }; + /// A step's requests, encoded and registered for the host to take + /// ([`take_pending`]). + static PENDING: Cell> = const { Cell::new(None) }; +} + +/// Send `request` through the host: record it for the step being polled, +/// and resolve to the host's response once [`resume`] delivers it. +/// +/// The request is recorded on the first poll, which must happen inside a +/// step ([`start`] or [`resume`]); otherwise it resolves to [`NotInStep`]. +pub fn request(request: Request) -> impl Future> { + HostRequest::Unsent(Some(request)) +} + +enum HostRequest { + Unsent(Option), + Sent { io: Rc, index: usize }, + Finished, +} + +impl Future for HostRequest { + type Output = Result; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + match this { + HostRequest::Unsent(request) => { + let Some(io) = CURRENT.with(|current| current.borrow().clone()) else { + *this = HostRequest::Finished; + return Poll::Ready(Err(NotInStep)); + }; + let index = { + let mut exchanges = io.exchanges.borrow_mut(); + exchanges.push(Exchange { + request: request.take(), + response: None, + waker: Some(cx.waker().clone()), + }); + exchanges.len() - 1 + }; + *this = HostRequest::Sent { io, index }; + Poll::Pending + } + HostRequest::Sent { io, index } => { + let response = { + let mut exchanges = io.exchanges.borrow_mut(); + let exchange = &mut exchanges[*index]; + let response = exchange.response.take(); + if response.is_none() { + exchange.waker = Some(cx.waker().clone()); + } + response + }; + match response { + Some(response) => { + *this = HostRequest::Finished; + Poll::Ready(Ok(response)) + } + None => Poll::Pending, + } + } + HostRequest::Finished => Poll::Ready(Err(NotInStep)), + } + } +} + +/// Run an operation's first step. +/// +/// # Errors +/// +/// The operation's own failure, or `STATUS_STATE` when [`MAX_OPEN`] +/// operations are already paused (the operation is dropped), or +/// `STATUS_INTERNAL` when it paused without asking for anything. +pub fn start(work: Operation) -> Result { + clear_pending(); + step(None, work, Rc::default()) +} + +/// Hand a paused operation the responses to the requests its last step +/// asked for, in that order, and run its next step. +/// +/// The handle is spent whatever the outcome: a paused operation comes back +/// under a new [`Step::NeedsIo`] with the same handle. +/// +/// # Errors +/// +/// `STATUS_STATE` for an unknown handle; `STATUS_ENCODING` when the number +/// of responses is not the number of requests (the operation is dropped); +/// otherwise as [`start`]. +pub fn resume(handle: u64, responses: Vec) -> Result { + clear_pending(); + let Some(open) = OPEN.with(|open| open.borrow_mut().remove(&handle)) else { + return Err(unknown_handle()); + }; + if responses.len() != open.awaiting.len() { + drop(open); + return Err(last_error::malformed( + "the number of responses is not the number of requests", + )); + } + { + let mut exchanges = open.io.exchanges.borrow_mut(); + for (index, response) in open.awaiting.iter().zip(responses) { + let exchange = &mut exchanges[*index]; + exchange.response = Some(response); + if let Some(waker) = exchange.waker.take() { + waker.wake(); + } + } + } + step(Some(handle), open.work, open.io) +} + +/// Drop a paused operation, wiping everything it held. +/// +/// # Errors +/// +/// `STATUS_STATE` for an unknown handle. +pub fn abort(handle: u64) -> Result<(), u32> { + match OPEN.with(|open| open.borrow_mut().remove(&handle)) { + Some(open) => { + drop(open); + Ok(()) + } + None => Err(unknown_handle()), + } +} + +/// Drop every paused operation: what a guest's shutdown does before it +/// drops the state they share. +pub fn drop_all() { + let open = OPEN.with(|open| core::mem::take(&mut *open.borrow_mut())); + drop(open); + clear_pending(); +} + +/// How many operations are paused. +pub fn open_count() -> usize { + OPEN.with(|open| open.borrow().len()) +} + +/// Poll `work` once, as operation `handle` if it has one, and say what came +/// of it. +fn step(handle: Option, mut work: Operation, io: Rc) -> Result { + let previous = CURRENT.with(|current| current.replace(Some(io.clone()))); + let polled = work.as_mut().poll(&mut Context::from_waker(Waker::noop())); + CURRENT.with(|current| *current.borrow_mut() = previous); + if let Poll::Ready(result) = polled { + return result.map(Step::Done); + } + // Every request made since the last step, in the order made. + let (awaiting, requests): (Vec, Vec) = { + let mut exchanges = io.exchanges.borrow_mut(); + exchanges + .iter_mut() + .enumerate() + .filter_map(|(index, exchange)| exchange.request.take().map(|r| (index, r))) + .unzip() + }; + if requests.is_empty() { + // Nothing to wait for, so nothing would ever wake it. + drop(work); + return Err(last_error::internal( + "an operation paused without asking the host for anything", + )); + } + let handle = match handle { + Some(handle) => handle, + None => { + if open_count() >= MAX_OPEN { + drop(work); + return Err(last_error::out_of_order( + "too many operations are paused: resume or abort one first", + )); + } + issue()? + } + }; + OPEN.with(|open| { + let _ = open + .borrow_mut() + .insert(handle, Open { work, io, awaiting }); + }); + Ok(Step::NeedsIo { handle, requests }) +} + +/// The next handle. +fn issue() -> Result { + NEXT.with(|next| { + let handle = next.get(); + let following = handle + .checked_add(1) + .ok_or_else(|| last_error::out_of_order("this instance has issued every handle"))?; + next.set(following); + Ok(handle) + }) +} + +fn unknown_handle() -> u32 { + last_error::out_of_order("no paused operation has this handle") +} + +/// Register a step's encoded requests for the host to take with +/// [`take_pending`], replacing (and wiping) any it did not take. +pub fn set_pending(bytes: Vec) { + clear_pending(); + let len = bytes.len(); + let ptr = buffers::register(bytes); + PENDING.with(|pending| pending.set(Some((ptr, len)))); +} + +/// Hand the registered requests over, emptying the slot: what the guest's +/// export returns to the host, which releases the buffer with `se_dealloc`. +pub fn take_pending() -> Option<(*mut u8, usize)> { + PENDING.with(Cell::take) +} + +/// Wipe and free registered requests the host never took. +fn clear_pending() { + if let Some((ptr, len)) = PENDING.with(Cell::take) { + // SAFETY: the pair is the one `buffers::register` returned in + // `set_pending`, and nothing else holds it: the host is only given + // it by `take_pending`, which empties the slot first. + unsafe { buffers::dealloc(ptr, len) }; + } +} + +/// Drop the slot without freeing, for a registry wipe about to free it. +pub(crate) fn forget_pending() { + PENDING.with(|pending| pending.set(None)); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::status::{STATUS_ENCODING, STATUS_INTERNAL, STATUS_STATE}; + + fn get(url: &str) -> Request { + Request::new( + "POST", + url.to_owned(), + Zeroizing::new(b"authorization: Bearer t\n".to_vec()), + Zeroizing::new(b"{}".to_vec()), + ) + } + + fn ok(body: &[u8]) -> Response { + Response::new(200, Vec::new(), Zeroizing::new(body.to_vec())) + } + + /// Starts a test from an empty table and slot, and leaves them empty: + /// a recorded error or a paused operation is a registered buffer, which + /// Miri reports as leaked if a test ends holding one. + #[must_use] + struct Reset; + + impl Drop for Reset { + fn drop(&mut self) { + drop_all(); + last_error::clear(); + } + } + + fn reset() -> Reset { + drop_all(); + last_error::clear(); + Reset + } + + /// Work that makes `n` requests one after another and outputs their + /// bodies joined. + fn sequential(n: usize) -> Operation { + Box::pin(async move { + let mut out = Vec::new(); + for i in 0..n { + let response = request(get(&format!("https://kms/{i}"))) + .await + .map_err(|_| STATUS_INTERNAL)?; + out.extend_from_slice(&response.body); + } + Ok(out) + }) + } + + fn needs_io(step: Result) -> (u64, Vec) { + match step { + Ok(Step::NeedsIo { handle, requests }) => (handle, requests), + other => panic!("expected NeedsIo, got {other:?}"), + } + } + + #[test] + fn work_that_needs_no_request_is_done_without_a_handle() { + let _reset = reset(); + let step = start(Box::pin(async { Ok(b"out".to_vec()) })).unwrap(); + assert!(matches!(step, Step::Done(out) if out == b"out")); + assert_eq!(open_count(), 0); + } + + #[test] + fn a_failure_is_the_status_and_holds_no_handle() { + let _reset = reset(); + assert_eq!( + start(Box::pin(async { Err(STATUS_ENCODING) })).unwrap_err(), + STATUS_ENCODING + ); + assert_eq!(open_count(), 0); + } + + #[test] + fn each_request_is_one_step_and_the_responses_reach_the_work() { + let _reset = reset(); + let (handle, requests) = needs_io(start(sequential(2))); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].url, "https://kms/0"); + let (again, requests) = needs_io(resume(handle, vec![ok(b"a")])); + assert_eq!(again, handle, "a paused operation keeps its handle"); + assert_eq!(requests[0].url, "https://kms/1"); + let done = resume(handle, vec![ok(b"b")]).unwrap(); + assert!(matches!(done, Step::Done(out) if out == b"ab")); + assert_eq!(open_count(), 0); + assert_eq!( + resume(handle, vec![]).unwrap_err(), + STATUS_STATE, + "a finished handle is spent" + ); + } + + /// `buffered` polls an inner future again only once it is woken: the + /// stored waker is what keeps the operation from stalling after a + /// response. + #[test] + fn a_combinator_that_waits_on_its_wakers_is_woken() { + use futures::stream::{self, StreamExt}; + let _reset = reset(); + let work: Operation = Box::pin(async { + let bodies: Vec<_> = stream::iter(0..3) + .map(|i| request(get(&format!("https://kms/{i}")))) + .buffered(1) + .collect() + .await; + let mut out = Vec::new(); + for body in bodies { + out.extend_from_slice(&body.map_err(|_| STATUS_INTERNAL)?.body); + } + Ok(out) + }); + let (handle, _) = needs_io(start(work)); + needs_io(resume(handle, vec![ok(b"x")])); + needs_io(resume(handle, vec![ok(b"y")])); + let done = resume(handle, vec![ok(b"z")]).unwrap(); + assert!(matches!(done, Step::Done(out) if out == b"xyz")); + } + + #[test] + fn several_requests_in_one_step_take_their_responses_in_order() { + use futures::future::join; + let _reset = reset(); + let work: Operation = Box::pin(async { + let (a, b) = join(request(get("https://kms/a")), request(get("https://kms/b"))).await; + let mut out = a.map_err(|_| STATUS_INTERNAL)?.body.to_vec(); + out.extend_from_slice(&b.map_err(|_| STATUS_INTERNAL)?.body); + Ok(out) + }); + let (handle, requests) = needs_io(start(work)); + let urls: Vec<_> = requests.iter().map(|r| r.url.as_str()).collect(); + assert_eq!(urls, ["https://kms/a", "https://kms/b"]); + let done = resume(handle, vec![ok(b"1"), ok(b"2")]).unwrap(); + assert!(matches!(done, Step::Done(out) if out == b"12")); + } + + #[test] + fn handles_interleave_and_are_never_reused() { + let _reset = reset(); + let (first, _) = needs_io(start(sequential(1))); + let (second, _) = needs_io(start(sequential(1))); + assert!(second > first); + let done = resume(second, vec![ok(b"2")]).unwrap(); + assert!(matches!(done, Step::Done(out) if out == b"2")); + let done = resume(first, vec![ok(b"1")]).unwrap(); + assert!(matches!(done, Step::Done(out) if out == b"1")); + let (third, _) = needs_io(start(sequential(1))); + assert!(third > second, "a spent handle is not issued again"); + abort(third).unwrap(); + } + + #[test] + fn an_aborted_operation_is_dropped_and_its_handle_spent() { + let _reset = reset(); + let dropped = Rc::new(Cell::new(false)); + struct Flag(Rc>); + impl Drop for Flag { + fn drop(&mut self) { + self.0.set(true); + } + } + let flag = Flag(dropped.clone()); + let (handle, _) = needs_io(start(Box::pin(async move { + let _flag = flag; + request(get("https://kms")) + .await + .map_err(|_| STATUS_INTERNAL)?; + Ok(Vec::new()) + }))); + assert!(!dropped.get()); + abort(handle).unwrap(); + assert!(dropped.get(), "aborting drops the work"); + assert_eq!(abort(handle).unwrap_err(), STATUS_STATE); + assert_eq!(resume(handle, vec![ok(b"")]).unwrap_err(), STATUS_STATE); + } + + #[test] + fn the_wrong_number_of_responses_is_refused_and_drops_the_operation() { + let _reset = reset(); + let (handle, _) = needs_io(start(sequential(1))); + assert_eq!(resume(handle, vec![]).unwrap_err(), STATUS_ENCODING); + assert_eq!(open_count(), 0); + } + + #[test] + fn no_more_than_max_open_operations_pause() { + let _reset = reset(); + for _ in 0..MAX_OPEN { + needs_io(start(sequential(1))); + } + assert_eq!(start(sequential(1)).unwrap_err(), STATUS_STATE); + assert_eq!(open_count(), MAX_OPEN); + // A step that completes needs no handle, so it still runs. + assert!(matches!( + start(Box::pin(async { Ok(Vec::new()) })), + Ok(Step::Done(_)) + )); + drop_all(); + assert_eq!(open_count(), 0); + } + + #[test] + fn work_that_pauses_without_a_request_is_an_internal_failure() { + let _reset = reset(); + let work: Operation = Box::pin(futures::future::pending()); + assert_eq!(start(work).unwrap_err(), STATUS_INTERNAL); + assert_eq!(open_count(), 0); + } + + #[test] + fn a_request_outside_a_step_says_so() { + let mut work = Box::pin(request(get("https://kms"))); + let polled = work.as_mut().poll(&mut Context::from_waker(Waker::noop())); + assert!(matches!(polled, Poll::Ready(Err(NotInStep)))); + } + + /// A registry wipe frees the step's buffer with every other, so the slot + /// must not go on naming it. + #[test] + fn a_registry_wipe_empties_the_slot() { + let _reset = reset(); + set_pending(b"step".to_vec()); + buffers::wipe_all(); + assert!(take_pending().is_none()); + } + + #[test] + fn pending_requests_are_handed_over_once_and_replaced_wiped() { + let _reset = reset(); + set_pending(b"one".to_vec()); + set_pending(b"two".to_vec()); + let (ptr, len) = take_pending().unwrap(); + // SAFETY: registered above and handed over once. + let bytes = unsafe { buffers::take(ptr, len) }.unwrap(); + assert_eq!(bytes, b"two"); + assert!(take_pending().is_none()); + } +}