diff --git a/docs/hub.md b/docs/hub.md index c0259fd707..001b330fbe 100644 --- a/docs/hub.md +++ b/docs/hub.md @@ -18,7 +18,8 @@ WebSocket without pairing again. The current foundation supports one Hub relatio 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. A hello message's `clientType` is descriptive and grants no authority. +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. ## Hub session scope @@ -57,8 +58,9 @@ scopes, and a sanitized reason for status reporting. `paseo hub disconnect` disables socket reconnect before requesting remote revocation. If the Hub is offline, the daemon persists `disconnecting` and retries revocation across daemon restarts without -opening a Hub socket. `--force` removes local authority immediately and warns that remote revocation -may still be pending. +opening a Hub socket. This also covers an enrollment whose request may have succeeded but whose +response was lost. `--force` removes local authority immediately and warns that remote revocation may +still be pending. ## Cross-repository dependency diff --git a/packages/cli/src/commands/hub/index.test.ts b/packages/cli/src/commands/hub/index.test.ts new file mode 100644 index 0000000000..e6d3088b17 --- /dev/null +++ b/packages/cli/src/commands/hub/index.test.ts @@ -0,0 +1,30 @@ +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 81872bf5f8..2831393437 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 } from "../../utils/client.js"; +import { connectToDaemon, resolveDaemonTarget } from "../../utils/client.js"; interface HubRow { state: string; @@ -66,6 +66,26 @@ 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( @@ -74,6 +94,7 @@ 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/bootstrap.ts b/packages/server/src/server/bootstrap.ts index c8bf8eb5b1..7aaad4041e 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1520,7 +1520,9 @@ export async function createPaseoDaemon( // Freeze both ingress and registration before taking the agent closure snapshot. wsServer?.prepareForShutdown(); agentManager.prepareForShutdown(); - await closeAllAgents(logger, agentManager); + await closeAllAgents(logger, agentManager, (relationshipId) => + hubRelationships.hasReconnectableRelationship(relationshipId), + ); await agentManager.flushForShutdown().catch(() => undefined); detachAgentStoragePersistence(); await agentStorage.flush().catch(() => undefined); @@ -1564,13 +1566,19 @@ export async function createPaseoDaemon( }; } -async function closeAllAgents(logger: Logger, agentManager: AgentManager): Promise { +async function closeAllAgents( + logger: Logger, + agentManager: AgentManager, + hasReconnectableHubRelationship: (relationshipId: string) => boolean, +): Promise { const agents = agentManager.listAgents(); await Promise.all( agents.map(async (agent) => { try { await agentManager.closeAgent(agent.id, { - persistClosedState: agent.owner?.kind !== "hub", + persistClosedState: + agent.owner?.kind !== "hub" || + !hasReconnectableHubRelationship(agent.owner.relationshipId), }); } catch (err) { logger.error({ err, agentId: agent.id }, "Failed to close agent"); diff --git a/packages/server/src/server/hub/relationship-controller.test.ts b/packages/server/src/server/hub/relationship-controller.test.ts index de89cec3f4..7e63efd1e2 100644 --- a/packages/server/src/server/hub/relationship-controller.test.ts +++ b/packages/server/src/server/hub/relationship-controller.test.ts @@ -150,6 +150,48 @@ describe("Hub relationship", () => { expect(relationship.relationshipFile()).toBeNull(); }); + test.each(["{not-json", JSON.stringify({ version: 1, state: "unknown" })])( + "an invalid relationship file is quarantined without blocking daemon startup", + async (contents) => { + relationship = await HubRelationshipHarness.start(); + await relationship.corruptRelationshipFile(contents); + + await relationship.startStoppedDaemon(); + + expect(await relationship.status()).toMatchObject({ state: "not_connected" }); + expect(relationship.relationshipFile()).toBeNull(); + expect(await relationship.quarantinedRelationshipFiles()).toHaveLength(1); + }, + ); + + test("disconnect revokes an ambiguous pending enrollment before removing local authority", async () => { + relationship = await HubRelationshipHarness.start(); + relationship.holdEnrollment(); + const connecting = relationship.beginConnect("one-time-token"); + const enrollment = await relationship.enrollmentBegins(); + const credential = relationship.relationshipFile()?.credential?.secret; + relationship.loseEnrollmentResponse(); + await connecting.result; + relationship.failRevocations(2); + + const disconnected = await relationship.disconnect(); + expect(disconnected.state).toBe("disconnecting"); + expect(relationship.relationshipFile()?.state).toBe("disconnecting"); + + await relationship.restartDaemon(); + await relationship.retry(); + + expect(relationship.revocationAttempts()).toBe(3); + expect(relationship.latestRevocation()).toEqual( + expect.objectContaining({ + relationshipId: enrollment.relationshipId, + hubOrigin: enrollment.hubOrigin, + credential, + }), + ); + expect(relationship.relationshipFile()).toBeNull(); + }); + test("daemon restart reconnects the same durable relationship", async () => { relationship = await HubRelationshipHarness.start(); await relationship.beginConnect().result; @@ -427,4 +469,18 @@ describe("Hub relationship", () => { expect(forced.warning).toContain("remote revocation"); expect(relationship.relationshipFile()).toBeNull(); }); + + test("daemon shutdown closes owned agents after their Hub relationship is removed", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + relationship.connectLatestSocket(); + relationship.beginOwnedCreate("orphan-create", "execution-orphan", { prompt: "sleep 30" }); + const created = await relationship.ownedCreateResult("orphan-create"); + expect(created).toMatchObject({ payload: { agent: { status: "running" } } }); + + await relationship.disconnect(true); + await relationship.restartDaemon(); + + expect(await relationship.storedOwnedStatus(created.payload.agentId!)).toBe("closed"); + }); }); diff --git a/packages/server/src/server/hub/relationship-controller.ts b/packages/server/src/server/hub/relationship-controller.ts index e060f84fad..177e5b1cec 100644 --- a/packages/server/src/server/hub/relationship-controller.ts +++ b/packages/server/src/server/hub/relationship-controller.ts @@ -1,5 +1,5 @@ import { createHash, randomBytes, randomUUID } from "node:crypto"; -import { existsSync, readFileSync, rmSync } from "node:fs"; +import { existsSync, readFileSync, renameSync, rmSync } from "node:fs"; import path from "node:path"; import type pino from "pino"; import { z } from "zod"; @@ -40,11 +40,18 @@ const PendingSchema = z.object({ }); const ActiveSchema = z.object({ version: z.literal(1), - state: z.enum(["active", "disconnecting"]), + state: z.literal("active"), relationship: RelationshipSchema, credential: CredentialSchema, transport: TransportSchema, }); +const DisconnectingSchema = z.object({ + version: z.literal(1), + state: z.literal("disconnecting"), + relationship: RelationshipSchema, + credential: CredentialSchema, + transport: TransportSchema.optional(), +}); const RevokedSchema = z.object({ version: z.literal(1), state: z.literal("revoked"), @@ -52,9 +59,15 @@ const RevokedSchema = z.object({ transport: TransportSchema.optional(), reason: z.string().optional(), }); -const RecordSchema = z.discriminatedUnion("state", [PendingSchema, ActiveSchema, RevokedSchema]); +const RecordSchema = z.discriminatedUnion("state", [ + PendingSchema, + ActiveSchema, + DisconnectingSchema, + RevokedSchema, +]); type PendingRecord = z.infer; type ActiveRecord = z.infer; +type DisconnectingRecord = z.infer; type RevokedRecord = z.infer; type HubRelationshipRecord = z.infer; @@ -190,6 +203,10 @@ export class HubRelationshipController implements HubRelationshipManagement { }; } + hasReconnectableRelationship(relationshipId: string): boolean { + return this.record?.state === "active" && this.record.relationship.id === relationshipId; + } + async connect(input: { hubUrl: string; token: string }): Promise { if (this.record?.state === "pending") { if (normalizeHubUrl(input.hubUrl) !== this.record.relationship.hubOrigin) { @@ -239,11 +256,13 @@ export class HubRelationshipController implements HubRelationshipManagement { warning: "Local Hub credential removed; remote revocation may remain pending.", }; } - if (this.record.state === "pending") { - this.remove(); - return { status: this.status() }; - } - const disconnecting: ActiveRecord = { ...this.record, state: "disconnecting" }; + const disconnecting: DisconnectingRecord = { + version: 1, + state: "disconnecting", + relationship: this.record.relationship, + credential: this.record.credential, + ...(this.record.state === "active" ? { transport: this.record.transport } : {}), + }; this.persist(disconnecting); this.record = disconnecting; this.state = "disconnecting"; @@ -365,7 +384,7 @@ export class HubRelationshipController implements HubRelationshipManagement { this.schedule(() => void this.tryEnrollment(record)); } - private async tryRevocation(record: ActiveRecord): Promise { + private async tryRevocation(record: DisconnectingRecord): Promise { const generation = this.generation; try { await this.options.remote.revoke({ @@ -433,7 +452,22 @@ export class HubRelationshipController implements HubRelationshipManagement { private load(): HubRelationshipRecord | null { if (!existsSync(this.filePath)) return null; - const record = RecordSchema.parse(JSON.parse(readFileSync(this.filePath, "utf8"))); + let record: HubRelationshipRecord; + try { + record = RecordSchema.parse(JSON.parse(readFileSync(this.filePath, "utf8"))); + } catch (error) { + const quarantinePath = path.join( + path.dirname(this.filePath), + `hub-relationship.invalid-${this.clock.now().getTime()}-${randomUUID()}.json`, + ); + renameSync(this.filePath, quarantinePath); + ensurePrivateFile(quarantinePath); + this.options.logger.error( + { err: error, quarantinePath }, + "Quarantined invalid Hub relationship authority", + ); + return null; + } ensurePrivateFile(this.filePath); return record; } diff --git a/packages/server/src/server/hub/relationship-remote.test.ts b/packages/server/src/server/hub/relationship-remote.test.ts index a2be94dc2e..62c715e02e 100644 --- a/packages/server/src/server/hub/relationship-remote.test.ts +++ b/packages/server/src/server/hub/relationship-remote.test.ts @@ -50,6 +50,31 @@ test("transient revocation failures remain retryable", async () => { ).rejects.toThrow("Hub revocation failed (503)"); }); +test.each(["enrollment", "revocation"])("%s HTTP calls are bounded", async (operation) => { + const hubOrigin = await startStalledHub(); + const remote = new DirectHubRelationshipRemote({ requestTimeoutMs: 25 }); + + const request = + operation === "enrollment" + ? remote.enroll({ + relationshipId: "relationship-1", + idempotencyKey: "ceremony-1", + hubOrigin, + token: "token", + serverId: "server-1", + daemonPublicKey: "public-key", + credentialVerifier: "verifier", + scopes: ["hub.*"], + }) + : remote.revoke({ + relationshipId: "relationship-1", + hubOrigin, + credential: "credential", + }); + + await expect(request).rejects.toThrow("Hub request timed out"); +}); + test.each([401, 403] as const)( "socket authentication status %s is terminal and releases the failed upgrade", async (status) => { @@ -164,6 +189,17 @@ async function startHubReturning(status: number): Promise { return `http://127.0.0.1:${address.port}`; } +async function startStalledHub(): Promise { + const server = createServer(() => undefined); + openServers.push(server); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address() as AddressInfo; + return `http://127.0.0.1:${address.port}`; +} + function closeServer(server: ReturnType): Promise { return new Promise((resolve, reject) => { server.close((error) => { diff --git a/packages/server/src/server/hub/relationship-remote.ts b/packages/server/src/server/hub/relationship-remote.ts index 38d430897c..a0f2e07304 100644 --- a/packages/server/src/server/hub/relationship-remote.ts +++ b/packages/server/src/server/hub/relationship-remote.ts @@ -62,39 +62,54 @@ const EnrollmentResultSchema = z.object({ }); export class DirectHubRelationshipRemote implements HubRelationshipRemote { + private readonly requestTimeoutMs: number; + + constructor(options: { requestTimeoutMs?: number } = {}) { + this.requestTimeoutMs = options.requestTimeoutMs ?? 15_000; + } + async enroll(input: HubEnrollment): Promise { - const response = await fetch(`${input.hubOrigin}/api/daemon-relationships/enroll`, { - method: "POST", - headers: { - "content-type": "application/json", - authorization: `Bearer ${input.token}`, - }, - body: JSON.stringify({ - relationshipId: input.relationshipId, - idempotencyKey: input.idempotencyKey, - serverId: input.serverId, - daemonPublicKey: input.daemonPublicKey, - credentialVerifier: input.credentialVerifier, - scopes: input.scopes, - }), - }); - if (!response.ok) { - if (response.status >= 400 && response.status < 500) { - throw new HubEnrollmentRejectedError(response.status); + return this.withRequestTimeout(async (signal) => { + const response = await fetch(`${input.hubOrigin}/api/daemon-relationships/enroll`, { + method: "POST", + headers: { + "content-type": "application/json", + authorization: `Bearer ${input.token}`, + }, + body: JSON.stringify({ + relationshipId: input.relationshipId, + idempotencyKey: input.idempotencyKey, + serverId: input.serverId, + daemonPublicKey: input.daemonPublicKey, + credentialVerifier: input.credentialVerifier, + scopes: input.scopes, + }), + signal, + }); + if (!response.ok) { + if (response.status >= 400 && response.status < 500) { + throw new HubEnrollmentRejectedError(response.status); + } + throw new Error(`Hub enrollment failed (${response.status})`); } - throw new Error(`Hub enrollment failed (${response.status})`); - } - return EnrollmentResultSchema.parse(await response.json()); + return EnrollmentResultSchema.parse(await response.json()); + }); } async revoke(input: HubRevocation): Promise { - const response = await fetch( - `${input.hubOrigin}/api/daemon-relationships/${encodeURIComponent(input.relationshipId)}`, - { method: "DELETE", headers: { authorization: `Bearer ${input.credential}` } }, - ); - if (!response.ok && ![401, 403, 404].includes(response.status)) { - throw new Error(`Hub revocation failed (${response.status})`); - } + await this.withRequestTimeout(async (signal) => { + const response = await fetch( + `${input.hubOrigin}/api/daemon-relationships/${encodeURIComponent(input.relationshipId)}`, + { + method: "DELETE", + headers: { authorization: `Bearer ${input.credential}` }, + signal, + }, + ); + if (!response.ok && ![401, 403, 404].includes(response.status)) { + throw new Error(`Hub revocation failed (${response.status})`); + } + }); } openSocket(input: HubSocketCredentials, events: HubSocketEvents): HubSocketConnection { @@ -134,4 +149,18 @@ export class DirectHubRelationshipRemote implements HubRelationshipRemote { }); return socket; } + + private async withRequestTimeout(operation: (signal: AbortSignal) => Promise): Promise { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), this.requestTimeoutMs); + timeout.unref?.(); + try { + return await operation(controller.signal); + } catch (error) { + if (controller.signal.aborted) throw new Error("Hub request timed out", { cause: error }); + throw error; + } finally { + clearTimeout(timeout); + } + } } 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 799e85bf15..ec4608636e 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -2,7 +2,7 @@ import { execFile } from "node:child_process"; import { execFileSync } from "node:child_process"; import { EventEmitter } from "node:events"; import { existsSync, readFileSync, statSync, watch } from "node:fs"; -import { mkdir, mkdtemp, rm } from "node:fs/promises"; +import { mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises"; import { networkInterfaces, tmpdir } from "node:os"; import path from "node:path"; import { promisify } from "node:util"; @@ -993,6 +993,10 @@ export class HubRelationshipHarness { return this.remote.revocations.length; } + latestRevocation(): HubRevocation | null { + return this.remote.revocations.at(-1) ?? null; + } + enrollmentAttempts(): HubEnrollment[] { return this.remote.enrollments.map((input) => ({ ...input, scopes: input.scopes.slice() })); } @@ -1027,6 +1031,25 @@ export class HubRelationshipHarness { return statSync(path.join(this.paseoHome, "hub-relationship.json")).mode & 0o777; } + async corruptRelationshipFile(contents = "{not-json"): Promise { + await this.stopDaemon(); + await writeFile(path.join(this.paseoHome, "hub-relationship.json"), contents, "utf8"); + } + + async quarantinedRelationshipFiles(): Promise { + return (await readdir(this.paseoHome)).filter((file) => + file.startsWith("hub-relationship.invalid-"), + ); + } + + async startStoppedDaemon(): Promise { + await this.startDaemon(); + } + + async storedOwnedStatus(agentId: string): Promise { + return (await this.daemon!.agentStorage.get(agentId))?.lastStatus ?? null; + } + private captureRelationship(): RelationshipInvocationSnapshot { const record = this.relationshipFile(); if (!record) throw new Error("Relationship authority was not persisted before invocation");