fix(hosts): validate safe targets and verify encrypted relay routes

packages/server/src/server/connected-hosts/hosts.e2e.test.ts

@@ -19,6 +19,8 @@ async function api(instance: TestPaseoDaemon) {
     reconnect: { enabled: false },
   });
   await client.connect();
+  expect(client.getLastServerInfoMessage()).toMatchObject({ sourceId: "owner" });
+  expect(client.getLastServerInfoMessage()).not.toHaveProperty("principalId");
   const paseo = createPaseoApi(client);
   cleanup.push(async () => {
     await paseo.dispose();
@@ -46,6 +48,16 @@ test("SDK saves unreachable hosts, rejects syntax and preserves source identity
   await expect(paseo.hosts.add({ connection: "ssh://-bad" })).rejects.toMatchObject({
     code: "invalid_connection",
   });
+  for (const connection of [
+    "alice:example-secret@localhost:6767",
+    "not-a-scheme://localhost:6767",
+    "localhost/path:6767",
+  ]) {
+    const failure = await paseo.hosts.add({ connection }).catch((error: unknown) => error);
+    expect(failure).toMatchObject({ code: "invalid_connection" });
+    expect(String(failure)).not.toContain("example-secret");
+    expect((await paseo.hosts.list()).hosts).toEqual([]);
+  }
   const saved = await paseo.hosts.add({
     connection: "tcp://127.0.0.1:1?password=private-secret",
     label: "Unavailable",
@@ -241,6 +253,14 @@ class RoutePeer {
           ).length,
       )
       .toBe(1);
+    const info = peer.messages.find(
+      (message) => message.type === "status" && message.payload.status === "server_info",
+    );
+    if (info?.type !== "status") throw new Error("Missing routed server info");
+    expect(info.payload).toMatchObject({
+      sourceId: `route:owner:${origin.getLastServerInfoMessage()!.serverId}:owner`,
+    });
+    expect(info.payload).not.toHaveProperty("principalId");
     return peer;
   }
   send(message: SessionInboundMessage) {
@@ -436,6 +456,13 @@ test("CLI host commands manage the selected daemon and emit secret-free structur
         )
       ).stdout,
     );
+  const malformed = await cli("add", "alice:example-secret@localhost:6767").catch(
+    (error: unknown) => error,
+  );
+  expect(malformed).toHaveProperty("stderr");
+  expect((malformed as { stderr: string }).stderr).toContain("invalid_connection");
+  expect((malformed as { stderr: string }).stderr).not.toContain("example-secret");
+  expect(await cli("ls")).toEqual([]);
   const saved = await cli(
     "add",
     "tcp://127.0.0.1:1?password=cli-secret",

packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts

@@ -1,3 +1,14 @@
+import { createPaseoApi, type PaseoApi } from "@getpaseo/client";
+import { createHostClient } from "@getpaseo/client/node";
+import { hashSync } from "bcryptjs";
+import { once } from "node:events";
+import {
+  decodeTerminalStreamFrame,
+  encodeTerminalStreamFrame,
+  TerminalStreamOpcode,
+} from "@getpaseo/protocol/binary-frames/index";
+import type { DaemonClient as HostClient } from "@getpaseo/client/internal/daemon-client";
+import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
 import { describe, expect, test } from "vitest";
 import { withTimeout } from "../../utils/promise-timeout.js";
 import { DaemonClient } from "../test-utils/daemon-client.js";
@@ -194,9 +205,9 @@ async function waitForCapturedLog(
   let relayProcess: ChildProcess | null = null;
   let relayStdoutLines: string[] = [];
 
-  const startRelay = async (options: { useLocalRelay?: boolean } = {}) => {
+  const startRelay = async (options: { useLocalRelay?: boolean; port?: number } = {}) => {
     relayStdoutLines = [];
-    relayPort = await getAvailablePort();
+    relayPort = options.port ?? (await getAvailablePort());
     const relayDir = path.resolve(process.cwd(), "../relay");
     const relayArgs = [
       "wrangler",
@@ -255,10 +266,142 @@ async function waitForCapturedLog(
 
   const stopRelay = async () => {
     if (!relayProcess) return;
-    relayProcess.kill("SIGTERM");
+    const process = relayProcess;
     relayProcess = null;
+    const exited = once(process, "exit");
+    process.kill("SIGTERM");
+    await withTimeout(exited, 10000, "Local relay did not stop");
   };
 
+  test("connected hosts route RPC and binary traffic through local E2EE relay and recover scopes", async () => {
+    let a: Awaited<ReturnType<typeof createTestPaseoDaemon>> | undefined;
+    let b: Awaited<ReturnType<typeof createTestPaseoDaemon>> | undefined;
+    let origin: ReturnType<typeof createHostClient> | undefined;
+    let api: PaseoApi | undefined;
+    let remote: PaseoApi | undefined;
+    let sibling: PaseoApi | undefined;
+    let terminalRoute: Awaited<ReturnType<typeof openConnectedHostTerminal>> | undefined;
+    try {
+      await startRelay({ useLocalRelay: true });
+      const port = relayPort;
+      b = await createTestPaseoDaemon({
+        relayEnabled: true,
+        relayEndpoint: `127.0.0.1:${port}`,
+        relayUseTls: false,
+        relayPublicUseTls: false,
+        auth: { password: hashSync("relay-password", 4) },
+      });
+      a = await createTestPaseoDaemon();
+      origin = createHostClient(`127.0.0.1:${a.port}`, { clientId: crypto.randomUUID() });
+      await origin.connect();
+      api = createPaseoApi(origin);
+      const offerUrl = await getPairingOfferUrl({
+        paseoHome: b.paseoHome,
+        relayEnabled: true,
+        relayEndpoint: b.config.relayEndpoint,
+      });
+      const { serverId, daemonPublicKeyB64 } = decodeOfferFromFragmentUrl(offerUrl);
+      const connection = new URL(`relay://127.0.0.1:${port}/${serverId}`);
+      connection.searchParams.set("key", daemonPublicKeyB64);
+      connection.searchParams.set("password", "wrong-password");
+      const saved = await api.hosts.add({ connection: connection.toString() });
+      await expect
+        .poll(async () => (await api!.hosts.list()).hosts[0]?.error?.code, { timeout: 20000 })
+        .toBe("authentication_failed");
+      expect(JSON.stringify(await api.hosts.list())).not.toMatch(/wrong-password|relay-password/);
+      connection.searchParams.set("password", "relay-password");
+      await api.hosts.update(saved.registrationId, { connection: connection.toString() });
+      await expect
+        .poll(async () => (await api!.hosts.list()).hosts[0]?.status, { timeout: 20000 })
+        .toBe("online");
+      remote = await api.hosts.client(serverId);
+      sibling = await api.hosts.client(serverId);
+      expect((await remote.workspaces.list()).entries).toEqual([]);
+      const observed = await remote.workspaces.list({ subscribe: {} });
+      let snapshots = 0;
+      observed.subscription.subscribe({
+        snapshot: () => {
+          snapshots++;
+        },
+      });
+      const workspace = await remote.workspaces.create({
+        source: { kind: "directory", path: b.staticDir },
+        title: "Relayed workspace",
+      });
+      const terminal = await workspace.terminals.create({
+        command: process.execPath,
+        args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"],
+      });
+      terminalRoute = await openConnectedHostTerminal(origin, serverId, terminal.id);
+      terminalRoute.write("ENCRYPTED_BINARY_BEFORE\n");
+      await expect.poll(() => terminalRoute!.output()).toContain("ENCRYPTED_BINARY_BEFORE");
+      terminalRoute.close();
+      terminalRoute = undefined;
+      const before = snapshots;
+      await stopRelay();
+      await expect
+        .poll(
+          async () => {
+            try {
+              await remote!.workspaces.list();
+              return "online";
+            } catch (error) {
+              return (error as { code: string }).code;
+            }
+          },
+          { timeout: 15000 },
+        )
+        .toBe("host_offline");
+      await startRelay({ useLocalRelay: true, port });
+      await expect
+        .poll(
+          async () => {
+            try {
+              return (await remote!.workspaces.list()).entries.length;
+            } catch {
+              return -1;
+            }
+          },
+          { timeout: 30000 },
+        )
+        .toBe(1);
+      await expect.poll(() => snapshots, { timeout: 15000 }).toBeGreaterThan(before);
+      terminalRoute = await openConnectedHostTerminal(origin, serverId, terminal.id);
+      terminalRoute.write("ENCRYPTED_BINARY_AFTER\n");
+      await expect.poll(() => terminalRoute!.output()).toContain("ENCRYPTED_BINARY_AFTER");
+      terminalRoute.close();
+      terminalRoute = undefined;
+      await remote.dispose();
+      remote = undefined;
+      expect((await sibling.workspaces.list()).entries).toHaveLength(1);
+      await sibling.terminals.ref(terminal.id).kill();
+      await api.dispose();
+      await expect(sibling.workspaces.list()).rejects.toThrow();
+      const direct = createHostClient(`127.0.0.1:${b.port}`, {
+        clientId: crypto.randomUUID(),
+        password: "relay-password",
+        reconnect: { enabled: false },
+      });
+      try {
+        await direct.connect();
+        const directApi = createPaseoApi(direct);
+        expect((await directApi.workspaces.list()).entries).toHaveLength(1);
+        await directApi.dispose();
+      } finally {
+        await direct.close();
+      }
+    } finally {
+      terminalRoute?.close();
+      await remote?.dispose();
+      await sibling?.dispose();
+      await api?.dispose();
+      await origin?.close();
+      await a?.close();
+      await b?.close();
+      await stopRelay();
+    }
+  }, 120000);
+
   test("oversized attachments fail without disconnecting or resubmitting over the relay", async () => {
     await startRelay({ useLocalRelay: true });
     const daemon = await createTestPaseoDaemon({
@@ -932,3 +1075,73 @@ async function waitForCapturedLog(
     }
   }, 90000);
 });
+
+async function openConnectedHostTerminal(origin: HostClient, serverId: string, terminalId: string) {
+  const routeId = await origin.openHostRoute(serverId, crypto.randomUUID());
+  const messages: SessionOutboundMessage[] = [];
+  let output = "";
+  const detach = origin.subscribeRawMessages((message) => {
+    if (message.type !== "hosts.route.frame" || message.payload.subscriptionId !== routeId) return;
+    if (message.payload.binary) {
+      const frame = decodeTerminalStreamFrame(Buffer.from(message.payload.data, "base64"));
+      if (frame?.opcode === TerminalStreamOpcode.Output)
+        output += new TextDecoder().decode(frame.payload);
+    } else {
+      const frame = WSOutboundMessageSchema.parse(JSON.parse(message.payload.data));
+      if (frame.type === "session") messages.push(frame.message);
+    }
+  });
+  const close = () => {
+    detach();
+    origin.closeHostRoute(routeId);
+  };
+  try {
+    origin.sendHostFrame(
+      routeId,
+      JSON.stringify({
+        type: "hello",
+        clientId: "relay-binary",
+        clientType: "cli",
+        protocolVersion: 1,
+      }),
+      false,
+    );
+    await expect
+      .poll(() =>
+        messages.some(
+          (message) => message.type === "status" && message.payload.status === "server_info",
+        ),
+      )
+      .toBe(true);
+    origin.sendHostFrame(
+      routeId,
+      JSON.stringify({
+        type: "session",
+        message: { type: "subscribe_terminal_request", terminalId, requestId: "binary" },
+      }),
+      false,
+    );
+    await expect
+      .poll(() => messages.some((message) => message.type === "subscribe_terminal_response"))
+      .toBe(true);
+    const response = messages.find((message) => message.type === "subscribe_terminal_response");
+    if (response?.type !== "subscribe_terminal_response" || response.payload.slot == null)
+      throw new Error("Missing binary stream slot");
+    const slot = response.payload.slot;
+    return {
+      close,
+      output: () => output,
+      write: (payload: string) =>
+        origin.sendHostFrame(
+          routeId,
+          Buffer.from(
+            encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, slot, payload }),
+          ).toString("base64"),
+          true,
+        ),
+    };
+  } catch (error) {
+    close();
+    throw error;
+  }
+}

packages/server/src/server/session.ts

@@ -677,12 +677,12 @@ export class Session {
   isRouted(): boolean {
     return this.routed;
   }
-  getPrincipalId(): string {
-    return this.hostPrincipalId;
+  getHostSourceId(): string {
+    return this.hostSourceId;
   }
-  private hostPrincipalId: string;
+  private hostSourceId: string;
   setHostSource(sourceId: string): void {
-    this.hostPrincipalId = sourceId;
+    this.hostSourceId = sourceId;
     this.connectedHostSession?.setSource(sourceId);
   }
   setRouted(routed: boolean): void {
@@ -882,7 +882,7 @@ export class Session {
     this.appVersion = appVersion ?? null;
     this.clientCapabilities = parseClientCapabilities(clientCapabilities);
     this.sessionId = uuidv4();
-    this.hostPrincipalId = hostSource(options);
+    this.hostSourceId = hostSource(options);
     this.connectedHostSession = createHostSession(options, this);
     this.onMessage = onMessage;
     this.onMessageToSource = onMessageToSource ?? null;

packages/server/src/server/websocket-server.relay-reconnect.test.ts

@@ -71,6 +71,18 @@ const sessionMock = vi.hoisted(() => {
     wantsSourceEvent = (source: object) => !this.delivery.isModern(source);
     getSessionId = vi.fn(() => "mock-session-id");
     getPermissions = vi.fn(() => this.args.permissions as string[]);
+    setPermissions = (permissions: string[]) => {
+      this.args.permissions = permissions;
+    };
+    private routed = false;
+    setRouted = (routed: boolean) => {
+      this.routed = routed;
+    };
+    isRouted = () => this.routed;
+    setHostSource = (sourceId: string) => {
+      this.args.sourceId = sourceId;
+    };
+    getHostSourceId = () => this.args.sourceId as string;
     allowsInbound = vi.fn(() => true);
     allowsPermission = vi.fn(() => true);
     publish = vi.fn((message: unknown) => {
@@ -473,16 +485,12 @@ function holdNextSessionMessage(session: (typeof sessionMock.instances)[number])
 function holdSessionCleanup(session: (typeof sessionMock.instances)[number]): {
   finish: () => void;
 } {
-  let finish = () => {};
-  session.cleanup.mockImplementationOnce(
-    () =>
-      new Promise<void>((resolve) => {
-        finish = resolve;
-      }),
-  );
-  return {
-    finish: () => finish(),
-  };
+  let finish!: () => void;
+  const pending = new Promise<void>((resolve) => {
+    finish = resolve;
+  });
+  session.cleanup.mockImplementationOnce(() => pending);
+  return { finish };
 }
 
 describe("relay external socket reconnect behavior", () => {

packages/server/src/server/websocket-server.ts

@@ -1846,7 +1846,7 @@ export class VoiceAssistantWebSocketServer {
       serverId: this.serverId,
       hostname: getHostName(),
       version: this.daemonVersion,
-      principalId: session.getPrincipalId(),
+      sourceId: session.getHostSourceId(),
       permissions: session.getPermissions(),
       // COMPAT(desktopManaged): added in v0.1.X, remove optional parsing after 2027-01-16.
       desktopManaged: this.daemonRuntimeConfig?.desktopManaged === true,