fix(hub): correct relationship trust boundaries

Authenticated daemon sessions own relationship management regardless of transport. The separate Hub session remains operation-allowlisted, rejects malformed execution inputs, and uses bounded outbound handshakes.
This commit is contained in:
Mohamed Boudra
2026-07-13 17:47:40 +02:00
parent 3772a8a118
commit dfcaa9bf89
11 changed files with 142 additions and 188 deletions
+3 -5
View File
@@ -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
@@ -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");
});
});
+1 -22
View File
@@ -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<T>(
}
}
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),
);
@@ -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<OwnedAgentSnapshot> {
this.creates++;
@@ -24,6 +25,7 @@ class InMemoryHubExecutions implements HubExecutions {
}
async reconcile(_executionId: string): Promise<OwnedAgentSnapshot | null> {
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<void> {
async createWith(input: HubCreateInput): Promise<void> {
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<void> {
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",
@@ -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<void> {
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<void> {
let result: Awaited<ReturnType<HubExecutions["reconcile"]>> = 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`);
}
@@ -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 () => {
@@ -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<void>();
private attempts = 0;
private constructor(private readonly statuses: number[]) {
private constructor(private readonly statuses: Array<number | "stall">) {
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<UpgradeRejectingHub> {
static async start(statuses: Array<number | "stall">): Promise<UpgradeRejectingHub> {
const hub = new UpgradeRejectingHub(statuses);
openUpgradeHubs.push(hub);
openServers.push(hub.server);
@@ -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;
@@ -512,7 +512,7 @@ export class HubRelationshipHarness {
}
}
async attemptExternalManagementAsCli(): Promise<SessionOutboundMessage[]> {
async manageRelationshipFromExternalSocket(): Promise<SessionOutboundMessage[]> {
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<SessionOutboundMessage> {
async connectFromBrowserSocket(): Promise<SessionOutboundMessage> {
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`, {
+3 -58
View File
@@ -395,15 +395,6 @@ export interface SessionFileSystem {
isDirectory(path: string): Promise<boolean>;
}
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<void> {
public async handleMessage(msg: SessionInboundMessage): Promise<void> {
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<void> {
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<void> {
const promise =
this.dispatchVoiceAndControlMessage(msg) ??
this.dispatchAgentRewindMessage(msg) ??
+2 -22
View File
@@ -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;