fix: proxmox guest sync, startup snippet sync, drop dead cloudflare ssh transport

This commit is contained in:
LukeGus committed 2026-09-28 21:59:37 -05:00
1 parent 44062e30d1
commit ffa872ebd3
19 files changed
+270 -292

No files matched your search

-64
View File
@@ -1,64 +0,0 @@
import { Duplex } from "node:stream";
import type { RawData, WebSocket } from "ws";
export function waitForWebSocketOpen(
socket: WebSocket,
timeoutMs: number,
): Promise<void> {
return new Promise((resolve, reject) => {
const cleanup = () => {
clearTimeout(timer);
socket.off("open", onOpen);
socket.off("error", onError);
};
const onOpen = () => {
cleanup();
resolve();
};
const onError = (error: Error) => {
cleanup();
socket.terminate();
reject(error);
};
const timer = setTimeout(() => {
cleanup();
socket.terminate();
reject(new Error("Cloudflare tunnel timeout"));
}, timeoutMs);
socket.once("open", onOpen);
socket.once("error", onError);
});
}
export function createWebSocketDuplex(socket: WebSocket): Duplex {
const duplex = new Duplex({
read() {},
write(chunk, _encoding, callback) {
try {
socket.send(chunk, (error) => callback(error || undefined));
} catch (error) {
callback(error instanceof Error ? error : new Error(String(error)));
}
},
destroy(error, callback) {
cleanup();
socket.terminate();
callback(error);
},
});
const onMessage = (data: RawData) => duplex.push(data);
const onClose = () => duplex.destroy();
const onError = () => duplex.destroy();
const cleanup = () => {
socket.off("message", onMessage);
socket.off("close", onClose);
socket.off("error", onError);
};
socket.on("message", onMessage);
socket.on("close", onClose);
socket.on("error", onError);
return duplex;
}
+4 -44
View File
@@ -1,6 +1,6 @@
/**
* How the bytes get to a host: port knocking, a Cloudflare Access tunnel, a
* jump host chain or a SOCKS5 proxy, in that order. Sets config.sock and
* How the bytes get to a host: port knocking, a jump host chain or a SOCKS5
* proxy, in that order. Sets config.sock and
* returns the jump client so the caller can close it with the connection.
*/
@@ -39,7 +39,7 @@ function getHostSocks5Config(host: SshConnectHost): SOCKS5Config | null {
class SshTransportError extends Error {
constructor(
message: string,
readonly stage: "cloudflare" | "jump-host" | "jump-forward" | "proxy",
readonly stage: "jump-host" | "jump-forward" | "proxy",
options?: { cause?: unknown },
) {
super(message, options);
@@ -50,8 +50,6 @@ class SshTransportError extends Error {
export interface OpenTransportOptions {
/** Knock before connecting when the host has a sequence. Default true. */
portKnock?: boolean;
/** Use terminalConfig.cfAccessClient* when present. Default true. */
cloudflare?: boolean;
/** Resolve DNS up front for a direct connection. Default true. */
resolveDns?: boolean;
log?: SshAuthLog;
@@ -59,7 +57,7 @@ export interface OpenTransportOptions {
export interface OpenedTransport {
jumpClient: Client | null;
via: "direct" | "proxy" | "jump" | "cloudflare";
via: "direct" | "proxy" | "jump";
}
function forwardThrough(
@@ -94,8 +92,6 @@ export async function openSshTransport(
config: MutableConnectConfig,
options: OpenTransportOptions = {},
): Promise<OpenedTransport> {
const log = options.log ?? (() => {});
if (
options.portKnock !== false &&
Array.isArray(host.portKnockSequence) &&
@@ -113,42 +109,6 @@ export async function openSshTransport(
let via: OpenedTransport["via"] = "direct";
const terminalConfig = (host.terminalConfig ?? {}) as Record<string, unknown>;
if (
options.cloudflare !== false &&
terminalConfig.cfAccessClientId &&
terminalConfig.cfAccessClientSecret
) {
try {
const WebSocket = (await import("ws")).default;
const { createWebSocketDuplex, waitForWebSocketOpen } =
await import("../cloudflare-websocket.js");
const cfHostname = (terminalConfig.cfTunnelHostname as string) || host.ip;
const cfWs = new WebSocket(
`wss://${cfHostname}/cdn-cgi/access/ssh-connect`,
{
headers: {
"CF-Access-Client-Id": terminalConfig.cfAccessClientId as string,
"CF-Access-Client-Secret":
terminalConfig.cfAccessClientSecret as string,
},
},
);
await waitForWebSocketOpen(cfWs, 30000);
config.sock = createWebSocketDuplex(
cfWs,
) as unknown as typeof config.sock;
via = "cloudflare";
log("info", "Connected via Cloudflare Tunnel");
} catch (error) {
throw new SshTransportError(
"Cloudflare tunnel connection failed: " + getErrorMessage(error),
"cloudflare",
{ cause: error },
);
}
}
const jumpUserId = host.userId || "";
if (host.jumpHosts && host.jumpHosts.length > 0 && jumpUserId) {
const jumpClient = await createJumpHostChain(host.jumpHosts, jumpUserId);
+1
View File
@@ -58,6 +58,7 @@ const HOST_REFERENCES = [
},
{ field: "jumpHosts[].hostId", entityType: "hosts" },
{ field: "quickActions[].snippetId", entityType: "commandSnippet" },
{ field: "terminalConfig.startupSnippetId", entityType: "commandSnippet" },
] as const;
/** Columns that only mean something on the device that holds them. */
@@ -1,74 +0,0 @@
import { EventEmitter } from "node:events";
import { afterEach, describe, expect, it, vi } from "vitest";
import type { WebSocket } from "ws";
import {
createWebSocketDuplex,
waitForWebSocketOpen,
} from "../../hosts/cloudflare-websocket.js";
function setupSocket() {
const socket = new EventEmitter() as EventEmitter & {
terminate: ReturnType<typeof vi.fn>;
send: ReturnType<typeof vi.fn>;
};
socket.terminate = vi.fn();
socket.send = vi.fn();
return socket as unknown as WebSocket;
}
describe("waitForWebSocketOpen", () => {
afterEach(() => vi.useRealTimers());
it("clears listeners and leaves an opened socket alive", async () => {
vi.useFakeTimers();
const socket = setupSocket();
const result = waitForWebSocketOpen(socket, 30_000);
socket.emit("open");
await expect(result).resolves.toBeUndefined();
expect(socket.terminate).not.toHaveBeenCalled();
expect(socket.listenerCount("open")).toBe(0);
expect(socket.listenerCount("error")).toBe(0);
expect(vi.getTimerCount()).toBe(0);
});
it("terminates a socket that does not open before the deadline", async () => {
vi.useFakeTimers();
const socket = setupSocket();
const result = waitForWebSocketOpen(socket, 30_000);
const rejection = expect(result).rejects.toThrow(
"Cloudflare tunnel timeout",
);
await vi.advanceTimersByTimeAsync(30_000);
await rejection;
expect(socket.terminate).toHaveBeenCalledOnce();
expect(socket.listenerCount("open")).toBe(0);
expect(socket.listenerCount("error")).toBe(0);
});
it("terminates the socket when the SSH duplex is destroyed", () => {
const socket = setupSocket();
const duplex = createWebSocketDuplex(socket);
duplex.destroy();
expect(socket.terminate).toHaveBeenCalledOnce();
expect(socket.listenerCount("message")).toBe(0);
expect(socket.listenerCount("close")).toBe(0);
expect(socket.listenerCount("error")).toBe(0);
});
it("handles established socket errors without an unhandled event", () => {
const socket = setupSocket();
const duplex = createWebSocketDuplex(socket);
socket.emit("error", new Error("connection lost"));
expect(duplex.destroyed).toBe(true);
expect(socket.terminate).toHaveBeenCalledOnce();
expect(socket.listenerCount("error")).toBe(0);
});
});
@@ -1,5 +1,5 @@
import { describe, expect, it, vi } from "vitest";
import { HostSessionStatus } from "./host-session-status.js";
import { HostSessionStatus } from "../../hosts/host-session-status.js";
describe("HostSessionStatus", () => {
it("reports only the first connection and last disconnection per host", () => {
@@ -1,5 +1,5 @@
import { describe, expect, it } from "vitest";
import { getJumpHostSocks5Config } from "./jump-host-proxy.js";
import { getJumpHostSocks5Config } from "../../hosts/jump-host-proxy.js";
describe("getJumpHostSocks5Config", () => {
it("uses the first jump host proxy settings", () => {
@@ -1,5 +1,5 @@
import { describe, expect, it } from "vitest";
import { resolveSshKeepalive } from "./ssh-keepalive.js";
import { resolveSshKeepalive } from "../../hosts/ssh-keepalive.js";
describe("resolveSshKeepalive", () => {
it("preserves zero to disable SSH keepalives", () => {
@@ -197,6 +197,16 @@ describe("core sync entities", () => {
);
});
it("translates every snippet a host points at", () => {
const fields = (getEntity("hosts")?.references ?? [])
.filter((reference) => reference.entityType === "commandSnippet")
.map((reference) => reference.field);
expect(fields).toEqual([
"quickActions[].snippetId",
"terminalConfig.startupSnippetId",
]);
});
it("encrypts host and credential secrets", () => {
expect(getEntity("hosts")?.encryptedFields).toContain("password");
expect(getEntity("sshCredentials")?.encryptedFields).toContain(
+24
View File
@@ -35,6 +35,7 @@ beforeEach(() => {
references: [
{ field: "parentId", syncField: "parentSyncId", entityType: "things" },
{ field: "hops[].thingId", entityType: "things" },
{ field: "config.startId", entityType: "things" },
{
field: "bindings[].action.snippetId",
entityType: "things",
@@ -111,6 +112,29 @@ describe("serializing references", () => {
expect(back.bindings).toEqual([{ action: { snippetId: "12" } }]);
});
it("rewrites an id nested in a JSON object column", async () => {
const wire = await serializeReferences(
entity("things"),
{ config: JSON.stringify({ theme: "dark", startId: 5 }) },
toSync,
);
expect(JSON.parse(wire.config as string)).toEqual({
theme: "dark",
startId: "sync-5",
});
const back = await deserializeReferences(
entity("things"),
wire,
toId,
null,
);
expect(JSON.parse(back.config as string)).toEqual({
theme: "dark",
startId: 5,
});
});
it("keeps the current value when a referenced row is not here", async () => {
const back = await deserializeReferences(
entity("things"),