diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 09ec3034a9bc9..bfe77a5bbe651 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -78,6 +78,7 @@ import * as ModelManifest from "../../provider/ModelManifest.ts"; import * as PortScanner from "../../preview/PortScanner.ts"; import { resolveProviderCompatibility } from "../../provider/providerCompatibility.ts"; import * as CodexAdapterV2 from "./CodexAdapterV2.ts"; +import { requestBrowserAccess } from "../../provider/testFixtures/browserAccessClient.ts"; import { withCodexReplayChildMetadata } from "./CodexAdapterV2.testkit.ts"; import * as CodexAdapterV2Testkit from "./CodexAdapterV2.testkit.ts"; @@ -1372,9 +1373,11 @@ describe("CodexAdapterV2 post-settle continuation", () => { promptSuggestions: Effect.Effect = Effect.succeed(false), onNotification: (method: string) => Effect.Effect = () => Effect.void, models?: CodexAdapterV2.CodexAdapterV2Options["models"], + environment: NodeJS.ProcessEnv = {}, ) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; + let launchEnvironment: NodeJS.ProcessEnv = {}; const idAllocator = yield* IdAllocator.IdAllocatorV2; const serverConfig = yield* ProviderReplayHarness.makeReplayServerConfig( transcript.scenario, @@ -1389,6 +1392,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { const clientFactory: CodexAdapterV2.CodexAppServerClientFactoryShape = { open: (openInput) => Layer.build(CodexReplay.layerReplayWithDriver(driver)).pipe( + Effect.tap(() => Effect.sync(() => (launchEnvironment = openInput.environment))), Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterOpenSessionError({ @@ -1431,7 +1435,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { const adapter = yield* CodexAdapterV2.makeCodexAdapterV2({ instanceId: CodexAdapterV2.CODEX_DEFAULT_INSTANCE_ID, settings: DEFAULT_CODEX_SETTINGS, - environment: {}, + environment, clientFactory, ...(onUsageLimits ? { onUsageLimits } : {}), ...(models ? { models } : {}), @@ -1502,6 +1506,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { providerThread, threadId, serverConfig, + launchEnvironment, events, continuationRequests, terminalEvents, @@ -1581,6 +1586,128 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("answers the browser plugin's access check only for this session's running turns", () => + Effect.gen(function* () { + const scenario = "browser-access"; + const nativeThreadId = "browser-access-thread"; + const nativeTurnId = "browser-access-turn"; + const completeTurn = yield* Deferred.make(); + const sessions = yield* McpProviderSessions.McpProviderSessions; + const browserAccessWithdrawn = yield* Deferred.make(); + const mcpSession = { + environmentId: EnvironmentId.make("environment-browser-access"), + threadId: ThreadId.make(`thread-${scenario}`), + providerSessionId: "browser-access-mcp", + providerInstanceId: CodexAdapterV2.CODEX_DEFAULT_INSTANCE_ID, + endpoint: "http://127.0.0.1:1/mcp", + authorizationHeader: "Bearer fixture", + browserToolsAvailable: true, + browserAccessAllowed: Deferred.isDone(browserAccessWithdrawn).pipe( + Effect.map((done) => !done), + ), + }; + yield* sessions.set(mcpSession); + const entries = codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "work" }); + entries[5] = { + type: "expect_outbound", + label: "turn/start", + frame: { + id: 3, + method: "turn/start", + params: yield* CodexAdapterV2.buildCodexTurnStartParams({ + nativeThreadId, + codexInput: [{ type: "text", text: "work" }], + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + modelSelection: CODEX_TEST_MODEL_SELECTION, + hasT3Mcp: true, + }), + }, + }; + entries.push({ + type: "emit_inbound", + label: "browser turn completed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }, + }, + }); + const inherited = { + BROWSER_USE_ACCESS_EXECUTOR_ID: "other-app-executor", + BROWSER_USE_ACCESS_PIPE_PATH: "/other-app/socket", + BROWSER_USE_ACCESS_VERIFYING_IDENTITY: "other-app-identity", + }; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ scenario, entries }), + undefined, + undefined, + undefined, + (entry) => + entry.label === "browser turn completed" ? Deferred.await(completeTurn) : Effect.void, + undefined, + undefined, + undefined, + undefined, + inherited, + ); + const environment = harness.launchEnvironment; + assert.notEqual( + environment.BROWSER_USE_ACCESS_EXECUTOR_ID, + inherited.BROWSER_USE_ACCESS_EXECUTOR_ID, + ); + assert.notEqual( + environment.BROWSER_USE_ACCESS_PIPE_PATH, + inherited.BROWSER_USE_ACCESS_PIPE_PATH, + ); + const allowed = ( + request: { sessionId?: string; turnId?: string; executorId?: string } = {}, + ) => + requestBrowserAccess(environment, { + sessionId: nativeThreadId, + turnId: nativeTurnId, + ...request, + }); + + assert.isFalse(yield* allowed()); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("browser-access-attempt"), + text: "work", + }), + ); + assert.isTrue(yield* allowed()); + assert.isFalse(yield* allowed({ sessionId: "other-session" })); + assert.isFalse(yield* allowed({ turnId: "other-turn" })); + assert.isFalse(yield* allowed({ executorId: "other-executor" })); + yield* Deferred.succeed(browserAccessWithdrawn, undefined); + assert.isFalse(yield* allowed()); + yield* sessions.set({ ...mcpSession, browserToolsAvailable: false }); + assert.isFalse(yield* allowed()); + yield* sessions.clear(mcpSession.threadId); + assert.isFalse(yield* allowed()); + yield* sessions.set({ ...mcpSession, browserAccessAllowed: Effect.succeed(true) }); + assert.isTrue(yield* allowed()); + yield* Deferred.succeed(completeTurn, undefined); + yield* harness.firstTerminal; + assert.isFalse(yield* allowed()); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + it.effect.each(["before", "after"] as const)( "rejects a previous run's native turn when it completes %s the new start response", (completionOrder) => diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index efa5d0a9c28f1..633b5171933fd 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -17,6 +17,7 @@ import { import type { ManagedServerProvider } from "@t3tools/provider-core/server/snapshot"; import type { CodexEffectiveRuntime } from "../../provider/CodexManagedRuntime.ts"; import { classifyCodexManagedError } from "../../provider/CodexManagedErrors.ts"; +import { makeCodexBrowserAccessHost } from "../../provider/CodexBrowserAccessHost.ts"; import { buildCodexInitializeParams, codexAppServerEnvironment, @@ -1872,13 +1873,49 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }), ), ); + const activeTurns = yield* Ref.make(new Map()); + // OpenAI's browser plugin asks the host before each turn; never inherit another app's host. + const browserAccessEnvironment = yield* makeCodexBrowserAccessHost( + Effect.fn("CodexAdapterV2.authorizeBrowserAccess")(function* (request) { + const context = (yield* Ref.get(activeTurns)).get(request.turnId); + if (context === undefined) return false; + let root = context; + while (root.subagent !== null) root = root.subagent.parentContext; + // Codex reports the root session id for subagent turns. + if ( + context.providerThread.nativeThreadRef?.nativeId !== request.sessionId && + root.providerThread.nativeThreadRef?.nativeId !== request.sessionId + ) + return false; + const session = yield* readMcpSession(context.input.threadId); + if (session?.browserToolsAvailable !== true) return false; + return session.browserAccessAllowed === undefined + ? true + : yield* session.browserAccessAllowed; + }), + ).pipe( + Effect.catch((cause) => + Effect.logWarning("Browser plugin access is unavailable for this session.", { + cause, + }).pipe( + Effect.as({ + BROWSER_USE_ACCESS_EXECUTOR_ID: "", + BROWSER_USE_ACCESS_PIPE_PATH: "", + BROWSER_USE_ACCESS_VERIFYING_IDENTITY: "", + }), + ), + ), + ); const client = yield* clientFactory.open({ instanceId: adapterOptions.instanceId, threadId: input.threadId, providerSessionId: input.providerSessionId, runtimePolicy: input.runtimePolicy, settings: resolvedRuntime?.config ?? adapterOptions.settings, - environment: resolvedRuntime?.environment ?? adapterOptions.environment, + environment: { + ...(resolvedRuntime?.environment ?? adapterOptions.environment), + ...browserAccessEnvironment, + }, }); const additionalContextByThread = yield* Ref.make( new Map>(), @@ -1995,7 +2032,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const limitedTurnItems = yield* Ref.make( new Map>(), ); - const activeTurns = yield* Ref.make(new Map()); // Codex reports MCP startup on thread start, usually before the turn context exists. const mcpStartupFailures = new Map>(); const turnTokenUsageByThread = new Map(); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index c2f7eb767945d..097a68c863788 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -605,6 +605,9 @@ export const layerWithOptions = ( }); yield* mcpSessions.set({ ...credential.config, + browserAccessAllowed: agentAccessSettings(threadId).pipe( + Effect.map((access) => access.browser), + ), hasPendingWake: projectionStore .getThreadRecords(threadId, ["runs", "subagents"]) .pipe( diff --git a/apps/server/src/provider/CodexBrowserAccessHost.test.ts b/apps/server/src/provider/CodexBrowserAccessHost.test.ts new file mode 100644 index 0000000000000..d9f5f5728a587 --- /dev/null +++ b/apps/server/src/provider/CodexBrowserAccessHost.test.ts @@ -0,0 +1,30 @@ +// @effect-diagnostics nodeBuiltinImport:off -- Checks the real private socket directory and teardown. +import * as NodeFSP from "node:fs/promises"; +import * as NodePath from "node:path"; +import { assert, describe, it } from "@effect/vitest"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Effect from "effect/Effect"; +import { makeCodexBrowserAccessHost } from "./CodexBrowserAccessHost.ts"; + +describe("CodexBrowserAccessHost", () => { + it.effect("keeps its socket in an owner-only directory that closing removes", () => + Effect.gen(function* () { + if ((yield* HostProcess.Platform) === "win32") return; + const directory = yield* Effect.scoped( + Effect.gen(function* () { + const environment = yield* makeCodexBrowserAccessHost(() => Effect.succeed(true)); + const directory = NodePath.dirname(environment.BROWSER_USE_ACCESS_PIPE_PATH); + assert.equal((yield* Effect.promise(() => NodeFSP.stat(directory))).mode & 0o777, 0o700); + return directory; + }), + ); + const exists = yield* Effect.promise(() => + NodeFSP.access(directory).then( + () => true, + () => false, + ), + ); + assert.isFalse(exists); + }), + ); +}); diff --git a/apps/server/src/provider/CodexBrowserAccessHost.ts b/apps/server/src/provider/CodexBrowserAccessHost.ts new file mode 100644 index 0000000000000..0c1e5f47b43c7 --- /dev/null +++ b/apps/server/src/provider/CodexBrowserAccessHost.ts @@ -0,0 +1,168 @@ +// @effect-diagnostics nodeBuiltinImport:off - Private signed IPC answering the OpenAI browser plugin's access check. +import * as NodeCrypto from "node:crypto"; +import * as NodeFSP from "node:fs/promises"; +import * as NodeNet from "node:net"; +import * as NodeOS from "node:os"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Effect from "effect/Effect"; +import * as FiberSet from "effect/FiberSet"; +import * as Schema from "effect/Schema"; +import { acquirePrivateIpcEndpoint } from "./privateIpcEndpoint.ts"; + +// The plugin sends one request of about 250 bytes per turn and needs one reply. +const MAX_REQUEST_BYTES = 1024; +const MAX_CONNECTIONS = 32; +const CONNECTION_TIMEOUT = "5 seconds"; + +const AccessRequest = Schema.Struct({ + executorId: Schema.String.check(Schema.isMaxLength(128)), + sessionId: Schema.NonEmptyString.check(Schema.isMaxLength(128)), + turnId: Schema.NonEmptyString.check(Schema.isMaxLength(128)), + nonce: Schema.String.check(Schema.isPattern(/^[a-f0-9]{64}$/u)), +}); +const decodeRequest = Schema.decodeUnknownEffect( + Schema.fromJsonString( + Schema.Struct({ + jsonrpc: Schema.Literal("2.0"), + id: Schema.Int, + method: Schema.Literal("browser/access"), + params: AccessRequest, + }), + ), + { onExcessProperty: "error" }, +); +const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); + +class BrowserAccessHostError extends Schema.TaggedError()( + "BrowserAccessHostError", + { cause: Schema.Defect() }, +) { + override get message() { + return "The private browser access host failed."; + } +} + +// The plugin's native pipe frames are a native-endian uint32 length, then UTF-8 JSON. +const readLength = (bytes: Buffer) => + NodeOS.endianness() === "LE" ? bytes.readUInt32LE() : bytes.readUInt32BE(); + +const readFrame = (socket: NodeNet.Socket) => + Effect.callback((resume) => { + let bytes = Buffer.alloc(0); + const fail = () => resume(Effect.fail(new BrowserAccessHostError({ cause: "Invalid frame" }))); + const onData = (chunk: Buffer) => { + bytes = Buffer.concat([bytes, chunk]); + if (bytes.length < 4) return; + const size = readLength(bytes); + if (size > MAX_REQUEST_BYTES || bytes.length > size + 4) return fail(); + if (bytes.length === size + 4) resume(Effect.succeed(bytes.subarray(4).toString("utf8"))); + }; + socket.on("data", onData); + socket.once("error", fail); + socket.once("end", fail); + socket.once("close", fail); + return Effect.sync(() => { + socket.off("data", onData); + socket.off("error", fail); + socket.off("end", fail); + socket.off("close", fail); + }); + }); + +const writeFrame = (socket: NodeNet.Socket, json: string) => + Effect.callback((resume) => { + const payload = Buffer.from(json); + const header = Buffer.alloc(4); + if (NodeOS.endianness() === "LE") header.writeUInt32LE(payload.length); + else header.writeUInt32BE(payload.length); + const fail = () => resume(Effect.fail(new BrowserAccessHostError({ cause: "Peer closed" }))); + socket.once("error", fail); + socket.once("close", fail); + socket.end(Buffer.concat([header, payload]), () => resume(Effect.void)); + return Effect.sync(() => { + socket.off("error", fail); + socket.off("close", fail); + }); + }); + +/** + * Serves `browser/access` for one Codex process until the scope closes. Codex forwards the + * returned variables to the browser plugin, which verifies each signed decision. + */ +export const makeCodexBrowserAccessHost = Effect.fn("CodexBrowserAccessHost.make")(function* ( + isAllowed: (request: { + readonly sessionId: string; + readonly turnId: string; + }) => Effect.Effect, +) { + const platform = yield* HostProcess.Platform; + const endpoint = yield* acquirePrivateIpcEndpoint("t3-browser").pipe( + Effect.mapError((cause) => new BrowserAccessHostError({ cause })), + ); + const { publicKey, privateKey } = yield* Effect.try({ + try: () => NodeCrypto.generateKeyPairSync("ed25519"), + catch: (cause) => new BrowserAccessHostError({ cause }), + }); + const executorId = NodeCrypto.randomUUID(); + const connections = yield* FiberSet.make(); + const run = yield* FiberSet.runtime(connections)(); + const sockets = new Set(); + let closing = false; + const server = NodeNet.createServer((socket) => { + socket.on("error", () => {}); + if (closing || sockets.size >= MAX_CONNECTIONS) { + socket.destroy(); + return; + } + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); + run( + Effect.gen(function* () { + // Windows pipes admit read-only peers; reply only to a complete, valid request. + const { id, params } = yield* readFrame(socket).pipe(Effect.flatMap(decodeRequest)); + const allowed = params.executorId === executorId && (yield* isAllowed(params)); + const payload = encodeJson({ domain: "codex.browserAccess.v1", request: params, allowed }); + const signature = NodeCrypto.sign(null, Buffer.from(payload), privateKey); + yield* writeFrame( + socket, + encodeJson({ + jsonrpc: "2.0", + id, + result: { payload, signature: signature.toString("base64") }, + }), + ); + }).pipe( + Effect.timeout(CONNECTION_TIMEOUT), + Effect.ignoreCause, + Effect.ensuring(Effect.sync(() => socket.destroy())), + ), + ); + }); + server.on("error", () => {}); + yield* Effect.addFinalizer(() => + Effect.gen(function* () { + closing = true; + for (const socket of sockets) socket.destroy(); + yield* Effect.promise(() => new Promise((resolve) => server.close(() => resolve()))); + }), + ); + yield* Effect.callback((resume) => { + const onError = (cause: Error) => resume(Effect.fail(new BrowserAccessHostError({ cause }))); + server.once("error", onError); + server.listen(endpoint, () => resume(Effect.void)); + return Effect.sync(() => server.off("error", onError)); + }); + if (platform !== "win32") { + yield* Effect.tryPromise({ + try: () => NodeFSP.chmod(endpoint, 0o600), + catch: (cause) => new BrowserAccessHostError({ cause }), + }); + } + return { + BROWSER_USE_ACCESS_EXECUTOR_ID: executorId, + BROWSER_USE_ACCESS_PIPE_PATH: endpoint, + BROWSER_USE_ACCESS_VERIFYING_IDENTITY: publicKey + .export({ type: "spki", format: "der" }) + .toString("base64"), + }; +}); diff --git a/apps/server/src/provider/CodexManagedTokenBroker.ts b/apps/server/src/provider/CodexManagedTokenBroker.ts index 4a1e861827142..151f5b58eef48 100644 --- a/apps/server/src/provider/CodexManagedTokenBroker.ts +++ b/apps/server/src/provider/CodexManagedTokenBroker.ts @@ -1,9 +1,6 @@ // @effect-diagnostics nodeBuiltinImport:off - Server-owned private pipe/Unix socket, never a public listener. import * as NodeFSP from "node:fs/promises"; import * as NodeNet from "node:net"; -import * as NodeOS from "node:os"; -import * as NodePath from "node:path"; -import * as NodeCrypto from "node:crypto"; import { EnvironmentId, ProviderInstanceId } from "@t3tools/contracts"; import * as HostProcess from "@t3tools/shared/HostProcess"; import * as Context from "effect/Context"; @@ -16,11 +13,10 @@ import * as Schema from "effect/Schema"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import { makeCodexChatGptAuth } from "./CodexChatGptAuth.ts"; import * as Protocol from "./CodexManagedTokenProtocol.ts"; +import { acquirePrivateIpcEndpoint } from "./privateIpcEndpoint.ts"; const MAX_CONNECTIONS = 32; const MAX_JOBS = 16; -// Stay below macOS's 104-byte and Linux's 108-byte sockaddr_un limits, including NUL. -const UNIX_SOCKET_PATH_BYTES = 100; const encodeResponse = Schema.encodeSync(Schema.fromJsonString(Protocol.Response)); class BrokerStartError extends Schema.TaggedError()( "ManagedTokenBrokerStartError", @@ -46,59 +42,8 @@ const acquireListener = Effect.gen(function* () { const identity = yield* ServerEnvironment.ServerEnvironmentIdentity; const environmentId = yield* identity.getEnvironmentId; const platform = yield* HostProcess.Platform; - const uid = yield* HostProcess.UserId; - const owned = yield* Effect.acquireRelease( - io(async () => { - if (platform === "win32") - return { endpoint: `\\\\.\\pipe\\t3-token-${NodeCrypto.randomUUID()}` }; - if (uid === undefined) throw new Error("Missing IPC owner"); - const acquireDirectory = async (candidate: string) => { - const parent = await NodeFSP.realpath(candidate); - const stat = await NodeFSP.lstat(parent); - if ( - !stat.isDirectory() || - stat.isSymbolicLink() || - !( - (stat.uid === uid && (stat.mode & 0o022) === 0) || - (stat.uid === 0 && ((stat.mode & 0o022) === 0 || (stat.mode & 0o1000) !== 0)) - ) - ) - throw new Error("Untrusted IPC parent"); - const directory = await NodeFSP.mkdtemp(NodePath.join(parent, "t3-token-")); - try { - await NodeFSP.chmod(directory, 0o700); - const created = await NodeFSP.lstat(directory); - if ( - !created.isDirectory() || - created.isSymbolicLink() || - created.uid !== uid || - (created.mode & 0o777) !== 0o700 - ) - throw new Error("Unprotected IPC directory"); - const endpoint = NodePath.join(directory, "s"); - if (Buffer.byteLength(endpoint) > UNIX_SOCKET_PATH_BYTES) - throw new Error("IPC path too long"); - return { endpoint, directory, created }; - } catch (error) { - await NodeFSP.rmdir(directory); - throw error; - } - }; - return acquireDirectory(NodeOS.tmpdir()).catch(() => acquireDirectory("/tmp")); - }), - (owned) => - Effect.promise(async () => { - if (!("directory" in owned)) return; - // Never recursively remove a replaced directory or another process's files. - const current = await NodeFSP.lstat(owned.directory).catch(() => undefined); - if ( - current && - current.dev === owned.created.dev && - current.ino === owned.created.ino && - !current.isSymbolicLink() - ) - await NodeFSP.rmdir(owned.directory).catch(() => undefined); - }), + const endpoint = yield* acquirePrivateIpcEndpoint("t3-token").pipe( + Effect.mapError((cause) => new BrokerStartError({ cause })), ); const jobs = yield* FiberSet.make(); const runJob = @@ -198,14 +143,14 @@ const acquireListener = Effect.gen(function* () { const onError = (cause: Error) => resume(Effect.fail(new BrokerStartError({ cause }))); server.once("error", onError); try { - server.listen(owned.endpoint, () => resume(Effect.void)); + server.listen(endpoint, () => resume(Effect.void)); } catch (cause) { resume(Effect.fail(new BrokerStartError({ cause }))); } return Effect.sync(() => server.off("error", onError)); }); - if (platform !== "win32") yield* io(() => NodeFSP.chmod(owned.endpoint, 0o600)); - return { endpoint: owned.endpoint, awaitIdle: FiberSet.awaitEmpty(jobs) }; + if (platform !== "win32") yield* io(() => NodeFSP.chmod(endpoint, 0o600)); + return { endpoint, awaitIdle: FiberSet.awaitEmpty(jobs) }; }); const make = Effect.gen(function* () { const scope = yield* Scope.fork(yield* Effect.scope); diff --git a/apps/server/src/provider/privateIpcEndpoint.ts b/apps/server/src/provider/privateIpcEndpoint.ts new file mode 100644 index 0000000000000..a60161f02c500 --- /dev/null +++ b/apps/server/src/provider/privateIpcEndpoint.ts @@ -0,0 +1,85 @@ +// @effect-diagnostics nodeBuiltinImport:off - Owner-only pipe/Unix socket paths for server-owned listeners. +import * as NodeCrypto from "node:crypto"; +import * as NodeFSP from "node:fs/promises"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +// Stay below macOS's 104-byte and Linux's 108-byte sockaddr_un limits, including NUL. +const UNIX_SOCKET_PATH_BYTES = 100; + +class PrivateIpcEndpointError extends Schema.TaggedError()( + "PrivateIpcEndpointError", + { cause: Schema.Defect() }, +) { + override get message() { + return "Could not reserve a private IPC endpoint."; + } +} + +/** Reserves a random pipe name, or a socket path inside a fresh 0700 directory removed on release. */ +export const acquirePrivateIpcEndpoint = Effect.fn("PrivateIpcEndpoint.acquire")(function* ( + prefix: string, +) { + const platform = yield* HostProcess.Platform; + const uid = yield* HostProcess.UserId; + const owned = yield* Effect.acquireRelease( + Effect.tryPromise({ + try: async () => { + if (platform === "win32") + return { endpoint: `\\\\.\\pipe\\${prefix}-${NodeCrypto.randomUUID()}` }; + if (uid === undefined) throw new Error("Missing IPC owner"); + const acquireDirectory = async (candidate: string) => { + const parent = await NodeFSP.realpath(candidate); + const stat = await NodeFSP.lstat(parent); + if ( + !stat.isDirectory() || + stat.isSymbolicLink() || + !( + (stat.uid === uid && (stat.mode & 0o022) === 0) || + (stat.uid === 0 && ((stat.mode & 0o022) === 0 || (stat.mode & 0o1000) !== 0)) + ) + ) + throw new Error("Untrusted IPC parent"); + const directory = await NodeFSP.mkdtemp(NodePath.join(parent, `${prefix}-`)); + try { + await NodeFSP.chmod(directory, 0o700); + const created = await NodeFSP.lstat(directory); + if ( + !created.isDirectory() || + created.isSymbolicLink() || + created.uid !== uid || + (created.mode & 0o777) !== 0o700 + ) + throw new Error("Unprotected IPC directory"); + const endpoint = NodePath.join(directory, "s"); + if (Buffer.byteLength(endpoint) > UNIX_SOCKET_PATH_BYTES) + throw new Error("IPC path too long"); + return { endpoint, directory, created }; + } catch (error) { + await NodeFSP.rmdir(directory); + throw error; + } + }; + return acquireDirectory(NodeOS.tmpdir()).catch(() => acquireDirectory("/tmp")); + }, + catch: (cause) => new PrivateIpcEndpointError({ cause }), + }), + (owned) => + Effect.promise(async () => { + if (!("directory" in owned)) return; + // Never recursively remove a replaced directory or another process's files. + const current = await NodeFSP.lstat(owned.directory).catch(() => undefined); + if ( + current && + current.dev === owned.created.dev && + current.ino === owned.created.ino && + !current.isSymbolicLink() + ) + await NodeFSP.rmdir(owned.directory).catch(() => undefined); + }), + ); + return owned.endpoint; +}); diff --git a/apps/server/src/provider/testFixtures/browserAccessClient.ts b/apps/server/src/provider/testFixtures/browserAccessClient.ts new file mode 100644 index 0000000000000..86f0fa143f9d5 --- /dev/null +++ b/apps/server/src/provider/testFixtures/browserAccessClient.ts @@ -0,0 +1,88 @@ +// @effect-diagnostics nodeBuiltinImport:off -- Speaks the browser plugin's native pipe framing in tests. +import * as NodeCrypto from "node:crypto"; +import * as NodeNet from "node:net"; +import * as NodeOS from "node:os"; +import * as Effect from "effect/Effect"; + +const header = (length: number) => { + const bytes = Buffer.alloc(4); + if (NodeOS.endianness() === "LE") bytes.writeUInt32LE(length); + else bytes.writeUInt32BE(length); + return bytes; +}; + +/** + * Asks the host like the browser plugin does and returns its decision, after checking the + * signature against the published identity. Undefined when the host hangs up. + */ +export const requestBrowserAccess = ( + environment: NodeJS.ProcessEnv, + request: { readonly sessionId: string; readonly turnId: string; readonly executorId?: string }, +) => { + const params = { + executorId: request.executorId ?? environment.BROWSER_USE_ACCESS_EXECUTOR_ID, + sessionId: request.sessionId, + turnId: request.turnId, + nonce: "a".repeat(64), + }; + const body = Buffer.from( + JSON.stringify({ jsonrpc: "2.0", method: "browser/access", params, id: 1 }), + ); + return Effect.promise( + () => + new Promise< + | { + readonly jsonrpc: unknown; + readonly id: unknown; + readonly result: { readonly payload: string; readonly signature: string }; + } + | undefined + >((resolve) => { + const frame = Buffer.concat([header(body.length), body]); + const socket = NodeNet.createConnection( + environment.BROWSER_USE_ACCESS_PIPE_PATH ?? "", + () => { + // Split the length prefix so the host must reassemble partial frames. + socket.write(frame.subarray(0, 3)); + socket.write(frame.subarray(3)); + }, + ); + let bytes = Buffer.alloc(0); + socket.on("data", (chunk: Buffer) => { + bytes = Buffer.concat([bytes, chunk]); + }); + socket.on("error", () => resolve(undefined)); + socket.on("close", () => { + if (bytes.length < 4) return resolve(undefined); + const size = NodeOS.endianness() === "LE" ? bytes.readUInt32LE() : bytes.readUInt32BE(); + resolve(JSON.parse(bytes.subarray(4, 4 + size).toString("utf8"))); + }); + }), + ).pipe( + Effect.map((envelope) => { + if (envelope === undefined) return undefined; + if (envelope.jsonrpc !== "2.0" || envelope.id !== 1) + throw new Error("Browser access reply does not answer the request"); + const response = envelope.result; + const key = NodeCrypto.createPublicKey({ + key: Buffer.from(environment.BROWSER_USE_ACCESS_VERIFYING_IDENTITY ?? "", "base64"), + type: "spki", + format: "der", + }); + const signed = NodeCrypto.verify( + null, + Buffer.from(response.payload), + key, + Buffer.from(response.signature, "base64"), + ); + const decision = JSON.parse(response.payload); + if ( + !signed || + decision.domain !== "codex.browserAccess.v1" || + JSON.stringify(decision.request) !== JSON.stringify(params) + ) + throw new Error("Browser access decision failed verification"); + return decision.allowed as boolean; + }), + ); +}; diff --git a/packages/provider-core/src/server/mcpSession.ts b/packages/provider-core/src/server/mcpSession.ts index 20d17df610fd7..43d713781e9f4 100644 --- a/packages/provider-core/src/server/mcpSession.ts +++ b/packages/provider-core/src/server/mcpSession.ts @@ -16,6 +16,8 @@ export interface McpProviderSessionConfig { readonly authorizationHeader: string; /** Omit preview_* instructions when the user withholds agent browser access. */ readonly browserToolsAvailable: boolean; + /** Recheck permission because a provider session can outlive a settings change. */ + readonly browserAccessAllowed?: Effect.Effect; readonly capabilities?: ReadonlySet; readonly agentDeviceEnvironment?: Readonly>; /** True while a delegated task this thread started will wake it with its result. */