diff --git a/docs/content/docs/cloud/monitoring.mdx b/docs/content/docs/cloud/monitoring.mdx index e0b9682..d67a462 100644 --- a/docs/content/docs/cloud/monitoring.mdx +++ b/docs/content/docs/cloud/monitoring.mdx @@ -103,7 +103,7 @@ reported rather than freezing at its last value: **Lifecycle operations**: `orchestrator_operations_total{op,result}` and `orchestrator_operations_duration_seconds{op}` cover `create`, `destroy`, -`fork`, `migrate`, `migrate_live`, `snapshot` and `restore`. `destroy` counts +`fork`, `migrate`, `migrate_live`, `snapshot`, `restore` and `rotate_token`. `destroy` counts explicit destroys, from the API or a migrate removing its source. Teardowns the reconcile loop does on its own (stuck, idle, orphaned) and the teardown after a task completes aren't included; the reconcile counters above cover the first diff --git a/docs/content/docs/cloud/orchestrator-deploy.mdx b/docs/content/docs/cloud/orchestrator-deploy.mdx index 7c375f9..aa86139 100644 --- a/docs/content/docs/cloud/orchestrator-deploy.mdx +++ b/docs/content/docs/cloud/orchestrator-deploy.mdx @@ -71,7 +71,7 @@ range covers the fleet. | `FUSE_RESTATE_INGRESS_URL` | | | Restate ingress URL, required unless the mode is `off`. `http://127.0.0.1:18080` with `ops/restate/compose.yaml`. | | `FUSE_RESTATE_API_KEY` / `FUSE_RESTATE_IDENTITY_KEY` | | | Mode `cloud` only: the Cloud API key and the `publickeyv1_...` signing key. Other cloud settings are on [Restate Cloud](/docs/cloud/restate-cloud#settings). | | `ORCH_RESTATE_LISTEN` | | `127.0.0.1:9080` | Restate handler endpoint (h2c). Only restate-server, or the TLS proxy in front of it in cloud public mode, should reach it. | -| `FUSE_RESTATE_RETRY__INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate service; `` is `ENVIRONMENT`, `HOST`, `ARTIFACTS` or `ARTIFACTPIN`. The `ARTIFACTS` policy also bounds the retries of an artifact move's pull; the `ENVIRONMENT` policy bounds the host steps of a destroy, snapshot or restore, and the retries of a snapshot gc deletion. | +| `FUSE_RESTATE_RETRY__INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate service; `` is `ENVIRONMENT`, `HOST`, `ARTIFACTS` or `ARTIFACTPIN`. The `ARTIFACTS` policy also bounds the retries of an artifact move's pull; the `ENVIRONMENT` policy bounds the host steps of a destroy, snapshot, restore or token rotation upload (a rotation's state store write retries without bound), and the retries of a snapshot gc deletion. | With `FIRECRACKER_BASE_URL` unset, the orchestrator runs the in-memory stub provider (fine for local dev, not for production). With `DATABASE_URL` unset, state is @@ -79,7 +79,7 @@ in-memory and does not survive a restart. ## Restate -Restate is optional and off by default. With it on, destroys, artifact moves, snapshots, restores, the checkpoint chain and snapshot gc run on it. To run it next to the orchestrator: +Restate is optional and off by default. With it on, destroys, artifact moves, snapshots, restores, the checkpoint chain, snapshot gc and token rotation run on it. To run it next to the orchestrator: ```bash docker compose -f ops/restate/compose.yaml up -d diff --git a/docs/runbooks/restate.mdx b/docs/runbooks/restate.mdx index 872ef56..835b384 100644 --- a/docs/runbooks/restate.mdx +++ b/docs/runbooks/restate.mdx @@ -186,6 +186,22 @@ retention gc marks an expired snapshot `deleting` and sends (`snapshot.gc_failed`); the next sweep marks it again. gc skips a snapshot any pin holds, in process or in `ArtifactPin`. +token rotation runs as `Environment//RotateToken`: generate, upload, +persist. generate is journaled, so a restart between the upload and the +persist writes the token already in the guest to the state store instead of +making a second one. the upload retries under the `ENVIRONMENT` policy; the +persist has no attempt bound, so a state store that stays down holds the +rotation (and the vm's other environment ops) until it is back. the journaled value is ciphertext under +`TOKEN_ENCRYPTION_KEY` (the token, cert and key), and the steps decrypt it +themselves; no plaintext token reaches the journal. a rotation left unfinished +across a `TOKEN_ENCRYPTION_KEY` change fails, since its ciphertext no longer +decrypts; run it again. a rotation's journal can be read with the query below. rotations are counted as +`orchestrator_operations_total{op="rotate_token"}`. + +```sh +q "SELECT j.index, j.name, j.entry_json FROM sys_journal j JOIN sys_invocation i ON j.id = i.id WHERE i.target_handler_name = 'RotateToken'" +``` + copies are metered by `orchestrator_artifact_move_duration_seconds`, `orchestrator_artifact_move_bytes_total` and `orchestrator_artifact_move_failures_total`, labeled `source_host` and diff --git a/internal/durable/server.go b/internal/durable/server.go index 25b6e7d..05541ad 100644 --- a/internal/durable/server.go +++ b/internal/durable/server.go @@ -44,6 +44,7 @@ type Fleet interface { Artifacts Snapshots Checkpoints + Tokens } // snapshotAbortTimeout is how long restate lets a snapshot, restore or @@ -105,6 +106,10 @@ func NewEndpoint(cfg Config, fleet Fleet, logger *slog.Logger) *server.Restate { Handler("RetireChain", restate.NewObjectHandler( func(ctx restate.ObjectContext, _ restate.Void) (restate.Void, error) { return restate.Void{}, dropChain(ctx, fleet) + })). + Handler("RotateToken", restate.NewObjectHandler( + func(ctx restate.ObjectContext, _ restate.Void) (restate.Void, error) { + return restate.Void{}, rotateToken(ctx, fleet, servicePolicy(cfg, EnvironmentObject)) })) host := restate.NewObject(HostObject, retryOption(cfg, HostObject)). diff --git a/internal/durable/server_test.go b/internal/durable/server_test.go index 9f1b2bd..d9bf0aa 100644 --- a/internal/durable/server_test.go +++ b/internal/durable/server_test.go @@ -99,6 +99,16 @@ func (fakeFleet) ChainAdvanced(context.Context, string, orchestrator.ChainState, } func (fakeFleet) MirrorChain(string, orchestrator.ChainState, bool) {} +func (fakeFleet) RotateTokenGenerate(context.Context, string) (orchestrator.RotatedCredentials, error) { + return orchestrator.RotatedCredentials{}, nil +} +func (fakeFleet) RotateTokenUpload(context.Context, string, orchestrator.RotatedCredentials) error { + return nil +} +func (fakeFleet) RotateTokenPersist(context.Context, string, orchestrator.RotatedCredentials) error { + return nil +} + var testLogger = slog.New(slog.NewTextHandler(io.Discard, nil)) // discovery is the manifest restate-server reads on registration. checking it @@ -157,7 +167,7 @@ func TestDiscoveryManifest(t *testing.T) { want := map[string]string{"Ping": "SHARED"} switch s.Name { case EnvironmentObject: - for _, h := range []string{"Destroy", "Snapshot", "Restore", "CollectSnapshot", "Checkpoint", "ChainMoved", "HoldChain", "ReleaseChain", "RetireChain"} { + for _, h := range []string{"Destroy", "Snapshot", "Restore", "CollectSnapshot", "Checkpoint", "ChainMoved", "HoldChain", "ReleaseChain", "RetireChain", "RotateToken"} { want[h] = "EXCLUSIVE" } case ArtifactsObject: diff --git a/internal/durable/token.go b/internal/durable/token.go new file mode 100644 index 0000000..4b8886c --- /dev/null +++ b/internal/durable/token.go @@ -0,0 +1,83 @@ +package durable + +import ( + "context" + "errors" + + restate "github.com/restatedev/sdk-go" + "github.com/restatedev/sdk-go/ingress" + + "github.com/folsomintel/fuse/internal/orchestrator" +) + +// opRotateToken is the op label on token rotation's step spans and metrics. +const opRotateToken = "rotate_token" + +// rotateRefusedCode is the error code a token rotation's own refusal (vm not +// running, no encryption key) crosses restate as, so the client hands the api +// the fleet's message unchanged. 480-486 are the snapshot sentinels. +const rotateRefusedCode restate.Code = 487 + +// Tokens is the part of the fleet the token rotation handler drives, one +// method per step. *FleetManager satisfies it. +type Tokens interface { + RotateTokenGenerate(ctx context.Context, vmID string) (orchestrator.RotatedCredentials, error) + RotateTokenUpload(ctx context.Context, vmID string, rotated orchestrator.RotatedCredentials) error + RotateTokenPersist(ctx context.Context, vmID string, rotated orchestrator.RotatedCredentials) error +} + +// rotatePermanent marks err terminal when a retry cannot change it: the +// rotation's own refusals under rotateRefusedCode, the rest the way snapshot +// steps mark theirs (vm not found keeps its sentinel, an agent 4xx is final). +func rotatePermanent(err error) error { + if orchestrator.TokenRotationRefused(err) { + return restate.ToTerminalError(err, restate.WithErrorCode(rotateRefusedCode)) + } + return permanent(err) +} + +// rotateToken is Environment/{vm_id}.RotateToken: generate, upload, persist. +// the upload retries under policy; the persist retries until it succeeds. +// generate is journaled, so a replay after a crash uploads and persists the +// token it already made instead of a second one, and the stored token ends up +// the one in the guest. its journaled value is ciphertext only: the plaintext +// token is decrypted inside the upload and persist steps and never returned. +func rotateToken(ctx restate.ObjectContext, fleet Tokens, policy RetryPolicy) error { + vmID := restate.Key(ctx) + retry := retryUnder(policy) + rotated, err := Run(ctx, opRotateToken, "generate", func(c context.Context) (orchestrator.RotatedCredentials, error) { + rotated, err := fleet.RotateTokenGenerate(c, vmID) + return rotated, rotatePermanent(err) + }) + if err != nil { + return err + } + if err := runStep(ctx, opRotateToken, "upload", func(c context.Context) error { + return rotatePermanent(fleet.RotateTokenUpload(c, vmID, rotated)) + }, retry...); err != nil { + return err + } + // the guest holds the new token now, so the persist has no attempt + // bound: a state store that stays down holds the rotation here until it + // is back, instead of ending it with the old token stored. + return runStep(ctx, opRotateToken, "persist", func(c context.Context) error { + return rotatePermanent(fleet.RotateTokenPersist(c, vmID, rotated)) + }, restate.WithInitialRetryInterval(policy.InitialInterval), + restate.WithMaxRetryInterval(policy.MaxInterval)) +} + +// RotateToken runs Environment/{vmID}.RotateToken and waits for it. the +// rotation may queue behind a long snapshot on the same vm, so it goes through +// the client with no request timeout of its own; ctx bounds the wait, and the +// rotation keeps running if ctx ends first. +func (c *Client) RotateToken(ctx context.Context, vmID string) error { + _, err := ingress.Object[restate.Void, restate.Void](c.longIngress, EnvironmentObject, vmID, "RotateToken"). + Request(ctx, restate.Void{}) + var ierr ingress.Error + if errors.As(err, &ierr) && ierr.Code() == rotateRefusedCode { + return errors.New(ierr.Message()) + } + return fleetError(err) +} + +var _ orchestrator.DurableTokenRotator = (*Client)(nil) diff --git a/internal/durable/token_test.go b/internal/durable/token_test.go new file mode 100644 index 0000000..0f323f6 --- /dev/null +++ b/internal/durable/token_test.go @@ -0,0 +1,303 @@ +package durable + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/folsomintel/fuse/internal/orchestrator" + "github.com/folsomintel/fuse/internal/secrets" +) + +var tokenTestKey = bytes.Repeat([]byte{7}, 32) + +// tokenGuest is the state a rotation changes, shared by the fleets of an +// endpoint and its restart: the token in the guest's files and the one the +// state store holds. +type tokenGuest struct { + mu sync.Mutex + inGuest string + stored string + generated []string + sealed [][]byte +} + +func (g *tokenGuest) tokens() (inGuest, stored string, generated []string) { + g.mu.Lock() + defer g.mu.Unlock() + return g.inGuest, g.stored, append([]string(nil), g.generated...) +} + +// tokenFleet scripts the rotation steps over a tokenGuest, with real +// encryption so the journal carries what the fleet's steps really return. +type tokenFleet struct { + *snapFleet + guest *tokenGuest +} + +func newTokenFleet(g *tokenGuest) *tokenFleet { + return &tokenFleet{snapFleet: newSnapFleet(), guest: g} +} + +func (f *tokenFleet) RotateTokenGenerate(ctx context.Context, vm string) (orchestrator.RotatedCredentials, error) { + if err := f.enter(ctx, "generate"); err != nil { + return orchestrator.RotatedCredentials{}, err + } + creds, err := secrets.GenerateVMCredentials(vm) + if err != nil { + return orchestrator.RotatedCredentials{}, err + } + sealed, err := secrets.EncryptToken(creds.AuthToken, tokenTestKey) + if err != nil { + return orchestrator.RotatedCredentials{}, err + } + f.guest.mu.Lock() + f.guest.generated = append(f.guest.generated, creds.AuthToken) + f.guest.sealed = append(f.guest.sealed, sealed) + f.guest.mu.Unlock() + return orchestrator.RotatedCredentials{Sealed: sealed, TokenEncrypted: sealed}, nil +} + +func (f *tokenFleet) RotateTokenUpload(ctx context.Context, _ string, r orchestrator.RotatedCredentials) error { + if err := f.enter(ctx, "upload"); err != nil { + return err + } + token, err := secrets.DecryptToken(r.Sealed, tokenTestKey) + if err != nil { + return err + } + f.guest.mu.Lock() + f.guest.inGuest = token + f.guest.mu.Unlock() + return nil +} + +func (f *tokenFleet) RotateTokenPersist(ctx context.Context, _ string, r orchestrator.RotatedCredentials) error { + if err := f.enter(ctx, "persist"); err != nil { + return err + } + token, err := secrets.DecryptToken(r.TokenEncrypted, tokenTestKey) + if err != nil { + return err + } + f.guest.mu.Lock() + f.guest.stored = token + f.guest.mu.Unlock() + return nil +} + +// TestRotateTokenThroughRestate: an upload that fails twice is retried inside +// its step, generate runs once, and the stored token is the guest's. +func TestRotateTokenThroughRestate(t *testing.T) { + guest := &tokenGuest{} + fleet := newTokenFleet(guest) + fleet.fail["upload"] = []error{errors.New("agent 503"), errors.New("agent 503")} + client := serveFleet(t, fleet, fastPolicy) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + if err := client.RotateToken(ctx, uniqueVM("rotate")); err != nil { + t.Fatalf("rotate: %v", err) + } + if got, want := fleet.steps(), "generate,upload,upload,upload,persist"; got != want { + t.Fatalf("steps %s, want %s", got, want) + } + inGuest, stored, generated := guest.tokens() + if len(generated) != 1 || inGuest != generated[0] || stored != generated[0] { + t.Fatalf("guest %q, stored %q, generated %v; want one token in both", inGuest, stored, generated) + } +} + +// TestRotateTokenPersistOutlastsThePolicy: a state store down for longer than +// the policy's attempts holds the rotation at the persist until it is back, +// rather than ending it with the guest's token never stored. +func TestRotateTokenPersistOutlastsThePolicy(t *testing.T) { + guest := &tokenGuest{} + fleet := newTokenFleet(guest) + var downs []error + for range fastPolicy.MaxAttempts + 3 { + downs = append(downs, errors.New("store down")) + } + fleet.fail["persist"] = downs + client := serveFleet(t, fleet, fastPolicy) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + if err := client.RotateToken(ctx, uniqueVM("rotate-store-down")); err != nil { + t.Fatalf("rotate: %v", err) + } + want := "generate,upload," + strings.Repeat("persist,", len(downs)+1) + if got := fleet.steps(); got != strings.TrimSuffix(want, ",") { + t.Fatalf("steps %s, want %s", got, want) + } + inGuest, stored, _ := guest.tokens() + if inGuest == "" || stored != inGuest { + t.Fatalf("guest %q, stored %q; want the guest's token stored", inGuest, stored) + } +} + +// TestRotateTokenRefusals: a refusal crosses restate with the fleet's own +// message, and a vm the fleet does not know keeps its sentinel. +func TestRotateTokenRefusals(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + + notRunning := fmt.Errorf("vm x in state stopped: %w", orchestrator.ErrRotateNotRunning) + fleet := newTokenFleet(&tokenGuest{}) + fleet.fail["generate"] = []error{notRunning} + client := serveFleet(t, fleet, fastPolicy) + err := client.RotateToken(ctx, uniqueVM("rotate-refused")) + if err == nil || err.Error() != notRunning.Error() { + t.Fatalf("rotate error %v, want %q", err, notRunning) + } + if got := fleet.steps(); got != "generate" { + t.Fatalf("steps %s, want only generate", got) + } + + gone := newTokenFleet(&tokenGuest{}) + gone.fail["upload"] = []error{fmt.Errorf("%w: fuse-gone", orchestrator.ErrVMNotFound)} + client = serveFleet(t, gone, fastPolicy) + if err := client.RotateToken(ctx, uniqueVM("rotate-gone")); !errors.Is(err, orchestrator.ErrVMNotFound) { + t.Fatalf("rotate error %v, want ErrVMNotFound", err) + } + if got := gone.steps(); got != "generate,upload" { + t.Fatalf("steps %s, want generate,upload", got) + } +} + +// TestRotateTokenSurvivesEndpointRestart stands in for kill -9 of the +// orchestrator between the upload and the persist: the endpoint drops while +// the persist runs, and the restarted one persists the token already in the +// guest instead of generating another. while the invocation is held there, +// its journal is read through the admin api: it holds the generate step's +// ciphertext and nowhere the plaintext token. +func TestRotateTokenSurvivesEndpointRestart(t *testing.T) { + guest := &tokenGuest{} + first := newTokenFleet(guest) + first.block["persist"] = make(chan struct{}) + defer close(first.block["persist"]) + client, addr, stop := serveFleetAt(t, first, "") + vm := uniqueVM("rotate-restart") + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- client.RotateToken(ctx, vm) }() + <-first.entered + + _, _, generated := guest.tokens() + if len(generated) != 1 { + t.Fatalf("generated %v before the persist, want one token", generated) + } + guest.mu.Lock() + sealed := guest.sealed[0] + guest.mu.Unlock() + journal := readJournal(t, ctx, vm) + if !bytes.Contains(journal, []byte(base64.StdEncoding.EncodeToString(sealed))) { + t.Fatalf("journal of %s does not hold the generate step's ciphertext; the check below would be vacuous", vm) + } + for _, leak := range [][]byte{ + []byte(generated[0]), + []byte(base64.StdEncoding.EncodeToString([]byte(generated[0]))), + } { + if bytes.Contains(journal, leak) { + t.Fatalf("journal of %s holds the plaintext token", vm) + } + } + + stop() + second := newTokenFleet(guest) + _, _, _ = serveFleetAt(t, second, addr) + if err := <-done; err != nil { + t.Fatalf("rotate: %v", err) + } + if got := second.steps(); got != "persist" { + t.Fatalf("steps after the restart %s, want only persist", got) + } + inGuest, stored, generated := guest.tokens() + if len(generated) != 1 || inGuest != generated[0] || stored != inGuest { + t.Fatalf("guest %q, stored %q, generated %v; want the first token in both", inGuest, stored, generated) + } +} + +// readJournal returns the journal entries of vm's in-flight invocations from +// restate's sql introspection, with every payload (rendered as an array of +// bytes) decoded back into its bytes. +func readJournal(t *testing.T, ctx context.Context, vm string) []byte { + t.Helper() + admin, _ := restateTestURLs(t) + query := fmt.Sprintf("SELECT j.name, j.entry_json FROM sys_journal j JOIN sys_invocation i ON j.id = i.id "+ + "WHERE i.target_service_name = '%s' AND i.target_service_key = '%s'", EnvironmentObject, vm) + body, _ := json.Marshal(map[string]string{"query": query}) + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, admin+"/query", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + res, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer func() { _ = res.Body.Close() }() + raw, _ := io.ReadAll(res.Body) + if res.StatusCode != http.StatusOK { + t.Fatalf("journal query: http %d: %s", res.StatusCode, raw) + } + var out struct { + Rows []struct { + Name string `json:"name"` + EntryJSON string `json:"entry_json"` + } `json:"rows"` + } + if err := json.Unmarshal(raw, &out); err != nil { + t.Fatalf("journal query: %v: %s", err, raw) + } + var names []string + all := append([]byte(nil), raw...) + for _, r := range out.Rows { + names = append(names, r.Name) + var entry any + if err := json.Unmarshal([]byte(r.EntryJSON), &entry); err != nil { + t.Fatalf("journal entry %q: %v", r.EntryJSON, err) + } + all = appendPayloads(all, entry) + } + if !strings.Contains(strings.Join(names, ","), "generate") { + t.Fatalf("journal of %s has entries %v, want the generate step", vm, names) + } + return all +} + +// appendPayloads appends every array of numbers under v as the bytes it +// stands for. +func appendPayloads(out []byte, v any) []byte { + switch v := v.(type) { + case map[string]any: + for _, x := range v { + out = appendPayloads(out, x) + } + case []any: + b := make([]byte, 0, len(v)) + for _, x := range v { + f, ok := x.(float64) + if !ok { + // not a payload: a list of entries or headers. + for _, y := range v { + out = appendPayloads(out, y) + } + return out + } + b = append(b, byte(f)) + } + out = append(out, ' ') + out = append(out, b...) + } + return out +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 67bdc06..6e7a13a 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -208,7 +208,7 @@ func NewPrometheusMetrics(reg prometheus.Registerer) *PrometheusMetrics { Namespace: "orchestrator", Subsystem: "operations", Name: "total", - Help: "Lifecycle operations by op (create, destroy, fork, migrate, migrate_live, snapshot, restore) and result (ok, error).", + Help: "Lifecycle operations by op (create, destroy, fork, migrate, migrate_live, snapshot, restore, rotate_token) and result (ok, error).", }, []string{"op", "result"}), operationDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: "orchestrator", diff --git a/internal/orchestrator/fleet.go b/internal/orchestrator/fleet.go index c0128da..0451203 100644 --- a/internal/orchestrator/fleet.go +++ b/internal/orchestrator/fleet.go @@ -326,7 +326,7 @@ type EgressMetrics interface { // OperationMetrics is an optional extension of ReconcileMetrics for // lifecycle operations: create, destroy, fork, migrate, migrate_live, -// snapshot and restore. destroy is DestroyVM only: the reconcile loop's own +// snapshot, restore and rotate_token. destroy is DestroyVM only: the reconcile loop's own // teardowns and the one after CompleteTask go through destroyAndRemove and are // not counted. snapshot includes the snapshots fork, migrate and checkpoint // chains take internally, since each is a real checkpoint the host had to @@ -630,6 +630,10 @@ type FleetManager struct { // checkpoints_durable.go. durableSnapshots DurableSnapshots + // durableTokens runs token rotation on restate; nil in mode off. see + // token_rotation.go. + durableTokens DurableTokenRotator + cancel context.CancelFunc done chan struct{} } diff --git a/internal/orchestrator/token_rotation.go b/internal/orchestrator/token_rotation.go index 577c6dc..0d64a3c 100644 --- a/internal/orchestrator/token_rotation.go +++ b/internal/orchestrator/token_rotation.go @@ -2,15 +2,80 @@ package orchestrator import ( "context" + "encoding/json" + "errors" "fmt" + "maps" + "slices" "time" "github.com/folsomintel/fuse/internal/secrets" ) +// the refusals a token rotation fails with when a retry cannot help. +// TokenRotationRefused reports them. +var ( + // ErrRotateNotRunning is a vm that is not running. + ErrRotateNotRunning = errors.New("token rotation requires running") + // errRotateNoKey is an orchestrator with no TOKEN_ENCRYPTION_KEY. + errRotateNoKey = errors.New("token rotation requires a 32-byte encryption key (TOKEN_ENCRYPTION_KEY)") + // errRotateSealed is rotated credentials that no longer decrypt, as + // after TOKEN_ENCRYPTION_KEY changed under an unfinished rotation. + errRotateSealed = errors.New("rotated credentials do not decrypt under TOKEN_ENCRYPTION_KEY") +) + +// TokenRotationRefused reports whether a token rotation step failed with one +// of the rotation's own refusals (vm not running, no encryption key, +// credentials that do not decrypt). a retry cannot change any of them. +func TokenRotationRefused(err error) bool { + return errors.Is(err, ErrRotateNotRunning) || errors.Is(err, errRotateNoKey) || errors.Is(err, errRotateSealed) +} + +// RotatedCredentials is the output of RotateTokenGenerate and the input of +// the later steps. both fields are ciphertext under TOKEN_ENCRYPTION_KEY: the +// durable handler journals this value, so neither the plaintext token nor the +// tls key ever reaches restate. +type RotatedCredentials struct { + // Sealed is the cert, key and token as json, encrypted. the upload step + // decrypts it. + Sealed []byte `json:"sealed"` + // TokenEncrypted is the token as the vm row stores it. + TokenEncrypted []byte `json:"token_encrypted"` +} + +// sealedCredentials is the plaintext inside RotatedCredentials.Sealed. +type sealedCredentials struct { + AuthToken string `json:"auth_token"` + CertPEM []byte `json:"cert_pem"` + KeyPEM []byte `json:"key_pem"` +} + +// DurableTokenRotator runs a token rotation as the restate +// Environment/{vm_id}.RotateToken handler and waits for it. *durable.Client +// satisfies it. +type DurableTokenRotator interface { + RotateToken(ctx context.Context, vmID string) error +} + +// SetDurableTokenRotator routes token rotation through restate. nil (mode +// off, the default) keeps it in process. set it before Start. +func (fm *FleetManager) SetDurableTokenRotator(d DurableTokenRotator) { + fm.mu.Lock() + fm.durableTokens = d + fm.mu.Unlock() +} + +func (fm *FleetManager) durableTokenRotator() DurableTokenRotator { + fm.mu.RLock() + defer fm.mu.RUnlock() + return fm.durableTokens +} + // RotateToken generates new TLS credentials and an auth token for a // running VM, uploads them to the guest filesystem, and persists the -// new encrypted token. +// new encrypted token. on restate it runs as the +// Environment/{vm_id}.RotateToken handler, so a crash between the upload and +// the persist resumes at the persist with the same token. // // KNOWN GAP: fused reads --auth-token-file exactly once at process start // (fused/main.go) and has no credential poller, so a live fused keeps @@ -24,55 +89,156 @@ import ( // by the agent profile). New inbound connections use the rotated // credentials; existing connections retain the old cert until they reconnect. func (fm *FleetManager) RotateToken(ctx context.Context, vmID string) error { + ctx, end := fm.startOperation(ctx, "rotate_token", vmID) + err := fm.rotateToken(ctx, vmID) + end(err) + return err +} + +func (fm *FleetManager) rotateToken(ctx context.Context, vmID string) error { + if d := fm.durableTokenRotator(); d != nil { + ctx, cancel := durableWait(ctx) + defer cancel() + return d.RotateToken(ctx, vmID) + } + rotated, err := fm.RotateTokenGenerate(ctx, vmID) + if err != nil { + return err + } + if err := fm.RotateTokenUpload(ctx, vmID, rotated); err != nil { + return err + } + if err := fm.storeRotatedToken(vmID, rotated); err != nil { + return err + } + // mode off keeps the background persist it always had. + fm.persistVMBackground(vmID) + fm.appendEvent(ctx, "vm", vmID, "vm.token_rotated", nil) + fm.logger.InfoContext(ctx, "vm token rotated", "vm", vmID) + return nil +} + +// the methods below are the steps of a token rotation, in order. the +// durable handler runs each as a journaled step; mode off runs them in turn. + +// rotateTarget returns vmID's environment handle when the vm can take a +// rotation. +func (fm *FleetManager) rotateTarget(vmID string) (Environment, error) { fm.mu.RLock() v, ok := fm.vms[vmID] if !ok { fm.mu.RUnlock() - return fmt.Errorf("%w: %s", ErrVMNotFound, vmID) + return nil, fmt.Errorf("%w: %s", ErrVMNotFound, vmID) } env := v.env state := v.state fm.mu.RUnlock() if state != VMStateRunning { - return fmt.Errorf("vm %s in state %s: token rotation requires running", vmID, state) + return nil, fmt.Errorf("vm %s in state %s: %w", vmID, state, ErrRotateNotRunning) } if env == nil { - return fmt.Errorf("vm %s has no active environment handle", vmID) + return nil, fmt.Errorf("%w: vm %s", errNoEnvironment, vmID) } if len(fm.tokenEncryptionKey) != 32 { - return fmt.Errorf("token rotation requires a 32-byte encryption key (TOKEN_ENCRYPTION_KEY)") + return nil, errRotateNoKey } + return env, nil +} - // Generate fresh credentials. +// RotateTokenGenerate checks the vm can take a rotation and generates its +// new credentials, returned encrypted. on restate its result is journaled, so +// a replay reuses the same token instead of generating another. +func (fm *FleetManager) RotateTokenGenerate(_ context.Context, vmID string) (RotatedCredentials, error) { + if _, err := fm.rotateTarget(vmID); err != nil { + return RotatedCredentials{}, err + } creds, err := secrets.GenerateVMCredentials(vmID) if err != nil { - return fmt.Errorf("generate credentials: %w", err) + return RotatedCredentials{}, fmt.Errorf("generate credentials: %w", err) + } + raw, err := json.Marshal(sealedCredentials{AuthToken: creds.AuthToken, CertPEM: creds.CertPEM, KeyPEM: creds.KeyPEM}) + if err != nil { + return RotatedCredentials{}, fmt.Errorf("encode credentials: %w", err) + } + sealed, err := secrets.EncryptBytes(raw, fm.tokenEncryptionKey) + if err != nil { + return RotatedCredentials{}, fmt.Errorf("encrypt credentials: %w", err) + } + token, err := secrets.EncryptToken(creds.AuthToken, fm.tokenEncryptionKey) + if err != nil { + return RotatedCredentials{}, fmt.Errorf("encrypt token: %w", err) } + return RotatedCredentials{Sealed: sealed, TokenEncrypted: token}, nil +} - // Upload the agent's credential files (cert, key, auth token) to the - // guest. Paths are owned by the agent profile (fusedCredentialFiles). - if err := uploadFiles(ctx, env, fusedCredentialFiles(creds)); err != nil { - return fmt.Errorf("upload credentials: %w", err) +// RotateTokenUpload writes the credential files (cert, key, auth token) into +// the guest, at the paths the agent profile owns (fusedCredentialFiles). the +// same files are written in the same order on every attempt, so a retry is +// harmless. +func (fm *FleetManager) RotateTokenUpload(ctx context.Context, vmID string, rotated RotatedCredentials) error { + env, err := fm.rotateTarget(vmID) + if err != nil { + return err + } + raw, err := secrets.DecryptBytes(rotated.Sealed, fm.tokenEncryptionKey) + if err != nil { + return errRotateSealed + } + var sealed sealedCredentials + if err := json.Unmarshal(raw, &sealed); err != nil { + return errRotateSealed + } + creds := &secrets.VMCredentials{AuthToken: sealed.AuthToken, CertPEM: sealed.CertPEM, KeyPEM: sealed.KeyPEM} + // one at a time in path order, not concurrently like uploadFiles: on + // restate each upload is keyed by its place in the step (key/1, key/2, + // ...), and the agent replays a key's first answer whatever file it now + // carries, so a retried step has to send the same file under each key. + files := fusedCredentialFiles(creds) + for _, path := range slices.Sorted(maps.Keys(files)) { + if err := env.Upload(ctx, files[path], path); err != nil { + return fmt.Errorf("upload credentials: upload %s: %w", path, err) + } } setTokenIfSupported(env, creds) + return nil +} - // Encrypt and persist the new token. - encrypted, err := secrets.EncryptToken(creds.AuthToken, fm.tokenEncryptionKey) - if err != nil { - return fmt.Errorf("encrypt token: %w", err) +// RotateTokenPersist makes the new token the vm's own: in memory, on the +// environment handle (which a restart between the upload and this step left +// on the old one), and in the state store, written before it returns. the +// durable path's replacement for persistVMBackground. +func (fm *FleetManager) RotateTokenPersist(ctx context.Context, vmID string, rotated RotatedCredentials) error { + if err := fm.storeRotatedToken(vmID, rotated); err != nil { + return err } + if err := fm.persistVMByID(ctx, vmID); err != nil { + return fmt.Errorf("persist rotated token: %w", err) + } + fm.appendEvent(ctx, "vm", vmID, "vm.token_rotated", nil) + fm.logger.InfoContext(ctx, "vm token rotated", "vm", vmID) + return nil +} +// storeRotatedToken records the rotated token on vmID in memory and on its +// environment handle. +func (fm *FleetManager) storeRotatedToken(vmID string, rotated RotatedCredentials) error { + token, err := secrets.DecryptToken(rotated.TokenEncrypted, fm.tokenEncryptionKey) + if err != nil { + return errRotateSealed + } fm.mu.Lock() - if v, ok := fm.vms[vmID]; ok { - v.authTokenEncrypted = encrypted - v.updatedAt = time.Now() + v, ok := fm.vms[vmID] + if !ok { + fm.mu.Unlock() + return fmt.Errorf("%w: %s", ErrVMNotFound, vmID) } + v.authTokenEncrypted = rotated.TokenEncrypted + v.updatedAt = time.Now() + env := v.env fm.mu.Unlock() - - fm.persistVMBackground(vmID) - fm.appendEvent(ctx, "vm", vmID, "vm.token_rotated", nil) - fm.logger.Info("vm token rotated", "vm", vmID) - + if ts, ok := env.(TokenSetter); ok { + ts.SetToken(token) + } return nil } diff --git a/internal/orchestrator/token_rotation_test.go b/internal/orchestrator/token_rotation_test.go new file mode 100644 index 0000000..efcbb6a --- /dev/null +++ b/internal/orchestrator/token_rotation_test.go @@ -0,0 +1,238 @@ +package orchestrator + +import ( + "context" + "encoding/json" + "errors" + "slices" + "strings" + "testing" + + "github.com/folsomintel/fuse/internal/secrets" +) + +// fakeTokenRotator stands in for the restate client and records the vms it +// was asked to rotate. +type fakeTokenRotator struct{ rotated []string } + +func (f *fakeTokenRotator) RotateToken(_ context.Context, vmID string) error { + f.rotated = append(f.rotated, vmID) + return nil +} + +func newRotateTestFleet(t *testing.T) (*FleetManager, *credForkProvider, *MemoryStateStore, string) { + t.Helper() + provider := newCredForkProvider() + store := NewMemoryStateStore() + fm := NewFleetManager(FleetConfig{ + Provider: provider, + StateStore: store, + Prefix: "fuse-", + TokenEncryptionKey: testEncryptionKey(), + }) + return fm, provider, store, provisionSnapshotTestVM(t, fm, "task-1") +} + +// storedToken decrypts the token the state store holds for vmID. +func storedToken(t *testing.T, store *MemoryStateStore, vmID string) string { + t.Helper() + records, err := store.ListVMs(context.Background()) + if err != nil { + t.Fatal(err) + } + for _, r := range records { + if r.ID == vmID { + plain, err := secrets.DecryptToken(r.AuthTokenEncrypted, testEncryptionKey()) + if err != nil { + t.Fatalf("decrypt stored token: %v", err) + } + return plain + } + } + t.Fatalf("no stored record for %s", vmID) + return "" +} + +// TestRotateToken_inProcess: mode off writes a new token into the guest and +// makes it the vm's own, in memory and on the environment handle. +func TestRotateToken_inProcess(t *testing.T) { + fm, provider, _, vmID := newRotateTestFleet(t) + env := provider.envs[vmID] + before := string(env.files[fuseAuthTokenPath]) + + if err := fm.RotateToken(context.Background(), vmID); err != nil { + t.Fatalf("rotate: %v", err) + } + inGuest := string(env.files[fuseAuthTokenPath]) + if inGuest == "" || inGuest == before { + t.Fatalf("guest token %q, want a new one (was %q)", inGuest, before) + } + if env.Token() != inGuest { + t.Fatal("environment handle does not carry the rotated token") + } + fm.mu.RLock() + enc := fm.vms[vmID].authTokenEncrypted + fm.mu.RUnlock() + if plain, err := secrets.DecryptToken(enc, testEncryptionKey()); err != nil || plain != inGuest { + t.Fatalf("vm token %q (%v), want the guest's", plain, err) + } +} + +// TestRotateToken_durable: with restate configured the rotation runs on the +// Environment object, and nothing happens in process. +func TestRotateToken_durable(t *testing.T) { + fm, provider, _, vmID := newRotateTestFleet(t) + d := &fakeTokenRotator{} + fm.SetDurableTokenRotator(d) + before := string(provider.envs[vmID].files[fuseAuthTokenPath]) + + if err := fm.RotateToken(context.Background(), vmID); err != nil { + t.Fatalf("rotate: %v", err) + } + if len(d.rotated) != 1 || d.rotated[0] != vmID { + t.Fatalf("durable rotations %v, want [%s]", d.rotated, vmID) + } + if got := string(provider.envs[vmID].files[fuseAuthTokenPath]); got != before { + t.Fatal("durable rotation touched the guest in process") + } +} + +// TestRotateTokenSteps_restartBeforePersist runs the durable steps the way a +// replay after a crash between the upload and the persist does: the vm comes +// back with its old token, and the persist alone makes the guest's token the +// stored one. the generate step's output, which restate journals, never +// carries the plaintext token or tls key. +func TestRotateTokenSteps_restartBeforePersist(t *testing.T) { + fm, provider, store, vmID := newRotateTestFleet(t) + env := provider.envs[vmID] + ctx := context.Background() + if err := fm.persistVMByID(ctx, vmID); err != nil { + t.Fatal(err) + } + old := storedToken(t, store, vmID) + + rotated, err := fm.RotateTokenGenerate(ctx, vmID) + if err != nil { + t.Fatalf("generate: %v", err) + } + if err := fm.RotateTokenUpload(ctx, vmID, rotated); err != nil { + t.Fatalf("upload: %v", err) + } + inGuest := string(env.files[fuseAuthTokenPath]) + journaled, _ := json.Marshal(rotated) + for _, secret := range []string{inGuest, string(env.files[fuseTLSKeyPath])} { + if strings.Contains(string(journaled), secret) { + t.Fatal("generate's output carries a plaintext credential") + } + } + + // the restart: the recovered handle and store still hold the old token. + env.SetToken(old) + if got := storedToken(t, store, vmID); got != old { + t.Fatalf("store changed before the persist: %q", got) + } + + if err := fm.RotateTokenPersist(ctx, vmID, rotated); err != nil { + t.Fatalf("persist: %v", err) + } + if got := storedToken(t, store, vmID); got != inGuest { + t.Fatalf("stored token %q, want the guest's %q", got, inGuest) + } + if env.Token() != inGuest { + t.Fatal("persist did not put the rotated token back on the environment handle") + } +} + +// TestRotateTokenSteps_refusals: what a retry cannot change is reported as a +// refusal, and a vm the fleet does not know keeps its sentinel. +func TestRotateTokenSteps_refusals(t *testing.T) { + fm, _, _, vmID := newRotateTestFleet(t) + ctx := context.Background() + + if _, err := fm.RotateTokenGenerate(ctx, "fuse-missing"); !errors.Is(err, ErrVMNotFound) { + t.Fatalf("missing vm: %v, want ErrVMNotFound", err) + } + + rotated, err := fm.RotateTokenGenerate(ctx, vmID) + if err != nil { + t.Fatal(err) + } + tampered := RotatedCredentials{Sealed: []byte("not ciphertext"), TokenEncrypted: rotated.TokenEncrypted} + if err := fm.RotateTokenUpload(ctx, vmID, tampered); !TokenRotationRefused(err) { + t.Fatalf("undecryptable credentials: %v, want a refusal", err) + } + + fm.mu.Lock() + fm.vms[vmID].state = VMStateDraining + fm.mu.Unlock() + _, err = fm.RotateTokenGenerate(ctx, vmID) + if !TokenRotationRefused(err) || !strings.Contains(err.Error(), "token rotation requires running") { + t.Fatalf("draining vm: %v, want the not-running refusal", err) + } + + noKey := NewFleetManager(FleetConfig{Provider: newCredForkProvider(), Prefix: "fuse-"}) + id := provisionSnapshotTestVM(t, noKey, "task-2") + if _, err := noKey.RotateTokenGenerate(ctx, id); !errors.Is(err, errRotateNoKey) { + t.Fatalf("no key: %v, want errRotateNoKey", err) + } +} + +// TestRotateToken_metered: every rotation is counted as rotate_token, a +// refused one as an error, in both modes. +func TestRotateToken_metered(t *testing.T) { + metrics := &opMetrics{} + fm := NewFleetManager(FleetConfig{ + Provider: newCredForkProvider(), + Prefix: "fuse-", + Metrics: metrics, + TokenEncryptionKey: testEncryptionKey(), + }) + ctx := context.Background() + vmID := provisionSnapshotTestVM(t, fm, "task-1") + if err := fm.RotateToken(ctx, vmID); err != nil { + t.Fatal(err) + } + if err := fm.RotateToken(ctx, "fuse-missing"); err == nil { + t.Fatal("rotating an unknown vm succeeded") + } + fm.SetDurableTokenRotator(&fakeTokenRotator{}) + if err := fm.RotateToken(ctx, vmID); err != nil { + t.Fatal(err) + } + want := []string{"create:ok", "rotate_token:ok", "rotate_token:error", "rotate_token:ok"} + metrics.mu.Lock() + defer metrics.mu.Unlock() + if !slices.Equal(metrics.ops, want) { + t.Fatalf("operations = %v, want %v", metrics.ops, want) + } +} + +// TestRotateTokenUpload_stableOrder: every attempt sends the credential files +// one at a time in the same order, so on restate a retried upload keys each +// file the way the first attempt did. +func TestRotateTokenUpload_stableOrder(t *testing.T) { + fm, provider, _, vmID := newRotateTestFleet(t) + env := provider.envs[vmID] + ctx := context.Background() + rotated, err := fm.RotateTokenGenerate(ctx, vmID) + if err != nil { + t.Fatal(err) + } + var attempts [][]string + for range 3 { + env.mu.Lock() + env.uploadedPaths = nil + env.mu.Unlock() + if err := fm.RotateTokenUpload(ctx, vmID, rotated); err != nil { + t.Fatal(err) + } + env.mu.Lock() + attempts = append(attempts, append([]string(nil), env.uploadedPaths...)) + env.mu.Unlock() + } + for _, got := range attempts { + if !slices.IsSorted(got) || len(got) != 3 || !slices.Equal(got, attempts[0]) { + t.Fatalf("upload orders %v, want the same sorted three files each attempt", attempts) + } + } +} diff --git a/orchestrator/main.go b/orchestrator/main.go index c1366b2..4ab3d5d 100644 --- a/orchestrator/main.go +++ b/orchestrator/main.go @@ -572,6 +572,7 @@ func run() error { fm.SetDurableDestroyer(restateClient) fm.SetDurableArtifacts(restateClient) fm.SetDurableSnapshots(restateClient) + fm.SetDurableTokenRotator(restateClient) } fm.Start(rootCtx) defer fm.Stop()