diff --git a/docs/hub.md b/docs/hub.md index 86cc06b07e..029083ff26 100644 --- a/docs/hub.md +++ b/docs/hub.md @@ -15,11 +15,9 @@ The daemon persists a relationship ID and private connection credential before e relationship is independent of its current transport, so a future transport can replace the direct WebSocket without pairing again. The current foundation supports one Hub relationship per daemon. -Only a trusted session arriving through a server-observed local, non-browser daemon transport may -run the `hub.relationship.connect`, `hub.relationship.get_status`, and -`hub.relationship.disconnect` RPCs. Remote, relay, browser-origin, and Hub sessions cannot manage -the relationship. The CLI also rejects an explicit non-loopback connect target before sending the -one-time token. A hello message's `clientType` is descriptive and grants no authority. +Normal authenticated daemon sessions may run the `hub.relationship.connect`, +`hub.relationship.get_status`, and `hub.relationship.disconnect` RPCs regardless of transport. +The restricted Hub session cannot manage the relationship. ## Hub session scope diff --git a/packages/cli/src/commands/hub/index.test.ts b/packages/cli/src/commands/hub/index.test.ts deleted file mode 100644 index e6d3088b17..0000000000 --- a/packages/cli/src/commands/hub/index.test.ts +++ /dev/null @@ -1,30 +0,0 @@ -import { describe, expect, test } from "vitest"; -import { assertLocalHubManagementTarget } from "./index.js"; - -describe("Hub CLI target authority", () => { - test.each(["relay.paseo.test:443", "tcp://10.0.0.5:6767?ssl=true"])( - "rejects non-local explicit target %s before connect", - (host) => { - expect(() => assertLocalHubManagementTarget(host, {})).toThrow( - "Hub relationship management requires a local daemon target", - ); - }, - ); - - test.each([ - undefined, - "localhost:6767", - "127.0.0.1:6767", - "[::1]:6767", - "[0:0:0:0:0:0:0:1]:6767", - "unix:///tmp/paseo.sock", - ])("accepts local target %s", (host) => { - expect(() => assertLocalHubManagementTarget(host, {})).not.toThrow(); - }); - - test("rejects a non-local PASEO_HOST target", () => { - expect(() => - assertLocalHubManagementTarget(undefined, { PASEO_HOST: "hub.test:6767" }), - ).toThrow("Hub relationship management requires a local daemon target"); - }); -}); diff --git a/packages/cli/src/commands/hub/index.ts b/packages/cli/src/commands/hub/index.ts index 2831393437..81872bf5f8 100644 --- a/packages/cli/src/commands/hub/index.ts +++ b/packages/cli/src/commands/hub/index.ts @@ -1,7 +1,7 @@ import { Command } from "commander"; import { withOutput, type ListResult, type OutputSchema } from "../../output/index.js"; import { addJsonAndDaemonHostOptions } from "../../utils/command-options.js"; -import { connectToDaemon, resolveDaemonTarget } from "../../utils/client.js"; +import { connectToDaemon } from "../../utils/client.js"; interface HubRow { state: string; @@ -66,26 +66,6 @@ async function withClient( } } -export function assertLocalHubManagementTarget( - host: string | undefined, - env: NodeJS.ProcessEnv = process.env, -): void { - const explicitHost = host ?? env.PASEO_HOST; - if (!explicitHost) return; - const target = resolveDaemonTarget(explicitHost); - if (target.type === "ipc") return; - const hostname = new URL(target.url).hostname.replace(/^\[|\]$/g, "").toLowerCase(); - if ( - hostname === "localhost" || - hostname === "::1" || - hostname === "0:0:0:0:0:0:0:1" || - /^127(?:\.\d{1,3}){3}$/.test(hostname) - ) { - return; - } - throw new Error("Hub relationship management requires a local daemon target"); -} - export function createHubCommand(): Command { const hub = new Command("hub").description("Manage this daemon's Paseo Hub relationship"); addJsonAndDaemonHostOptions( @@ -94,7 +74,6 @@ export function createHubCommand(): Command { withOutput(async (...args) => { const url = args[0] as string; const options = args.at(-2) as { token: string; host?: string }; - assertLocalHubManagementTarget(options.host); return withClient(options.host, async (client) => result((await client.connectHub(url, options.token)).status), ); diff --git a/packages/server/src/server/hub/hub-session.test.ts b/packages/server/src/server/hub/hub-session.test.ts index 6a0981f09b..3ad2ac08a4 100644 --- a/packages/server/src/server/hub/hub-session.test.ts +++ b/packages/server/src/server/hub/hub-session.test.ts @@ -17,6 +17,7 @@ class InMemoryHubExecutions implements HubExecutions { private readonly listeners = new Set<(event: OwnedAgentEvent) => void>(); private reconcileError: Error | null = null; private creates = 0; + private reconciles = 0; async create(_input: HubAgentCreateInput): Promise { this.creates++; @@ -24,6 +25,7 @@ class InMemoryHubExecutions implements HubExecutions { } async reconcile(_executionId: string): Promise { + this.reconciles++; if (this.reconcileError) throw this.reconcileError; return { executionId: "execution-1", agent: createAgentSnapshot() }; } @@ -48,6 +50,16 @@ class InMemoryHubExecutions implements HubExecutions { createCount(): number { return this.creates; } + + reconcileCount(): number { + return this.reconciles; + } +} + +interface HubCreateInput { + executionId?: string; + prompt: string; + cwd: string; } class HubSessionBoundary { @@ -75,11 +87,11 @@ class HubSessionBoundary { }); } - async createWith(input: { prompt: string; cwd: string }): Promise { + async createWith(input: HubCreateInput): Promise { await this.session.handleMessage({ type: "hub.agent.create.request", requestId: "create-1", - executionId: "execution-1", + executionId: input.executionId ?? "execution-1", provider: "codex", workspaceId: "workspace-1", ...input, @@ -90,6 +102,18 @@ class HubSessionBoundary { return this.executions.createCount(); } + async reconcileWith(executionId: string): Promise { + await this.session.handleMessage({ + type: "hub.execution.reconcile.request", + requestId: "reconcile-1", + executionId, + }); + } + + reconcileCount(): number { + return this.executions.reconcileCount(); + } + deliveredMessages(): SessionOutboundMessage[] { return this.messages.slice(); } @@ -151,6 +175,59 @@ test.each([ ]); }); +test("blank Hub execution IDs reject create and reconcile before execution ownership", async () => { + const hub = new HubSessionBoundary(); + + await hub.createWith({ executionId: " ", prompt: "Do the work", cwd: "/workspace" }); + await hub.reconcileWith("\t"); + + expect(hub.createCount()).toBe(0); + expect(hub.reconcileCount()).toBe(0); + expect(hub.deliveredMessages()).toEqual([ + { + type: "hub.agent.create.response", + payload: { + requestId: "create-1", + executionId: " ", + agentId: null, + agent: null, + success: false, + error: "Hub agent executionId cannot be blank", + }, + }, + { + type: "hub.execution.reconcile.response", + payload: { + requestId: "reconcile-1", + executionId: "\t", + agentId: null, + agent: null, + }, + }, + ]); +}); + +test("relative Hub working directories reject create before execution ownership", async () => { + const hub = new HubSessionBoundary(); + + await hub.createWith({ prompt: "Do the work", cwd: "repo" }); + + expect(hub.createCount()).toBe(0); + expect(hub.deliveredMessages()).toEqual([ + { + type: "hub.agent.create.response", + payload: { + requestId: "create-1", + executionId: "execution-1", + agentId: null, + agent: null, + success: false, + error: "Hub agent cwd must be absolute", + }, + }, + ]); +}); + function createAgentSnapshot(): AgentSnapshotPayload { return AgentSnapshotPayloadSchema.parse({ id: "agent-1", diff --git a/packages/server/src/server/hub/hub-session.ts b/packages/server/src/server/hub/hub-session.ts index e9b6b3c633..197d12bc6b 100644 --- a/packages/server/src/server/hub/hub-session.ts +++ b/packages/server/src/server/hub/hub-session.ts @@ -1,3 +1,4 @@ +import { isAbsolute } from "node:path"; import type { HubAgentCreateRequest, HubExecutionReconcileRequest, @@ -49,8 +50,10 @@ export class HubSession { private async createAgent(message: HubAgentCreateRequest): Promise { try { + requireNonBlankHubAgentField("executionId", message.executionId); requireNonBlankHubAgentField("prompt", message.prompt); requireNonBlankHubAgentField("cwd", message.cwd); + if (!isAbsolute(message.cwd)) throw new Error("Hub agent cwd must be absolute"); const result = await this.executions.create({ executionId: message.executionId, provider: message.provider, @@ -94,6 +97,7 @@ export class HubSession { private async reconcile(message: HubExecutionReconcileRequest): Promise { let result: Awaited> = null; try { + requireNonBlankHubAgentField("executionId", message.executionId); result = await this.executions.reconcile(message.executionId); } catch { // Reconcile has no error variant on the wire. A terminal empty response @@ -133,7 +137,10 @@ export class HubSession { } } -function requireNonBlankHubAgentField(field: "prompt" | "cwd", value: string): void { +function requireNonBlankHubAgentField( + field: "executionId" | "prompt" | "cwd", + value: string, +): void { if (value.trim().length === 0) { throw new Error(`Hub agent ${field} cannot be blank`); } diff --git a/packages/server/src/server/hub/relationship-controller.test.ts b/packages/server/src/server/hub/relationship-controller.test.ts index d8a8d9e13b..6c499cb664 100644 --- a/packages/server/src/server/hub/relationship-controller.test.ts +++ b/packages/server/src/server/hub/relationship-controller.test.ts @@ -54,60 +54,28 @@ describe("Hub relationship", () => { expect(relationship.enrollmentAttempts()).toEqual([]); }); - test("an external socket cannot manage Hub relationships even when it claims to be the CLI", async () => { + test("an authenticated external socket can manage the Hub relationship", async () => { relationship = await HubRelationshipHarness.start(); - const denials = await relationship.attemptExternalManagementAsCli(); + const responses = await relationship.manageRelationshipFromExternalSocket(); - expect(denials).toEqual([ - { - type: "rpc_error", - payload: { - requestId: "external-hub-connect", - requestType: "hub.relationship.connect.request", - error: "Hub relationship management requires a local daemon connection", - code: "local_management_required", - }, - }, - { - type: "rpc_error", - payload: { - requestId: "external-hub-status", - requestType: "hub.relationship.get_status.request", - error: "Hub relationship management requires a local daemon connection", - code: "local_management_required", - }, - }, - { - type: "rpc_error", - payload: { - requestId: "external-hub-disconnect", - requestType: "hub.relationship.disconnect.request", - error: "Hub relationship management requires a local daemon connection", - code: "local_management_required", - }, - }, + expect(responses.map((response) => response.type)).toEqual([ + "hub.relationship.connect.response", + "hub.relationship.get_status.response", + "hub.relationship.disconnect.response", ]); - expect(relationship.enrollmentAttempts()).toEqual([]); + expect(relationship.enrollmentAttempts()).toHaveLength(1); expect(relationship.relationshipFile()).toBeNull(); }); - test("a loopback browser socket cannot manage Hub relationships by claiming to be the CLI", async () => { + test("an authenticated browser socket can manage the Hub relationship", async () => { relationship = await HubRelationshipHarness.start(); - const denial = await relationship.attemptLoopbackBrowserConnectAsCli(); + const response = await relationship.connectFromBrowserSocket(); - expect(denial).toEqual({ - type: "rpc_error", - payload: { - requestId: "browser-hub-connect", - requestType: "hub.relationship.connect.request", - error: "Hub relationship management requires a local daemon connection", - code: "local_management_required", - }, - }); - expect(relationship.enrollmentAttempts()).toEqual([]); - expect(relationship.relationshipFile()).toBeNull(); + expect(response.type).toBe("hub.relationship.connect.response"); + expect(relationship.enrollmentAttempts()).toHaveLength(1); + expect(relationship.relationshipFile()?.state).toBe("active"); }); test("persists private generated authority before enrollment and active before dialing", async () => { diff --git a/packages/server/src/server/hub/relationship-remote.test.ts b/packages/server/src/server/hub/relationship-remote.test.ts index af1c03639e..dc0e0cc1f6 100644 --- a/packages/server/src/server/hub/relationship-remote.test.ts +++ b/packages/server/src/server/hub/relationship-remote.test.ts @@ -197,6 +197,30 @@ test("a transient socket response releases the failed upgrade and reports one co expect(hub.openConnections()).toBe(0); }); +test("a stalled socket opening handshake times out and releases the connection", async () => { + const hub = await UpgradeRejectingHub.start(["stall"]); + const outcomes = new SocketOutcomes(); + const remote = new DirectHubRelationshipRemote({ requestTimeoutMs: 25 }); + + const socket = remote.openSocket( + { + relationshipId: "relationship-1", + webSocketUrl: hub.webSocketUrl, + credential: "credential", + }, + outcomes.events, + ) as HubSocketConnection & WebSocket; + + expect(await outcomes.next()).toEqual({ + type: "failed", + message: "Opening handshake has timed out", + }); + expect(await outcomes.afterCleanup()).toEqual([ + { type: "failed", message: "Opening handshake has timed out" }, + ]); + expect(socket.readyState).toBe(WebSocket.CLOSED); +}); + test.each([401, 403] as const)( "controller treats socket authentication status %s as terminal without redialing", async (status) => { @@ -375,7 +399,7 @@ class UpgradeRejectingHub { private attemptObserved = deferred(); private attempts = 0; - private constructor(private readonly statuses: number[]) { + private constructor(private readonly statuses: Array) { this.server.on("upgrade", (_request, socket) => { const attempt = ++this.attempts; this.attemptObserved.resolve(); @@ -388,6 +412,10 @@ class UpgradeRejectingHub { released.resolve(); }); const status = this.statuses[attempt - 1] ?? this.statuses.at(-1) ?? 503; + if (status === "stall") { + socket.resume(); + return; + } if (status === 0) { socket.destroy(); return; @@ -398,7 +426,7 @@ class UpgradeRejectingHub { }); } - static async start(statuses: number[]): Promise { + static async start(statuses: Array): Promise { const hub = new UpgradeRejectingHub(statuses); openUpgradeHubs.push(hub); openServers.push(hub.server); diff --git a/packages/server/src/server/hub/relationship-remote.ts b/packages/server/src/server/hub/relationship-remote.ts index fcc22f6266..5c109c0fe4 100644 --- a/packages/server/src/server/hub/relationship-remote.ts +++ b/packages/server/src/server/hub/relationship-remote.ts @@ -133,6 +133,7 @@ export class DirectHubRelationshipRemote implements HubRelationshipRemote { openSocket(input: HubSocketCredentials, events: HubSocketEvents): HubSocketConnection { const socket = new WebSocket(input.webSocketUrl, { + handshakeTimeout: this.requestTimeoutMs, headers: { authorization: `Bearer ${input.credential}`, "x-paseo-relationship-id": input.relationshipId, @@ -164,6 +165,7 @@ export class DirectHubRelationshipRemote implements HubRelationshipRemote { socket.once("error", (error) => { if (settled) return; settled = true; + socket.terminate(); events.failed(error); }); return socket; diff --git a/packages/server/src/server/hub/test-utils/relationship-harness.ts b/packages/server/src/server/hub/test-utils/relationship-harness.ts index b7d3b3f8d1..6b439335d1 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -512,7 +512,7 @@ export class HubRelationshipHarness { } } - async attemptExternalManagementAsCli(): Promise { + async manageRelationshipFromExternalSocket(): Promise { const address = Object.values(networkInterfaces()) .flat() .find((candidate) => candidate?.family === "IPv4" && !candidate.internal)?.address; @@ -544,7 +544,7 @@ export class HubRelationshipHarness { return responses; } - async attemptLoopbackBrowserConnectAsCli(): Promise { + async connectFromBrowserSocket(): Promise { const target = this.daemon?.getListenTarget(); if (!target || target.type !== "tcp") throw new Error("Daemon did not bind TCP"); const socket = await this.openClaimedCliSocket(`ws://127.0.0.1:${target.port}/ws`, { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 06206cf3c2..e504a163f4 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -395,15 +395,6 @@ export interface SessionFileSystem { isDirectory(path: string): Promise; } -export type SessionPeer = - | { - transport: "direct"; - peer: "loopback" | "local_ipc" | "external"; - browserOrigin: boolean; - } - | { transport: "relay"; peer: "external" } - | { transport: "internal"; peer: "internal" }; - const nodeSessionFileSystem: SessionFileSystem = { async isDirectory(path) { const stats = await stat(path).catch(() => null); @@ -515,34 +506,6 @@ function parseClientCapabilities( return new Set(result); } -type HubRelationshipManagementRequest = Extract< - SessionInboundMessage, - { - type: - | "hub.relationship.connect.request" - | "hub.relationship.get_status.request" - | "hub.relationship.disconnect.request"; - } ->; - -function isHubRelationshipManagementRequest( - message: SessionInboundMessage, -): message is HubRelationshipManagementRequest { - return ( - message.type === "hub.relationship.connect.request" || - message.type === "hub.relationship.get_status.request" || - message.type === "hub.relationship.disconnect.request" - ); -} - -function canManageHubRelationships(peer: SessionPeer): boolean { - return ( - peer.transport === "direct" && - !peer.browserOrigin && - (peer.peer === "loopback" || peer.peer === "local_ipc") - ); -} - interface AgentTimelineProjectionSelection { timeline: AgentTimelineFetchResult; entries: TimelineProjectionEntry[]; @@ -1386,10 +1349,7 @@ export class Session { /** * Main entry point for processing session messages */ - public async handleMessage( - msg: SessionInboundMessage, - peer: SessionPeer = { transport: "internal", peer: "internal" }, - ): Promise { + public async handleMessage(msg: SessionInboundMessage): Promise { this.inflightRequests++; if (this.inflightRequests > this.peakInflightRequests) { this.peakInflightRequests = this.inflightRequests; @@ -1403,7 +1363,7 @@ export class Session { "agent.session.inbound", ); try { - await this.dispatchInboundMessage(msg, peer); + await this.dispatchInboundMessage(msg); } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); this.sessionLogger.error({ err }, "Error handling message"); @@ -1441,22 +1401,7 @@ export class Session { } } - private async dispatchInboundMessage( - msg: SessionInboundMessage, - peer: SessionPeer, - ): Promise { - if (isHubRelationshipManagementRequest(msg) && !canManageHubRelationships(peer)) { - this.emit({ - type: "rpc_error", - payload: { - requestId: msg.requestId, - requestType: msg.type, - error: "Hub relationship management requires a local daemon connection", - code: "local_management_required", - }, - }); - return; - } + private async dispatchInboundMessage(msg: SessionInboundMessage): Promise { const promise = this.dispatchVoiceAndControlMessage(msg) ?? this.dispatchAgentRewindMessage(msg) ?? diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 5bcceb22a8..2d6d01bb12 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -30,12 +30,7 @@ import { asUint8Array, decodeBinaryFrame } from "@getpaseo/protocol/binary-frame import type { TerminalActivity } from "@getpaseo/protocol/terminal-activity"; import type { HostnamesConfig } from "./hostnames.js"; import { isHostnameAllowed } from "./hostnames.js"; -import { - Session, - type SessionLifecycleIntent, - type SessionPeer, - type SessionRuntimeMetrics, -} from "./session.js"; +import { Session, type SessionLifecycleIntent, type SessionRuntimeMetrics } from "./session.js"; import { HubSession } from "./hub/hub-session.js"; import type { HubRelationshipManagement } from "./hub/relationship-controller.js"; import type { HubExecutions } from "./hub/relationship-owned-executions.js"; @@ -1849,11 +1844,7 @@ export class VoiceAssistantWebSocketServer { } const startMs = performance.now(); - const identity = this.socketIdentities.get(ws); - const peer = identity - ? toSessionPeer(identity) - : ({ transport: "internal", peer: "internal" } satisfies SessionPeer); - await activeConnection.session.handleMessage(message.message, peer); + await activeConnection.session.handleMessage(message.message); const durationMs = performance.now() - startMs; this.recordRequestLatency(message.message.type, durationMs); @@ -2254,17 +2245,6 @@ function resolveConnectionPeer( return isLoopbackAddress(requestMetadata.remoteAddress) ? "loopback" : "external"; } -function toSessionPeer(identity: WebSocketConnectionIdentity): SessionPeer { - if (identity.transport === "relay") { - return { transport: "relay", peer: "external" }; - } - return { - transport: "direct", - peer: identity.peer, - browserOrigin: identity.browserOrigin, - }; -} - function isLoopbackAddress(address: string): boolean { const normalized = address.toLowerCase(); if (normalized === "::1" || normalized === "0:0:0:0:0:0:0:1") return true;