mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-31 21:52:54 +00:00
feat(hosts): connect to saved hosts through a daemon
This commit is contained in:
Mohamed Boudra
@@ -0,0 +1,22 @@
|
||||
/** Counts bytes until the socket's send callback, shared by all links owned by a daemon. */
|
||||
export class HostTransportBudget {
|
||||
private used = 0;
|
||||
constructor(readonly limit = 64 * 1024 * 1024) {}
|
||||
get bufferedBytes(): number {
|
||||
return this.used;
|
||||
}
|
||||
reserve(bytes: number): () => void {
|
||||
if (bytes > this.limit - this.used)
|
||||
throw Object.assign(new Error("Connected host buffer limit reached"), {
|
||||
code: "resource_limit",
|
||||
});
|
||||
this.used += bytes;
|
||||
let released = false;
|
||||
return () => {
|
||||
if (!released) {
|
||||
released = true;
|
||||
this.used -= bytes;
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
import { once } from "node:events";
|
||||
import { createServer } from "node:http";
|
||||
import { afterEach, expect, test } from "vitest";
|
||||
import { WebSocketServer, WebSocket } from "ws";
|
||||
import { createHostTransportFactory, HostTransportBudget } from "./index.js";
|
||||
const cleanup: (() => Promise<unknown>)[] = [];
|
||||
afterEach(async () => {
|
||||
for (const close of cleanup.splice(0).toReversed()) await close();
|
||||
});
|
||||
async function endpoint() {
|
||||
const server = createServer();
|
||||
const wss = new WebSocketServer({ server });
|
||||
server.listen(0, "127.0.0.1");
|
||||
await once(server, "listening");
|
||||
cleanup.push(async () => {
|
||||
for (const socket of wss.clients) socket.terminate();
|
||||
await new Promise<void>((resolve) => wss.close(() => server.close(() => resolve())));
|
||||
});
|
||||
const address = server.address();
|
||||
if (!address || typeof address === "string") throw new Error("Missing port");
|
||||
return { wss, url: `ws://127.0.0.1:${address.port}/ws` };
|
||||
}
|
||||
test("connector retains per-attempt headers and WebSocket subprotocols", async () => {
|
||||
const server = await endpoint();
|
||||
const connected = once(server.wss, "connection");
|
||||
const transport = createHostTransportFactory(server.url)({
|
||||
url: server.url,
|
||||
headers: { Authorization: "Bearer example" },
|
||||
protocols: ["paseo-test"],
|
||||
});
|
||||
cleanup.push(async () => transport.close());
|
||||
const [socket, request] = (await connected) as [WebSocket, import("node:http").IncomingMessage];
|
||||
expect(request.headers.authorization).toBe("Bearer example");
|
||||
expect(socket.protocol).toBe("paseo-test");
|
||||
});
|
||||
test("a slow consumer cannot exceed the shared byte budget and close releases reserved bytes", async () => {
|
||||
const server = await endpoint();
|
||||
const budget = new HostTransportBudget(4 * 1024 * 1024);
|
||||
const connected = once(server.wss, "connection");
|
||||
const transport = createHostTransportFactory(server.url, budget)({ url: server.url });
|
||||
cleanup.push(async () => transport.close());
|
||||
const ready = new Promise<void>((resolve) => transport.onOpen(resolve));
|
||||
const [socket] = (await connected) as [WebSocket];
|
||||
socket.pause();
|
||||
await ready;
|
||||
const frame = new Uint8Array(1024 * 1024);
|
||||
for (let index = 0; index < 4; index++) transport.send(frame);
|
||||
expect(budget.bufferedBytes).toBe(4 * 1024 * 1024);
|
||||
expect(() => transport.send(frame)).toThrow("Connected host buffer limit reached");
|
||||
expect(budget.bufferedBytes).toBeLessThanOrEqual(budget.limit);
|
||||
transport.close();
|
||||
await expect.poll(() => budget.bufferedBytes).toBe(0);
|
||||
});
|
||||
@@ -0,0 +1,190 @@
|
||||
import { HostTransportBudget } from "./buffering.js";
|
||||
export { HostTransportBudget } from "./buffering.js";
|
||||
import { homedir } from "node:os";
|
||||
import { WebSocket } from "ws";
|
||||
import { parseHostTarget, type HostTarget } from "@getpaseo/protocol/host-target";
|
||||
import { createSshTunnel, type SshTunnel } from "./ssh-tunnel.js";
|
||||
import { createWebSocketTransportFactory } from "../daemon-client-websocket-transport.js";
|
||||
import { createRelayE2eeTransportFactory } from "../daemon-client-relay-e2ee-transport.js";
|
||||
import type {
|
||||
DaemonTransport,
|
||||
DaemonTransportFactory,
|
||||
WebSocketLike,
|
||||
} from "../daemon-client-transport-types.js";
|
||||
import { DaemonClient, type DaemonClientConfig } from "../daemon-client.js";
|
||||
|
||||
export { parseHostTarget };
|
||||
export type { SshTunnel } from "./ssh-tunnel.js";
|
||||
export { createSshTunnel, resolveSshFailureDetail } from "./ssh-tunnel.js";
|
||||
|
||||
/** A factory owns one attempt, including a fresh SSH process and its teardown. */
|
||||
export function createHostTransportFactory(
|
||||
connection: string,
|
||||
budget = new HostTransportBudget(),
|
||||
): DaemonTransportFactory {
|
||||
const target = parseHostTarget(connection);
|
||||
return (attemptOptions) => {
|
||||
const opens = new Set<() => void>();
|
||||
const closes = new Set<(event?: unknown) => void>();
|
||||
const errors = new Set<(event?: unknown) => void>();
|
||||
const messages = new Set<(data: unknown, binary: boolean) => void>();
|
||||
let closed = false;
|
||||
let closeEvent: unknown;
|
||||
let transport: DaemonTransport | undefined;
|
||||
let teardown = () => {};
|
||||
let sshTunnel: SshTunnel | null = null;
|
||||
const close = (code?: number, reason?: string) => {
|
||||
if (closed) return;
|
||||
closed = true;
|
||||
try {
|
||||
transport?.close(code, reason);
|
||||
} finally {
|
||||
try {
|
||||
teardown();
|
||||
} finally {
|
||||
for (const handler of closes) handler(closeEvent);
|
||||
}
|
||||
}
|
||||
};
|
||||
void (async () => {
|
||||
let socketTarget: Extract<HostTarget, { kind: "socket" }>;
|
||||
if (target.kind === "ssh") {
|
||||
const tunnel = await createSshTunnel(target.ssh);
|
||||
sshTunnel = tunnel;
|
||||
teardown = () => tunnel.close();
|
||||
socketTarget = parseHostTarget(tunnel.endpoint) as Extract<HostTarget, { kind: "socket" }>;
|
||||
} else socketTarget = target;
|
||||
if (closed) {
|
||||
teardown();
|
||||
return;
|
||||
}
|
||||
const socketPath = socketTarget.socketPath?.replace(/^~(?=\/)/, homedir());
|
||||
let factory = createWebSocketTransportFactory((url, options) => {
|
||||
const socket = new WebSocket(
|
||||
socketPath && url.startsWith("ws+unix:") ? `ws+unix://${socketPath}:/ws` : url,
|
||||
options?.protocols,
|
||||
{
|
||||
headers: options?.headers,
|
||||
...(socketPath ? { socketPath } : {}),
|
||||
maxPayload: 24 * 1024 * 1024,
|
||||
},
|
||||
);
|
||||
const pending = new Set<() => void>();
|
||||
const send = socket.send.bind(socket);
|
||||
socket.send = (data, ...args: unknown[]) => {
|
||||
const bytes =
|
||||
typeof data === "string"
|
||||
? Buffer.byteLength(data)
|
||||
: Buffer.byteLength(data as Uint8Array);
|
||||
if (socket.bufferedAmount + bytes > 24 * 1024 * 1024) {
|
||||
socket.terminate();
|
||||
throw Object.assign(new Error("Host socket buffer limit reached"), {
|
||||
code: "resource_limit",
|
||||
});
|
||||
}
|
||||
let release: () => void;
|
||||
try {
|
||||
release = budget.reserve(bytes);
|
||||
} catch (error) {
|
||||
socket.terminate();
|
||||
throw error;
|
||||
}
|
||||
pending.add(release);
|
||||
const callback = args.find(
|
||||
(value): value is (error?: Error) => void => typeof value === "function",
|
||||
);
|
||||
const sendOptions = args.find((value) => value !== null && typeof value === "object") as
|
||||
| { binary?: boolean }
|
||||
| undefined;
|
||||
try {
|
||||
send(data, sendOptions ?? {}, (error) => {
|
||||
release();
|
||||
pending.delete(release);
|
||||
callback?.(error);
|
||||
});
|
||||
} catch (error) {
|
||||
release();
|
||||
pending.delete(release);
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
socket.once("close", () => {
|
||||
for (const release of pending) release();
|
||||
pending.clear();
|
||||
});
|
||||
return socket as unknown as WebSocketLike;
|
||||
});
|
||||
if (socketTarget.offer)
|
||||
factory = createRelayE2eeTransportFactory({
|
||||
baseFactory: factory,
|
||||
daemonPublicKeyB64: socketTarget.offer.daemonPublicKeyB64,
|
||||
logger: { warn() {} },
|
||||
});
|
||||
transport = factory({ ...attemptOptions, url: socketTarget.url });
|
||||
transport.onOpen(() => {
|
||||
if (!closed) for (const handler of opens) handler();
|
||||
});
|
||||
transport.onClose((event) => {
|
||||
closeEvent = event;
|
||||
close();
|
||||
});
|
||||
transport.onError((error) => {
|
||||
const detail = sshTunnel?.failureDetail() ?? "";
|
||||
const failure =
|
||||
/permission denied|authentication failed|host key verification failed|REMOTE HOST IDENTIFICATION HAS CHANGED/i.test(
|
||||
detail,
|
||||
)
|
||||
? Object.assign(
|
||||
new Error(
|
||||
"SSH authentication or host verification failed. Check the SSH configuration on this daemon.",
|
||||
),
|
||||
{ code: "authentication_failed" },
|
||||
)
|
||||
: error;
|
||||
for (const handler of errors) handler(failure);
|
||||
close();
|
||||
});
|
||||
transport.onMessage((data, binary) => {
|
||||
if (!closed) for (const handler of messages) handler(data, binary);
|
||||
});
|
||||
})().catch(() => {
|
||||
for (const handler of errors) handler(new Error("Host transport failed"));
|
||||
close();
|
||||
});
|
||||
return {
|
||||
send(data) {
|
||||
if (closed || !transport) throw new Error("Host transport is unavailable");
|
||||
transport.send(data);
|
||||
},
|
||||
close,
|
||||
onOpen(handler) {
|
||||
opens.add(handler);
|
||||
return () => opens.delete(handler);
|
||||
},
|
||||
onClose(handler) {
|
||||
closes.add(handler);
|
||||
return () => closes.delete(handler);
|
||||
},
|
||||
onError(handler) {
|
||||
errors.add(handler);
|
||||
return () => errors.delete(handler);
|
||||
},
|
||||
onMessage(handler) {
|
||||
messages.add(handler);
|
||||
return () => messages.delete(handler);
|
||||
},
|
||||
};
|
||||
};
|
||||
}
|
||||
export function createHostClient(
|
||||
connection: string,
|
||||
config: Omit<DaemonClientConfig, "url" | "transportFactory">,
|
||||
): DaemonClient {
|
||||
const target = parseHostTarget(connection);
|
||||
return new DaemonClient({
|
||||
...config,
|
||||
url: "ws://host-connector/ws",
|
||||
...(target.password ? { password: target.password } : {}),
|
||||
transportFactory: createHostTransportFactory(connection),
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import { createServer, type Server, type Socket } from "node:net";
|
||||
import { buildSshTunnelArgs, type SshTransportTarget } from "@getpaseo/protocol/ssh-transport";
|
||||
|
||||
const SSH_STDERR_LIMIT = 8192;
|
||||
|
||||
export interface SshTunnel {
|
||||
endpoint: string;
|
||||
close(): void;
|
||||
failureDetail(): string | null;
|
||||
}
|
||||
|
||||
function formatSshFailure(
|
||||
stderr: string,
|
||||
code: number | null,
|
||||
signal: NodeJS.Signals | null,
|
||||
): string {
|
||||
const detail = stderr.trim();
|
||||
if (detail) return detail;
|
||||
if (signal) return `ssh exited with signal ${signal}`;
|
||||
return `ssh exited with code ${code ?? "unknown"}`;
|
||||
}
|
||||
|
||||
export function resolveSshFailureDetail(failure: string | null, stderr: string): string | null {
|
||||
return failure ?? (stderr.trim() || null);
|
||||
}
|
||||
|
||||
export function createSshTunnel(target: SshTransportTarget): Promise<SshTunnel> {
|
||||
let server: Server | null = null;
|
||||
let socket: Socket | null = null;
|
||||
let child: ChildProcessWithoutNullStreams | null = null;
|
||||
let stderr = "";
|
||||
let failure: string | null = null;
|
||||
|
||||
function close(): void {
|
||||
server?.close();
|
||||
server = null;
|
||||
socket?.destroy();
|
||||
socket = null;
|
||||
if (child && !child.killed) child.kill();
|
||||
child = null;
|
||||
}
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
server = createServer((acceptedSocket) => {
|
||||
socket = acceptedSocket;
|
||||
server?.close();
|
||||
server = null;
|
||||
|
||||
child = spawn("ssh", buildSshTunnelArgs(target), {
|
||||
stdio: ["pipe", "pipe", "pipe"],
|
||||
windowsHide: true,
|
||||
});
|
||||
child.stderr.on("data", (chunk: Buffer | string) => {
|
||||
stderr = `${stderr}${chunk.toString()}`.slice(-SSH_STDERR_LIMIT);
|
||||
});
|
||||
child.on("error", (error) => {
|
||||
failure = error.message;
|
||||
acceptedSocket.destroy(error);
|
||||
});
|
||||
child.on("exit", (code, signal) => {
|
||||
if (code !== 0 || signal) failure = formatSshFailure(stderr, code, signal);
|
||||
acceptedSocket.destroy(failure ? new Error(failure) : undefined);
|
||||
});
|
||||
acceptedSocket.on("error", () => undefined);
|
||||
acceptedSocket.on("close", () => {
|
||||
if (child && !child.killed) child.kill();
|
||||
});
|
||||
acceptedSocket.pipe(child.stdin);
|
||||
child.stdout.pipe(acceptedSocket);
|
||||
});
|
||||
server.once("error", (error) => {
|
||||
close();
|
||||
reject(error);
|
||||
});
|
||||
server.listen(0, "127.0.0.1", () => {
|
||||
const address = server?.address();
|
||||
if (!address || typeof address === "string") {
|
||||
close();
|
||||
reject(new Error("Failed to allocate the SSH tunnel port"));
|
||||
return;
|
||||
}
|
||||
resolve({
|
||||
endpoint: `127.0.0.1:${address.port}`,
|
||||
close,
|
||||
failureDetail: () => resolveSshFailureDetail(failure, stderr),
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user