From 7e65b226e7a74ecd13bf8b4738ea6bdd096cb1ba Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 22 Sep 2026 17:01:42 -0700 Subject: [PATCH] feat(auth): share provider sign-in flows and credential bindings (#12983) Co-authored-by: Claude Opus 5.5 --- apps/server/src/auth/RpcAuthorization.ts | 1 + .../Layers/ProviderAdapterRegistry.test.ts | 98 ++++ .../Layers/ProviderAdapterRegistry.ts | 76 ++- .../Layers/ProviderAuthService.test.ts | 319 ++++++++++- .../provider/Layers/ProviderAuthService.ts | 163 +++++- .../src/provider/ProviderAuthFlow.test.ts | 520 ++++++++++++++++++ apps/server/src/provider/ProviderAuthFlow.ts | 463 ++++++++++++++++ .../provider/ProviderCredentialStore.test.ts | 49 ++ .../src/provider/ProviderCredentialStore.ts | 21 + .../provider/Services/ProviderAuthService.ts | 28 +- apps/server/src/ws.ts | 9 + packages/client-runtime/src/state/server.ts | 4 + packages/contracts/src/providerSetup.test.ts | 53 ++ packages/contracts/src/providerSetup.ts | 96 +++- packages/contracts/src/rpc.ts | 12 +- 15 files changed, 1870 insertions(+), 42 deletions(-) create mode 100644 apps/server/src/provider/ProviderAuthFlow.test.ts create mode 100644 apps/server/src/provider/ProviderAuthFlow.ts create mode 100644 apps/server/src/provider/ProviderCredentialStore.test.ts create mode 100644 apps/server/src/provider/ProviderCredentialStore.ts create mode 100644 packages/contracts/src/providerSetup.test.ts diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 68a16708376240ce9da001fe28c4d14ab1270493..8ab1520a6f348ffc6a30c39ea33607c3f5e2aa22 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -37,6 +37,7 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.providerAuthStart]: AuthOrchestrationOperateScope, [WS_METHODS.providerConsumeResetCredit]: AuthOrchestrationOperateScope, [WS_METHODS.providerAuthComplete]: AuthOrchestrationOperateScope, + [WS_METHODS.providerAuthRespond]: AuthOrchestrationOperateScope, [WS_METHODS.providerAuthCancel]: AuthOrchestrationOperateScope, [WS_METHODS.providerAuthLogout]: AuthOrchestrationOperateScope, [WS_METHODS.providerAuthSubscribe]: AuthOrchestrationOperateScope, diff --git a/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts b/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts index bb8ff7835c8fd76b4a0561ca277c9f2c51c889d8..6a24b49dc9307f949620ea93e194e45afc210a55 100644 --- a/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts @@ -1,15 +1,21 @@ import { defaultInstanceIdForDriver, ProviderDriverKind, + ThreadId, type ServerProvider, } from "@t3tools/contracts"; import { it, assert, vi } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; +import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Stream from "effect/Stream"; +import * as ProviderAuthFlow from "../ProviderAuthFlow.ts"; + import type * as ClaudeAdapter from "../Services/ClaudeAdapter.ts"; import type * as CodexAdapter from "../Services/CodexAdapter.ts"; import type * as CursorAdapter from "../Services/CursorAdapter.ts"; @@ -185,3 +191,95 @@ it.layer(layer)("ProviderAdapterRegistryLive", (it) => { ]); })); }); + +it.effect("blocks shared credential session startup and preserves guarded adapter identity", () => + Effect.gen(function* () { + const target = fakeInstances[0]!; + const peer = fakeInstances[1]!; + const auth = yield* ProviderAuthFlow.make({ + instanceId: target.instanceId, + credentialBinding: { owner: "t3", key: "shared-auth" }, + methods: Effect.succeed([ + { id: "browser", name: "Browser", description: null, type: "agent" }, + ]), + authenticate: () => Effect.never, + logout: Effect.void, + }); + const peerAuth = yield* ProviderAuthFlow.make({ + instanceId: peer.instanceId, + credentialBinding: { owner: "t3", key: "shared-auth" }, + methods: Effect.succeed([]), + authenticate: () => Effect.void, + logout: Effect.void, + }); + const session = { + threadId: ThreadId.make("new-session"), + provider: peer.driverKind, + providerInstanceId: peer.instanceId, + status: "ready" as const, + runtimeMode: "approval-required" as const, + createdAt: "2026-09-21T00:00:00.000Z", + updatedAt: "2026-09-21T00:00:00.000Z", + }; + const start = vi.fn(() => Effect.succeed(session)); + const instances = [ + { ...target, auth }, + { ...peer, auth: peerAuth, adapter: { ...peer.adapter, startSession: start } }, + ]; + const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistry.pipe( + Effect.provide( + ProviderAdapterRegistryLayer.ProviderAdapterRegistryLive.pipe( + Layer.provide( + Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ + getInstance: (id) => + Effect.succeed(instances.find((instance) => instance.instanceId === id)), + listInstances: Effect.succeed(instances), + }), + ), + ), + ), + ); + const guarded = yield* registry.getByInstance(peer.instanceId); + assert.strictEqual(yield* registry.getByInstance(peer.instanceId), guarded); + const flow = yield* auth.start("owner"); + const error = yield* guarded + .startSession({ + threadId: session.threadId, + providerInstanceId: peer.instanceId, + runtimeMode: "approval-required", + }) + .pipe(Effect.flip); + assert.strictEqual(error._tag, "ProviderAdapterValidationError"); + assert.strictEqual(start.mock.calls.length, 0); + yield* auth.cancel("owner", flow.flowId!); + assert.deepStrictEqual( + yield* guarded.startSession({ + threadId: session.threadId, + providerInstanceId: peer.instanceId, + runtimeMode: "approval-required", + }), + session, + ); + assert.strictEqual(start.mock.calls.length, 1); + const entered = yield* Deferred.make(); + const stopped = yield* Deferred.make(); + start.mockImplementation(() => + Effect.gen(function* () { + yield* Deferred.succeed(entered, undefined); + return yield* Effect.never; + }).pipe(Effect.ensuring(Deferred.succeed(stopped, undefined))), + ); + const startup = yield* guarded + .startSession({ + threadId: session.threadId, + providerInstanceId: peer.instanceId, + runtimeMode: "approval-required", + }) + .pipe(Effect.forkChild); + yield* Deferred.await(entered); + // Signing out through another instance must drain its peer's startup too. + yield* auth.logout(Effect.void); + yield* Deferred.await(stopped); + assert.strictEqual(Exit.isFailure(yield* Fiber.await(startup)), true); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/provider/Layers/ProviderAdapterRegistry.ts b/apps/server/src/provider/Layers/ProviderAdapterRegistry.ts index 9e8e3c5d1f9025fddabe562e37886f4909ff1b64..5d70ee0d3d75114619ed4907bd161866ee02833b 100644 --- a/apps/server/src/provider/Layers/ProviderAdapterRegistry.ts +++ b/apps/server/src/provider/Layers/ProviderAdapterRegistry.ts @@ -15,19 +15,89 @@ * * @module ProviderAdapterRegistryLive */ -import { ProviderInstanceId } from "@t3tools/contracts"; +import { ProviderInstanceId, ProviderSetupError, type ProviderSession } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; import * as Layer from "effect/Layer"; +import type * as Scope from "effect/Scope"; -import { ProviderUnsupportedError } from "../Errors.ts"; +import { + ProviderUnsupportedError, + ProviderAdapterValidationError, + type ProviderAdapterError, +} from "../Errors.ts"; import { ProviderInstanceRegistry } from "../Services/ProviderInstanceRegistry.ts"; import { ProviderAdapterRegistry, type ProviderAdapterRegistryShape, } from "../Services/ProviderAdapterRegistry.ts"; +import type { ProviderInstance } from "../ProviderDriver.ts"; +import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts"; + +const isSetupError = Schema.is(ProviderSetupError); + const makeProviderAdapterRegistry = Effect.fn("makeProviderAdapterRegistry")(function* () { const registry = yield* ProviderInstanceRegistry; + // Stable identity keeps ProviderService's event subscriptions attached once. + const guarded = new WeakMap>(); + const guard = (instance: ProviderInstance) => { + const auth = instance.auth; + if (!auth || (!auth.withAccess && !auth.isChangingCredentials && !auth.credentialBinding)) + return instance.adapter; + const cached = guarded.get(instance); + if (cached) return cached; + const adapter: ProviderAdapterShape = { + ...instance.adapter, + startSession: (input) => { + const start = Effect.gen(function* () { + const binding = auth.credentialBinding; + const related = binding + ? (yield* registry.listInstances).filter( + (peer) => + peer.auth?.credentialBinding?.key === binding.key && + peer.auth.credentialBinding.owner === binding.owner, + ) + : [instance]; + for (const peer of related) { + if (peer.auth?.isChangingCredentials && (yield* peer.auth.isChangingCredentials)) { + return yield* new ProviderSetupError({ + instanceId: instance.instanceId, + operation: "session", + detail: "Provider sign-in is changing. Try again after it finishes.", + }); + } + } + let admitted: Effect.Effect< + ProviderSession, + ProviderAdapterError | ProviderSetupError, + Scope.Scope + > = instance.adapter.startSession(input); + // Every shared owner holds the startup scope. A credential change + // interrupts admitted startup before it can escape the session drain. + for (const peer of related) { + if (peer.auth?.withAccess) admitted = peer.auth.withAccess(admitted); + } + return yield* Effect.scoped(admitted); + }); + // Adapters own established session lifetimes. This scope guards startup; + // ProviderAuthService drains routed sessions before changing credentials. + return start.pipe( + Effect.mapError((cause) => + isSetupError(cause) + ? new ProviderAdapterValidationError({ + provider: instance.driverKind, + operation: "startSession", + issue: cause.detail, + }) + : cause, + ), + ); + }, + }; + guarded.set(instance, adapter); + return adapter; + }; const getByInstance: ProviderAdapterRegistryShape["getByInstance"] = (instanceId) => registry.getInstance(instanceId).pipe( @@ -38,7 +108,7 @@ const makeProviderAdapterRegistry = Effect.fn("makeProviderAdapterRegistry")(fun provider: instanceId, }), ) - : Effect.succeed(instance.adapter), + : Effect.succeed(guard(instance)), ), ); diff --git a/apps/server/src/provider/Layers/ProviderAuthService.test.ts b/apps/server/src/provider/Layers/ProviderAuthService.test.ts index fb961b710486fdf4e49c0e9291a25fa2428e2341..8430ad3e1319d3b8d2c1b935ff04274172e461bb 100644 --- a/apps/server/src/provider/Layers/ProviderAuthService.test.ts +++ b/apps/server/src/provider/Layers/ProviderAuthService.test.ts @@ -112,6 +112,14 @@ const makeHarness = Effect.fn("ProviderAuthService.test.makeHarness")(function* directoryError?: ProviderSessionDirectoryPersistenceError; stopError?: ProviderServiceError; logoutError?: ProviderSetupError; + sharedCredentials?: boolean; + sharedBusy?: boolean; + sharedBusyEffect?: Effect.Effect; + responds?: boolean; + beforeLogout?: Effect.Effect; + beforeStop?: Effect.Effect; + beforeListSessions?: Effect.Effect; + onLookup?: Effect.Effect; } = {}, ) { const actions: string[] = []; @@ -138,6 +146,18 @@ const makeHarness = Effect.fn("ProviderAuthService.test.makeHarness")(function* }); const auth: ProviderAuthController = { + ...(input.sharedCredentials + ? { credentialBinding: { owner: "provider" as const, key: "shared" } } + : {}), + ...(input.responds + ? { + respond: Effect.fn(function* (ownerSessionId, request) { + yield* checkOwner(ownerSessionId, request.flowId, "respond"); + actions.push(`respond:${request.response.type}`); + return state; + }), + } + : {}), start: Effect.fn(function* (ownerSessionId, stopSessions) { gateClosed = true; actions.push("close-gate"); @@ -169,6 +189,7 @@ const makeHarness = Effect.fn("ProviderAuthService.test.makeHarness")(function* actions.push("close-gate"); yield* stopSessions; if (input.logoutError) return yield* input.logoutError; + yield* input.beforeLogout ?? Effect.void; actions.push("native-logout"); state = idle; flowOwner = undefined; @@ -181,13 +202,34 @@ const makeHarness = Effect.fn("ProviderAuthService.test.makeHarness")(function* const instances = [ makeInstance({ instanceId, enabled: input.enabled ?? true, auth }), makeInstance({ instanceId: unsupportedInstanceId, enabled: true }), + ...(input.sharedCredentials + ? [ + makeInstance({ + instanceId: otherInstanceId, + enabled: true, + auth: { + ...auth, + isChangingCredentials: + input.sharedBusyEffect ?? Effect.succeed(input.sharedBusy ?? false), + invalidate: Effect.sync(() => { + actions.push("invalidate-shared"); + }), + }, + }), + ] + : []), ]; const service = yield* makeProviderAuthService.pipe( Effect.provide( Layer.mergeAll( Layer.mock(ProviderInstanceRegistry)({ getInstance: (id) => - Effect.succeed(instances.find((instance) => instance.instanceId === id)), + Effect.gen(function* () { + const found = instances.find((instance) => instance.instanceId === id); + yield* input.onLookup ?? Effect.void; + return found; + }), + listInstances: Effect.succeed(instances), subscribeChanges: PubSub.subscribe(registryChanges), }), Layer.mock(ProviderSessionDirectory)({ @@ -202,26 +244,44 @@ const makeHarness = Effect.fn("ProviderAuthService.test.makeHarness")(function* }), Layer.mock(ProviderService)({ listSessions: () => - Effect.sync(() => { + Effect.gen(function* () { assert.isTrue(gateClosed); actions.push("list-sessions"); + yield* input.beforeListSessions ?? Effect.void; return [...sessions.values()]; }), stopSession: ({ threadId }) => - Effect.suspend(() => { + Effect.gen(function* () { assert.isTrue(gateClosed); actions.push(`stop:${threadId}`); - if (input.stopError) return Effect.fail(input.stopError); + yield* input.beforeStop ?? Effect.void; + if (input.stopError) return yield* input.stopError; sessions.delete(threadId); const binding = bindings.get(threadId); if (binding) bindings.set(threadId, { ...binding, status: "stopped" }); - return Effect.void; }), }), ), ), ); - return { service, actions, sessions, bindings }; + return { + service, + actions, + sessions, + bindings, + auth, + addInstance: (instance: ProviderInstance) => instances.push(instance), + replaceInstance: (replacement: ProviderInstance) => { + const index = instances.findIndex( + (instance) => instance.instanceId === replacement.instanceId, + ); + assert.isAtLeast(index, 0); + instances[index] = replacement; + }, + replaceAuth: (next: ProviderAuthController) => { + instances[0] = makeInstance({ instanceId, enabled: input.enabled ?? true, auth: next }); + }, + }; }); const makeStreamingController = Effect.fn("ProviderAuthService.test.makeStreamingController")( @@ -305,6 +365,56 @@ const observeAuth = Effect.fn("ProviderAuthService.test.observeAuth")(function* }); describe("ProviderAuthService", () => { + it.effect( + "stops sessions sharing credentials and invalidates their processes before logout", + () => + Effect.gen(function* () { + const { service, actions, sessions } = yield* makeHarness({ + sharedCredentials: true, + sessions: [ + makeSession("target"), + makeSession("shared", otherInstanceId), + makeSession("unrelated", unsupportedInstanceId), + ], + }); + yield* service.logout({ instanceId }); + assert.deepStrictEqual([...sessions.keys()], [ThreadId.make("unrelated")]); + assert.isBelow(actions.indexOf("stop:shared"), actions.indexOf("invalidate-shared")); + assert.isBelow(actions.indexOf("invalidate-shared"), actions.indexOf("native-logout")); + }), + ); + it.effect("rejects overlapping changes to a shared sign-in", () => + Effect.gen(function* () { + const { service, actions } = yield* makeHarness({ + sharedCredentials: true, + sharedBusy: true, + }); + for (const task of [service.start({ instanceId }, owner), service.logout({ instanceId })]) { + const error = yield* task.pipe(Effect.flip); + assert.include(error.detail, "shared sign-in"); + } + assert.deepStrictEqual(actions, []); + }), + ); + it.effect("routes typed interactions to the flow owner and rejects unsupported controllers", () => + Effect.gen(function* () { + const { service, actions } = yield* makeHarness({ responds: true }); + yield* service.start({ instanceId }, owner); + const request = { + instanceId, + flowId, + interactionId: "consent", + response: { type: "browser" as const, action: "accept" as const }, + }; + const rejected = yield* service.respond(request, otherOwner).pipe(Effect.flip); + assert.strictEqual(rejected.operation, "respond"); + yield* service.respond(request, owner); + assert.strictEqual(actions.at(-1), "respond:browser"); + const unsupported = yield* makeHarness(); + const error = yield* unsupported.service.respond(request, owner).pipe(Effect.flip); + assert.include(error.detail, "does not accept"); + }), + ); it.effect("stops routed sessions before sign-in, including for a disabled instance", () => Effect.gen(function* () { const { service, actions, sessions } = yield* makeHarness({ @@ -606,3 +716,200 @@ describe("ProviderAuthService", () => { }), ); }); + +it.effect("queued logout prompts resolve the current controller after provider replacement", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + const lookup = yield* Deferred.make(); + let observeLookup = false; + const harness = yield* makeHarness({ + beforeLogout: Deferred.succeed(entered, undefined).pipe( + Effect.andThen(Deferred.await(release)), + ), + onLookup: Effect.suspend(() => + observeLookup ? Deferred.succeed(lookup, undefined).pipe(Effect.asVoid) : Effect.void, + ), + }); + const first = yield* harness.service.logout({ instanceId }).pipe(Effect.forkChild); + yield* Deferred.await(entered); + observeLookup = true; + const queued = yield* harness.service + .tryHandlePromptCommand({ instanceId, text: "/logout", hasAttachments: false }) + .pipe(Effect.forkChild); + yield* Deferred.await(lookup); + let replacementLoggedOut = false; + harness.replaceAuth({ + ...harness.auth, + logout: (stopSessions) => + stopSessions.pipe( + Effect.andThen( + Effect.sync(() => { + replacementLoggedOut = true; + return idleAuthState; + }), + ), + ), + }); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(first); + assert.isTrue(yield* Fiber.join(queued)); + assert.isTrue(replacementLoggedOut); + assert.strictEqual(harness.actions.filter((action) => action === "native-logout").length, 1); + }).pipe(Effect.scoped), +); + +it.effect.each(["start", "logout", "prompt"] as const)( + "%s uses the same credential controller for shared exclusion and mutation during replacement", + (action) => + Effect.gen(function* () { + const checked = yield* Deferred.make(); + const continueCheck = yield* Deferred.make(); + const replacementPeerId = ProviderInstanceId.make("replacement-shared-peer"); + const harness = yield* makeHarness({ + sharedCredentials: true, + sharedBusyEffect: Deferred.succeed(checked, undefined).pipe( + Effect.andThen(Deferred.await(continueCheck)), + Effect.as(false), + ), + sessions: [ + makeSession("old-shared", otherInstanceId), + makeSession("replacement-shared", replacementPeerId), + ], + }); + harness.addInstance( + makeInstance({ + instanceId: replacementPeerId, + enabled: true, + auth: { + ...harness.auth, + credentialBinding: { owner: "provider", key: "replacement-binding" }, + isChangingCredentials: Effect.succeed(true), + invalidate: Effect.die( + "The unrelated replacement credential binding must stay intact.", + ), + }, + }), + ); + const operation = + action === "start" + ? harness.service.start({ instanceId }, owner) + : action === "logout" + ? harness.service.logout({ instanceId }) + : harness.service.tryHandlePromptCommand({ + instanceId, + text: "/logout", + hasAttachments: false, + }); + const running = yield* operation.pipe(Effect.forkChild); + yield* Deferred.await(checked); + let replacementMutations = 0; + harness.replaceAuth({ + ...harness.auth, + credentialBinding: { owner: "provider", key: "replacement-binding" }, + start: () => + Effect.sync(() => { + replacementMutations++; + return waitingAuthState; + }), + logout: () => + Effect.sync(() => { + replacementMutations++; + return idleAuthState; + }), + }); + yield* Deferred.succeed(continueCheck, undefined); + yield* Fiber.join(running); + assert.equal(replacementMutations, 0); + assert.include(harness.actions, action === "start" ? "start-sign-in" : "native-logout"); + assert.isFalse(harness.sessions.has(ThreadId.make("old-shared"))); + assert.isTrue(harness.sessions.has(ThreadId.make("replacement-shared"))); + const blocked = yield* Effect.flip(harness.service.logout({ instanceId })); + assert.include(blocked.detail, "shared sign-in"); + assert.equal(replacementMutations, 0); + }).pipe(Effect.scoped), +); + +it.effect.each([ + { owner: "provider" as const, key: "different-binding" }, + { owner: "t3" as const, key: "shared" }, +])( + "does not invalidate a peer that switches credential binding during session draining %#", + (binding) => + Effect.gen(function* () { + const draining = yield* Deferred.make(); + const continueDrain = yield* Deferred.make(); + const harness = yield* makeHarness({ + sharedCredentials: true, + sessions: [makeSession("shared-draining", otherInstanceId)], + beforeStop: Deferred.succeed(draining, undefined).pipe( + Effect.andThen(Deferred.await(continueDrain)), + ), + }); + const logout = yield* harness.service.logout({ instanceId }).pipe(Effect.forkChild); + yield* Deferred.await(draining); + let replacementInvalidated = false; + harness.replaceInstance( + makeInstance({ + instanceId: otherInstanceId, + enabled: true, + auth: { + ...harness.auth, + credentialBinding: binding, + invalidate: Effect.sync(() => { + replacementInvalidated = true; + }), + }, + }), + ); + yield* Deferred.succeed(continueDrain, undefined); + yield* Fiber.join(logout); + assert.isFalse(replacementInvalidated); + assert.isFalse(harness.sessions.has(ThreadId.make("shared-draining"))); + assert.include(harness.actions, "native-logout"); + }).pipe(Effect.scoped), +); + +it.effect.each(["selection", "drain"] as const)( + "preserves replacement peer sessions when credentials change during session %s", + (phase) => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const proceed = yield* Deferred.make(); + const block = Deferred.succeed(entered, undefined).pipe( + Effect.andThen(Deferred.await(proceed)), + ); + const harness = yield* makeHarness({ + sharedCredentials: true, + sessions: [ + makeSession("target-draining"), + makeSession("peer-replacement", otherInstanceId), + ], + bindings: [ + makeBinding("target-draining", "running"), + makeBinding("peer-persisted", "running", otherInstanceId), + ], + ...(phase === "selection" ? { beforeListSessions: block } : { beforeStop: block }), + }); + const logout = yield* harness.service.logout({ instanceId }).pipe(Effect.forkChild); + yield* Deferred.await(entered); + harness.replaceInstance( + makeInstance({ + instanceId: otherInstanceId, + enabled: true, + auth: { + ...harness.auth, + credentialBinding: { owner: "provider", key: "replacement-credentials" }, + invalidate: Effect.die("The replacement peer's credentials must not be invalidated."), + }, + }), + ); + yield* Deferred.succeed(proceed, undefined); + yield* Fiber.join(logout); + assert.isTrue(harness.sessions.has(ThreadId.make("peer-replacement"))); + assert.notInclude(harness.actions, "stop:peer-replacement"); + assert.notInclude(harness.actions, "stop:peer-persisted"); + assert.isFalse(harness.sessions.has(ThreadId.make("target-draining"))); + assert.include(harness.actions, "native-logout"); + }).pipe(Effect.scoped), +); diff --git a/apps/server/src/provider/Layers/ProviderAuthService.ts b/apps/server/src/provider/Layers/ProviderAuthService.ts index 3b69c5082528e0eab184641334831334f34e3412..cec2a5878880fe02f3be01875d2d1bd34280ffdd 100644 --- a/apps/server/src/provider/Layers/ProviderAuthService.ts +++ b/apps/server/src/provider/Layers/ProviderAuthService.ts @@ -2,8 +2,9 @@ import { ProviderSetupError, type ProviderInstanceId } from "@t3tools/contracts" import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; +import * as Semaphore from "effect/Semaphore"; -import { ProviderAuthService } from "../Services/ProviderAuthService.ts"; +import * as ProviderAuthService from "../Services/ProviderAuthService.ts"; import { ProviderInstanceRegistry } from "../Services/ProviderInstanceRegistry.ts"; import { ProviderService } from "../Services/ProviderService.ts"; import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; @@ -12,6 +13,7 @@ export const makeProviderAuthService = Effect.gen(function* () { const registry = yield* ProviderInstanceRegistry; const providers = yield* ProviderService; const directory = yield* ProviderSessionDirectory; + const credentialChanges = yield* Semaphore.make(1); const getController = Effect.fn("ProviderAuthService.getController")(function* ( instanceId: ProviderInstanceId, @@ -30,9 +32,25 @@ export const makeProviderAuthService = Effect.gen(function* () { return instance.auth; }); + // Native sessions may still belong to the previous provider after the + // selected model changes. Read session bindings, not the selected model, + // when invalidating credentials for sign-in or sign-out. const stopSessions = Effect.fn("ProviderAuthService.stopSessions")(function* ( instanceId: ProviderInstanceId, + binding: ProviderAuthService.ProviderAuthController["credentialBinding"], ) { + const affectedIds = new Set([ + instanceId, + ...(binding === undefined + ? [] + : (yield* registry.listInstances) + .filter( + (instance) => + instance.auth?.credentialBinding?.key === binding.key && + instance.auth.credentialBinding.owner === binding.owner, + ) + .map((instance) => instance.instanceId)), + ]); const bindings = yield* directory.listBindings().pipe( Effect.mapError( () => @@ -44,39 +62,111 @@ export const makeProviderAuthService = Effect.gen(function* () { ), ); const sessions = yield* providers.listSessions(); - const threadIds = new Set( - bindings - .filter( - (binding) => binding.providerInstanceId === instanceId && binding.status !== "stopped", - ) - .map((binding) => binding.threadId), + const sessionsToStop = new Map( + bindings.flatMap((session) => + session.providerInstanceId !== undefined && + affectedIds.has(session.providerInstanceId) && + session.status !== "stopped" + ? [[session.threadId, session.providerInstanceId] as const] + : [], + ), ); for (const session of sessions) { - if (session.providerInstanceId === instanceId) { - threadIds.add(session.threadId); + if (session.providerInstanceId !== undefined && affectedIds.has(session.providerInstanceId)) { + sessionsToStop.set(session.threadId, session.providerInstanceId); } } yield* Effect.forEach( - threadIds, - (threadId) => - providers.stopSession({ threadId }).pipe( - Effect.mapError( - () => - new ProviderSetupError({ - instanceId, - operation: "stopSessions", - detail: "Could not stop all sessions for this provider. Try again.", - }), - ), - ), + sessionsToStop, + ([threadId, sessionInstanceId]) => + Effect.gen(function* () { + if (sessionInstanceId !== instanceId) { + const current = yield* registry.getInstance(sessionInstanceId); + if ( + !binding || + current?.auth?.credentialBinding?.key !== binding.key || + current.auth.credentialBinding.owner !== binding.owner + ) + return; + } + yield* providers.stopSession({ threadId }).pipe( + Effect.mapError( + () => + new ProviderSetupError({ + instanceId, + operation: "stopSessions", + detail: "Could not stop all sessions for this provider. Try again.", + }), + ), + ); + }), { discard: true }, ); + if (binding) { + yield* Effect.forEach( + (yield* registry.listInstances).filter( + (instance) => + instance.instanceId !== instanceId && + affectedIds.has(instance.instanceId) && + instance.auth?.credentialBinding?.key === binding.key && + instance.auth.credentialBinding.owner === binding.owner, + ), + (instance) => instance.auth?.invalidate ?? Effect.void, + { discard: true }, + ); + } }); - return ProviderAuthService.of({ + const checkSharedBinding = Effect.fnUntraced(function* ( + instanceId: ProviderInstanceId, + operation: "start" | "logout", + auth: ProviderAuthService.ProviderAuthController, + ) { + const binding = auth.credentialBinding; + if (!binding) return; + const instances = yield* registry.listInstances; + for (const instance of instances) { + if ( + instance.instanceId !== instanceId && + instance.auth?.credentialBinding?.key === binding.key && + instance.auth.credentialBinding.owner === binding.owner && + instance.auth.isChangingCredentials && + (yield* instance.auth.isChangingCredentials) + ) { + return yield* new ProviderSetupError({ + instanceId, + operation, + detail: + "Another provider instance is changing this shared sign-in. Finish or cancel it first.", + }); + } + } + }); + + return ProviderAuthService.ProviderAuthService.of({ start: Effect.fn("ProviderAuthService.start")(function* (input, ownerSessionId) { - const auth = yield* getController(input.instanceId, "start"); - return yield* auth.start(ownerSessionId, stopSessions(input.instanceId)); + return yield* credentialChanges.withPermit( + Effect.gen(function* () { + const auth = yield* getController(input.instanceId, "start"); + yield* checkSharedBinding(input.instanceId, "start", auth); + return yield* auth.start( + ownerSessionId, + stopSessions(input.instanceId, auth.credentialBinding), + input.methodId, + ); + }), + ); + }), + respond: Effect.fn("ProviderAuthService.respond")(function* (input, ownerSessionId) { + const auth = yield* getController(input.instanceId, "respond"); + if (!auth.respond) { + return yield* new ProviderSetupError({ + instanceId: input.instanceId, + operation: "respond", + detail: "This provider does not accept this sign-in interaction.", + }); + } + return yield* auth.respond(ownerSessionId, input); }), complete: Effect.fn("ProviderAuthService.complete")(function* (input, ownerSessionId) { const auth = yield* getController(input.instanceId, "complete"); @@ -87,8 +177,13 @@ export const makeProviderAuthService = Effect.gen(function* () { return yield* auth.cancel(ownerSessionId, input.flowId); }), logout: Effect.fn("ProviderAuthService.logout")(function* (input) { - const auth = yield* getController(input.instanceId, "logout"); - return yield* auth.logout(stopSessions(input.instanceId)); + return yield* credentialChanges.withPermit( + Effect.gen(function* () { + const auth = yield* getController(input.instanceId, "logout"); + yield* checkSharedBinding(input.instanceId, "logout", auth); + return yield* auth.logout(stopSessions(input.instanceId, auth.credentialBinding)); + }), + ); }), subscribe: (input, ownerSessionId) => Effect.gen(function* () { @@ -110,11 +205,21 @@ export const makeProviderAuthService = Effect.gen(function* () { if (!instance?.auth?.isLogoutPrompt?.(input.text, input.hasAttachments)) { return false; } - yield* instance.auth.logout(stopSessions(input.instanceId)); - return true; + return yield* credentialChanges.withPermit( + Effect.gen(function* () { + const auth = yield* getController(input.instanceId, "logout"); + if (!auth.isLogoutPrompt?.(input.text, input.hasAttachments)) return false; + yield* checkSharedBinding(input.instanceId, "logout", auth); + yield* auth.logout(stopSessions(input.instanceId, auth.credentialBinding)); + return true; + }), + ); }, ), }); }); -export const ProviderAuthServiceLive = Layer.effect(ProviderAuthService, makeProviderAuthService); +export const ProviderAuthServiceLive = Layer.effect( + ProviderAuthService.ProviderAuthService, + makeProviderAuthService, +); diff --git a/apps/server/src/provider/ProviderAuthFlow.test.ts b/apps/server/src/provider/ProviderAuthFlow.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..d12b6967c82e7f8ca1da5b3a679675d3bd207af2 --- /dev/null +++ b/apps/server/src/provider/ProviderAuthFlow.test.ts @@ -0,0 +1,520 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + ProviderInstanceId, + ProviderSetupError, + type ProviderAuthInteraction, + type ProviderAuthResponse, + type ProviderAuthState, +} from "@t3tools/contracts"; +import * as Fiber from "effect/Fiber"; +import * as Exit from "effect/Exit"; +import * as Scope from "effect/Scope"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; + +import * as ProviderAuthFlow from "./ProviderAuthFlow.ts"; + +const instanceId = ProviderInstanceId.make("auth-flow-test"); +const method = { id: "browser", name: "Browser", description: null, type: "agent" as const }; + +it.effect("distinguishes pending method discovery from an agent with no sign-in methods", () => + Effect.gen(function* () { + const discovered = yield* Deferred.make>(); + const controller = yield* ProviderAuthFlow.make({ + instanceId, + credentialBinding: { owner: "provider", key: "shared-agent" }, + methods: Deferred.await(discovered), + authenticate: () => Effect.void, + logout: Effect.void, + }); + const pending = yield* controller + .subscribe("owner") + .pipe(Stream.runHead, Effect.map(Option.getOrThrow)); + assert.isUndefined(pending.methods); + yield* Deferred.succeed(discovered, []); + const ready = yield* controller.subscribe("owner").pipe( + Stream.filter((state) => state.methods !== undefined), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + assert.deepEqual(ready.methods, []); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +const makeHarness = Effect.gen(function* () { + const approved = yield* Deferred.make(); + const verified = yield* Deferred.make(); + const started = yield* Deferred.make(); + let attempts = 0; + const controller = yield* ProviderAuthFlow.make({ + instanceId, + credentialBinding: { owner: "provider", key: "shared-agent" }, + methods: Effect.succeed([method]), + authenticate: (_, context) => + Effect.gen(function* () { + attempts++; + yield* context.setInteraction( + { + type: "browser", + id: "consent", + url: "https://example.com/login", + requiresConsent: true, + }, + (response) => + response.type === "browser" && response.action === "accept" + ? Deferred.succeed(approved, undefined).pipe(Effect.asVoid) + : Effect.void, + ); + yield* Deferred.succeed(started, undefined); + yield* Deferred.await(approved); + yield* context.verifying; + yield* Deferred.await(verified); + }), + logout: Effect.void, + }); + const phase = (phase: ProviderAuthState["phase"], owner = "owner") => + controller.subscribe(owner).pipe( + Stream.filter((state) => state.phase === phase), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + return { controller, phase, started, verified, attempts: () => attempts }; +}); + +it.effect("requires owner consent and provider verification before success", () => + Effect.gen(function* () { + const { controller, phase, started, verified, attempts } = yield* makeHarness; + const state = yield* controller.start("owner"); + yield* Deferred.await(started); + const response = { + instanceId, + flowId: state.flowId!, + interactionId: "consent", + response: { type: "browser" as const, action: "accept" as const }, + }; + const other = yield* phase("waiting", "other"); + assert.isNull(other.interaction); + assert.isNull(other.flowId); + assert.isTrue( + (yield* controller.respond!("other", response).pipe(Effect.result))._tag === "Failure", + ); + assert.isTrue( + (yield* controller.respond!("owner", { ...response, interactionId: "stale" }).pipe( + Effect.result, + ))._tag === "Failure", + ); + yield* controller.start("owner"); + assert.strictEqual(attempts(), 1); + yield* controller.respond!("owner", response); + yield* phase("verifying"); + yield* Deferred.succeed(verified, undefined); + assert.strictEqual((yield* phase("succeeded")).phase, "succeeded"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect("cancellation rejects late consent and permits another login", () => + Effect.gen(function* () { + const { controller, started, phase } = yield* makeHarness; + const state = yield* controller.start("owner"); + yield* Deferred.await(started); + yield* controller.cancel("owner", state.flowId!); + assert.strictEqual((yield* phase("cancelled")).phase, "cancelled"); + assert.isTrue( + (yield* controller.respond!("owner", { + instanceId, + flowId: state.flowId!, + interactionId: "consent", + response: { type: "browser", action: "accept" }, + }).pipe(Effect.result))._tag === "Failure", + ); + const next = yield* controller.start("other"); + assert.notStrictEqual(next.flowId, state.flowId); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect("expires pending login and rejects its response", () => + Effect.gen(function* () { + const { controller, started, phase } = yield* makeHarness; + const state = yield* controller.start("owner"); + yield* Deferred.await(started); + yield* TestClock.adjust(300_001); + assert.include((yield* phase("failed")).message ?? "", "expired"); + assert.isTrue( + (yield* controller.respond!("owner", { + instanceId, + flowId: state.flowId!, + interactionId: "consent", + response: { type: "browser", action: "accept" }, + }).pipe(Effect.result))._tag === "Failure", + ); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect("closes admitted provider processes and blocks new ones while login is pending", () => + Effect.gen(function* () { + const { controller, started } = yield* makeHarness; + let closed = false; + yield* controller.withAccess!( + Effect.addFinalizer(() => + Effect.sync(() => { + closed = true; + }), + ), + ); + yield* controller.start("owner"); + yield* Deferred.await(started); + assert.isTrue(closed); + assert.isTrue( + (yield* controller.withAccess!(Effect.void).pipe(Effect.result))._tag === "Failure", + ); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect("shared credential invalidation closes admitted processes before permitting reuse", () => + Effect.gen(function* () { + const { controller } = yield* makeHarness; + let closed = false; + yield* controller.withAccess!( + Effect.addFinalizer(() => + Effect.sync(() => { + closed = true; + }), + ), + ); + yield* controller.invalidate!; + assert.isTrue(closed); + assert.strictEqual(yield* controller.withAccess!(Effect.succeed("new process")), "new process"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect.each([ + { + interaction: { type: "terminal", id: "terminal", output: "Log in" }, + response: { type: "terminal", data: "input", size: { cols: 80, rows: 24 } }, + }, + { + interaction: { + type: "credentials", + id: "credentials", + fields: [{ name: "token", label: "Token", secret: true }], + }, + response: { type: "credentials", values: { token: "private-token" } }, + }, +] satisfies ReadonlyArray<{ + interaction: ProviderAuthInteraction; + response: ProviderAuthResponse; +}>)( + "keeps $interaction.type responses private and waits for adapter verification", + ({ interaction, response }) => + Effect.gen(function* () { + const received = yield* Deferred.make(); + const controller = yield* ProviderAuthFlow.make({ + instanceId, + credentialBinding: { owner: "t3", key: "binding" }, + methods: Effect.succeed([method]), + authenticate: (_, context) => + Effect.gen(function* () { + yield* context.setInteraction(interaction, (response) => + Deferred.succeed(received, response).pipe(Effect.asVoid), + ); + yield* Deferred.await(received); + yield* context.verifying; + return yield* Effect.never; + }), + logout: Effect.void, + }); + const state = yield* controller.start("owner"); + const pending = yield* controller.subscribe("owner").pipe( + Stream.filter((state) => state.phase === "waiting"), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + assert.deepStrictEqual(pending.interaction, interaction); + const hidden = yield* controller + .subscribe("other") + .pipe(Stream.runHead, Effect.map(Option.getOrThrow)); + assert.isNull(hidden.interaction); + yield* controller.respond!("owner", { + instanceId, + flowId: state.flowId!, + interactionId: interaction.id, + response, + }); + assert.deepStrictEqual(yield* Deferred.await(received), response); + const verifying = yield* controller.subscribe("owner").pipe( + Stream.filter((state) => state.phase === "verifying"), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + assert.isNull(verifying.interaction); + assert.notInclude(verifying.message ?? "", "private-token"); + yield* controller.cancel("owner", state.flowId!); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect("shows a device code only to its owner without treating it as authenticated", () => + Effect.gen(function* () { + const controller = yield* ProviderAuthFlow.make({ + instanceId, + credentialBinding: { owner: "provider", key: "device" }, + methods: Effect.succeed([method]), + authenticate: (_, context) => + context + .setInteraction({ + type: "deviceCode", + id: "code", + url: "https://example.com/device", + userCode: "ABCD-EFGH", + }) + .pipe(Effect.andThen(Effect.never)), + logout: Effect.void, + }); + const start = yield* controller.start("owner"); + const waiting = yield* controller.subscribe("owner").pipe( + Stream.filter((state) => state.phase === "waiting"), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + assert.strictEqual(waiting.interaction?.type, "deviceCode"); + assert.strictEqual(waiting.authorizationUrl, "https://example.com/device"); + const other = yield* controller + .subscribe("other") + .pipe(Stream.runHead, Effect.map(Option.getOrThrow)); + assert.isNull(other.interaction); + assert.isNull(other.authorizationUrl); + yield* controller.cancel("owner", start.flowId!); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect.each([ + { + failure: Effect.fail( + new ProviderSetupError({ + instanceId, + operation: "authenticate", + detail: "This account cannot sign in. Choose another account.", + }), + ), + message: "This account cannot sign in. Choose another account.", + }, + { + failure: Effect.die(new Error("private-token in native diagnostics")), + message: "Sign-in failed. Start again.", + }, +])("publishes only safe authentication failure text %#", ({ failure, message }) => + Effect.gen(function* () { + const controller = yield* ProviderAuthFlow.make({ + instanceId, + credentialBinding: { owner: "t3", key: "failure" }, + methods: Effect.succeed([method]), + authenticate: () => failure, + logout: Effect.void, + }); + yield* controller.start("owner"); + const failed = yield* controller.subscribe("owner").pipe( + Stream.filter((state) => state.phase === "failed"), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + assert.strictEqual(failed.message, message); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +const makeBlockingResponseHarness = Effect.gen(function* () { + const entered = yield* Deferred.make(); + const cleanupStarted = yield* Deferred.make(); + const cleanupReleased = yield* Deferred.make(); + const cleanupFinished = yield* Deferred.make(); + const authenticationFinished = yield* Deferred.make(); + let responses = 0; + const controller = yield* ProviderAuthFlow.make({ + instanceId, + credentialBinding: { owner: "t3", key: "blocked-response" }, + methods: Effect.succeed([method]), + authenticate: (_, context) => + Effect.gen(function* () { + yield* context.setInteraction( + { + type: "browser", + id: "blocked", + url: "https://example.com/login", + requiresConsent: true, + }, + () => + Effect.gen(function* () { + responses++; + yield* Deferred.succeed(entered, undefined); + return yield* Effect.never; + }).pipe( + Effect.ensuring( + Effect.gen(function* () { + yield* Deferred.succeed(cleanupStarted, undefined); + yield* Deferred.await(cleanupReleased); + yield* Deferred.succeed(cleanupFinished, undefined); + }), + ), + ), + ); + yield* Deferred.await(authenticationFinished); + }), + logout: Effect.void, + }); + const start = yield* controller.start("owner"); + yield* controller.subscribe("owner").pipe( + Stream.filter((state) => state.phase === "waiting"), + Stream.runHead, + ); + const input = { + instanceId, + flowId: start.flowId!, + interactionId: "blocked", + response: { type: "browser" as const, action: "accept" as const }, + }; + const response = yield* controller.respond!("owner", input).pipe(Effect.exit, Effect.forkChild); + yield* Deferred.await(entered); + return { + controller, + input, + response, + cleanupStarted, + cleanupReleased, + cleanupFinished, + authenticationFinished, + responses: () => responses, + }; +}); + +it.effect.each(["cancel", "logout"] as const)( + "%s interrupts and awaits a blocked adapter response before admitting another sign-in", + (action) => + Effect.gen(function* () { + const harness = yield* makeBlockingResponseHarness; + const duplicate = yield* Effect.flip(harness.controller.respond!("owner", harness.input)); + assert.include(duplicate.detail, "already in progress"); + assert.equal(harness.responses(), 1); + const stopping = yield* ( + action === "cancel" + ? harness.controller.cancel("owner", harness.input.flowId) + : harness.controller.logout(Effect.void) + ).pipe(Effect.forkChild); + yield* Deferred.await(harness.cleanupStarted); + assert.isTrue(yield* harness.controller.isChangingCredentials!); + assert.isUndefined(stopping.pollUnsafe()); + const premature = yield* Effect.flip(harness.controller.start("other")); + assert.include(premature.detail, "in progress"); + yield* Deferred.succeed(harness.cleanupReleased, undefined); + yield* Fiber.join(stopping); + assert.isTrue(yield* Deferred.isDone(harness.cleanupFinished)); + assert.isTrue(Exit.isFailure(yield* Fiber.join(harness.response))); + assert.isFalse(yield* harness.controller.isChangingCredentials!); + assert.notEqual((yield* harness.controller.start("other")).flowId, harness.input.flowId); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect("timeout interrupts and drains a blocked response before publishing failure", () => + Effect.gen(function* () { + const harness = yield* makeBlockingResponseHarness; + yield* TestClock.adjust(300_001); + yield* Deferred.await(harness.cleanupStarted); + assert.isTrue(yield* harness.controller.isChangingCredentials!); + const duringCleanup = yield* harness.controller + .subscribe("owner") + .pipe(Stream.runHead, Effect.map(Option.getOrThrow)); + assert.notEqual(duringCleanup.phase, "failed"); + yield* Deferred.succeed(harness.cleanupReleased, undefined); + const failed = yield* harness.controller.subscribe("owner").pipe( + Stream.filter((state) => state.phase === "failed"), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + assert.include(failed.message ?? "", "expired"); + assert.isTrue(yield* Deferred.isDone(harness.cleanupFinished)); + assert.isTrue(Exit.isFailure(yield* Fiber.join(harness.response))); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect( + "successful authentication drains a still-running response before admitting provider access", + () => + Effect.gen(function* () { + const harness = yield* makeBlockingResponseHarness; + yield* Deferred.succeed(harness.authenticationFinished, undefined); + yield* Deferred.await(harness.cleanupStarted); + const denied = yield* Effect.flip(harness.controller.withAccess!(Effect.void)); + assert.include(denied.detail, "changing"); + yield* Deferred.succeed(harness.cleanupReleased, undefined); + const succeeded = yield* harness.controller.subscribe("owner").pipe( + Stream.filter((state) => state.phase === "succeeded"), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + assert.equal(succeeded.phase, "succeeded"); + assert.isTrue(yield* Deferred.isDone(harness.cleanupFinished)); + assert.isTrue(Exit.isFailure(yield* Fiber.join(harness.response))); + yield* harness.controller.withAccess!(Effect.void); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect("closing the controller scope interrupts and drains its adapter response", () => + Effect.gen(function* () { + const scope = yield* Scope.make(); + const harness = yield* makeBlockingResponseHarness.pipe( + Effect.provideService(Scope.Scope, scope), + ); + const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild); + yield* Deferred.await(harness.cleanupStarted); + assert.isUndefined(closing.pollUnsafe()); + yield* Deferred.succeed(harness.cleanupReleased, undefined); + yield* Fiber.join(closing); + assert.isTrue(yield* Deferred.isDone(harness.cleanupFinished)); + assert.isTrue(Exit.isFailure(yield* Fiber.join(harness.response))); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect("a failed method refresh keeps the last discovered methods", () => + Effect.gen(function* () { + let fail = false; + const controller = yield* ProviderAuthFlow.make({ + instanceId, + credentialBinding: { owner: "provider", key: "shared-agent" }, + methods: Effect.suspend(() => + fail + ? Effect.fail( + new ProviderSetupError({ instanceId, operation: "status", detail: "interrupted" }), + ) + : Effect.succeed([method]), + ), + authenticate: () => Effect.void, + logout: Effect.void, + }); + yield* controller.refreshMethods!; + fail = true; + yield* controller.refreshMethods!; + const state = yield* controller + .subscribe("owner") + .pipe(Stream.runHead, Effect.map(Option.getOrThrow)); + assert.deepEqual(state.methods, [method]); + assert.strictEqual(state.message, "interrupted"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + +it.effect("rebuilding a controller leaves sessions it admitted to their owners", () => + Effect.gen(function* () { + const scope = yield* Scope.make(); + const { controller } = yield* makeHarness.pipe(Effect.provideService(Scope.Scope, scope)); + let closed = false; + yield* controller.withAccess!( + Effect.addFinalizer(() => + Effect.sync(() => { + closed = true; + }), + ), + ); + yield* Scope.close(scope, Exit.void); + assert.isFalse(closed); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/provider/ProviderAuthFlow.ts b/apps/server/src/provider/ProviderAuthFlow.ts new file mode 100644 index 0000000000000000000000000000000000000000..f01ce92e3fb5456da8ba17488d5ba1f6f6e81558 --- /dev/null +++ b/apps/server/src/provider/ProviderAuthFlow.ts @@ -0,0 +1,463 @@ +import { + ProviderSetupError, + type ProviderAuthInteraction, + type ProviderAuthMethod, + type ProviderAuthResponse, + type ProviderAuthState, + type ProviderInstanceId, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as Clock from "effect/Clock"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Scope from "effect/Scope"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; + +import type * as ProviderAuthService from "./Services/ProviderAuthService.ts"; + +export interface ProviderAuthFlowContext { + readonly flowId: string; + readonly setInteraction: ( + interaction: ProviderAuthInteraction, + respond?: (response: ProviderAuthResponse) => Effect.Effect, + ) => Effect.Effect; + readonly verifying: Effect.Effect; +} + +const isSetupError = Schema.is(ProviderSetupError); + +const failureMessage = (cause: Cause.Cause) => { + const error = Cause.findErrorOption(cause); + return Option.isSome(error) && isSetupError(error.value) + ? error.value.detail + : "Sign-in failed. Start again."; +}; + +interface Flow { + readonly id: string; + readonly owner: string; + readonly expiresAt: number; + fiber?: Fiber.Fiber; + responseFiber?: Fiber.Fiber; + respond: + | ((response: ProviderAuthResponse) => Effect.Effect) + | undefined; +} + +/** Adapters do login and credential handling; this owns client consent and flow lifetime. */ +export const make = Effect.fn("ProviderAuthFlow.make")(function* (options: { + readonly instanceId: ProviderInstanceId; + readonly credentialBinding: NonNullable< + ProviderAuthService.ProviderAuthController["credentialBinding"] + >; + readonly methods: Effect.Effect, ProviderSetupError>; + readonly defaultMethodId?: string; + /** Fail with ProviderSetupError containing safe text for the user, never native token data. */ + readonly authenticate: ( + methodId: string, + context: ProviderAuthFlowContext, + ) => Effect.Effect; + readonly logout: Effect.Effect; + readonly timeoutMs?: number; +}) { + const scope = yield* Scope.Scope; + const crypto = yield* Crypto.Crypto; + const lock = yield* Semaphore.make(1); + const timeoutMs = options.timeoutMs ?? 300_000; + const empty: ProviderAuthState = { + instanceId: options.instanceId, + phase: "idle", + flowId: null, + authorizationUrl: null, + expiresAt: null, + message: null, + interaction: null, + credentialOwner: options.credentialBinding.owner, + }; + const snapshot = yield* SubscriptionRef.make({ owner: null as string | null, state: empty }); + let active: Flow | undefined; + let operation: "idle" | "auth" | "stopping" | "closed" = "idle"; + const sessions = new Set(); + const stopOwnedSessions = Effect.suspend(() => + Effect.forEach(Array.from(sessions), (session) => Scope.close(session, Exit.void), { + discard: true, + concurrency: "unbounded", + }), + ); + const publish = (flow: Flow, patch: Partial) => + Effect.suspend(() => + active === flow + ? SubscriptionRef.update(snapshot, (current) => ({ + owner: flow.owner, + state: { ...current.state, ...patch }, + })) + : Effect.void, + ); + + const requireFlow = Effect.fnUntraced(function* (owner: string, id: string) { + if ( + operation !== "auth" || + !active || + active.owner !== owner || + active.id !== id || + (yield* Clock.currentTimeMillis) >= active.expiresAt + ) { + return yield* new ProviderSetupError({ + instanceId: options.instanceId, + operation: "respond", + detail: "This sign-in is no longer active in this client.", + }); + } + return active; + }); + + // Method discovery does not start a sign-in flow. + const refreshMethods = options.methods.pipe( + Effect.flatMap((methods) => + SubscriptionRef.update(snapshot, (current) => ({ + ...current, + state: { ...current.state, methods }, + })), + ), + // Keep the last known methods when a refresh fails or is interrupted. + Effect.catch((error) => + SubscriptionRef.update(snapshot, (current) => ({ + ...current, + state: { ...current.state, methods: current.state.methods ?? [], message: error.detail }, + })), + ), + ); + yield* refreshMethods.pipe(Effect.forkIn(scope)); + + const controller: ProviderAuthService.ProviderAuthController = { + credentialBinding: options.credentialBinding, + refreshMethods, + invalidate: lock.withPermit( + Effect.gen(function* () { + if (operation !== "idle") return; + yield* stopOwnedSessions; + yield* SubscriptionRef.set(snapshot, { + owner: null, + state: { + ...empty, + methods: snapshot.value.state.methods ?? [], + message: "This provider's shared sign-in changed.", + }, + }); + }), + ), + isChangingCredentials: Effect.sync(() => operation !== "idle"), + withAccess: (task) => + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const parent = yield* Scope.Scope; + const child = yield* lock.withPermit( + Effect.gen(function* () { + if (operation !== "idle") + return yield* new ProviderSetupError({ + instanceId: options.instanceId, + operation: "session", + detail: "Provider sign-in is changing. Try again after it finishes.", + }); + const child = yield* Scope.make(); + sessions.add(child); + yield* Scope.addFinalizer( + child, + Effect.sync(() => { + sessions.delete(child); + }), + ); + yield* Scope.addFinalizer(parent, Scope.close(child, Exit.void)); + return child; + }), + ); + const fiber = yield* restore(task).pipe( + Effect.provideService(Scope.Scope, child), + Effect.forkIn(child), + ); + return yield* restore(Fiber.await(fiber)).pipe( + Effect.flatMap((result) => result), + Effect.onExit((result) => + Exit.isFailure(result) ? Scope.close(child, Exit.void) : Effect.void, + ), + ); + }), + ), + start: (owner, stopSessions = Effect.void, selectedMethodId) => + lock.withPermit( + Effect.gen(function* () { + if (operation === "auth" && active?.owner === owner) return snapshot.value.state; + if (operation !== "idle") + return yield* new ProviderSetupError({ + instanceId: options.instanceId, + operation: "start", + detail: "Provider setup is already in progress.", + }); + const id = yield* crypto.randomUUIDv4.pipe( + Effect.mapError( + () => + new ProviderSetupError({ + instanceId: options.instanceId, + operation: "start", + detail: "Could not start sign-in. Try again.", + }), + ), + ); + const flow: Flow = { + id, + owner, + expiresAt: (yield* Clock.currentTimeMillis) + timeoutMs, + respond: undefined, + }; + active = flow; + operation = "auth"; + const state: ProviderAuthState = { + ...empty, + methods: snapshot.value.state.methods ?? [], + phase: "starting", + flowId: id, + expiresAt: DateTime.formatIso(DateTime.makeUnsafe(flow.expiresAt)), + message: "Starting sign-in.", + }; + yield* SubscriptionRef.set(snapshot, { owner, state }); + flow.fiber = yield* Effect.gen(function* () { + const methods = yield* options.methods; + yield* publish(flow, { methods }); + const methodId = selectedMethodId ?? options.defaultMethodId ?? methods[0]?.id; + if (!methodId || !methods.some((method) => method.id === methodId)) + return yield* new ProviderSetupError({ + instanceId: options.instanceId, + operation: "start", + detail: "The provider did not advertise this sign-in method.", + }); + yield* stopSessions.pipe(Effect.ensuring(stopOwnedSessions)); + yield* options.authenticate(methodId, { + flowId: id, + setInteraction: (interaction, respond) => + Effect.gen(function* () { + if (active !== flow) return; + flow.respond = respond; + yield* publish(flow, { + phase: "waiting", + interaction, + authorizationUrl: + interaction.type === "browser" || interaction.type === "deviceCode" + ? interaction.url + : null, + message: "Complete sign-in to continue.", + }); + }), + verifying: Effect.gen(function* () { + flow.respond = undefined; + yield* publish(flow, { + phase: "verifying", + interaction: null, + authorizationUrl: null, + message: "Checking provider sign-in.", + }); + }), + }); + }).pipe( + Effect.scoped, + Effect.timeoutOrElse({ + duration: timeoutMs, + orElse: () => + Effect.fail( + new ProviderSetupError({ + instanceId: options.instanceId, + operation: "start", + detail: "Sign-in expired. Start again.", + }), + ), + }), + Effect.exit, + Effect.flatMap((result) => + Effect.gen(function* () { + const finishing = yield* lock.withPermit( + Effect.sync(() => { + if (active !== flow) return false; + operation = "stopping"; + return true; + }), + ); + if (!finishing) return; + if (flow.responseFiber) yield* Fiber.interrupt(flow.responseFiber); + yield* lock.withPermit( + Effect.gen(function* () { + if (active !== flow) return; + yield* publish(flow, { + phase: Exit.isSuccess(result) ? "succeeded" : "failed", + interaction: null, + authorizationUrl: null, + expiresAt: null, + message: Exit.isSuccess(result) + ? "Sign-in complete." + : failureMessage(result.cause), + }); + active = undefined; + operation = "idle"; + }), + ); + }), + ), + Effect.interruptible, + Effect.forkIn(scope), + ); + return state; + }).pipe(Effect.uninterruptible), + ), + respond: (owner, input) => + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const { flow, fiber, ready } = yield* lock.withPermit( + Effect.gen(function* () { + const flow = yield* requireFlow(owner, input.flowId); + const interaction = snapshot.value.state.interaction; + if ( + !interaction || + interaction.id !== input.interactionId || + interaction.type !== input.response.type || + !flow.respond + ) + return yield* new ProviderSetupError({ + instanceId: options.instanceId, + operation: "respond", + detail: "This sign-in interaction is no longer available.", + }); + if (flow.responseFiber) + return yield* new ProviderSetupError({ + instanceId: options.instanceId, + operation: "respond", + detail: "A sign-in response is already in progress.", + }); + const ready = yield* Deferred.make(); + const callback = flow.respond; + const fiber = yield* Deferred.await(ready).pipe( + Effect.andThen(Effect.suspend(() => callback(input.response))), + Effect.interruptible, + Effect.forkIn(scope), + ); + flow.responseFiber = fiber; + return { flow, fiber, ready }; + }), + ); + yield* Deferred.succeed(ready, undefined); + yield* restore(Fiber.join(fiber)).pipe( + Effect.onInterrupt(() => Fiber.interrupt(fiber).pipe(Effect.asVoid)), + Effect.ensuring( + lock.withPermit( + Effect.sync(() => { + if (flow.responseFiber === fiber) delete flow.responseFiber; + }), + ), + ), + ); + return snapshot.value.state; + }), + ), + complete: () => + Effect.fail( + new ProviderSetupError({ + instanceId: options.instanceId, + operation: "complete", + detail: "This provider does not accept a pasted redirect URL.", + }), + ), + cancel: (owner, id) => + Effect.gen(function* () { + const flow = yield* lock.withPermit( + Effect.gen(function* () { + const flow = yield* requireFlow(owner, id); + yield* publish(flow, { + phase: "cancelled", + interaction: null, + authorizationUrl: null, + expiresAt: null, + message: "Sign-in cancelled.", + }); + active = undefined; + operation = "stopping"; + return flow; + }), + ); + if (flow.responseFiber) yield* Fiber.interrupt(flow.responseFiber); + if (flow.fiber) yield* Fiber.interrupt(flow.fiber); + operation = "idle"; + return snapshot.value.state; + }).pipe(Effect.uninterruptible), + logout: (stopSessions) => + Effect.gen(function* () { + const flow = yield* lock.withPermit( + Effect.gen(function* () { + if (operation !== "idle" && operation !== "auth") + return yield* new ProviderSetupError({ + instanceId: options.instanceId, + operation: "logout", + detail: "Provider setup is already stopping.", + }); + operation = "stopping"; + const flow = active; + active = undefined; + return flow; + }), + ); + const result = yield* Effect.gen(function* () { + if (flow?.responseFiber) yield* Fiber.interrupt(flow.responseFiber); + if (flow?.fiber) yield* Fiber.interrupt(flow.fiber); + yield* stopSessions.pipe(Effect.ensuring(stopOwnedSessions)); + yield* options.logout; + }).pipe(Effect.exit); + const state: ProviderAuthState = { + ...empty, + methods: snapshot.value.state.methods ?? [], + phase: Exit.isSuccess(result) ? "idle" : "failed", + message: Exit.isSuccess(result) ? "Signed out." : "Could not sign out. Try again.", + }; + yield* lock.withPermit( + Effect.gen(function* () { + yield* SubscriptionRef.set(snapshot, { owner: null, state }); + operation = "idle"; + }), + ); + if (Exit.isFailure(result)) return yield* Effect.failCause(result.cause); + return state; + }).pipe(Effect.uninterruptible), + subscribe: (owner) => + SubscriptionRef.changes(snapshot).pipe( + Stream.map((current) => + current.owner === null || current.owner === owner + ? current.state + : { + ...current.state, + flowId: null, + authorizationUrl: null, + interaction: null, + expiresAt: null, + message: active + ? "Sign-in is in progress in another client." + : current.state.message, + }, + ), + ), + }; + yield* Effect.addFinalizer(() => + Effect.gen(function* () { + operation = "closed"; + const flow = active; + active = undefined; + if (flow?.responseFiber) yield* Fiber.interrupt(flow.responseFiber); + if (flow?.fiber) yield* Fiber.interrupt(flow.fiber); + // Settings edits rebuild instances; admitted sessions (including those of + // peers sharing this binding) end with their own scopes, not this one. + }), + ); + return controller; +}); diff --git a/apps/server/src/provider/ProviderCredentialStore.test.ts b/apps/server/src/provider/ProviderCredentialStore.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..0a5ad906e126475ab12f783ff63936257604c95a --- /dev/null +++ b/apps/server/src/provider/ProviderCredentialStore.test.ts @@ -0,0 +1,49 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import { ServerSecretStore } from "../auth/ServerSecretStore.ts"; +import * as ProviderCredentialStore from "./ProviderCredentialStore.ts"; + +it.effect("isolates provider bindings and preserves opaque credentials", () => + Effect.gen(function* () { + const data = new Map(); + const secretStore = ServerSecretStore.of({ + get: (name) => Effect.sync(() => Option.fromUndefinedOr(data.get(name))), + set: (name, value) => + Effect.sync(() => { + data.set(name, value); + }), + remove: (name) => + Effect.sync(() => { + data.delete(name); + }), + create: () => Effect.die("unused"), + getOrCreateRandom: () => Effect.die("unused"), + }); + const a = yield* ProviderCredentialStore.make("cursor", "../../personal").pipe( + Effect.provideService(ServerSecretStore, secretStore), + ); + const b = yield* ProviderCredentialStore.make("cursor", "work").pipe( + Effect.provideService(ServerSecretStore, secretStore), + ); + const c = yield* ProviderCredentialStore.make("other", "../../personal").pipe( + Effect.provideService(ServerSecretStore, secretStore), + ); + const bytes = Uint8Array.from([0, 255, 128, 1]); + yield* a.set(bytes); + assert.deepStrictEqual(Option.getOrThrow(yield* a.get), bytes); + assert.isTrue(Option.isNone(yield* b.get)); + assert.isTrue(Option.isNone(yield* c.get)); + assert.isFalse(a.binding.key.includes("/")); + const long = yield* ProviderCredentialStore.make("a".repeat(64), "b".repeat(64)).pipe( + Effect.provideService(ServerSecretStore, secretStore), + ); + assert.isBelow(long.binding.key.length, 255); + const delimiter = yield* ProviderCredentialStore.make("cur", "sor../../personal").pipe( + Effect.provideService(ServerSecretStore, secretStore), + ); + assert.notStrictEqual(delimiter.binding.key, a.binding.key); + yield* a.remove; + assert.isTrue(Option.isNone(yield* a.get)); + }), +); diff --git a/apps/server/src/provider/ProviderCredentialStore.ts b/apps/server/src/provider/ProviderCredentialStore.ts new file mode 100644 index 0000000000000000000000000000000000000000..259b441654c916190cadeea9a93ca8a05aecb57c --- /dev/null +++ b/apps/server/src/provider/ProviderCredentialStore.ts @@ -0,0 +1,21 @@ +import * as NodeCrypto from "node:crypto"; +import * as Effect from "effect/Effect"; +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; + +/** A provider binding stores opaque bytes; only its adapter decodes or refreshes them. */ +export const make = Effect.fn("ProviderCredentialStore.make")(function* ( + driver: string, + bindingId: string, +) { + const secrets = yield* ServerSecretStore.ServerSecretStore; + // Hash the tuple so arbitrary bindings cannot escape or exceed a filename. + const key = `provider-auth-${NodeCrypto.createHash("sha256") + .update(`${driver.length}:${driver}${bindingId}`) + .digest("hex")}`; + return { + binding: { owner: "t3" as const, key }, + get: secrets.get(key), + set: (credentials: Uint8Array) => secrets.set(key, credentials), + remove: secrets.remove(key), + }; +}); diff --git a/apps/server/src/provider/Services/ProviderAuthService.ts b/apps/server/src/provider/Services/ProviderAuthService.ts index ea89edf78af516ee35d3cc78920d64a11bd47ad0..93fd39fb3e3fde2e2f0551ba9487b72f2f662968 100644 --- a/apps/server/src/provider/Services/ProviderAuthService.ts +++ b/apps/server/src/provider/Services/ProviderAuthService.ts @@ -1,12 +1,28 @@ -import type { ProviderAuthState, ProviderInstanceId, ProviderSetupError } from "@t3tools/contracts"; +import type { + ProviderAuthRespondInput, + ProviderAuthStartInput, + ProviderAuthState, + ProviderInstanceId, + ProviderSetupError, +} from "@t3tools/contracts"; import * as Context from "effect/Context"; import type * as Effect from "effect/Effect"; import type * as Stream from "effect/Stream"; +import type * as Scope from "effect/Scope"; export interface ProviderAuthController { + /** Equal keys mean these instances share credentials on this environment. */ + readonly credentialBinding?: { readonly owner: "provider" | "t3"; readonly key: string }; + readonly isChangingCredentials?: Effect.Effect; + readonly invalidate?: Effect.Effect; + readonly refreshMethods?: Effect.Effect; + readonly withAccess?: ( + task: Effect.Effect, + ) => Effect.Effect; readonly start: ( ownerSessionId: string, stopSessions?: Effect.Effect, + methodId?: string, ) => Effect.Effect; readonly complete: ( ownerSessionId: string, @@ -16,6 +32,10 @@ export interface ProviderAuthController { ownerSessionId: string, flowId: string, ) => Effect.Effect; + readonly respond?: ( + ownerSessionId: string, + input: ProviderAuthRespondInput, + ) => Effect.Effect; /** The controller closes process admission before it stops routed sessions. */ readonly logout: ( stopSessions: Effect.Effect, @@ -30,13 +50,17 @@ interface ProviderAuthTarget { export interface ProviderAuthServiceShape { readonly start: ( - input: ProviderAuthTarget, + input: ProviderAuthStartInput, ownerSessionId: string, ) => Effect.Effect; readonly complete: ( input: ProviderAuthTarget & { readonly flowId: string; readonly callbackUrl: string }, ownerSessionId: string, ) => Effect.Effect; + readonly respond: ( + input: ProviderAuthRespondInput, + ownerSessionId: string, + ) => Effect.Effect; readonly cancel: ( input: ProviderAuthTarget & { readonly flowId: string }, ownerSessionId: string, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 87ed56ddd46d05c8a37c9a64d6618dd036280e3f..17500460d9badb2008290a3481350f05afec2198 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2497,6 +2497,15 @@ const makeWsRpcLayer = ( providerAuth.start(input, currentSessionId), { "rpc.aggregate": "provider" }, ), + [WS_METHODS.providerAuthRespond]: (input) => + observeRpcEffect( + WS_METHODS.providerAuthRespond, + providerAuth.respond(input, currentSessionId), + { + "rpc.aggregate": "provider", + instanceId: input.instanceId, + }, + ), [WS_METHODS.providerAuthComplete]: (input) => observeRpcEffect( WS_METHODS.providerAuthComplete, diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 18127076ff19fe97fce28dea547aa9ce61951bda..a7476ec17157db2b563aa48e0a91177e63bb61b5 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -978,6 +978,10 @@ export function createServerEnvironmentAtoms( key: ({ environmentId, input }) => JSON.stringify([environmentId, input]), }, }), + respondProviderAuth: createEnvironmentRpcCommand(runtime, { + label: "environment-data:provider:auth-respond", + tag: WS_METHODS.providerAuthRespond, + }), completeProviderAuth: createEnvironmentRpcCommand(runtime, { label: "environment-data:provider:auth-complete", tag: WS_METHODS.providerAuthComplete, diff --git a/packages/contracts/src/providerSetup.test.ts b/packages/contracts/src/providerSetup.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..13fb3fca1f11cd3d7f2ffaf1ca30219ce7874c7a --- /dev/null +++ b/packages/contracts/src/providerSetup.test.ts @@ -0,0 +1,53 @@ +import { describe, expect, it } from "vite-plus/test"; +import * as Schema from "effect/Schema"; + +import { ProviderAuthResponse, ProviderAuthState } from "./providerSetup.ts"; + +const decodeResponse = Schema.decodeUnknownSync(ProviderAuthResponse); +const decodeState = Schema.decodeUnknownSync(ProviderAuthState); + +describe("provider credential responses", () => { + it("accepts the advertised field limit and rejects oversized or invalid fields", () => { + const values = Object.fromEntries( + Array.from({ length: 16 }, (_, i) => [`field_${i}`, "value"]), + ); + expect(decodeResponse({ type: "credentials", values })).toEqual({ + type: "credentials", + values, + }); + expect(() => + decodeResponse({ type: "credentials", values: { ...values, extra: "value" } }), + ).toThrow(); + expect(() => decodeResponse({ type: "credentials", values: { "": "value" } })).toThrow(); + expect(() => + decodeResponse({ type: "credentials", values: { token: "x".repeat(16_385) } }), + ).toThrow(); + }); +}); + +describe("provider auth state", () => { + it("drops auth variants from newer servers instead of rejecting the state", () => { + const method = { id: "browser", name: "Browser", description: null, type: "agent" }; + expect( + decodeState({ + instanceId: "cursor", + phase: "waiting", + flowId: null, + authorizationUrl: null, + expiresAt: null, + message: null, + methods: [method, { ...method, id: "passkey", type: "passkey" }], + interaction: { type: "passkey", id: "passkey" }, + credentialOwner: "keychain", + }), + ).toEqual({ + instanceId: "cursor", + phase: "waiting", + flowId: null, + authorizationUrl: null, + expiresAt: null, + message: null, + methods: [method], + }); + }); +}); diff --git a/packages/contracts/src/providerSetup.ts b/packages/contracts/src/providerSetup.ts index db6ce1f79aec1fa6acf9c1cb4beca19cbcc4722d..04261e934c89aa8551dfee82c2ae25f2d16e5642 100644 --- a/packages/contracts/src/providerSetup.ts +++ b/packages/contracts/src/providerSetup.ts @@ -1,6 +1,11 @@ import * as Schema from "effect/Schema"; -import { IsoDateTime, TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { + ForwardCompatibleArray, + ForwardCompatibleOptional, + IsoDateTime, + TrimmedNonEmptyString, +} from "./baseSchemas.ts"; import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; export const ProviderSetupInput = Schema.Struct({ @@ -10,6 +15,88 @@ export type ProviderSetupInput = typeof ProviderSetupInput.Type; const SetupOperationId = TrimmedNonEmptyString.check(Schema.isMaxLength(128)); +export const ProviderAuthMethod = Schema.Struct({ + id: SetupOperationId, + name: TrimmedNonEmptyString, + description: Schema.NullOr(Schema.String), + type: Schema.Literals(["agent", "terminal", "credentials"]), +}); +export type ProviderAuthMethod = typeof ProviderAuthMethod.Type; + +// These describe client interactions, not OAuth grant types. The provider +// adapter remains responsible for credentials, callbacks, and refresh. +export const ProviderAuthInteraction = Schema.Union([ + Schema.Struct({ + type: Schema.Literal("browser"), + id: SetupOperationId, + url: TrimmedNonEmptyString.check(Schema.isMaxLength(16_384)), + requiresConsent: Schema.Boolean, + acceptsCallback: Schema.optionalKey(Schema.Boolean), + }), + Schema.Struct({ + type: Schema.Literal("deviceCode"), + id: SetupOperationId, + url: TrimmedNonEmptyString.check(Schema.isMaxLength(16_384)), + userCode: TrimmedNonEmptyString.check(Schema.isMaxLength(256)), + }), + Schema.Struct({ + type: Schema.Literal("terminal"), + id: SetupOperationId, + output: Schema.String.check(Schema.isMaxLength(16_384)), + outputOffset: Schema.optionalKey(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))), + }), + Schema.Struct({ + type: Schema.Literal("credentials"), + id: SetupOperationId, + fields: Schema.Array( + Schema.Struct({ + name: SetupOperationId, + label: TrimmedNonEmptyString, + secret: Schema.Boolean, + }), + ).check(Schema.isMaxLength(16)), + }), +]); +export type ProviderAuthInteraction = typeof ProviderAuthInteraction.Type; + +export const ProviderAuthResponse = Schema.Union([ + Schema.Struct({ + type: Schema.Literal("browser"), + action: Schema.Literals(["accept", "decline"]), + }), + Schema.Struct({ + type: Schema.Literal("terminal"), + data: Schema.String.check(Schema.isMaxLength(4_096)), + size: Schema.optionalKey( + Schema.Struct({ + cols: Schema.Int.check(Schema.isBetween({ minimum: 1, maximum: 500 })), + rows: Schema.Int.check(Schema.isBetween({ minimum: 1, maximum: 200 })), + }), + ), + }), + Schema.Struct({ + type: Schema.Literal("credentials"), + values: Schema.Record(SetupOperationId, Schema.String.check(Schema.isMaxLength(16_384))).check( + Schema.isMaxProperties(16), + ), + }), +]); +export type ProviderAuthResponse = typeof ProviderAuthResponse.Type; + +export const ProviderAuthStartInput = Schema.Struct({ + instanceId: ProviderInstanceId, + methodId: Schema.optionalKey(SetupOperationId), +}); +export type ProviderAuthStartInput = typeof ProviderAuthStartInput.Type; + +export const ProviderAuthRespondInput = Schema.Struct({ + instanceId: ProviderInstanceId, + flowId: SetupOperationId, + interactionId: SetupOperationId, + response: ProviderAuthResponse, +}); +export type ProviderAuthRespondInput = typeof ProviderAuthRespondInput.Type; + export const ProviderAuthState = Schema.Struct({ instanceId: ProviderInstanceId, phase: Schema.Literals([ @@ -25,6 +112,13 @@ export const ProviderAuthState = Schema.Struct({ authorizationUrl: Schema.NullOr(Schema.String), expiresAt: Schema.NullOr(IsoDateTime), message: Schema.NullOr(Schema.String), + // Newer servers may add method types, interactions, or owners; older + // clients drop what they cannot decode instead of rejecting the state. + methods: Schema.optionalKey( + ForwardCompatibleArray(ProviderAuthMethod).check(Schema.isMaxLength(32)), + ), + interaction: ForwardCompatibleOptional(Schema.NullOr(ProviderAuthInteraction)), + credentialOwner: ForwardCompatibleOptional(Schema.Literals(["provider", "t3"])), }); export type ProviderAuthState = typeof ProviderAuthState.Type; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 5a421052207386b7897edcdb5379087a0610d740..dbc143048a721768891b30b113b57114a23e856e 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -6,6 +6,8 @@ import { ProviderAuthCancelInput, ProviderAuthCompleteInput, ProviderAuthState, + ProviderAuthStartInput, + ProviderAuthRespondInput, ProviderInstallCancelInput, ProviderInstallState, ProviderSetupError, @@ -299,6 +301,7 @@ export const WS_METHODS = { providerAuthStart: "provider.auth.start", providerConsumeResetCredit: "provider.consumeResetCredit", providerAuthComplete: "provider.auth.complete", + providerAuthRespond: "provider.auth.respond", providerAuthCancel: "provider.auth.cancel", providerAuthLogout: "provider.auth.logout", providerAuthSubscribe: "provider.auth.subscribe", @@ -499,7 +502,13 @@ const WsProviderConsumeResetCreditRpc = Rpc.make(WS_METHODS.providerConsumeReset }); const WsProviderAuthStartRpc = Rpc.make(WS_METHODS.providerAuthStart, { - payload: ProviderSetupInput, + payload: ProviderAuthStartInput, + success: ProviderAuthState, + error: ProviderSetupRpcError, +}); + +const WsProviderAuthRespondRpc = Rpc.make(WS_METHODS.providerAuthRespond, { + payload: ProviderAuthRespondInput, success: ProviderAuthState, error: ProviderSetupRpcError, }); @@ -1390,6 +1399,7 @@ export const WsRpcGroup = RpcGroup.make( WsProviderConsumeResetCreditRpc, WsProviderAuthStartRpc, WsProviderAuthCompleteRpc, + WsProviderAuthRespondRpc, WsProviderAuthCancelRpc, WsProviderAuthLogoutRpc, WsProviderAuthSubscribeRpc, -- 2.54.0