diff --git a/control/reconnect.go b/control/reconnect.go new file mode 100644 index 0000000..08cd6c7 --- /dev/null +++ b/control/reconnect.go @@ -0,0 +1,68 @@ +package control + +import ( + "context" + "errors" + "time" +) + +// GuestReconnectScope is host-authenticated authority, never a guest request. +// The store must lock and compare BOTH current placement and runner connection +// generation, including connected/lifecycle state, throughout each mutation. +// Hosted adapters additionally hold current membership/policy authorization. +type GuestReconnectScope struct { + WorkspaceID WorkspaceID + PoolID PoolID + SessionID SessionID + RunnerID RunnerID + PlacementGeneration uint64 + ConnectionGeneration uint64 +} + +// GuestReconnectIdentity is enrolled only by spending a fresh boot capability. +// It contains the public verification key, never guest private key material. +type GuestReconnectIdentity struct { + BootEpoch string + PublicKey string +} + +// GuestReconnectAttempt is generated by the control plane, not by a runner or +// guest. One pending attempt per session bounds storage. Beginning another +// attempt replaces only the pending challenge, never the active connection or +// bootstrap capability. ExpiresAt is no more than five seconds after issuance. +type GuestReconnectAttempt struct { + ID string + Challenge string + ExpiresAt time.Time +} + +var ( + ErrReconnectInvalid = errors.New("control: invalid guest reconnect") + ErrReconnectFenced = errors.New("control: guest reconnect fenced") + ErrReconnectExpired = errors.New("control: guest reconnect expired") +) + +// GuestReconnectStore is an optional persistence port. It does not enable a +// driver capability. Cryptographic proof validation belongs above this port; +// Consume must compare the EXACT identity and attempt that were verified. +// All failures leave bootstrap capabilities and connection epochs unchanged. +// Errors must not contain tokens, keys, challenges or untrusted payloads. +type GuestReconnectStore interface { + // EnrollGuest spends hash and pins identity in one atomic operation. An + // existing enrollment cannot be replaced. Ordinary PutSessionBootstrap + // invalidates enrollment for a new boot; reconnect consumption preserves it. + EnrollGuest(context.Context, GuestReconnectScope, string, GuestReconnectIdentity, time.Time) error + // BeginGuestReconnect returns the pinned identity and stores attempt under + // this exact scope. No enrollment means refusal, never legacy fallback. + BeginGuestReconnect(context.Context, GuestReconnectScope, GuestReconnectAttempt, time.Time) (GuestReconnectIdentity, error) + // ReadGuestReconnect reads only the named pending attempt for proof + // verification. This preliminary read does not authorize the connection or + // guarantee current runner/lifecycle authority; Consume rechecks every field + // under a transaction before granting a token or connection epoch. + ReadGuestReconnect(context.Context, GuestReconnectScope, string) (GuestReconnectIdentity, GuestReconnectAttempt, error) + // ConsumeGuestReconnect is a single-use compare-and-swap after proof + // verification. It increments and returns a connection epoch and replaces + // the bootstrap hash atomically. No retry returns a cached success/token. + // Delivery and old-relay fencing must still enforce the returned epoch. + ConsumeGuestReconnect(context.Context, GuestReconnectScope, GuestReconnectIdentity, GuestReconnectAttempt, SessionBootstrap, time.Time) (uint64, error) +} diff --git a/controlapp/reconnect.go b/controlapp/reconnect.go new file mode 100644 index 0000000..f1b8452 --- /dev/null +++ b/controlapp/reconnect.go @@ -0,0 +1,117 @@ +package controlapp + +import ( + "context" + "crypto/ed25519" + "crypto/rand" + "encoding/base64" + "strconv" + "time" + + "github.com/tokencanopy/rainier/control" + "github.com/tokencanopy/rainier/protocol/runner" +) + +// GuestReconnect authorizes continuity of an enrolled guest. Callers supply an +// authenticated workload scope and (for hosted use) hold current membership and +// policy authorization. This service does not attach relays or persist RAM. +// Driver capability advertisement must wait for both host and guest integration. +type GuestReconnect struct { + Store control.GuestReconnectStore + Clock control.Clock +} + +func (s GuestReconnect) available() bool { return s.Store != nil && s.Clock != nil } + +// Enroll binds a process-memory public key to a fresh, single-use boot token. +// Private keys never cross this interface. Failed enrollment cannot downgrade +// to a reconnect path that merely replays the old token. +func (s GuestReconnect) Enroll(ctx context.Context, b control.GuestReconnectScope, token, bootEpoch, publicKey string) error { + if !s.available() { + return control.ErrUnavailable + } + if len(publicKey) != 43 || len(token) != 43 { + return control.ErrReconnectInvalid + } + key, err := base64.RawURLEncoding.Strict().DecodeString(publicKey) + if err != nil || len(key) != ed25519.PublicKeySize || base64.RawURLEncoding.EncodeToString(key) != publicKey || len(token) != 43 { + return control.ErrReconnectInvalid + } + // Use the same field validation as the signed protocol, not a second + // interpretation of which boot identities can later be proved. + probe := reconnectChallenge(b, control.GuestReconnectIdentity{BootEpoch: bootEpoch}, control.GuestReconnectAttempt{ID: "validate", Challenge: "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"}) + if _, err := probe.SigningMessage(); err != nil { + return control.ErrReconnectInvalid + } + return s.Store.EnrollGuest(ctx, b, HashSessionBootstrapToken(token), control.GuestReconnectIdentity{BootEpoch: bootEpoch, PublicKey: publicKey}, s.Clock.Now()) +} + +// Begin mints a bounded pending attempt. It cannot evict an active relay or +// replace a bootstrap token. A new attempt invalidates only the prior challenge. +func (s GuestReconnect) Begin(ctx context.Context, b control.GuestReconnectScope) (runner.GuestReconnectChallenge, error) { + if !s.available() { + return runner.GuestReconnectChallenge{}, control.ErrUnavailable + } + attemptID, err := reconnectRandom() + if err != nil { + return runner.GuestReconnectChallenge{}, err + } + nonce, err := reconnectRandom() + if err != nil { + return runner.GuestReconnectChallenge{}, err + } + // PostgreSQL stores microseconds. Issuance and returned deadline use that + // same precision so a database round trip cannot change the CAS operand. + now := s.Clock.Now().UTC().Truncate(time.Microsecond) + a := control.GuestReconnectAttempt{ID: attemptID, Challenge: nonce, ExpiresAt: now.Add(5 * time.Second)} + id, err := s.Store.BeginGuestReconnect(ctx, b, a, now) + if err != nil { + return runner.GuestReconnectChallenge{}, err + } + c := reconnectChallenge(b, id, a) + if _, err := c.SigningMessage(); err != nil { + return runner.GuestReconnectChallenge{}, control.ErrReconnectInvalid + } + return c, nil +} + +// Accept verifies the exact durable challenge, then atomically spends it, +// advances connection epoch and replaces the bootstrap capability. A lost reply +// must start a NEW challenge; no successful response is cached or replayable. +func (s GuestReconnect) Accept(ctx context.Context, b control.GuestReconnectScope, attemptID, signature string) (uint64, string, error) { + if !s.available() { + return 0, "", control.ErrUnavailable + } + if len(attemptID) != 43 || len(signature) != 86 { + return 0, "", control.ErrReconnectInvalid + } + id, a, err := s.Store.ReadGuestReconnect(ctx, b, attemptID) + if err != nil { + return 0, "", err + } + if err := reconnectChallenge(b, id, a).VerifyProof(id.PublicKey, signature); err != nil { + return 0, "", control.ErrReconnectInvalid + } + token, err := reconnectRandom() + if err != nil { + return 0, "", err + } + now := s.Clock.Now() + epoch, err := s.Store.ConsumeGuestReconnect(ctx, b, id, a, control.SessionBootstrap{Hash: HashSessionBootstrapToken(token), PlacementGeneration: b.PlacementGeneration, ExpiresAt: now.Add(SessionBootstrapTTL)}, now) + if err != nil { + return 0, "", err + } + return epoch, token, nil +} + +func reconnectChallenge(b control.GuestReconnectScope, id control.GuestReconnectIdentity, a control.GuestReconnectAttempt) runner.GuestReconnectChallenge { + return runner.GuestReconnectChallenge{Protocol: runner.GuestReconnectProtocol, SessionID: string(b.SessionID), BootEpoch: id.BootEpoch, HostIncarnation: strconv.FormatUint(b.ConnectionGeneration, 10), AttemptID: a.ID, PlacementGeneration: b.PlacementGeneration, Challenge: a.Challenge} +} + +func reconnectRandom() (string, error) { + var b [32]byte + if _, err := rand.Read(b[:]); err != nil { + return "", control.ErrUnavailable + } + return base64.RawURLEncoding.EncodeToString(b[:]), nil +} diff --git a/controlapp/repotest/reconnect.go b/controlapp/repotest/reconnect.go new file mode 100644 index 0000000..b7bcaec --- /dev/null +++ b/controlapp/repotest/reconnect.go @@ -0,0 +1,254 @@ +package repotest + +import ( + "context" + "errors" + "github.com/tokencanopy/rainier/control" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// GuestReconnectStores extends the ordinary repository contract with optional +// guest authorization over the SAME durable state. No wire capability is implied. +type GuestReconnectStores struct { + Stores + Reconnects control.GuestReconnectStore +} + +// RunGuestReconnect is the adapter-neutral authorization contract. The factory +// supplies an empty store for each case, with live wall-clock expiry enforcement. +// Both self-hosted and hosted implementations must pass before reconnect is enabled. +func RunGuestReconnect(t *testing.T, open func(*testing.T) GuestReconnectStores) { + for _, tc := range []struct { + name string + run func(*testing.T, GuestReconnectStores) + }{ + {"durable single-use authorization", reconnectDurableAuthorization}, + {"disconnected and terminal authority", reconnectLifecycle}, + } { + t.Run(tc.name, func(t *testing.T) { + s := open(t) + for _, ws := range []control.WorkspaceID{Alpha, Beta} { + if err := s.Provision(context.Background(), ws); err != nil { + t.Fatal(err) + } + } + tc.run(t, s) + }) + } +} + +func reconnectDurableAuthorization(t *testing.T, st GuestReconnectStores) { + ctx := context.Background() + scope := control.GuestReconnectScope{WorkspaceID: Alpha, PoolID: PoolA, SessionID: "session.test", RunnerID: "runner.test", PlacementGeneration: 1, ConnectionGeneration: 7} + _, err := st.Sessions.CreateSession(ctx, scope.WorkspaceID, control.Session{ID: scope.SessionID, CreatorID: "actor.test", PoolID: scope.PoolID, RunnerID: scope.RunnerID, State: control.StateRunning, Spec: control.PortableSpec{Image: "image.test"}}) + if err != nil { + t.Fatal(err) + } + if err := st.Fleet.UpsertRunner(ctx, scope.PoolID, control.Runner{ID: scope.RunnerID, Generation: 7, Connected: true}); err != nil { + t.Fatal(err) + } + now := time.Now().UTC().Truncate(time.Microsecond) + token := control.SessionBootstrap{Hash: strings.Repeat("a", 64), PlacementGeneration: 1, ExpiresAt: now.Add(time.Minute)} + if err := st.Bootstraps.PutSessionBootstrap(ctx, scope.WorkspaceID, scope.SessionID, token); err != nil { + t.Fatal(err) + } + identity := control.GuestReconnectIdentity{BootEpoch: "boot.test", PublicKey: "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"} + r := st.Reconnects + bad := scope + bad.ConnectionGeneration = 6 + if err := r.EnrollGuest(ctx, bad, token.Hash, identity, now); !errors.Is(err, control.ErrReconnectFenced) { + t.Fatalf("stale enrollment: %v", err) + } + if err := r.EnrollGuest(ctx, scope, token.Hash, identity, now); err != nil { + t.Fatal(err) + } + if err := r.EnrollGuest(ctx, scope, token.Hash, identity, now); err == nil { + t.Fatal("bootstrap replay enrolled") + } + attempt := control.GuestReconnectAttempt{ID: "attempt.test", Challenge: "nonce.test", ExpiresAt: now.Add(5 * time.Second)} + enrolled, err := r.BeginGuestReconnect(ctx, scope, attempt, now) + if err != nil || enrolled != identity { + t.Fatalf("begin: %+v %v", enrolled, err) + } + fresh := control.SessionBootstrap{Hash: strings.Repeat("b", 64), PlacementGeneration: 1, ExpiresAt: now.Add(time.Minute)} + for name, changed := range map[string]control.GuestReconnectScope{ + "workspace": {WorkspaceID: "ws_other", PoolID: scope.PoolID, SessionID: scope.SessionID, RunnerID: scope.RunnerID, PlacementGeneration: 1, ConnectionGeneration: 7}, + "pool": {WorkspaceID: scope.WorkspaceID, PoolID: "pool_other", SessionID: scope.SessionID, RunnerID: scope.RunnerID, PlacementGeneration: 1, ConnectionGeneration: 7}, + "runner": {WorkspaceID: scope.WorkspaceID, PoolID: scope.PoolID, SessionID: scope.SessionID, RunnerID: "other.test", PlacementGeneration: 1, ConnectionGeneration: 7}, + "placement": {WorkspaceID: scope.WorkspaceID, PoolID: scope.PoolID, SessionID: scope.SessionID, RunnerID: scope.RunnerID, PlacementGeneration: 2, ConnectionGeneration: 7}, + } { + t.Run(name, func(t *testing.T) { + if _, err := r.ConsumeGuestReconnect(ctx, changed, identity, attempt, fresh, now); err == nil { + t.Fatal("foreign authority accepted") + } + }) + } + wrongIdentity := identity + wrongIdentity.BootEpoch = "other-boot.test" + if _, err := r.ConsumeGuestReconnect(ctx, scope, wrongIdentity, attempt, fresh, now); err == nil { + t.Fatal("wrong boot accepted") + } + wrongIdentity = identity + wrongIdentity.PublicKey = "other-key.test" + if _, err := r.ConsumeGuestReconnect(ctx, scope, wrongIdentity, attempt, fresh, now); err == nil { + t.Fatal("wrong pinned key accepted") + } + wrong := attempt + wrong.Challenge = "wrong.test" + if _, err := r.ConsumeGuestReconnect(ctx, scope, identity, wrong, fresh, now); err == nil { + t.Fatal("wrong challenge accepted") + } + var wins atomic.Int32 + var wg sync.WaitGroup + for i := 0; i < 12; i++ { + wg.Add(1) + go func() { + defer wg.Done() + epoch, err := r.ConsumeGuestReconnect(ctx, scope, identity, attempt, fresh, now) + if err == nil { + if epoch != 1 { + t.Errorf("epoch %d", epoch) + } + wins.Add(1) + } + }() + } + wg.Wait() + if wins.Load() != 1 { + t.Fatalf("concurrent successes: %d", wins.Load()) + } + if err := st.Bootstraps.ConsumeSessionBootstrap(ctx, scope.WorkspaceID, scope.SessionID, fresh.Hash, 1, now); err != nil { + t.Fatal(err) + } + attempt.ID = "second.test" + if _, err := r.BeginGuestReconnect(ctx, scope, attempt, now); err != nil { + t.Fatal(err) + } + if _, err := r.ConsumeGuestReconnect(ctx, scope, identity, attempt, fresh, attempt.ExpiresAt); !errors.Is(err, control.ErrReconnectExpired) { + t.Fatalf("expiry boundary: %v", err) + } + attempt.ID = "third.test" + if _, err := r.BeginGuestReconnect(ctx, scope, attempt, now); err != nil { + t.Fatal(err) + } + if err := st.Fleet.UpsertRunner(ctx, scope.PoolID, control.Runner{ID: scope.RunnerID, Generation: 8, Connected: true}); err != nil { + t.Fatal(err) + } + if _, err := r.ConsumeGuestReconnect(ctx, scope, identity, attempt, fresh, now); !errors.Is(err, control.ErrReconnectFenced) { + t.Fatalf("superseded runner: %v", err) + } + scope.ConnectionGeneration = 8 + attempt.ID = "fourth.test" + if _, err := r.BeginGuestReconnect(ctx, scope, attempt, now); err != nil { + t.Fatal(err) + } + if epoch, err := r.ConsumeGuestReconnect(ctx, scope, identity, attempt, fresh, now); err != nil || epoch != 2 { + t.Fatalf("next epoch: %d %v", epoch, err) + } + previous := attempt + previous.ID = "pending-a.test" + if _, err := r.BeginGuestReconnect(ctx, scope, previous, now); err != nil { + t.Fatal(err) + } + attempt.ID = "pending-b.test" + if _, err := r.BeginGuestReconnect(ctx, scope, attempt, now); err != nil { + t.Fatal(err) + } + if _, err := r.ConsumeGuestReconnect(ctx, scope, identity, previous, fresh, now); err == nil { + t.Fatal("replaced attempt accepted") + } + // Beginning either attempt and rejecting A cannot alter the accepted token. + if err := st.Bootstraps.ConsumeSessionBootstrap(ctx, scope.WorkspaceID, scope.SessionID, fresh.Hash, 1, now); err != nil { + t.Fatal("begin/refusal changed current bootstrap", err) + } + if epoch, err := r.ConsumeGuestReconnect(ctx, scope, identity, attempt, fresh, now); err != nil || epoch != 3 { + t.Fatalf("refusal changed epoch: %d %v", epoch, err) + } + attempt.ID = "pending-cold.test" + if _, err := r.BeginGuestReconnect(ctx, scope, attempt, now); err != nil { + t.Fatal(err) + } + // A stale caller clock cannot stretch a five-second attempt indefinitely. + expiredAttempt := control.GuestReconnectAttempt{ID: "old-clock.test", Challenge: "nonce.test", ExpiresAt: now.Add(-time.Minute)} + if _, err := r.BeginGuestReconnect(ctx, scope, expiredAttempt, now.Add(-time.Minute-5*time.Second)); !errors.Is(err, control.ErrReconnectExpired) { + t.Fatalf("stale issuance clock: %v", err) + } + // A cold boot invalidates enrollment, pending attempts, and old proof state. + if err := st.Bootstraps.PutSessionBootstrap(ctx, scope.WorkspaceID, scope.SessionID, token); err != nil { + t.Fatal(err) + } + if _, err := r.BeginGuestReconnect(ctx, scope, attempt, now); err == nil { + t.Fatal("cold mint retained enrollment") + } + if _, err := r.ConsumeGuestReconnect(ctx, scope, identity, attempt, fresh, now); err == nil { + t.Fatal("cold mint retained pending proof") + } + if err := st.Bootstraps.ConsumeSessionBootstrap(ctx, scope.WorkspaceID, scope.SessionID, token.Hash, 1, now); err != nil { + t.Fatal("rejected proof changed cold token", err) + } +} + +func reconnectLifecycle(t *testing.T, s GuestReconnectStores) { + ctx := context.Background() + b := control.GuestReconnectScope{WorkspaceID: Alpha, PoolID: PoolA, SessionID: "lifecycle.test", RunnerID: "runner.test", PlacementGeneration: 1, ConnectionGeneration: 7} + if _, err := s.Sessions.CreateSession(ctx, Alpha, control.Session{ID: b.SessionID, CreatorID: "actor.test", PoolID: PoolA, RunnerID: b.RunnerID, State: control.StateRunning}); err != nil { + t.Fatal(err) + } + if err := s.Fleet.UpsertRunner(ctx, PoolA, control.Runner{ID: b.RunnerID, Generation: 7, Connected: true}); err != nil { + t.Fatal(err) + } + now := time.Now().UTC().Truncate(time.Microsecond) + token := control.SessionBootstrap{Hash: strings.Repeat("a", 64), PlacementGeneration: 1, ExpiresAt: now.Add(time.Minute)} + if err := s.Bootstraps.PutSessionBootstrap(ctx, Alpha, b.SessionID, token); err != nil { + t.Fatal(err) + } + id := control.GuestReconnectIdentity{BootEpoch: "boot.test", PublicKey: "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"} + if err := s.Reconnects.EnrollGuest(ctx, b, token.Hash, id, now); err != nil { + t.Fatal(err) + } + a := control.GuestReconnectAttempt{ID: "first.test", Challenge: "nonce.test", ExpiresAt: now.Add(5 * time.Second)} + if _, err := s.Reconnects.BeginGuestReconnect(ctx, b, a, now); err != nil { + t.Fatal(err) + } + fresh := token + fresh.Hash = strings.Repeat("b", 64) + if epoch, err := s.Reconnects.ConsumeGuestReconnect(ctx, b, id, a, fresh, now); err != nil || epoch != 1 { + t.Fatalf("initial accept: %d %v", epoch, err) + } + a.ID = "next.test" + if _, err := s.Reconnects.BeginGuestReconnect(ctx, b, a, now); err != nil { + t.Fatal(err) + } + if err := s.Fleet.SetRunnerConnected(ctx, PoolA, b.RunnerID, false); err != nil { + t.Fatal(err) + } + if _, err := s.Reconnects.BeginGuestReconnect(ctx, b, a, now); !errors.Is(err, control.ErrReconnectFenced) { + t.Fatalf("disconnected begin: %v", err) + } + if _, err := s.Reconnects.ConsumeGuestReconnect(ctx, b, id, a, token, now); !errors.Is(err, control.ErrReconnectFenced) { + t.Fatalf("disconnected consume: %v", err) + } + if err := s.Fleet.SetRunnerConnected(ctx, PoolA, b.RunnerID, true); err != nil { + t.Fatal(err) + } + if epoch, err := s.Reconnects.ConsumeGuestReconnect(ctx, b, id, a, fresh, now); err != nil || epoch != 2 { + t.Fatalf("refusal advanced epoch: %d %v", epoch, err) + } + a.ID = "terminal.test" + if _, err := s.Reconnects.BeginGuestReconnect(ctx, b, a, now); err != nil { + t.Fatal(err) + } + if err := s.Sessions.Transition(ctx, Alpha, b.SessionID, []control.SessionState{control.StateRunning}, control.StateDestroyed, control.TransitionOpts{}); err != nil { + t.Fatal(err) + } + if _, err := s.Reconnects.ConsumeGuestReconnect(ctx, b, id, a, token, now); !errors.Is(err, control.ErrReconnectFenced) { + t.Fatalf("terminal consume: %v", err) + } + if err := s.Bootstraps.ConsumeSessionBootstrap(ctx, Alpha, b.SessionID, fresh.Hash, 1, now); err != nil { + t.Fatal("refusal changed capability", err) + } +} diff --git a/docs/design/2026-09-29-guest-reconnect-authorization.md b/docs/design/2026-09-29-guest-reconnect-authorization.md new file mode 100644 index 0000000..60bdaaa --- /dev/null +++ b/docs/design/2026-09-29-guest-reconnect-authorization.md @@ -0,0 +1,78 @@ +# Guest reconnection authorization foundation + +This change supplies protocol, application and PostgreSQL primitives for the +first stage of authenticated live-guest reconnection. It does **not** enable +reconnection, advertise a driver capability, expose a new RPC method, recover a +listener, or persist guest RAM. Existing guest and Docker paths keep their current +behavior. The staged production design lives in rainier-cloud PR143. + +## Authority and ownership + +`runnerplane.Binding.ConnectionGeneration` is taken from the registration that +owns the request's WebSocket. It is not a current-generation lookup or a field in +an announce/request body. Existing hosts may ignore it; new generation-sensitive +operations must compare it with durable fleet state. Propagating it alone does +not add authorization to existing session RPC methods. + +`controlapp.GuestReconnect` owns randomness, proof verification and bootstrap token +creation. The optional `control.GuestReconnectStore` owns enrollment, challenge +state and transactional consumption. Hosted adapters must hold current membership +and policy authorization in addition to implementing the placement/runner fences. +The reusable `controlapp/repotest.RunGuestReconnect` contract covers concurrency, +replay, scope, lifecycle, pending replacement and bootstrap preservation. The +PostgreSQL suite additionally exercises adapter reopen and expiry during lock +waits. Hosted qualification must run the same contract against its own adapter. + +No guest input selects the workspace, session, runner or placement scope. + +Enrollment spends the initial bootstrap token and pins a public Ed25519 key and +boot epoch atomically. Only the guest holds the private key. A normal cold-boot +bootstrap mint invalidates the enrollment and pending attempt. A reconnect token +replacement preserves enrollment. These are deliberately different operations. + +Beginning an attempt replaces only the pending challenge. It cannot revoke a live +connection or replace a bootstrap capability. Proof acceptance compares the exact +pinned identity, pending attempt, placement and connection generation, increments +the connection epoch, consumes the attempt and installs a new bootstrap hash in +one transaction. Concurrent consumers produce one success. A lost response cannot +be replayed; the caller must start a new attempt. + +The PostgreSQL adapter holds runner and session row locks until commit, checks +connected/live state and uses database time after acquiring authorization locks +for expiry. The caller's clock cannot extend a five-second challenge. Database +clock skew can cause a closed refusal; it must never extend validity. No failed +operation returns a token or an epoch. Database errors are fixed sentinels. + +The connection epoch is scoped to an enrolled boot. It is not the placement +number, runner connection generation, or terminal controller lease. A fresh boot +invalidates the old identity and starts a new epoch sequence. + +## Signed protocol + +The transcript is the bytes `rainier-guest-reconnect-v1`, NUL, then four +big-endian uint32-length-prefixed printable ASCII identifiers (session, boot, +host incarnation, attempt), a big-endian uint64 placement generation and a +32-byte random challenge. Identifiers are 1–256 bytes. The host incarnation is +the authenticated runner connection generation in decimal. Public keys, signatures +and nonces use canonical unpadded base64url. Golden-vector tests pin the format. + +Signature verification alone is not authorization: the corresponding pending +store row must still exist and be consumed under current authority. + +## Integration work required before enabling + +- Atomically enroll through the guest bootstrap exchange with capability + negotiation and strict bounded request decoding. +- Add runner-only challenge/proof RPC routing and hosted store support. +- Retain the guest signing key in sessiond memory; fail closed on reconnect + authorization failure and apply fresh configuration even without secrets. +- Bound unauthenticated peers and frames; authenticate before sending any boot + configuration or replacing a live relay. +- Fence old relay input and credential RPC immediately and order accepted + handoffs by connection epoch. +- Reauthorize recovered placements and verify the VMM, network slot and socket + ownership before restoring listeners. Do not unlink a live listener. +- Qualify PID, PTY, child and real-agent continuity on a disposable KVM guest. + +These gates remain unmet by this foundation. Memory persistence and any change to +credential custody require the separate policy amendments and lifecycle work. diff --git a/internal/controld/pgstore/bootstrap.go b/internal/controld/pgstore/bootstrap.go index 8ab6317..9eea022 100644 --- a/internal/controld/pgstore/bootstrap.go +++ b/internal/controld/pgstore/bootstrap.go @@ -41,6 +41,10 @@ func (r pgBootstraps) PutSessionBootstrap(ctx context.Context, ws control.Worksp placement_generation = EXCLUDED.placement_generation, expires_at = EXCLUDED.expires_at, consumed_at = NULL, + guest_boot_epoch = NULL, guest_public_key = NULL, + reconnect_attempt = NULL, reconnect_challenge = NULL, + reconnect_expires_at = NULL, reconnect_runner_generation = NULL, + guest_connection_epoch = 0, created_at = now()`, string(id), string(ws), rec.Hash, int64(rec.PlacementGeneration), rec.ExpiresAt) if err != nil { diff --git a/internal/controld/pgstore/migrations/0016_guest_reconnect.sql b/internal/controld/pgstore/migrations/0016_guest_reconnect.sql new file mode 100644 index 0000000..c40c95a --- /dev/null +++ b/internal/controld/pgstore/migrations/0016_guest_reconnect.sql @@ -0,0 +1,12 @@ +-- Optional live-guest authorization. No private keys or RAM are persisted. +-- State lives on the bootstrap row so cold boot invalidation and reconnect +-- token replacement cannot race across independent authorization records. +ALTER TABLE session_bootstraps + ADD COLUMN guest_boot_epoch text, + ADD COLUMN guest_public_key text, + ADD COLUMN reconnect_attempt text, + ADD COLUMN reconnect_challenge text, + ADD COLUMN reconnect_expires_at timestamptz, + ADD COLUMN reconnect_runner_generation bigint, + ADD COLUMN guest_connection_epoch bigint NOT NULL DEFAULT 0 CHECK (guest_connection_epoch >= 0), + ADD CONSTRAINT guest_identity_complete CHECK ((guest_boot_epoch IS NULL) = (guest_public_key IS NULL)); diff --git a/internal/controld/pgstore/pgstore_test.go b/internal/controld/pgstore/pgstore_test.go index 7bb056e..54c31bb 100644 --- a/internal/controld/pgstore/pgstore_test.go +++ b/internal/controld/pgstore/pgstore_test.go @@ -272,14 +272,14 @@ func TestMigrate0003To0004AddsColumnsToLegacyRows(t *testing.T) { if want := embeddedMigrationVersions(t); !slices.Equal(applied, want) { t.Fatalf("schema_migrations = %v, want every embedded migration in order %v", applied, want) } - // This release's head is 15: a database that stopped at 0003 runs the + // This release's head is 16: a database that stopped at 0003 runs the // expand step (0007), the contract step (0008), the events table // (0009), the agent credentials table (0010), the tombstone (0011), the // durable revoke fence (0012), the controller lease (0013), the exec - // event's command name (0014), and the session bootstrap token (0015) in - // the same start. - if head := applied[len(applied)-1]; head != 15 { - t.Fatalf("head migration = %d, want 15", head) + // event's command name (0014), the session bootstrap token (0015), and + // guest reconnect authorization (0016) in the same start. + if head := applied[len(applied)-1]; head != 16 { + t.Fatalf("head migration = %d, want 16", head) } // The legacy session survived, and its new columns read as "never exited" diff --git a/internal/controld/pgstore/reconnect.go b/internal/controld/pgstore/reconnect.go new file mode 100644 index 0000000..da28b7c --- /dev/null +++ b/internal/controld/pgstore/reconnect.go @@ -0,0 +1,164 @@ +package pgstore + +import ( + "context" + "errors" + "math" + "time" + + "github.com/jackc/pgx/v5" + "github.com/tokencanopy/rainier/control" +) + +type pgGuestReconnects struct{ s *Store } + +var _ control.GuestReconnectStore = pgGuestReconnects{} + +// GuestReconnects exposes durable authorization without enabling any wire path. +func (s *Store) GuestReconnects() control.GuestReconnectStore { return pgGuestReconnects{s} } + +func validReconnectScope(b control.GuestReconnectScope) bool { + return b.WorkspaceID != "" && b.PoolID != "" && b.SessionID != "" && b.RunnerID != "" && b.PlacementGeneration > 0 && b.PlacementGeneration <= math.MaxInt64 && b.ConnectionGeneration > 0 && b.ConnectionGeneration <= math.MaxInt64 +} + +// lockScope holds authority stable until the enclosing transaction commits. +// A read outside this transaction would authorize a takeover after re-placement +// or a newer runner registration. The row locks also serialize those changes. +func (r pgGuestReconnects) lockScope(ctx context.Context, b control.GuestReconnectScope) error { + if !validReconnectScope(b) { + return control.ErrReconnectInvalid + } + var found int + err := r.s.q(ctx).QueryRow(ctx, `SELECT 1 FROM runners WHERE pool_id=$1 AND name=$2 AND generation=$3 AND connected FOR SHARE`, string(b.PoolID), string(b.RunnerID), int64(b.ConnectionGeneration)).Scan(&found) + if errors.Is(err, pgx.ErrNoRows) { + return control.ErrReconnectFenced + } + if err != nil { + return control.ErrUnavailable + } + err = r.s.q(ctx).QueryRow(ctx, `SELECT 1 FROM sessions WHERE workspace_id=$1 AND id=$2 AND pool_id=$3 AND runner=$4 AND placement_generation=$5 AND state IN ('creating','running') FOR UPDATE`, string(b.WorkspaceID), string(b.SessionID), string(b.PoolID), string(b.RunnerID), int64(b.PlacementGeneration)).Scan(&found) + if errors.Is(err, pgx.ErrNoRows) { + return control.ErrReconnectFenced + } + if err != nil { + return control.ErrUnavailable + } + return nil +} + +func (r pgGuestReconnects) EnrollGuest(ctx context.Context, b control.GuestReconnectScope, hash string, id control.GuestReconnectIdentity, now time.Time) error { + if hash == "" || len(id.BootEpoch) == 0 || len(id.BootEpoch) > 256 || len(id.PublicKey) == 0 || len(id.PublicKey) > 256 || now.IsZero() { + return control.ErrReconnectInvalid + } + return r.s.Run(ctx, func(ctx context.Context) error { + if err := r.lockScope(ctx, b); err != nil { + return err + } + ct, err := r.s.q(ctx).Exec(ctx, `UPDATE session_bootstraps SET consumed_at=$1,guest_boot_epoch=$2,guest_public_key=$3 WHERE workspace_id=$4 AND session_id=$5 AND token_hash=$6 AND placement_generation=$7 AND expires_at>$1 AND expires_at>clock_timestamp() AND consumed_at IS NULL AND guest_public_key IS NULL`, now, id.BootEpoch, id.PublicKey, string(b.WorkspaceID), string(b.SessionID), hash, int64(b.PlacementGeneration)) + if err != nil { + return control.ErrUnavailable + } + if ct.RowsAffected() != 1 { + return control.ErrReconnectInvalid + } + return nil + }) +} + +func validReconnectAttempt(a control.GuestReconnectAttempt, now time.Time) bool { + return len(a.ID) > 0 && len(a.ID) <= 256 && len(a.Challenge) > 0 && len(a.Challenge) <= 256 && !now.IsZero() && a.ExpiresAt.After(now) && !a.ExpiresAt.After(now.Add(5*time.Second)) +} + +func (r pgGuestReconnects) BeginGuestReconnect(ctx context.Context, b control.GuestReconnectScope, a control.GuestReconnectAttempt, now time.Time) (control.GuestReconnectIdentity, error) { + var id control.GuestReconnectIdentity + if !validReconnectAttempt(a, now) { + return id, control.ErrReconnectInvalid + } + err := r.s.Run(ctx, func(ctx context.Context) error { + if err := r.lockScope(ctx, b); err != nil { + return err + } + var databaseNow time.Time + if err := r.s.q(ctx).QueryRow(ctx, `SELECT clock_timestamp()`).Scan(&databaseNow); err != nil { + return control.ErrUnavailable + } + if !databaseNow.Before(a.ExpiresAt) { + return control.ErrReconnectExpired + } + if a.ExpiresAt.After(databaseNow.Add(5 * time.Second)) { + return control.ErrReconnectInvalid + } + err := r.s.q(ctx).QueryRow(ctx, `UPDATE session_bootstraps SET reconnect_attempt=$1,reconnect_challenge=$2,reconnect_expires_at=$3,reconnect_runner_generation=$4 WHERE workspace_id=$5 AND session_id=$6 AND placement_generation=$7 AND guest_public_key IS NOT NULL RETURNING guest_boot_epoch,guest_public_key`, a.ID, a.Challenge, a.ExpiresAt, int64(b.ConnectionGeneration), string(b.WorkspaceID), string(b.SessionID), int64(b.PlacementGeneration)).Scan(&id.BootEpoch, &id.PublicKey) + if errors.Is(err, pgx.ErrNoRows) { + return control.ErrReconnectInvalid + } + if err != nil { + return control.ErrUnavailable + } + return nil + }) + if err != nil { + return control.GuestReconnectIdentity{}, err + } + return id, nil +} + +func (r pgGuestReconnects) ConsumeGuestReconnect(ctx context.Context, b control.GuestReconnectScope, id control.GuestReconnectIdentity, a control.GuestReconnectAttempt, token control.SessionBootstrap, now time.Time) (uint64, error) { + var epoch uint64 + if token.Hash == "" || token.PlacementGeneration != b.PlacementGeneration || !token.ExpiresAt.After(now) || token.ExpiresAt.After(now.Add(120*time.Second)) || now.IsZero() { + return 0, control.ErrReconnectInvalid + } + err := r.s.Run(ctx, func(ctx context.Context) error { + if err := r.lockScope(ctx, b); err != nil { + return err + } + var expiry time.Time + var current int64 + err := r.s.q(ctx).QueryRow(ctx, `SELECT reconnect_expires_at,guest_connection_epoch FROM session_bootstraps WHERE workspace_id=$1 AND session_id=$2 AND placement_generation=$3 AND guest_boot_epoch=$4 AND guest_public_key=$5 AND reconnect_attempt=$6 AND reconnect_challenge=$7 AND reconnect_runner_generation=$8 FOR UPDATE`, string(b.WorkspaceID), string(b.SessionID), int64(b.PlacementGeneration), id.BootEpoch, id.PublicKey, a.ID, a.Challenge, int64(b.ConnectionGeneration)).Scan(&expiry, ¤t) + if errors.Is(err, pgx.ErrNoRows) { + return control.ErrReconnectInvalid + } + if err != nil { + return control.ErrUnavailable + } + var databaseNow time.Time + if err := r.s.q(ctx).QueryRow(ctx, `SELECT clock_timestamp()`).Scan(&databaseNow); err != nil { + return control.ErrUnavailable + } + if !now.Before(expiry) || !databaseNow.Before(expiry) { + return control.ErrReconnectExpired + } + if !token.ExpiresAt.After(databaseNow) || token.ExpiresAt.After(databaseNow.Add(120*time.Second)) { + return control.ErrReconnectInvalid + } + if !expiry.Equal(a.ExpiresAt) || current == math.MaxInt64 { + return control.ErrReconnectInvalid + } + _, err = r.s.q(ctx).Exec(ctx, `UPDATE session_bootstraps SET guest_connection_epoch=guest_connection_epoch+1,token_hash=$1,expires_at=$2,consumed_at=NULL,reconnect_attempt=NULL,reconnect_challenge=NULL,reconnect_expires_at=NULL,reconnect_runner_generation=NULL WHERE workspace_id=$3 AND session_id=$4`, token.Hash, token.ExpiresAt, string(b.WorkspaceID), string(b.SessionID)) + if err != nil { + return control.ErrUnavailable + } + epoch = uint64(current + 1) + return nil + }) + if err != nil { + return 0, err + } + return epoch, nil +} + +func (r pgGuestReconnects) ReadGuestReconnect(ctx context.Context, b control.GuestReconnectScope, attemptID string) (control.GuestReconnectIdentity, control.GuestReconnectAttempt, error) { + var id control.GuestReconnectIdentity + var a control.GuestReconnectAttempt + if !validReconnectScope(b) || len(attemptID) == 0 || len(attemptID) > 256 { + return id, a, control.ErrReconnectInvalid + } + err := r.s.q(ctx).QueryRow(ctx, `SELECT guest_boot_epoch,guest_public_key,reconnect_attempt,reconnect_challenge,reconnect_expires_at FROM session_bootstraps WHERE workspace_id=$1 AND session_id=$2 AND placement_generation=$3 AND reconnect_runner_generation=$4 AND reconnect_attempt=$5 AND guest_public_key IS NOT NULL`, string(b.WorkspaceID), string(b.SessionID), int64(b.PlacementGeneration), int64(b.ConnectionGeneration), attemptID).Scan(&id.BootEpoch, &id.PublicKey, &a.ID, &a.Challenge, &a.ExpiresAt) + if errors.Is(err, pgx.ErrNoRows) { + return control.GuestReconnectIdentity{}, control.GuestReconnectAttempt{}, control.ErrReconnectInvalid + } + if err != nil { + return control.GuestReconnectIdentity{}, control.GuestReconnectAttempt{}, control.ErrUnavailable + } + return id, a, nil +} diff --git a/internal/controld/pgstore/reconnect_test.go b/internal/controld/pgstore/reconnect_test.go new file mode 100644 index 0000000..626ca37 --- /dev/null +++ b/internal/controld/pgstore/reconnect_test.go @@ -0,0 +1,166 @@ +package pgstore + +import ( + "context" + "crypto/ed25519" + "crypto/rand" + "encoding/base64" + "errors" + "fmt" + "github.com/tokencanopy/rainier/controlapp" + "github.com/tokencanopy/rainier/controlapp/repotest" + "testing" + "time" + + "github.com/tokencanopy/rainier/control" +) + +func TestGuestReconnectStoreContract(t *testing.T) { + dsn := startPostgres(t) + repotest.RunGuestReconnect(t, func(t *testing.T) repotest.GuestReconnectStores { + st := freshStore(t, dsn, t.Name()) + return repotest.GuestReconnectStores{Stores: repotest.Stores{Sessions: st.Sessions(), Fleet: st.Fleet(), Bootstraps: st.Bootstraps(), Provision: st.EnsureWorkspace}, Reconnects: st.GuestReconnects()} + }) +} + +func TestGuestReconnectProofWithPostgres(t *testing.T) { + ctx := context.Background() + st := freshStore(t, startPostgres(t), t.Name()) + b := control.GuestReconnectScope{WorkspaceID: "ws_self_hosted", PoolID: "pool_self_hosted", SessionID: "proof.test", RunnerID: "runner.test", PlacementGeneration: 1, ConnectionGeneration: 7} + if _, err := st.Sessions().CreateSession(ctx, b.WorkspaceID, control.Session{ID: b.SessionID, CreatorID: "actor.test", PoolID: b.PoolID, RunnerID: b.RunnerID, State: control.StateRunning}); err != nil { + t.Fatal(err) + } + if err := st.Fleet().UpsertRunner(ctx, b.PoolID, control.Runner{ID: b.RunnerID, Generation: 7, Connected: true}); err != nil { + t.Fatal(err) + } + clock := reconnectClock{time.Now().UTC().Truncate(time.Microsecond)} + mint := controlapp.SessionBootstrapMinter{Store: st.Bootstraps(), Clock: clock} + token, err := mint.Mint(ctx, b.WorkspaceID, b.SessionID, 1) + if err != nil { + t.Fatal(err) + } + pub, priv, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatal(err) + } + service := controlapp.GuestReconnect{Store: st.GuestReconnects(), Clock: clock} + if err := service.Enroll(ctx, b, token, "boot.test", base64.RawURLEncoding.EncodeToString(pub)); err != nil { + t.Fatal(err) + } + challenge, err := service.Begin(ctx, b) + if err != nil { + t.Fatal(err) + } + message, err := challenge.SigningMessage() + if err != nil { + t.Fatal(err) + } + signature := base64.RawURLEncoding.EncodeToString(ed25519.Sign(priv, message)) + wrong := b + wrong.SessionID = "another.test" + if _, _, err := service.Accept(ctx, wrong, challenge.AttemptID, signature); err == nil { + t.Fatal("cross-session proof accepted") + } + _, wrongPriv, _ := ed25519.GenerateKey(rand.Reader) + if _, _, err := service.Accept(ctx, b, challenge.AttemptID, base64.RawURLEncoding.EncodeToString(ed25519.Sign(wrongPriv, message))); err == nil { + t.Fatal("wrong key accepted") + } + epoch, fresh, err := service.Accept(ctx, b, challenge.AttemptID, signature) + if err != nil || epoch != 1 || fresh == "" || fresh == token { + t.Fatalf("accept: epoch=%d token-present=%t err=%v", epoch, fresh != "", err) + } + // Reopening the durable adapter cannot make a lost acceptance replayable. + reopened := reopen(t, st.pool.Config().ConnString()) + service.Store = reopened.GuestReconnects() + if epoch, token, err := service.Accept(ctx, b, challenge.AttemptID, signature); err == nil || epoch != 0 || token != "" { + t.Fatal("replay returned authority") + } + if err := st.Bootstraps().ConsumeSessionBootstrap(ctx, b.WorkspaceID, b.SessionID, controlapp.HashSessionBootstrapToken(fresh), 1, clock.Now()); err != nil { + t.Fatal(err) + } +} + +type reconnectClock struct{ now time.Time } + +func (c reconnectClock) Now() time.Time { return c.now } + +// A request can wait behind a lifecycle transaction longer than its challenge +// lives. The store must check database time AFTER that wait, not reuse the +// caller's timestamp or PostgreSQL's transaction-start timestamp. +func TestGuestReconnectExpiresWhileWaitingForAuthorityLock(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + st := freshStore(t, startPostgres(t), t.Name()) + b := control.GuestReconnectScope{WorkspaceID: "ws_self_hosted", PoolID: "pool_self_hosted", SessionID: "lock.test", RunnerID: "runner.test", PlacementGeneration: 1, ConnectionGeneration: 1} + if _, err := st.Sessions().CreateSession(ctx, b.WorkspaceID, control.Session{ID: b.SessionID, CreatorID: "actor.test", PoolID: b.PoolID, RunnerID: b.RunnerID, State: control.StateRunning}); err != nil { + t.Fatal(err) + } + if err := st.Fleet().UpsertRunner(ctx, b.PoolID, control.Runner{ID: b.RunnerID, Generation: 1, Connected: true}); err != nil { + t.Fatal(err) + } + now := time.Now().UTC().Truncate(time.Microsecond) + token := control.SessionBootstrap{Hash: "old-hash.test", PlacementGeneration: 1, ExpiresAt: now.Add(time.Minute)} + if err := st.Bootstraps().PutSessionBootstrap(ctx, b.WorkspaceID, b.SessionID, token); err != nil { + t.Fatal(err) + } + id := control.GuestReconnectIdentity{BootEpoch: "boot.test", PublicKey: "public-key.test"} + r := st.GuestReconnects() + if err := r.EnrollGuest(ctx, b, token.Hash, id, now); err != nil { + t.Fatal(err) + } + now = time.Now().UTC().Truncate(time.Microsecond) + a := control.GuestReconnectAttempt{ID: "attempt.test", Challenge: "nonce.test", ExpiresAt: now.Add(2 * time.Second)} + if _, err := r.BeginGuestReconnect(ctx, b, a, now); err != nil { + t.Fatal(err) + } + tx, err := st.pool.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + if _, err := tx.Exec(ctx, `SELECT 1 FROM sessions WHERE workspace_id=$1 AND id=$2 FOR UPDATE`, string(b.WorkspaceID), string(b.SessionID)); err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { + fresh := token + fresh.Hash = "new-hash.test" + epoch, err := r.ConsumeGuestReconnect(ctx, b, id, a, fresh, now) + if epoch != 0 { + err = fmt.Errorf("unexpected epoch %d", epoch) + } + done <- err + }() + deadline := time.Now().Add(3 * time.Second) + for { + var waiting bool + err := st.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE 'SELECT 1 FROM sessions%')`).Scan(&waiting) + if err != nil { + t.Fatal(err) + } + if waiting { + break + } + if time.Now().After(deadline) { + t.Fatal("consume did not wait on authority lock") + } + time.Sleep(5 * time.Millisecond) + } + if wait := time.Until(a.ExpiresAt.Add(20 * time.Millisecond)); wait > 0 { + time.Sleep(wait) + } + if err := tx.Commit(ctx); err != nil { + t.Fatal(err) + } + if err := <-done; !errors.Is(err, control.ErrReconnectExpired) { + t.Fatalf("expired during lock wait: %v", err) + } + var hash string + var epoch int64 + if err := st.pool.QueryRow(ctx, `SELECT token_hash,guest_connection_epoch FROM session_bootstraps WHERE workspace_id=$1 AND session_id=$2`, string(b.WorkspaceID), string(b.SessionID)).Scan(&hash, &epoch); err != nil { + t.Fatal(err) + } + if hash != token.Hash || epoch != 0 { + t.Fatal("expired request changed capability or epoch") + } +} diff --git a/protocol/runner/reconnect.go b/protocol/runner/reconnect.go new file mode 100644 index 0000000..7801195 --- /dev/null +++ b/protocol/runner/reconnect.go @@ -0,0 +1,90 @@ +package runner + +import ( + "crypto/ed25519" + "encoding/base64" + "encoding/binary" + "errors" +) + +// GuestReconnectProtocol is negotiated independently of the boot configuration. +// Its presence does not enable memory persistence or permit legacy token replay. +const GuestReconnectProtocol uint64 = 1 + +// GuestReconnectChallenge is the control-plane-issued context signed by a +// surviving guest. SessionID and placement are derived from the authenticated +// runner binding, never selected by a proof request. Challenges expire and are +// consumed in the authoritative store; signature verification alone is not an +// authorization decision. HostIncarnation identifies the runner connection's +// control-plane generation, not a guest-supplied host name. +type GuestReconnectChallenge struct { + Protocol uint64 `json:"protocol"` + SessionID string `json:"session_id"` + BootEpoch string `json:"boot_epoch"` + HostIncarnation string `json:"host_incarnation"` + AttemptID string `json:"attempt_id"` + PlacementGeneration uint64 `json:"placement_generation"` + Challenge string `json:"challenge"` +} + +var errGuestReconnect = errors.New("runner: invalid guest reconnect proof") + +// SigningMessage returns the versioned, domain-separated binary transcript. +// Identifiers are 1..256 printable ASCII bytes; nonce encoding is canonical +// unpadded base64url of exactly 32 bytes. Invalid input returns a fixed error +// that never contains peer data. Signatures must not be made over JSON encoding. +func (c GuestReconnectChallenge) SigningMessage() ([]byte, error) { + if c.Protocol != GuestReconnectProtocol || c.PlacementGeneration == 0 { + return nil, errGuestReconnect + } + nonce, ok := reconnectBytes(c.Challenge, 32) + if !ok { + return nil, errGuestReconnect + } + message := []byte("rainier-guest-reconnect-v1\x00") + for _, id := range []string{c.SessionID, c.BootEpoch, c.HostIncarnation, c.AttemptID} { + if len(id) == 0 || len(id) > 256 { + return nil, errGuestReconnect + } + for i := 0; i < len(id); i++ { + if id[i] < 0x21 || id[i] > 0x7e { + return nil, errGuestReconnect + } + } + message = binary.BigEndian.AppendUint32(message, uint32(len(id))) + message = append(message, id...) + } + message = binary.BigEndian.AppendUint64(message, c.PlacementGeneration) + return append(message, nonce...), nil +} + +// VerifyProof verifies a canonical base64url Ed25519 public key and signature +// over this exact challenge. It does NOT check expiry, enrollment, placement or +// single use; the authoritative store must atomically enforce those conditions +// before any configuration or bootstrap capability is returned. +func (c GuestReconnectChallenge) VerifyProof(publicKey, signature string) error { + message, err := c.SigningMessage() + if err != nil { + return err + } + key, ok := reconnectBytes(publicKey, ed25519.PublicKeySize) + if !ok { + return errGuestReconnect + } + sig, ok := reconnectBytes(signature, ed25519.SignatureSize) + if !ok { + return errGuestReconnect + } + if !ed25519.Verify(ed25519.PublicKey(key), message, sig) { + return errGuestReconnect + } + return nil +} + +func reconnectBytes(s string, size int) ([]byte, bool) { + if len(s) != base64.RawURLEncoding.EncodedLen(size) { + return nil, false + } + b, err := base64.RawURLEncoding.Strict().DecodeString(s) + return b, err == nil && len(b) == size && base64.RawURLEncoding.EncodeToString(b) == s +} diff --git a/protocol/runner/reconnect_test.go b/protocol/runner/reconnect_test.go new file mode 100644 index 0000000..4170659 --- /dev/null +++ b/protocol/runner/reconnect_test.go @@ -0,0 +1,100 @@ +package runner + +import ( + "crypto/ed25519" + "crypto/rand" + "encoding/base64" + "encoding/hex" + "strings" + "testing" +) + +func reconnectChallenge() GuestReconnectChallenge { + return GuestReconnectChallenge{Protocol: 1, SessionID: "session.test", BootEpoch: "boot.test", HostIncarnation: "host.test", AttemptID: "attempt.test", PlacementGeneration: 7, Challenge: base64.RawURLEncoding.EncodeToString(make([]byte, 32))} +} + +func TestGuestReconnectTranscript(t *testing.T) { + c := reconnectChallenge() + got, err := c.SigningMessage() + if err != nil { + t.Fatal(err) + } + // Fixed wire vector, independent of the implementation's encoder. + want := "7261696e6965722d67756573742d7265636f6e6e6563742d7631000000000c73657373696f6e2e7465737400000009626f6f742e7465737400000009686f73742e746573740000000c617474656d70742e746573740000000000000007" + strings.Repeat("00", 32) + if hex.EncodeToString(got) != want { + t.Fatalf("wire transcript differs: %x", got) + } +} + +func TestGuestReconnectProofBindsEveryField(t *testing.T) { + public, private, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatal(err) + } + original := reconnectChallenge() + message, err := original.SigningMessage() + if err != nil { + t.Fatal(err) + } + signature := base64.RawURLEncoding.EncodeToString(ed25519.Sign(private, message)) + key := base64.RawURLEncoding.EncodeToString(public) + if err := original.VerifyProof(key, signature); err != nil { + t.Fatal(err) + } + mutations := map[string]func(*GuestReconnectChallenge){ + "session": func(c *GuestReconnectChallenge) { c.SessionID = "another.test" }, + "boot": func(c *GuestReconnectChallenge) { c.BootEpoch = "another.test" }, + "host": func(c *GuestReconnectChallenge) { c.HostIncarnation = "another.test" }, + "attempt": func(c *GuestReconnectChallenge) { c.AttemptID = "another.test" }, + "placement": func(c *GuestReconnectChallenge) { c.PlacementGeneration++ }, + "challenge": func(c *GuestReconnectChallenge) { + c.Challenge = base64.RawURLEncoding.EncodeToString([]byte(strings.Repeat("x", 32))) + }, + "protocol": func(c *GuestReconnectChallenge) { c.Protocol = 2 }, + } + for name, mutate := range mutations { + t.Run(name, func(t *testing.T) { + c := original + mutate(&c) + if c.VerifyProof(key, signature) == nil { + t.Fatal("accepted a proof for another context") + } + }) + } + _, other, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatal(err) + } + if original.VerifyProof(base64.RawURLEncoding.EncodeToString(other.Public().(ed25519.PublicKey)), signature) == nil { + t.Fatal("accepted another guest key") + } +} + +func TestGuestReconnectRejectsMalformedInputs(t *testing.T) { + for name, mutate := range map[string]func(*GuestReconnectChallenge){ + "empty session": func(c *GuestReconnectChallenge) { c.SessionID = "" }, + "oversized id": func(c *GuestReconnectChallenge) { c.AttemptID = strings.Repeat("x", 257) }, + "control character": func(c *GuestReconnectChallenge) { c.BootEpoch = "boot\nsecret" }, + "zero placement": func(c *GuestReconnectChallenge) { c.PlacementGeneration = 0 }, + "short challenge": func(c *GuestReconnectChallenge) { c.Challenge = "AAAA" }, + "noncanonical challenge": func(c *GuestReconnectChallenge) { c.Challenge += "=" }, + } { + t.Run(name, func(t *testing.T) { + c := reconnectChallenge() + mutate(&c) + if _, err := c.SigningMessage(); err == nil { + t.Fatal("accepted malformed context") + } + }) + } + c := reconnectChallenge() + for _, key := range []string{"", "secret.test", strings.Repeat("A", 43)} { + for _, sig := range []string{"", "secret.test", strings.Repeat("A", 86)} { + if err := c.VerifyProof(key, sig); err == nil { + t.Fatal("accepted invalid proof") + } else if strings.Contains(err.Error(), "secret.test") { + t.Fatal("error discloses input") + } + } + } +} diff --git a/runnerplane/conn.go b/runnerplane/conn.go index 5047ea9..97046fb 100644 --- a/runnerplane/conn.go +++ b/runnerplane/conn.go @@ -377,7 +377,9 @@ func (p *Plane) routeSessionReq(ctx context.Context, rc *runnerConn, m runner.Fr // the sandbox is holding a pending entry (and, for a credential mint, a git // process) until one arrives. func (p *Plane) answerSessionRequest(ctx context.Context, rc *runnerConn, sessionID string, env runner.RPCEnvelope) { - ans := p.host.SessionRequest(ctx, rc.binding, control.SessionID(sessionID), env) + binding := rc.binding + binding.ConnectionGeneration = rc.gen + ans := p.host.SessionRequest(ctx, binding, control.SessionID(sessionID), env) // The id and the method are this layer's to set, never the host's: the id // is what the sandbox correlates against, and every answer is a "resp". ans.ID = env.ID diff --git a/runnerplane/host.go b/runnerplane/host.go index 614f546..2a8c094 100644 --- a/runnerplane/host.go +++ b/runnerplane/host.go @@ -15,6 +15,11 @@ type Binding struct { WorkspaceID control.WorkspaceID PoolID control.PoolID RunnerID control.RunnerID + // ConnectionGeneration is assigned by the plane for SessionRequest from + // this connection's registration. Hosts must compare it with durable fleet + // state when authorizing generation-sensitive operations. It is never + // taken from an announce or request body; zero means no such binding. + ConnectionGeneration uint64 } // Host is what a plane needs from its host. Every method is one dependency. diff --git a/runnerplane/reconnect_binding_test.go b/runnerplane/reconnect_binding_test.go new file mode 100644 index 0000000..797e308 --- /dev/null +++ b/runnerplane/reconnect_binding_test.go @@ -0,0 +1,55 @@ +package runnerplane + +import ( + "context" + "encoding/json" + "testing" + + "github.com/tokencanopy/rainier/control" + "github.com/tokencanopy/rainier/protocol/runner" +) + +func TestSessionRequestBindsOriginalConnectionGeneration(t *testing.T) { + h := newFakeHost() + var got Binding + h.answer = func(b Binding, _ control.SessionID, _ runner.RPCEnvelope) runner.RPCEnvelope { + got = b + return runner.RPCEnvelope{OK: true} + } + p := New(h, Options{}) + rc := newRunnerConn(Binding{WorkspaceID: testWorkspace, PoolID: testPool, RunnerID: "runner.test"}, nil) + rc.gen = 7 + // A newer store generation must not make this old connection authoritative. + h.repo.runners["runner.test"] = &control.Runner{Generation: 8} + p.answerSessionRequest(context.Background(), rc, "session.test", runner.RPCEnvelope{ID: 1, Method: "synthetic"}) + if got.ConnectionGeneration != 7 { + t.Fatalf("got generation %d, want original connection generation 7", got.ConnectionGeneration) + } + if got.WorkspaceID != testWorkspace || got.PoolID != testPool || got.RunnerID != "runner.test" { + t.Fatal("scope changed") + } +} + +// This drives the real WebSocket registration and session-request path. The +// payload's invented generation cannot replace the plane's connection binding. +func TestSessionRequestConnectionGenerationOverWebSocket(t *testing.T) { + h := newFakeHost() + got := make(chan Binding, 1) + h.answer = func(b Binding, _ control.SessionID, _ runner.RPCEnvelope) runner.RPCEnvelope { + got <- b + return runner.RPCEnvelope{OK: true} + } + p, ts := newTestPlaneOver(t, h) + f := startFakeRunner(t, ts, runnerScript{Name: "runner.test", Total: 1}) + waitConnected(t, p, "runner.test") + drainAccept(t, f) + f.write(t, runner.FromRunner{Type: "session_req", Session: "session.test", RPC: &runner.RPCEnvelope{ID: 44, Method: "synthetic", Payload: json.RawMessage(`{"connection_generation":999}`)}}) + response := nextSessionRPC(t, f) + if !response.RPC.OK || response.RPC.ID != 44 { + t.Fatalf("response: %+v", response) + } + b := <-got + if b.ConnectionGeneration != 1 || b.RunnerID != "runner.test" || b.WorkspaceID != testWorkspace || b.PoolID != testPool { + t.Fatalf("binding: %+v", b) + } +}