From db3f348faba2b2ff1bb075e490bfa77b55bd8589 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 9 May 2026 16:06:04 -0400 Subject: [PATCH] feat(runs): add sandbox terminal Expose a run-scoped websocket terminal for Docker and Daytona sandboxes, and add the web terminal route so sandbox-backed runs can be inspected interactively from the run detail page. --- Cargo.lock | 41 +- apps/fabro-web/app/app.css | 1 + apps/fabro-web/app/router.tsx | 2 + apps/fabro-web/app/routes/run-detail.test.ts | 20 + apps/fabro-web/app/routes/run-detail.tsx | 18 +- .../fabro-web/app/routes/run-terminal.test.ts | 34 + apps/fabro-web/app/routes/run-terminal.tsx | 297 +++++++ apps/fabro-web/package.json | 2 + bun.lock | 6 + lib/crates/fabro-sandbox/Cargo.toml | 5 +- lib/crates/fabro-sandbox/src/daytona/mod.rs | 4 + lib/crates/fabro-sandbox/src/docker.rs | 8 + lib/crates/fabro-sandbox/src/lib.rs | 3 + lib/crates/fabro-sandbox/src/terminal.rs | 730 ++++++++++++++++++ lib/crates/fabro-server/Cargo.toml | 4 +- .../src/server/handler/sandbox.rs | 263 ++++++- 16 files changed, 1424 insertions(+), 14 deletions(-) create mode 100644 apps/fabro-web/app/routes/run-terminal.test.ts create mode 100644 apps/fabro-web/app/routes/run-terminal.tsx create mode 100644 lib/crates/fabro-sandbox/src/terminal.rs diff --git a/Cargo.lock b/Cargo.lock index 8e51cbf1b..a58cba99f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -373,6 +373,7 @@ checksum = "8b52af3cb4058c895d37317bb27508dccc8e5f2d39454016b297bf4a400597b8" dependencies = [ "axum-core", "axum-macros", + "base64", "bytes", "form_urlencoded", "futures-util", @@ -391,8 +392,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1", "sync_wrapper", "tokio", + "tokio-tungstenite 0.28.0", "tower", "tower-layer", "tower-service", @@ -1235,7 +1238,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.26.2", "url", ] @@ -2103,11 +2106,13 @@ dependencies = [ "fabro-types", "fabro-util", "futures", + "futures-util", "git2", "glob", "httpmock", "rand 0.9.4", "reqwest-middleware", + "rustls", "serde", "serde_json", "shlex", @@ -2116,6 +2121,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tokio", + "tokio-tungstenite 0.26.2", "tokio-util", "toml 0.8.23", "tracing", @@ -2218,7 +2224,7 @@ dependencies = [ "serde_json", "thiserror 2.0.18", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.26.2", "toml 0.8.23", "tracing", "tracing-subscriber", @@ -6858,10 +6864,22 @@ dependencies = [ "rustls-pki-types", "tokio", "tokio-rustls", - "tungstenite", + "tungstenite 0.26.2", "webpki-roots 0.26.11", ] +[[package]] +name = "tokio-tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite 0.28.0", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -7103,6 +7121,23 @@ dependencies = [ "utf-8", ] +[[package]] +name = "tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.9.4", + "sha1", + "thiserror 2.0.18", + "utf-8", +] + [[package]] name = "twin-github" version = "0.228.0-nightly.0" diff --git a/apps/fabro-web/app/app.css b/apps/fabro-web/app/app.css index c4c2eae5b..da290d54d 100644 --- a/apps/fabro-web/app/app.css +++ b/apps/fabro-web/app/app.css @@ -1,3 +1,4 @@ +@import "@xterm/xterm/css/xterm.css"; @import "tailwindcss"; @plugin "@tailwindcss/typography"; diff --git a/apps/fabro-web/app/router.tsx b/apps/fabro-web/app/router.tsx index 28497177c..9811309e4 100644 --- a/apps/fabro-web/app/router.tsx +++ b/apps/fabro-web/app/router.tsx @@ -21,6 +21,7 @@ import * as RunLogs from "./routes/run-logs"; import * as RunEvents from "./routes/run-events"; import * as RunArtifacts from "./routes/run-artifacts"; import * as RunFiles from "./routes/run-files"; +import * as RunTerminal from "./routes/run-terminal"; import * as RunBilling from "./routes/run-billing"; import * as Insights from "./routes/insights"; import * as InsightsEditor from "./routes/insights-editor"; @@ -100,6 +101,7 @@ export const routes: RouteObject[] = [ route("events", RunEvents), route("artifacts", RunArtifacts), route("files", RunFiles), + route("terminal", RunTerminal), route("billing", RunBilling), ], }), diff --git a/apps/fabro-web/app/routes/run-detail.test.ts b/apps/fabro-web/app/routes/run-detail.test.ts index 49a07f29e..93ad4cf3a 100644 --- a/apps/fabro-web/app/routes/run-detail.test.ts +++ b/apps/fabro-web/app/routes/run-detail.test.ts @@ -12,6 +12,7 @@ import { ToastProvider } from "../components/toast"; import { DemoModeProvider } from "../lib/demo-mode"; let currentRunSummary: any = null; +let currentRunState: any = null; let currentQuestions: any[] = []; const mountedRenderers: TestRenderer.ReactTestRenderer[] = []; @@ -23,6 +24,9 @@ mock.module("../lib/queries", () => ({ useRunQuestions: () => ({ data: currentQuestions, }), + useRunState: () => ({ + data: currentRunState, + }), useRunFiles: () => ({ data: null, error: null, @@ -335,6 +339,7 @@ describe("RunDetail full-height child routes", () => { } }); currentRunSummary = null; + currentRunState = null; currentQuestions = []; delete (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT; }); @@ -373,6 +378,21 @@ describe("RunDetail full-height child routes", () => { expect(badges.map((badge) => badge.children.join(""))).toContain("7"); }); + test("shows the Terminal tab when the run has a sandbox", async () => { + currentRunState = { sandbox: { provider: "docker", id: "container-1" } }; + const renderer = await renderRunDetail({ + initialEntry: "/runs/run_1", + }); + + const terminalLinks = renderer.root.findAll( + (node) => + node.type === "a" && + node.props.href === "/runs/run_1/terminal" && + node.children.includes("Terminal"), + ); + expect(terminalLinks).toHaveLength(1); + }); + test("defers steer bar focus until after the Actions menu item click settles", async () => { const focusCalls: string[] = []; diff --git a/apps/fabro-web/app/routes/run-detail.tsx b/apps/fabro-web/app/routes/run-detail.tsx index 4e4487a64..157cf7d40 100644 --- a/apps/fabro-web/app/routes/run-detail.tsx +++ b/apps/fabro-web/app/routes/run-detail.tsx @@ -44,7 +44,7 @@ import { formatAbsoluteTs, formatRelativeTime } from "../lib/format"; import { queryKeys } from "../lib/query-keys"; import { useRunEvents } from "../lib/run-events"; import { useRunToasts } from "../hooks/use-run-toasts"; -import { useRun, useRunQuestions } from "../lib/queries"; +import { useRun, useRunQuestions, useRunState } from "../lib/queries"; import { canArchive, canCancel, @@ -61,6 +61,7 @@ import { const allTabs = [ { name: "Overview", path: "", count: null, demoOnly: false }, { name: "Stages", path: "/stages", count: null, demoOnly: false }, + { name: "Terminal", path: "/terminal", count: null, demoOnly: false, requiresSandbox: true }, { name: "Files Changed", path: "/files", count: null, demoOnly: false }, { name: "Billing", path: "/billing", count: null, demoOnly: false }, ]; @@ -135,6 +136,15 @@ export function lifecycleActionVisibility(status: string | null | undefined) { }; } +function runHasSandbox(runState: unknown): boolean { + return !!( + runState && + typeof runState === "object" && + "sandbox" in runState && + (runState as { sandbox?: unknown }).sandbox + ); +} + function buildRunDetailRun(summary: RunSummary): RunDetailRun { const item = mapRunSummaryToRunItem(summary); const rawStatus = summary.status; @@ -159,6 +169,7 @@ export function meta({ data }: any) { export default function RunDetail({ params }: { params: { id: string } }) { const demoMode = useDemoMode(); const runQuery = useRun(params.id); + const runStateQuery = useRunState(params.id); const run = runQuery.data ? buildRunDetailRun(runQuery.data) : null; const statusKind = runQuery.data?.status?.kind; const isBlocked = statusKind === "blocked"; @@ -178,11 +189,12 @@ export default function RunDetail({ params }: { params: { id: string } }) { const [deletePending, setDeletePending] = useState(false); const { push, dismiss } = useToast(); const filesCount = runQuery.data?.diff_summary?.files_changed ?? null; + const hasSandbox = runHasSandbox(runStateQuery.data); const tabs = allTabs .map((tab) => tab.name === "Files Changed" ? { ...tab, count: filesCount } : tab, ) - .filter((t) => !t.demoOnly || demoMode); + .filter((t) => (!t.demoOnly || demoMode) && (!t.requiresSandbox || hasSandbox)); const lifecycleToastStateRef = useRef(INITIAL_LIFECYCLE_TOAST_STATE); const steerBarRef = useRef(null); const now = useTickingNow(30_000); @@ -345,7 +357,7 @@ export default function RunDetail({ params }: { params: { id: string } }) { onFocusSteer={() => { focusSteerAfterMenuClose(() => steerBarRef.current?.focus()); }} - canPreview={!!run.sandboxId} + canPreview={hasSandbox} previewPending={previewPending} onPreview={() => void previewMutation.trigger({ port: 3000, diff --git a/apps/fabro-web/app/routes/run-terminal.test.ts b/apps/fabro-web/app/routes/run-terminal.test.ts new file mode 100644 index 000000000..0fc17f89d --- /dev/null +++ b/apps/fabro-web/app/routes/run-terminal.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, test } from "bun:test"; + +import { + buildTerminalWebSocketUrl, + parseTerminalServerMessage, +} from "./run-terminal"; + +function locationLike(url: string): Location { + return new URL(url) as unknown as Location; +} + +describe("run terminal route helpers", () => { + test("builds ws URLs for local HTTP", () => { + expect( + buildTerminalWebSocketUrl(locationLike("http://127.0.0.1:4187/runs/run_1"), "run_1"), + ).toBe("ws://127.0.0.1:4187/api/v1/runs/run_1/terminal"); + }); + + test("builds wss URLs for HTTPS", () => { + expect( + buildTerminalWebSocketUrl(locationLike("https://fabro.example/runs/run/1"), "run/1"), + ).toBe("wss://fabro.example/api/v1/runs/run%2F1/terminal"); + }); + + test("parses terminal server control messages", () => { + expect(parseTerminalServerMessage('{"type":"ready"}')).toEqual({ type: "ready" }); + expect(parseTerminalServerMessage('{"type":"error","message":"no sandbox"}')).toEqual({ + type: "error", + message: "no sandbox", + }); + expect(parseTerminalServerMessage('{"type":"unknown"}')).toBeNull(); + expect(parseTerminalServerMessage("{")).toBeNull(); + }); +}); diff --git a/apps/fabro-web/app/routes/run-terminal.tsx b/apps/fabro-web/app/routes/run-terminal.tsx new file mode 100644 index 000000000..38b0371a3 --- /dev/null +++ b/apps/fabro-web/app/routes/run-terminal.tsx @@ -0,0 +1,297 @@ +import { + useCallback, + useEffect, + useMemo, + useRef, + useState, +} from "react"; +import type { Terminal as XtermTerminal } from "@xterm/xterm"; +import type { FitAddon as XtermFitAddon } from "@xterm/addon-fit"; +import { + ArrowPathIcon, + ClipboardDocumentIcon, +} from "@heroicons/react/20/solid"; + +import { SECONDARY_BUTTON_CLASS } from "../components/ui"; +import { useToast } from "../components/toast"; +import { apiData, humanInTheLoopApi } from "../lib/api-client"; +import { useRunState } from "../lib/queries"; + +export const handle = { wide: true, fullHeight: true }; + +type ConnectionStatus = "connecting" | "ready" | "closed" | "error"; + +interface TerminalServerMessage { + type: "ready" | "error" | "closed"; + message?: string; +} + +const TERMINAL_THEME = { + background: "#0F1729", + foreground: "#E8EDF3", + cursor: "#67B2D7", + selectionBackground: "#357F9E66", + black: "#0F1729", + red: "#E86B6B", + green: "#5AC8A8", + yellow: "#F0A45B", + blue: "#67B2D7", + magenta: "#A8B5C5", + cyan: "#B5DDEF", + white: "#F7F9FB", +}; + +export function buildTerminalWebSocketUrl(location: Location, runId: string): string { + const protocol = location.protocol === "https:" ? "wss:" : "ws:"; + return `${protocol}//${location.host}/api/v1/runs/${encodeURIComponent(runId)}/terminal`; +} + +export function parseTerminalServerMessage(data: string): TerminalServerMessage | null { + try { + const parsed = JSON.parse(data); + if (!parsed || typeof parsed !== "object") return null; + const type = (parsed as { type?: unknown }).type; + if (type !== "ready" && type !== "error" && type !== "closed") return null; + const message = (parsed as { message?: unknown }).message; + return { + type, + message: typeof message === "string" ? message : undefined, + }; + } catch { + return null; + } +} + +function getObject(value: unknown, key: string): Record | null { + if (!value || typeof value !== "object") return null; + const child = (value as Record)[key]; + return child && typeof child === "object" ? child as Record : null; +} + +function getString(value: Record | null, key: string): string | null { + const child = value?.[key]; + return typeof child === "string" ? child : null; +} + +function sendResize(socket: WebSocket | null, fitAddon: XtermFitAddon | null) { + if (!socket || socket.readyState !== WebSocket.OPEN || !fitAddon) return; + const proposed = fitAddon.proposeDimensions(); + if (!proposed || proposed.cols <= 0 || proposed.rows <= 0) return; + socket.send(JSON.stringify({ + type: "resize", + cols: proposed.cols, + rows: proposed.rows, + })); +} + +function statusClasses(status: ConnectionStatus): string { + switch (status) { + case "ready": + return "bg-teal-500 text-on-primary"; + case "error": + return "bg-coral/20 text-coral"; + case "closed": + return "bg-overlay-strong text-fg-3"; + case "connecting": + return "bg-amber/20 text-amber"; + } +} + +function statusLabel(status: ConnectionStatus): string { + switch (status) { + case "ready": + return "Connected"; + case "error": + return "Error"; + case "closed": + return "Closed"; + case "connecting": + return "Connecting"; + } +} + +export default function RunTerminal({ params }: { params: { id: string } }) { + const { push } = useToast(); + const stateQuery = useRunState(params.id); + const sandbox = getObject(getObject(stateQuery.data, "run"), "sandbox") + ?? getObject(stateQuery.data, "sandbox"); + const provider = getString(sandbox, "provider"); + const canCopySsh = provider === "daytona"; + const [connectionKey, setConnectionKey] = useState(0); + const [status, setStatus] = useState("connecting"); + const [error, setError] = useState(null); + const terminalEl = useRef(null); + const terminalRef = useRef(null); + const fitRef = useRef(null); + const socketRef = useRef(null); + const terminalId = useMemo( + () => `run-terminal-${params.id}`, + [params.id], + ); + + const reconnect = useCallback(() => { + setConnectionKey((key) => key + 1); + }, []); + + const copySshCommand = useCallback(async () => { + if (!canCopySsh) return; + try { + const response = await apiData(() => + humanInTheLoopApi.createRunSshAccess(params.id, { ttl_minutes: 60 }), + ); + await navigator.clipboard.writeText(response.command); + push({ message: "SSH command copied." }); + } catch (err) { + push({ + tone: "error", + message: err instanceof Error ? err.message : "Could not copy SSH command.", + }); + } + }, [canCopySsh, params.id, push]); + + useEffect(() => { + if (!terminalEl.current) return undefined; + + let disposed = false; + let resizeObserver: ResizeObserver | null = null; + const textEncoder = new TextEncoder(); + const disposables: Array<{ dispose: () => void }> = []; + + async function connect() { + setStatus("connecting"); + setError(null); + + const [{ Terminal }, { FitAddon }] = await Promise.all([ + import("@xterm/xterm"), + import("@xterm/addon-fit"), + ]); + if (disposed || !terminalEl.current) return; + + const terminal = new Terminal({ + cursorBlink: true, + convertEol: true, + fontFamily: "\"JetBrains Mono\", ui-monospace, monospace", + fontSize: 13, + lineHeight: 1.45, + scrollback: 5000, + theme: TERMINAL_THEME, + }); + const fitAddon = new FitAddon(); + terminal.loadAddon(fitAddon); + terminal.open(terminalEl.current); + fitAddon.fit(); + terminal.focus(); + terminalRef.current = terminal; + fitRef.current = fitAddon; + + const socket = new WebSocket(buildTerminalWebSocketUrl(window.location, params.id)); + socket.binaryType = "arraybuffer"; + socketRef.current = socket; + + disposables.push(terminal.onData((data) => { + if (socket.readyState === WebSocket.OPEN) { + socket.send(textEncoder.encode(data)); + } + })); + + socket.addEventListener("open", () => { + sendResize(socket, fitAddon); + }); + socket.addEventListener("message", (event) => { + if (typeof event.data === "string") { + const message = parseTerminalServerMessage(event.data); + if (!message) return; + if (message.type === "ready") { + setStatus("ready"); + return; + } + if (message.type === "closed") { + setStatus("closed"); + return; + } + setStatus("error"); + setError(message.message ?? "Terminal session failed."); + return; + } + const bytes = event.data instanceof ArrayBuffer + ? new Uint8Array(event.data) + : event.data; + terminal.write(bytes); + }); + socket.addEventListener("close", () => { + setStatus((current) => current === "error" ? current : "closed"); + }); + socket.addEventListener("error", () => { + setStatus("error"); + setError("Terminal WebSocket connection failed."); + }); + + resizeObserver = new ResizeObserver(() => { + fitAddon.fit(); + sendResize(socket, fitAddon); + }); + resizeObserver.observe(terminalEl.current); + } + + void connect(); + + return () => { + disposed = true; + resizeObserver?.disconnect(); + for (const disposable of disposables) disposable.dispose(); + socketRef.current?.send(JSON.stringify({ type: "close" })); + socketRef.current?.close(); + socketRef.current = null; + terminalRef.current?.dispose(); + terminalRef.current = null; + fitRef.current = null; + }; + }, [connectionKey, params.id]); + + return ( +
+
+
+

+ Terminal +

+ {error ? ( +

{error}

+ ) : ( +

+ {provider ? `${provider} sandbox` : "Sandbox terminal"} +

+ )} +
+
+ + {statusLabel(status)} + + + {canCopySsh && ( + + )} +
+
+
+
+
+
+ ); +} diff --git a/apps/fabro-web/package.json b/apps/fabro-web/package.json index 61236b3b2..ac2201678 100644 --- a/apps/fabro-web/package.json +++ b/apps/fabro-web/package.json @@ -20,6 +20,8 @@ "@qltysh/fabro-api-client": "workspace:*", "@tailwindcss/typography": "^0.5.19", "@viz-js/viz": "^3.24.0", + "@xterm/addon-fit": "^0.11.0", + "@xterm/xterm": "^6.0.0", "axios": "^1.7.0", "marked": "^18.0.0", "react": "^19.2.4", diff --git a/bun.lock b/bun.lock index c7cf988e2..de38148f8 100644 --- a/bun.lock +++ b/bun.lock @@ -17,6 +17,8 @@ "@qltysh/fabro-api-client": "workspace:*", "@tailwindcss/typography": "^0.5.19", "@viz-js/viz": "^3.24.0", + "@xterm/addon-fit": "^0.11.0", + "@xterm/xterm": "^6.0.0", "axios": "^1.7.0", "marked": "^18.0.0", "react": "^19.2.4", @@ -605,6 +607,10 @@ "@webassemblyjs/wast-printer": ["@webassemblyjs/wast-printer@1.14.1", "", { "dependencies": { "@webassemblyjs/ast": "1.14.1", "@xtuc/long": "4.2.2" } }, "sha512-kPSSXE6De1XOR820C90RIo2ogvZG+c3KiHzqUoO/F34Y2shGzesfqv7o57xrxovZJH/MetF5UjroJ/R/3isoiw=="], + "@xterm/addon-fit": ["@xterm/addon-fit@0.11.0", "", {}, "sha512-jYcgT6xtVYhnhgxh3QgYDnnNMYTcf8ElbxxFzX0IZo+vabQqSPAjC3c1wJrKB5E19VwQei89QCiZZP86DCPF7g=="], + + "@xterm/xterm": ["@xterm/xterm@6.0.0", "", {}, "sha512-TQwDdQGtwwDt+2cgKDLn0IRaSxYu1tSUjgKarSDkUM0ZNiSRXFpjxEsvc/Zgc5kq5omJ+V0a8/kIM2WD3sMOYg=="], + "@xtuc/ieee754": ["@xtuc/ieee754@1.2.0", "", {}, "sha512-DX8nKgqcGwsc0eJSqYt5lwP4DH5FlHnmuWWBRy7X0NcaGR0ZtuyeESgMwTYVEtxmsNGY+qit4QYT/MIYTOTPeA=="], "@xtuc/long": ["@xtuc/long@4.2.2", "", {}, "sha512-NuHqBY1PB/D8xU6s/thBgOAiAP7HOYDQ32+BFZILJ8ivkUkAHQnWfn6WhL79Owj1qmUnoN/YPhktdIoucipkAQ=="], diff --git a/lib/crates/fabro-sandbox/Cargo.toml b/lib/crates/fabro-sandbox/Cargo.toml index 369dbc7ee..57302c4d8 100644 --- a/lib/crates/fabro-sandbox/Cargo.toml +++ b/lib/crates/fabro-sandbox/Cargo.toml @@ -10,7 +10,7 @@ description = "Sandbox trait and implementations for Fabro agent execution envir default = ["local"] local = [] docker = ["dep:bollard", "dep:tar", "dep:futures", "dep:fabro-github"] -daytona = ["dep:daytona-sdk", "dep:daytona-api-client", "dep:git2", "dep:fabro-github", "dep:fabro-config", "dep:fabro-http", "dep:reqwest-middleware", "dep:chrono", "dep:rand"] +daytona = ["dep:daytona-sdk", "dep:daytona-api-client", "dep:git2", "dep:fabro-github", "dep:fabro-config", "dep:fabro-http", "dep:reqwest-middleware", "dep:chrono", "dep:rand", "dep:tokio-tungstenite", "dep:futures-util", "dep:rustls"] test-support = [] [lib] @@ -57,6 +57,9 @@ daytona-api-client = { workspace = true, optional = true } git2 = { workspace = true, optional = true } fabro-http = { workspace = true, optional = true } reqwest-middleware = { version = "0.5", features = ["json", "multipart", "form", "query"], optional = true } +tokio-tungstenite = { workspace = true, optional = true } +futures-util = { workspace = true, optional = true } +rustls = { version = "0.23", default-features = false, features = ["std", "ring"], optional = true } [dev-dependencies] tokio = { workspace = true, features = ["test-util", "macros"] } diff --git a/lib/crates/fabro-sandbox/src/daytona/mod.rs b/lib/crates/fabro-sandbox/src/daytona/mod.rs index d333b1750..f9e4706c2 100644 --- a/lib/crates/fabro-sandbox/src/daytona/mod.rs +++ b/lib/crates/fabro-sandbox/src/daytona/mod.rs @@ -400,6 +400,10 @@ impl DaytonaSandbox { }) } + pub(crate) fn daytona_id(&self) -> crate::Result<&str> { + Ok(&self.sandbox()?.id) + } + fn repo_cloned(&self) -> bool { self.repo_cloned.get().copied().unwrap_or(false) } diff --git a/lib/crates/fabro-sandbox/src/docker.rs b/lib/crates/fabro-sandbox/src/docker.rs index 8ae8d8dc7..05b4c675f 100644 --- a/lib/crates/fabro-sandbox/src/docker.rs +++ b/lib/crates/fabro-sandbox/src/docker.rs @@ -174,6 +174,14 @@ impl DockerSandbox { }) } + pub(crate) fn container_identifier(&self) -> crate::Result<&str> { + self.container_id() + } + + pub(crate) fn docker_client(&self) -> Docker { + self.docker.clone() + } + fn resolve_container_path(path: &str) -> String { resolve_path(path, WORKING_DIRECTORY) } diff --git a/lib/crates/fabro-sandbox/src/lib.rs b/lib/crates/fabro-sandbox/src/lib.rs index a17296239..014bbb3bc 100644 --- a/lib/crates/fabro-sandbox/src/lib.rs +++ b/lib/crates/fabro-sandbox/src/lib.rs @@ -19,6 +19,8 @@ pub mod sandbox_record; pub mod worktree; +pub mod terminal; + pub mod local; #[cfg(feature = "docker")] @@ -45,4 +47,5 @@ pub use sandbox::{ pub use sandbox_provider::SandboxProvider; pub use sandbox_record::SandboxRecord; pub use sandbox_spec::SandboxSpec; +pub use terminal::{TerminalSession, TerminalSize, open_terminal_for_run}; pub use worktree::{WorktreeEvent, WorktreeEventCallback, WorktreeOptions, WorktreeSandbox}; diff --git a/lib/crates/fabro-sandbox/src/terminal.rs b/lib/crates/fabro-sandbox/src/terminal.rs new file mode 100644 index 000000000..88d1b5a4a --- /dev/null +++ b/lib/crates/fabro-sandbox/src/terminal.rs @@ -0,0 +1,730 @@ +use async_trait::async_trait; +#[cfg(feature = "daytona")] +use fabro_static::EnvVars; +use fabro_types::RunId; + +#[cfg(any(feature = "daytona", feature = "docker"))] +use crate::Sandbox; +#[cfg(feature = "daytona")] +use crate::daytona::{DEFAULT_DAYTONA_API_URL, DaytonaSandbox}; +#[cfg(feature = "docker")] +use crate::docker::DockerSandbox; +use crate::{SandboxProvider, SandboxRecord}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct TerminalSize { + pub cols: u16, + pub rows: u16, +} + +impl Default for TerminalSize { + fn default() -> Self { + Self { + cols: 120, + rows: 32, + } + } +} + +#[async_trait] +pub trait TerminalSession: Send + Sync { + async fn write_input(&self, bytes: &[u8]) -> crate::Result<()>; + async fn read_output(&self) -> crate::Result>>; + async fn resize(&self, size: TerminalSize) -> crate::Result<()>; + async fn close(&self) -> crate::Result<()>; +} + +pub async fn open_terminal_for_run( + record: &SandboxRecord, + daytona_api_key: Option, + daytona_organization_id: Option, + run_id: Option, + size: TerminalSize, +) -> crate::Result> { + #[cfg(not(feature = "daytona"))] + let _ = (&daytona_api_key, &daytona_organization_id); + #[cfg(not(feature = "docker"))] + let _ = &run_id; + #[cfg(not(any(feature = "daytona", feature = "docker")))] + let _ = size; + + match record.provider.as_str() { + #[cfg(feature = "daytona")] + "daytona" => { + let identifier = record.identifier.as_deref().ok_or_else(|| { + crate::Error::message("Daytona sandbox record is missing the sandbox identifier") + })?; + let repo_cloned = record.repo_cloned.ok_or_else(|| { + crate::Error::message("Daytona sandbox record is missing clone metadata") + })?; + let sandbox = DaytonaSandbox::reconnect( + identifier, + daytona_api_key.clone(), + repo_cloned, + record.clone_origin_url.clone(), + record.clone_branch.clone(), + ) + .await?; + sandbox.start().await?; + let api_key = resolve_daytona_api_key(daytona_api_key)?; + let organization_id = resolve_daytona_organization_id(daytona_organization_id); + let session = DaytonaTerminalSession::open( + &sandbox, + api_key, + organization_id, + daytona_api_base_url(), + size, + ) + .await?; + Ok(Box::new(session)) + } + #[cfg(feature = "docker")] + "docker" => { + let identifier = record.identifier.as_deref().ok_or_else(|| { + crate::Error::message("Docker sandbox record is missing the container identifier") + })?; + let repo_cloned = record.repo_cloned.ok_or_else(|| { + crate::Error::message("Docker sandbox record is missing clone metadata") + })?; + let sandbox = DockerSandbox::reconnect( + identifier, + repo_cloned, + record.clone_origin_url.clone(), + record.clone_branch.clone(), + run_id, + ) + .await?; + sandbox.start().await?; + let session = DockerTerminalSession::open(&sandbox, size).await?; + Ok(Box::new(session)) + } + provider if provider == SandboxProvider::Local.to_string() => Err(crate::Error::message( + "Local sandboxes do not support embedded terminals", + )), + other => Err(crate::Error::message(format!( + "Sandbox provider '{other}' does not support embedded terminals" + ))), + } +} + +#[cfg(feature = "daytona")] +#[expect( + clippy::disallowed_methods, + reason = "Terminal reconnect falls back to the process environment when no vault value was supplied." +)] +fn resolve_daytona_api_key(api_key: Option) -> crate::Result { + api_key + .or_else(|| std::env::var(EnvVars::DAYTONA_API_KEY).ok()) + .ok_or_else(|| crate::Error::message("DAYTONA_API_KEY is required for Daytona terminals")) +} + +#[cfg(feature = "daytona")] +#[expect( + clippy::disallowed_methods, + reason = "Daytona SDK configuration convention uses process environment fallbacks for API URLs." +)] +fn daytona_api_base_url() -> String { + std::env::var(EnvVars::DAYTONA_API_URL) + .or_else(|_| std::env::var(EnvVars::DAYTONA_SERVER_URL)) + .unwrap_or_else(|_| DEFAULT_DAYTONA_API_URL.to_string()) +} + +#[cfg(feature = "daytona")] +#[expect( + clippy::disallowed_methods, + reason = "Terminal reconnect falls back to the process environment when no vault value was supplied." +)] +fn resolve_daytona_organization_id(organization_id: Option) -> Option { + organization_id.or_else(|| std::env::var(EnvVars::DAYTONA_ORGANIZATION_ID).ok()) +} + +#[cfg(feature = "daytona")] +mod daytona_terminal { + use std::collections::HashMap; + use std::sync::Once; + + use async_trait::async_trait; + use futures_util::stream::{SplitSink, SplitStream}; + use futures_util::{SinkExt, StreamExt}; + use rustls::crypto::ring; + use serde::{Deserialize, Serialize}; + use tokio::net::TcpStream; + use tokio::runtime::Handle; + use tokio::sync::Mutex; + use tokio_tungstenite::tungstenite::error::ProtocolError; + use tokio_tungstenite::tungstenite::handshake::client; + use tokio_tungstenite::tungstenite::http::Request; + use tokio_tungstenite::tungstenite::protocol::Message as ProviderMessage; + use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, connect_async, tungstenite}; + + use super::{TerminalSession, TerminalSize}; + use crate::Sandbox; + use crate::daytona::DaytonaSandbox; + + type ProviderWs = WebSocketStream>; + type ProviderSink = SplitSink; + type ProviderStream = SplitStream; + + static RUSTLS_PROVIDER: Once = Once::new(); + + pub(super) struct DaytonaTerminalSession { + api_base_url: String, + api_key: String, + org_id: Option, + sandbox_id: String, + session_id: String, + write: Mutex>, + read: Mutex>, + closed: Mutex, + } + + #[derive(Serialize)] + #[serde(rename_all = "camelCase")] + struct DaytonaPtyCreateRequest { + cols: u16, + rows: u16, + cwd: String, + envs: HashMap, + lazy_start: bool, + } + + #[derive(Deserialize)] + #[serde(rename_all = "camelCase")] + struct DaytonaPtyCreateResponse { + session_id: String, + } + + #[derive(Serialize)] + struct DaytonaPtyResizeRequest { + cols: u16, + rows: u16, + } + + impl DaytonaTerminalSession { + pub(super) async fn open( + sandbox: &DaytonaSandbox, + api_key: String, + org_id: Option, + api_base_url: String, + size: TerminalSize, + ) -> crate::Result { + ensure_rustls_provider(); + let sandbox_id = sandbox.daytona_id()?.to_string(); + let session_id = create_pty_session( + &api_base_url, + &api_key, + org_id.as_deref(), + &sandbox_id, + sandbox.working_directory().to_string(), + size, + ) + .await?; + let ws_url = daytona_pty_ws_url(&api_base_url, &sandbox_id, &session_id)?; + let request = daytona_ws_request(&ws_url, &api_key, org_id.as_deref())?; + let (stream, _) = connect_async(request).await.map_err(|err| { + crate::Error::context("Failed to connect Daytona terminal WebSocket", err) + })?; + let (write, read) = stream.split(); + Ok(Self { + api_base_url, + api_key, + org_id, + sandbox_id, + session_id, + write: Mutex::new(Some(write)), + read: Mutex::new(Some(read)), + closed: Mutex::new(false), + }) + } + + async fn kill_session(&self) -> crate::Result<()> { + let url = format!( + "{}/toolbox/{}/toolbox/process/pty/{}", + trim_slash(&self.api_base_url), + url_component(&self.sandbox_id), + url_component(&self.session_id) + ); + let mut request = fabro_http::http_client() + .map_err(|err| crate::Error::context("Failed to build HTTP client", err))? + .delete(url) + .bearer_auth(&self.api_key); + if let Some(org_id) = self.org_id.as_deref() { + request = request.header("X-Daytona-Organization-ID", org_id); + } + let response = request.send().await.map_err(|err| { + crate::Error::context("Failed to delete Daytona PTY session", err) + })?; + if !response.status().is_success() + && response.status() != fabro_http::StatusCode::NOT_FOUND + { + return Err(crate::Error::message(format!( + "Failed to delete Daytona PTY session: HTTP {}", + response.status() + ))); + } + Ok(()) + } + } + + #[async_trait] + impl TerminalSession for DaytonaTerminalSession { + async fn write_input(&self, bytes: &[u8]) -> crate::Result<()> { + let mut write = self.write.lock().await; + let Some(write) = write.as_mut() else { + return Ok(()); + }; + write + .send(ProviderMessage::Binary(bytes.to_vec().into())) + .await + .map_err(|err| crate::Error::context("Failed to write Daytona terminal input", err)) + } + + async fn read_output(&self) -> crate::Result>> { + let mut read = self.read.lock().await; + let Some(read) = read.as_mut() else { + return Ok(None); + }; + while let Some(message) = read.next().await { + match message { + Ok(ProviderMessage::Binary(bytes)) => return Ok(Some(bytes.to_vec())), + Ok(ProviderMessage::Text(text)) => { + return Ok(Some(text.as_str().as_bytes().to_vec())); + } + Ok(ProviderMessage::Close(_)) + | Err(tungstenite::Error::Protocol( + ProtocolError::ResetWithoutClosingHandshake, + )) => return Ok(None), + Ok( + ProviderMessage::Ping(_) + | ProviderMessage::Pong(_) + | ProviderMessage::Frame(_), + ) => {} + Err(err) => { + return Err(crate::Error::context( + "Failed to read Daytona terminal output", + err, + )); + } + } + } + Ok(None) + } + + async fn resize(&self, size: TerminalSize) -> crate::Result<()> { + let url = format!( + "{}/toolbox/{}/toolbox/process/pty/{}/resize", + trim_slash(&self.api_base_url), + url_component(&self.sandbox_id), + url_component(&self.session_id) + ); + let mut request = fabro_http::http_client() + .map_err(|err| crate::Error::context("Failed to build HTTP client", err))? + .post(url) + .bearer_auth(&self.api_key) + .json(&DaytonaPtyResizeRequest { + cols: size.cols, + rows: size.rows, + }); + if let Some(org_id) = self.org_id.as_deref() { + request = request.header("X-Daytona-Organization-ID", org_id); + } + let response = request + .send() + .await + .map_err(|err| crate::Error::context("Failed to resize Daytona terminal", err))?; + if !response.status().is_success() { + return Err(crate::Error::message(format!( + "Failed to resize Daytona terminal: HTTP {}", + response.status() + ))); + } + Ok(()) + } + + async fn close(&self) -> crate::Result<()> { + let mut closed = self.closed.lock().await; + if *closed { + return Ok(()); + } + *closed = true; + drop(closed); + + if let Some(mut write) = self.write.lock().await.take() { + let _ = write.send(ProviderMessage::Close(None)).await; + } + let _ = self.read.lock().await.take(); + self.kill_session().await + } + } + + impl Drop for DaytonaTerminalSession { + fn drop(&mut self) { + let api_base_url = self.api_base_url.clone(); + let api_key = self.api_key.clone(); + let org_id = self.org_id.clone(); + let sandbox_id = self.sandbox_id.clone(); + let session_id = self.session_id.clone(); + if let Ok(handle) = Handle::try_current() { + handle.spawn(async move { + let url = format!( + "{}/toolbox/{}/toolbox/process/pty/{}", + trim_slash(&api_base_url), + url_component(&sandbox_id), + url_component(&session_id) + ); + let Ok(client) = fabro_http::http_client() else { + return; + }; + let mut request = client.delete(url).bearer_auth(api_key); + if let Some(org_id) = org_id.as_deref() { + request = request.header("X-Daytona-Organization-ID", org_id); + } + if let Err(err) = request.send().await { + tracing::warn!(error = %err, "failed to clean up Daytona terminal session"); + } + }); + } + } + } + + async fn create_pty_session( + api_base_url: &str, + api_key: &str, + org_id: Option<&str>, + sandbox_id: &str, + cwd: String, + size: TerminalSize, + ) -> crate::Result { + let mut envs = HashMap::new(); + envs.insert("TERM".to_string(), "xterm-256color".to_string()); + envs.insert("LANG".to_string(), "C.UTF-8".to_string()); + let url = format!( + "{}/toolbox/{}/toolbox/process/pty", + trim_slash(api_base_url), + url_component(sandbox_id) + ); + let mut request = fabro_http::http_client() + .map_err(|err| crate::Error::context("Failed to build HTTP client", err))? + .post(url) + .bearer_auth(api_key) + .json(&DaytonaPtyCreateRequest { + cols: size.cols, + rows: size.rows, + cwd, + envs, + lazy_start: false, + }); + if let Some(org_id) = org_id { + request = request.header("X-Daytona-Organization-ID", org_id); + } + let response = request + .send() + .await + .map_err(|err| crate::Error::context("Failed to create Daytona PTY session", err))?; + if !response.status().is_success() { + return Err(crate::Error::message(format!( + "Failed to create Daytona PTY session: HTTP {}", + response.status() + ))); + } + let body = response + .json::() + .await + .map_err(|err| crate::Error::context("Failed to decode Daytona PTY response", err))?; + Ok(body.session_id) + } + + fn daytona_pty_ws_url( + api_base_url: &str, + sandbox_id: &str, + session_id: &str, + ) -> crate::Result { + let base = trim_slash(api_base_url); + let ws_base = if let Some(rest) = base.strip_prefix("https://") { + format!("wss://{rest}") + } else if let Some(rest) = base.strip_prefix("http://") { + format!("ws://{rest}") + } else { + return Err(crate::Error::message( + "Daytona API URL must start with http:// or https://", + )); + }; + Ok(format!( + "{}/toolbox/{}/toolbox/process/pty/{}/connect", + ws_base, + url_component(sandbox_id), + url_component(session_id) + )) + } + + fn daytona_ws_request( + ws_url: &str, + api_key: &str, + org_id: Option<&str>, + ) -> crate::Result> { + let mut request = Request::builder() + .uri(ws_url) + .header("Host", extract_host(ws_url)) + .header("Connection", "Upgrade") + .header("Upgrade", "websocket") + .header("Sec-WebSocket-Version", "13") + .header("Sec-WebSocket-Key", client::generate_key()) + .header("Authorization", format!("Bearer {api_key}")) + .header("X-Daytona-Source", "fabro"); + if let Some(org_id) = org_id { + request = request.header("X-Daytona-Organization-ID", org_id); + } + request.body(()).map_err(|err| { + crate::Error::context("Failed to build Daytona terminal WebSocket request", err) + }) + } + + fn ensure_rustls_provider() { + RUSTLS_PROVIDER.call_once(|| { + let _ = ring::default_provider().install_default(); + }); + } + + pub(super) fn trim_slash(value: &str) -> &str { + value.trim_end_matches('/') + } + + pub(super) fn url_component(value: &str) -> String { + value.replace('/', "%2F") + } + + fn extract_host(ws_url: &str) -> String { + ws_url + .strip_prefix("wss://") + .or_else(|| ws_url.strip_prefix("ws://")) + .and_then(|rest| rest.split('/').next()) + .unwrap_or_default() + .to_string() + } + + #[cfg(test)] + mod tests { + use super::*; + + #[test] + fn builds_daytona_pty_websocket_url() { + assert_eq!( + daytona_pty_ws_url("https://app.daytona.io/api/", "sandbox/a", "pty-1").unwrap(), + "wss://app.daytona.io/api/toolbox/sandbox%2Fa/toolbox/process/pty/pty-1/connect" + ); + } + } +} + +#[cfg(feature = "daytona")] +use daytona_terminal::DaytonaTerminalSession; + +#[cfg(feature = "docker")] +mod docker_terminal { + use std::pin::Pin; + use std::sync::atomic::{AtomicU64, Ordering}; + + use async_trait::async_trait; + use bollard::Docker; + use bollard::container::LogOutput; + use bollard::errors::Error as DockerError; + use bollard::exec::{CreateExecOptions, ResizeExecOptions, StartExecResults}; + use futures::{Stream, StreamExt}; + use tokio::io::{AsyncWrite, AsyncWriteExt}; + use tokio::sync::Mutex; + + use super::{TerminalSession, TerminalSize}; + use crate::Sandbox; + use crate::docker::DockerSandbox; + + type DockerInput = Pin>; + type DockerOutput = Pin> + Send>>; + + pub(super) struct DockerTerminalSession { + docker: Docker, + container_id: String, + exec_id: String, + pid_file: String, + input: Mutex>, + output: Mutex>, + closed: Mutex, + } + + impl DockerTerminalSession { + pub(super) async fn open( + sandbox: &DockerSandbox, + size: TerminalSize, + ) -> crate::Result { + let docker = sandbox.docker_client(); + let container_id = sandbox.container_identifier()?.to_string(); + let pid_file = format!("/tmp/fabro-terminal-{}.pid", uuid_fragment()); + let exec_opts = docker_terminal_exec_options(sandbox.working_directory(), &pid_file); + let exec = docker + .create_exec(&container_id, exec_opts) + .await + .map_err(|err| { + crate::Error::context("Failed to create Docker terminal exec", err) + })?; + let exec_id = exec.id; + let start = docker.start_exec(&exec_id, None).await.map_err(|err| { + crate::Error::context("Failed to start Docker terminal exec", err) + })?; + let StartExecResults::Attached { output, input } = start else { + return Err(crate::Error::message("Docker terminal exec did not attach")); + }; + docker + .resize_exec(&exec_id, ResizeExecOptions { + height: size.rows, + width: size.cols, + }) + .await + .map_err(|err| { + crate::Error::context("Failed to resize Docker terminal exec", err) + })?; + Ok(Self { + docker, + container_id, + exec_id, + pid_file, + input: Mutex::new(Some(input)), + output: Mutex::new(Some(output)), + closed: Mutex::new(false), + }) + } + + async fn kill_shell(&self) -> crate::Result<()> { + let command = format!( + "if [ -f {pid_file} ]; then kill -TERM \"$(cat {pid_file})\" 2>/dev/null || true; rm -f {pid_file}; fi", + pid_file = crate::shell_quote(&self.pid_file), + ); + let exec = self + .docker + .create_exec(&self.container_id, CreateExecOptions { + cmd: Some(vec!["sh".to_string(), "-lc".to_string(), command]), + attach_stdout: Some(false), + attach_stderr: Some(false), + ..Default::default() + }) + .await + .map_err(|err| { + crate::Error::context("Failed to create Docker terminal cleanup exec", err) + })?; + self.docker + .start_exec(&exec.id, None) + .await + .map_err(|err| { + crate::Error::context("Failed to run Docker terminal cleanup exec", err) + })?; + Ok(()) + } + } + + #[async_trait] + impl TerminalSession for DockerTerminalSession { + async fn write_input(&self, bytes: &[u8]) -> crate::Result<()> { + let mut input = self.input.lock().await; + let Some(input) = input.as_mut() else { + return Ok(()); + }; + input + .write_all(bytes) + .await + .map_err(|err| crate::Error::context("Failed to write Docker terminal input", err)) + } + + async fn read_output(&self) -> crate::Result>> { + let mut output = self.output.lock().await; + let Some(output) = output.as_mut() else { + return Ok(None); + }; + match output.next().await { + Some(Ok(chunk)) => Ok(Some(chunk.into_bytes().to_vec())), + Some(Err(err)) => Err(crate::Error::context( + "Failed to read Docker terminal output", + err, + )), + None => Ok(None), + } + } + + async fn resize(&self, size: TerminalSize) -> crate::Result<()> { + self.docker + .resize_exec(&self.exec_id, ResizeExecOptions { + height: size.rows, + width: size.cols, + }) + .await + .map_err(|err| crate::Error::context("Failed to resize Docker terminal exec", err)) + } + + async fn close(&self) -> crate::Result<()> { + let mut closed = self.closed.lock().await; + if *closed { + return Ok(()); + } + *closed = true; + drop(closed); + let _ = self.input.lock().await.take(); + let _ = self.output.lock().await.take(); + self.kill_shell().await + } + } + + fn docker_terminal_exec_options( + working_directory: &str, + pid_file: &str, + ) -> CreateExecOptions { + let command = format!( + "printf '%s\\n' $$ > {pid_file}; exec sh -l", + pid_file = crate::shell_quote(pid_file), + ); + CreateExecOptions { + attach_stdin: Some(true), + attach_stdout: Some(true), + attach_stderr: Some(true), + tty: Some(true), + cmd: Some(vec!["sh".to_string(), "-lc".to_string(), command]), + working_dir: Some(working_directory.to_string()), + env: Some(vec![ + "TERM=xterm-256color".to_string(), + "LANG=C.UTF-8".to_string(), + ]), + ..Default::default() + } + } + + static DOCKER_TERMINAL_COUNTER: AtomicU64 = AtomicU64::new(1); + + fn uuid_fragment() -> String { + format!( + "{:016x}", + DOCKER_TERMINAL_COUNTER.fetch_add(1, Ordering::Relaxed) + ) + } + + #[cfg(test)] + mod tests { + use super::*; + + #[test] + fn docker_terminal_exec_options_attach_tty_and_workspace_env() { + let options = docker_terminal_exec_options("/workspace", "/tmp/fabro-terminal.pid"); + assert_eq!(options.attach_stdin, Some(true)); + assert_eq!(options.attach_stdout, Some(true)); + assert_eq!(options.attach_stderr, Some(true)); + assert_eq!(options.tty, Some(true)); + assert_eq!(options.working_dir.as_deref(), Some("/workspace")); + assert_eq!( + options.env, + Some(vec![ + "TERM=xterm-256color".to_string(), + "LANG=C.UTF-8".to_string() + ]) + ); + assert!(options.cmd.unwrap().join(" ").contains("exec sh -l")); + } + } +} + +#[cfg(feature = "docker")] +use docker_terminal::DockerTerminalSession; diff --git a/lib/crates/fabro-server/Cargo.toml b/lib/crates/fabro-server/Cargo.toml index a6dad9a56..e78933660 100644 --- a/lib/crates/fabro-server/Cargo.toml +++ b/lib/crates/fabro-server/Cargo.toml @@ -31,7 +31,7 @@ fabro-interview = { path = "../fabro-interview" } fabro-slack = { path = "../fabro-slack" } fabro-workflow = { path = "../fabro-workflow" } fabro-validate = { path = "../fabro-validate" } -fabro-sandbox = { path = "../fabro-sandbox", features = ["daytona"] } +fabro-sandbox = { path = "../fabro-sandbox", features = ["daytona", "docker"] } fabro-github = { path = "../fabro-github" } fabro-agent = { path = "../fabro-agent" } fabro-llm = { path = "../fabro-llm" } @@ -47,7 +47,7 @@ fabro-redact.workspace = true fabro-static.workspace = true chrono.workspace = true futures-util.workspace = true -axum.workspace = true +axum = { workspace = true, features = ["ws"] } axum-extra.workspace = true cookie.workspace = true dirs.workspace = true diff --git a/lib/crates/fabro-server/src/server/handler/sandbox.rs b/lib/crates/fabro-server/src/server/handler/sandbox.rs index 4493b96c8..a75f83f85 100644 --- a/lib/crates/fabro-server/src/server/handler/sandbox.rs +++ b/lib/crates/fabro-server/src/server/handler/sandbox.rs @@ -1,17 +1,24 @@ use std::sync::Arc; +use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; +use fabro_sandbox::{TerminalSize, open_terminal_for_run}; + use super::super::{ - ApiError, AppState, Bytes, DaytonaSandbox, EnvVars, IntoResponse, Json, NamedTempFile, Path, - PreviewUrlRequest, PreviewUrlResponse, Query, RequiredUser, Response, Router, RunId, Sandbox, - SandboxFileEntry, SandboxFileListResponse, SandboxProvider, SshAccessRequest, - SshAccessResponse, State, StatusCode, collect_causes, fs, get, octet_stream_response, - parse_run_id_path, post, reconnect_for_run, reject_if_archived, render_with_causes, + ApiError, AppState, Bytes, DaytonaSandbox, EnvVars, HeaderMap, IntoResponse, Json, + NamedTempFile, Path, PreviewUrlRequest, PreviewUrlResponse, Query, RequiredUser, Response, + Router, RunId, Sandbox, SandboxFileEntry, SandboxFileListResponse, SandboxProvider, + SshAccessRequest, SshAccessResponse, State, StatusCode, collect_causes, fs, get, + octet_stream_response, parse_run_id_path, post, reconnect_for_run, reject_if_archived, + render_with_causes, }; +const MAX_TERMINAL_CONTROL_BYTES: usize = 4096; + pub(super) fn routes() -> Router> { Router::new() .route("/runs/{id}/preview", post(generate_preview_url)) .route("/runs/{id}/ssh", post(create_ssh_access)) + .route("/runs/{id}/terminal", get(run_terminal)) .route("/runs/{id}/sandbox/files", get(list_sandbox_files)) .route( "/runs/{id}/sandbox/file", @@ -31,6 +38,203 @@ struct SandboxFileParams { path: String, } +#[derive(Debug, PartialEq, Eq)] +enum TerminalClientMessage { + Resize(TerminalSize), + Close, +} + +#[derive(serde::Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum TerminalClientControl { + Resize { cols: u16, rows: u16 }, + Close, +} + +fn parse_terminal_control_message(text: &str) -> Result { + if text.len() > MAX_TERMINAL_CONTROL_BYTES { + return Err("Terminal control message is too large."); + } + match serde_json::from_str::(text) { + Ok(TerminalClientControl::Resize { cols, rows }) if cols > 0 && rows > 0 => { + Ok(TerminalClientMessage::Resize(TerminalSize { cols, rows })) + } + Ok(TerminalClientControl::Resize { .. }) => { + Err("Terminal resize dimensions must be greater than zero.") + } + Ok(TerminalClientControl::Close) => Ok(TerminalClientMessage::Close), + Err(_) => Err("Invalid terminal control message."), + } +} + +fn terminal_server_text(message_type: &str, message: Option<&str>) -> WsMessage { + let payload = match message { + Some(message) => serde_json::json!({ "type": message_type, "message": message }), + None => serde_json::json!({ "type": message_type }), + }; + WsMessage::Text(payload.to_string().into()) +} + +#[expect( + clippy::disallowed_types, + reason = "The Origin header URL is parsed only for same-origin validation and is never logged." +)] +fn origin_allowed(headers: &HeaderMap) -> bool { + let Some(origin) = headers.get("origin").and_then(|value| value.to_str().ok()) else { + return true; + }; + let Some(host) = headers.get("host").and_then(|value| value.to_str().ok()) else { + return false; + }; + let Ok(origin_url) = url::Url::parse(origin) else { + return false; + }; + let Some(origin_host) = origin_url.host_str() else { + return false; + }; + let origin_authority = match origin_url.port_or_known_default() { + Some(port) => format!("{origin_host}:{port}"), + None => origin_host.to_string(), + }; + origin_authority.eq_ignore_ascii_case(host) +} + +async fn run_terminal( + _auth: RequiredUser, + State(state): State>, + Path(id): Path, + headers: HeaderMap, + ws: WebSocketUpgrade, +) -> Response { + let id = match parse_run_id_path(&id) { + Ok(id) => id, + Err(response) => return response, + }; + if !origin_allowed(&headers) { + return ApiError::new(StatusCode::FORBIDDEN, "WebSocket origin is not allowed.") + .into_response(); + } + ws.on_upgrade(move |socket| terminal_websocket(socket, state, id)) +} + +async fn terminal_websocket(mut socket: WebSocket, state: Arc, id: RunId) { + let record = match load_run_sandbox_record(&state, &id).await { + Ok(record) => record, + Err(response) => { + let message = terminal_error_from_status(response.status()); + let _ = socket + .send(terminal_server_text("error", Some(&message))) + .await; + return; + } + }; + let daytona_api_key = state.vault_or_env(EnvVars::DAYTONA_API_KEY); + let daytona_organization_id = state.vault_or_env(EnvVars::DAYTONA_ORGANIZATION_ID); + let session = match open_terminal_for_run( + &record, + daytona_api_key, + daytona_organization_id, + Some(id), + TerminalSize::default(), + ) + .await + { + Ok(session) => session, + Err(err) => { + let _ = socket + .send(terminal_server_text( + "error", + Some(&err.display_with_causes()), + )) + .await; + return; + } + }; + + if socket + .send(terminal_server_text("ready", None)) + .await + .is_err() + { + let _ = session.close().await; + return; + } + + loop { + tokio::select! { + message = socket.recv() => { + let Some(message) = message else { + break; + }; + match message { + Ok(WsMessage::Binary(bytes)) => { + if let Err(err) = session.write_input(&bytes).await { + let _ = socket + .send(terminal_server_text("error", Some(&err.display_with_causes()))) + .await; + break; + } + } + Ok(WsMessage::Text(text)) => { + match parse_terminal_control_message(text.as_str()) { + Ok(TerminalClientMessage::Resize(size)) => { + if let Err(err) = session.resize(size).await { + let _ = socket + .send(terminal_server_text("error", Some(&err.display_with_causes()))) + .await; + break; + } + } + Ok(TerminalClientMessage::Close) => { + let _ = socket.send(terminal_server_text("closed", None)).await; + break; + } + Err(message) => { + let _ = socket.send(terminal_server_text("error", Some(message))).await; + } + } + } + Ok(WsMessage::Close(_)) => break, + Ok(WsMessage::Ping(_) | WsMessage::Pong(_)) => {} + Err(err) => { + tracing::debug!(error = %err, run_id = %id, "run terminal websocket closed with error"); + break; + } + } + } + output = session.read_output() => { + match output { + Ok(Some(bytes)) => { + if socket.send(WsMessage::Binary(bytes.into())).await.is_err() { + break; + } + } + Ok(None) => { + let _ = socket.send(terminal_server_text("closed", None)).await; + break; + } + Err(err) => { + let _ = socket + .send(terminal_server_text("error", Some(&err.display_with_causes()))) + .await; + break; + } + } + } + } + } + if let Err(err) = session.close().await { + tracing::warn!(error = %err.display_with_causes(), run_id = %id, "failed to close run terminal session"); + } +} + +fn terminal_error_from_status(status: StatusCode) -> String { + status + .canonical_reason() + .unwrap_or("Terminal unavailable") + .to_string() +} + async fn generate_preview_url( _auth: RequiredUser, State(state): State>, @@ -284,3 +488,52 @@ async fn load_run_sandbox_record( Err(_) => Err(ApiError::not_found("Run not found.").into_response()), } } + +#[cfg(test)] +mod tests { + use axum::http::{HeaderMap, HeaderValue}; + + use super::*; + + #[test] + fn terminal_control_accepts_resize_and_close() { + assert_eq!( + parse_terminal_control_message(r#"{"type":"resize","cols":120,"rows":32}"#), + Ok(TerminalClientMessage::Resize(TerminalSize { + cols: 120, + rows: 32, + })) + ); + assert_eq!( + parse_terminal_control_message(r#"{"type":"close"}"#), + Ok(TerminalClientMessage::Close) + ); + } + + #[test] + fn terminal_control_rejects_malformed_oversized_and_zero_resize() { + assert!(parse_terminal_control_message("{").is_err()); + assert!(parse_terminal_control_message(r#"{"type":"resize","cols":0,"rows":32}"#).is_err()); + assert!( + parse_terminal_control_message(&"x".repeat(MAX_TERMINAL_CONTROL_BYTES + 1)).is_err() + ); + } + + #[test] + fn origin_validation_allows_absent_and_same_origin() { + assert!(origin_allowed(&HeaderMap::new())); + + let mut headers = HeaderMap::new(); + headers.insert("host", HeaderValue::from_static("127.0.0.1:4187")); + headers.insert("origin", HeaderValue::from_static("http://127.0.0.1:4187")); + assert!(origin_allowed(&headers)); + } + + #[test] + fn origin_validation_rejects_cross_origin_browser_origin() { + let mut headers = HeaderMap::new(); + headers.insert("host", HeaderValue::from_static("127.0.0.1:4187")); + headers.insert("origin", HeaderValue::from_static("https://evil.example")); + assert!(!origin_allowed(&headers)); + } +}