From 96f299bf8a00ae493a410b425e545aa430928403 Mon Sep 17 00:00:00 2001 From: Zef Hemel Date: Thu, 27 Mar 2025 17:04:15 +0100 Subject: [PATCH] Shell streaming support (Lua) --- common/shell_stream_client.ts | 254 +++++++++++ common/spaces/http_space_primitives.ts | 34 +- plug-api/syscalls/shell.ts | 34 ++ server/http_server.ts | 34 +- server/shell.test.ts | 583 +++++++++++++++++++++++++ server/shell_backend.ts | 209 +++++++++ server/shell_endpoints.ts | 125 ++++++ web/syscalls/shell.ts | 37 ++ 8 files changed, 1294 insertions(+), 16 deletions(-) create mode 100644 common/shell_stream_client.ts create mode 100644 server/shell.test.ts create mode 100644 server/shell_endpoints.ts diff --git a/common/shell_stream_client.ts b/common/shell_stream_client.ts new file mode 100644 index 00000000..9afdbc8e --- /dev/null +++ b/common/shell_stream_client.ts @@ -0,0 +1,254 @@ +import type { HttpSpacePrimitives } from "./spaces/http_space_primitives.ts"; + +/** + * Event types emitted by the ShellStreamClient + */ +export type ShellStreamEventType = + | "stdout" + | "stderr" + | "exit" + | "error" + | "open" + | "close"; + +/** + * Event object emitted by the ShellStreamClient + */ +export interface ShellStreamEvent { + type: ShellStreamEventType; + data: string | { code: number }; +} + +/** + * Options for creating a ShellStreamClient + */ +export interface ShellStreamOptions { + /** + * HTTP space primitives for authenticated requests + */ + httpSpacePrimitives: HttpSpacePrimitives; + + /** + * Command to execute + */ + cmd: string; + + /** + * Command arguments + */ + args?: string[]; +} + +/** + * Client for interacting with the shell stream WebSocket API + */ +export class ShellStreamClient { + private ws: WebSocket | null = null; + private eventHandlers = new Map< + ShellStreamEventType, + Array<(event: ShellStreamEvent) => void> + >(); + + constructor(private options: ShellStreamOptions) { + // Set default options + this.options.args = options.args ?? []; + } + + /** + * Connect to the shell stream WebSocket + */ + public async start(): Promise { + if (this.ws) { + return Promise.resolve(); // Already connected + } + + // Create an authenticated WebSocket connection + this.ws = await this.options.httpSpacePrimitives + .createAuthenticatedWebSocket( + ".shell/stream", + { + cmd: this.options.cmd, + args: JSON.stringify(this.options.args || []), + }, + ); + + // Set up event handlers + this.setupEventHandlers(); + + // Wait for the connection to be established + return new Promise((resolve, reject) => { + if (!this.ws) { + reject(new Error("WebSocket not initialized")); + return; + } + + const onOpen = () => { + this.ws?.removeEventListener("open", onOpen); + this.ws?.removeEventListener("error", onError); + resolve(); + }; + + const onError = (error: Event) => { + this.ws?.removeEventListener("open", onOpen); + this.ws?.removeEventListener("error", onError); + reject(new Error(`Failed to connect: ${error}`)); + }; + + this.ws.addEventListener("open", onOpen); + this.ws.addEventListener("error", onError); + }); + } + + /** + * Set up WebSocket event handlers + * @private + */ + private setupEventHandlers(): void { + if (!this.ws) return; + + // Set up event handlers + this.ws.onopen = () => { + this.emitEvent({ type: "open", data: "Connected" }); + }; + + this.ws.onmessage = (event) => { + try { + const message = JSON.parse(event.data); + // Special handling for exit events + if (message.type === "exit") { + // Parse the exit code from the JSON string if needed + let exitData = message.data; + if (typeof exitData === "string" && typeof exitData === "string") { + try { + // Try to parse as JSON in case it's a stringified object + const parsedData = JSON.parse(exitData); + if ( + typeof parsedData === "object" && parsedData !== null && + "code" in parsedData + ) { + exitData = parsedData; + } + } catch (e) { + // If it's not valid JSON, just use it as is + console.error("Failed to parse exit data:", e); + } + } + + this.emitEvent({ + type: message.type, + data: exitData, + }); + } else { + // Regular event + this.emitEvent({ + type: message.type, + data: message.data, + }); + } + } catch (e) { + const error = e as Error; + this.emitEvent({ + type: "error", + data: `Error parsing message: ${error.message}`, + }); + } + }; + + this.ws.onclose = () => { + this.ws = null; + this.emitEvent({ type: "close", data: "Disconnected" }); + }; + + this.ws.onerror = (error) => { + this.emitEvent({ + type: "error", + data: `WebSocket error: ${error}`, + }); + }; + } + + /** + * Close the WebSocket connection + */ + public close(): void { + if (this.ws) { + this.ws.close(); + this.ws = null; + } + } + + /** + * Send data to the process stdin + */ + public send(data: string): void { + if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { + throw new Error("WebSocket not connected"); + } + + this.ws.send(JSON.stringify({ + type: "stdin", + data, + })); + } + + /** + * Send a signal to the process + * @param signal The signal to send (e.g., "SIGTERM", "SIGINT", "SIGKILL", "SIGHUP") + * @throws Error if the WebSocket is not connected + */ + public kill(signal: string): void { + if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { + throw new Error("WebSocket not connected"); + } + + this.ws.send(JSON.stringify({ + type: "signal", + signal, + })); + } + + /** + * Register an event handler + */ + public addEventListener( + eventType: ShellStreamEventType, + handler: (event: ShellStreamEvent) => void, + ): void { + if (!this.eventHandlers.has(eventType)) { + this.eventHandlers.set(eventType, []); + } + + this.eventHandlers.get(eventType)!.push(handler); + } + + /** + * Remove an event handler + */ + public removeEventListener( + eventType: ShellStreamEventType, + handler: (event: ShellStreamEvent) => void, + ): void { + if (!this.eventHandlers.has(eventType)) { + return; + } + + const handlers = this.eventHandlers.get(eventType)!; + const index = handlers.indexOf(handler); + + if (index !== -1) { + handlers.splice(index, 1); + } + } + + /** + * Emit an event to all registered handlers + */ + private emitEvent(event: ShellStreamEvent): void { + // Call handlers for the specific event type + if (this.eventHandlers.has(event.type)) { + for (const handler of this.eventHandlers.get(event.type)!) { + handler(event); + } + } + } +} diff --git a/common/spaces/http_space_primitives.ts b/common/spaces/http_space_primitives.ts index ea766f59..151e945c 100644 --- a/common/spaces/http_space_primitives.ts +++ b/common/spaces/http_space_primitives.ts @@ -238,11 +238,43 @@ export class HttpSpacePrimitives implements SpacePrimitives { // Used to check if the server is reachable and the user is authenticated // If not: throws an error or invokes a redirect async ping() { - await this.authenticatedFetch(`${this.url}/.ping`, { + const response = await this.authenticatedFetch(`${this.url}/.ping`, { method: "GET", headers: { Accept: "application/json", }, }, 5000); + + // Consume the response body to avoid leaks + await response.text(); + } + + /** + * Create an authenticated WebSocket connection + * @param path The path to connect to + * @param queryParams Optional query parameters + * @returns A WebSocket connection + */ + async createAuthenticatedWebSocket( + path: string, + queryParams: Record = {}, + ): Promise { + // First make an authenticated request to ensure we have valid cookies + await this.ping(); + + // Build the WebSocket URL with query parameters + const queryString = Object.entries(queryParams) + .map(([key, value]) => + `${encodeURIComponent(key)}=${encodeURIComponent(value)}` + ) + .join("&"); + + // Create the WebSocket URL + const wsUrl = `${this.url.replace(/^http/, "ws")}/${path}${ + queryString ? `?${queryString}` : "" + }`; + + // Create and return the WebSocket + return new WebSocket(wsUrl); } } diff --git a/plug-api/syscalls/shell.ts b/plug-api/syscalls/shell.ts index 708b877e..12586c77 100644 --- a/plug-api/syscalls/shell.ts +++ b/plug-api/syscalls/shell.ts @@ -1,5 +1,26 @@ +import type { ShellStreamClient } from "$common/shell_stream_client.ts"; import { syscall } from "../syscall.ts"; +/** + * Interface for the shell stream client + */ +export interface ShellStream { + /** + * Send data to the process stdin + */ + send(data: string): void; + + /** + * Send a signal to the process + */ + kill(signal: string): void; + + /** + * Close the connection + */ + close(): void; +} + /** * Runs a shell command. * @param cmd the command to run @@ -12,3 +33,16 @@ export function run( ): Promise<{ stdout: string; stderr: string; code: number }> { return syscall("shell.run", cmd, args); } + +/** + * Runs a shell command with streaming I/O. + * @param cmd the command to run + * @param args the arguments to pass to the command + * @returns a shell stream client for interacting with the process + */ +export function spawn( + cmd: string, + args: string[], +): Promise { + return syscall("shell.spawn", cmd, args); +} diff --git a/server/http_server.ts b/server/http_server.ts index 9a751227..d1f80674 100644 --- a/server/http_server.ts +++ b/server/http_server.ts @@ -3,7 +3,10 @@ import { deleteCookie, getCookie, setCookie } from "hono/cookie"; import { validator } from "hono/validator"; import type { AssetBundle } from "$lib/asset_bundle/bundle.ts"; import type { FileMeta } from "@silverbulletmd/silverbullet/types"; -import type { ShellRequest } from "@silverbulletmd/silverbullet/type/rpc"; +import { + handleShellEndpoint, + handleShellStreamEndpoint, +} from "./shell_endpoints.ts"; import { SpaceServer } from "./space_server.ts"; import type { KvPrimitives } from "$lib/data/kv_primitives.ts"; import { extendedMarkdownLanguage } from "$common/markdown_parser/parser.ts"; @@ -397,20 +400,21 @@ export class HttpServer { }); // Shell command endpoint - this.app.post("/.shell", async (c) => { - const req = c.req; - const body = await req.json(); - try { - const shellCommand: ShellRequest = body; - // Note: in read-only this is set to NoShellSupport, so don't worry - const shellResponse = await this.spaceServer.shellBackend.handle( - shellCommand, - ); - return c.json(shellResponse); - } catch (e: any) { - console.log("Shell error", e); - return c.text(e.message, 500); - } + this.app.post("/.shell", (c) => { + return handleShellEndpoint( + c, + this.spaceServer.shellBackend, + this.spaceServer.readOnly, + ); + }); + + // Shell WebSocket endpoint + this.app.get("/.shell/stream", (c) => { + return handleShellStreamEndpoint( + c, + this.spaceServer.pagesPath, + this.spaceServer.readOnly, + ); }); // HTTP Proxy endpoint diff --git a/server/shell.test.ts b/server/shell.test.ts new file mode 100644 index 00000000..d7e24fa1 --- /dev/null +++ b/server/shell.test.ts @@ -0,0 +1,583 @@ +import { assert, assertEquals, assertExists } from "@std/assert"; +import { HttpServer, type ServerOptions } from "./http_server.ts"; +import { AssetBundle } from "$lib/asset_bundle/bundle.ts"; +import { MemoryKvPrimitives } from "$lib/data/memory_kv_primitives.ts"; +import { sleep } from "../lib/async.ts"; +import { + ShellStreamClient, + type ShellStreamEvent, +} from "../common/shell_stream_client.ts"; +import { HttpSpacePrimitives } from "../common/spaces/http_space_primitives.ts"; + +// Create a test script +async function createTestScript(dir: string): Promise { + const scriptContent = `#!/bin/sh +echo "Hello from test script" +echo "This is stderr output" >&2 +read input +echo "You said: $input" +exit 0 +`; + + const scriptPath = `${dir}/test_script.sh`; + await Deno.writeTextFile(scriptPath, scriptContent); + await Deno.chmod(scriptPath, 0o755); // Make executable + return scriptPath; +} + +// Create a signal-handling test script +async function createSignalTestScript(dir: string): Promise { + const scriptContent = `#!/bin/sh +echo "Starting signal test script" +# Set up signal handler +trap 'echo "Received SIGTERM"; exit 0' TERM +trap 'echo "Received SIGINT"; exit 0' INT +echo "Signal handlers set up" +# Wait indefinitely +while true; do + sleep 1 +done +`; + + const scriptPath = `${dir}/signal_test.sh`; + await Deno.writeTextFile(scriptPath, scriptContent); + await Deno.chmod(scriptPath, 0o755); // Make executable + return scriptPath; +} + +// Create a long-running test script that writes to a file when it exits +async function createLongRunningScript(dir: string): Promise { + const scriptContent = `#!/bin/sh +echo "Starting long-running script" +# Create a file to track the process ID +echo $$ > "${dir}/process.pid" +# Set up signal handler to detect proper cleanup +trap 'echo "Process terminated properly" > "${dir}/cleanup.log"; exit 0' TERM INT +echo "Ready for long-running operation" +# Wait indefinitely +while true; do + sleep 1 + echo "Still running..." +done +`; + + const scriptPath = `${dir}/long_running.sh`; + await Deno.writeTextFile(scriptPath, scriptContent); + await Deno.chmod(scriptPath, 0o755); // Make executable + return scriptPath; +} + +// Setup a minimal HttpServer for testing +async function setupTestServer( + tempDir: string, +): Promise<{ server: HttpServer; port: number }> { + // Create minimal asset bundles + const clientAssetBundle = new AssetBundle(); + const plugAssetBundle = new AssetBundle(); + + // Create a memory KV store + const kvPrimitives = new MemoryKvPrimitives(); + + // Find an available port + const port = 9000 + Math.floor(Math.random() * 1000); + + // Create server options + const options: ServerOptions = { + hostname: "localhost", + port, + clientAssetBundle, + plugAssetBundle, + baseKvPrimitives: kvPrimitives, + pagesPath: tempDir, + shellBackend: "local", // Use local shell backend + readOnly: false, + indexPage: "index", + enableSpaceScript: false, + }; + + // Create and start the server + const server = new HttpServer(options); + await server.start(); + + return { server, port }; +} + +// Helper function to create a script that fails with an error +async function createErrorScript(dir: string): Promise { + const scriptContent = `#!/bin/sh +echo "This is stdout before the error" +echo "This is stderr output" >&2 +exit 1 +`; + + const scriptPath = `${dir}/error_script.sh`; + await Deno.writeTextFile(scriptPath, scriptContent); + await Deno.chmod(scriptPath, 0o755); // Make executable + return scriptPath; +} + +// Run all tests in a single test function to avoid resource tracking issues +Deno.test("Shell endpoint tests", async () => { + // Test 1: Basic functionality with POST endpoint + { + // Test: Basic functionality with POST endpoint + // Create a temporary directory for testing + const tempDir = await Deno.makeTempDir(); + + try { + // Set up test server + const { server, port } = await setupTestServer(tempDir); + + try { + // Create a client with HttpSpacePrimitives + const httpSpacePrimitives = new HttpSpacePrimitives( + `http://localhost:${port}`, + ); + const client = new ShellStreamClient({ + httpSpacePrimitives, + cmd: "echo", + args: ["Hello, World!"], + }); + + // Collect events + const events: ShellStreamEvent[] = []; + client.addEventListener("stdout", (event) => events.push(event)); + client.addEventListener("stderr", (event) => events.push(event)); + client.addEventListener("exit", (event) => events.push(event)); + + // Connect to the server + await client.start(); + + // Wait for output and exit + await sleep(1000); + + // Close the connection and wait for cleanup + client.close(); + await sleep(500); + + // Verify events + assert(events.length >= 1, "Expected at least 1 event"); + + // Should have stdout message + const stdoutEvent = events.find((e) => + e.type === "stdout" && + typeof e.data === "string" && + e.data.includes("Hello, World!") + ); + assertExists(stdoutEvent, "Stdout event not found"); + } finally { + // Clean up server + server.stop(); + } + } finally { + // Clean up temporary directory + await Deno.remove(tempDir, { recursive: true }); + } + } + + // Cleanup on premature WebSocket close + { + // Test: Cleanup on premature WebSocket close + // Create a temporary directory for testing + const tempDir = await Deno.makeTempDir(); + + try { + // Create a long-running test script + const scriptPath = await createLongRunningScript(tempDir); + + // Set up test server + const { server, port } = await setupTestServer(tempDir); + + try { + // Create a client with HttpSpacePrimitives + const httpSpacePrimitives = new HttpSpacePrimitives( + `http://localhost:${port}`, + ); + const client = new ShellStreamClient({ + httpSpacePrimitives, + cmd: scriptPath, + args: [], + }); + + // Collect events + const events: ShellStreamEvent[] = []; + client.addEventListener("stdout", (event) => events.push(event)); + client.addEventListener("stderr", (event) => events.push(event)); + client.addEventListener("exit", (event) => events.push(event)); + + // Connect to the server + await client.start(); + + // Wait for the script to start + await sleep(1000); + + // Verify the process is running by checking if the PID file exists + const pidFileExists = await Deno.stat(`${tempDir}/process.pid`) + .then( + () => true, + () => false, + ); + assert(pidFileExists, "Process PID file should exist"); + + // Close the WebSocket connection prematurely + client.close(); + + // Wait for cleanup to happen + await sleep(2000); + + // Check if the cleanup log file was created, indicating proper termination + const cleanupFileExists = await Deno.stat(`${tempDir}/cleanup.log`) + .then( + () => true, + () => false, + ); + assert( + cleanupFileExists, + "Cleanup log file should exist, indicating proper process termination", + ); + + if (cleanupFileExists) { + const cleanupContent = await Deno.readTextFile( + `${tempDir}/cleanup.log`, + ); + assert( + cleanupContent.includes("Process terminated properly"), + "Process should have been terminated properly", + ); + } + + // Verify events + const startupEvent = events.find((e) => + e.type === "stdout" && + typeof e.data === "string" && + e.data.includes("Starting long-running script") + ); + assertExists(startupEvent, "Startup message not found"); + } finally { + // Clean up server + server.stop(); + } + } finally { + // Clean up temporary directory + await Deno.remove(tempDir, { recursive: true }); + } + } + + // Unified send method with POST endpoint + { + // Test: Unified send method with POST endpoint + // Create a temporary directory for testing + const tempDir = await Deno.makeTempDir(); + + try { + // Create a test script + const scriptPath = await createTestScript(tempDir); + + // Set up test server + const { server, port } = await setupTestServer(tempDir); + + try { + // Create a client with HttpSpacePrimitives + const httpSpacePrimitives = new HttpSpacePrimitives( + `http://localhost:${port}`, + ); + const client = new ShellStreamClient({ + httpSpacePrimitives, + cmd: scriptPath, + args: [], + }); + + // Collect events + const events: ShellStreamEvent[] = []; + client.addEventListener("stdout", (event) => events.push(event)); + client.addEventListener("stderr", (event) => events.push(event)); + client.addEventListener("exit", (event) => events.push(event)); + + // Connect to the server + await client.start(); + + // Wait for initial output + await sleep(500); + + // Send input using the unified send() method + client.send("Hello from test"); + + // Wait for response and exit + await sleep(1000); + + // Close the connection and wait for cleanup + client.close(); + await sleep(500); + + // Verify events + assert(events.length >= 3, "Expected at least 3 events"); + + // Should have stdout message + const stdoutEvent = events.find((e) => + e.type === "stdout" && + typeof e.data === "string" && + e.data.includes("Hello from test script") + ); + assertExists(stdoutEvent, "Stdout event not found"); + + // Should have stderr message + const stderrEvent = events.find((e) => + e.type === "stderr" && + typeof e.data === "string" && + e.data.includes("This is stderr output") + ); + assertExists(stderrEvent, "Stderr event not found"); + + // Should have response to input + const responseEvent = events.find((e) => + e.type === "stdout" && + typeof e.data === "string" && + e.data.includes("You said: Hello from test") + ); + assertExists(responseEvent, "Response event not found"); + } finally { + // Clean up server + server.stop(); + } + } finally { + // Clean up temporary directory + await Deno.remove(tempDir, { recursive: true }); + } + } + + // Send signal to process + { + // Test: Send signal to process + // Create a temporary directory for testing + const tempDir = await Deno.makeTempDir(); + + try { + // Create a signal-handling test script + const scriptPath = await createSignalTestScript(tempDir); + + // Set up test server + const { server, port } = await setupTestServer(tempDir); + + try { + // Create a client with HttpSpacePrimitives + const httpSpacePrimitives = new HttpSpacePrimitives( + `http://localhost:${port}`, + ); + const client = new ShellStreamClient({ + httpSpacePrimitives, + cmd: scriptPath, + args: [], + }); + + // Collect events + const events: ShellStreamEvent[] = []; + client.addEventListener("stdout", (event) => events.push(event)); + client.addEventListener("stderr", (event) => events.push(event)); + client.addEventListener("exit", (event) => events.push(event)); + + // Connect to the server + await client.start(); + + // Wait for the script to start and set up signal handlers + await sleep(1000); + + // Send a signal to the process + client.kill("SIGTERM"); + + // Wait for the process to handle the signal and exit + await sleep(1000); + + // Close the connection and wait for cleanup + client.close(); + await sleep(500); + + // Verify events + assert(events.length >= 3, "Expected at least 3 events"); + + // Should have initial stdout messages + const startupEvent = events.find((e) => + e.type === "stdout" && + typeof e.data === "string" && + e.data.includes("Starting signal test script") + ); + assertExists(startupEvent, "Startup message not found"); + + // Should have signal handler message + const signalEvent = events.find((e) => + e.type === "stdout" && + typeof e.data === "string" && + e.data.includes("Received SIGTERM") + ); + assertExists(signalEvent, "Signal handler message not found"); + + // Should have exit event + const exitEvent = events.find((e) => e.type === "exit"); + assertExists(exitEvent, "Exit event not found"); + } finally { + // Clean up server + server.stop(); + } + } finally { + // Clean up temporary directory + await Deno.remove(tempDir, { recursive: true }); + } + } + // .shell command with arguments + { + // Test: Command with arguments + // Create a temporary directory for testing + const tempDir = await Deno.makeTempDir(); + + try { + // Create a test script + const scriptPath = await createTestScript(tempDir); + + // Set up test server + const { server, port } = await setupTestServer(tempDir); + + try { + // Create HTTP space primitives for making requests + const httpSpacePrimitives = new HttpSpacePrimitives( + `http://localhost:${port}`, + ); + + // Test command with arguments + const response = await httpSpacePrimitives.authenticatedFetch( + `${httpSpacePrimitives.url}/.shell`, + { + method: "POST", + body: JSON.stringify({ + cmd: scriptPath, + args: [], + }), + }, + ); + + const result = await response.json(); + + // Verify response + assertEquals(result.code, 0, "Exit code should be 0"); + assert( + result.stdout.includes("Hello from test script"), + "Output should contain the script output", + ); + assert( + result.stderr.includes("This is stderr output"), + "Stderr should contain the error output", + ); + } finally { + // Clean up server + server.stop(); + } + } finally { + // Clean up temporary directory + await Deno.remove(tempDir, { recursive: true }); + } + } + + // Error handling + { + // Test: Error handling + // Create a temporary directory for testing + const tempDir = await Deno.makeTempDir(); + + try { + // Create an error script + const scriptPath = await createErrorScript(tempDir); + + // Set up test server + const { server, port } = await setupTestServer(tempDir); + + try { + // Create HTTP space primitives for making requests + const httpSpacePrimitives = new HttpSpacePrimitives( + `http://localhost:${port}`, + ); + + // Test command that fails + const response = await httpSpacePrimitives.authenticatedFetch( + `${httpSpacePrimitives.url}/.shell`, + { + method: "POST", + body: JSON.stringify({ + cmd: scriptPath, + args: [], + }), + }, + ); + + const result = await response.json(); + + // Verify response + assertEquals(result.code, 1, "Exit code should be 1"); + assert( + result.stdout.includes("This is stdout before the error"), + "Output should contain the stdout text", + ); + assert( + result.stderr.includes("This is stderr output"), + "Stderr should contain the error output", + ); + } finally { + // Clean up server + server.stop(); + } + } finally { + // Clean up temporary directory + await Deno.remove(tempDir, { recursive: true }); + } + } + + // Read-only mode + { + // Test: Read-only mode + // Create a temporary directory for testing + const tempDir = await Deno.makeTempDir(); + + try { + // Set up test server with read-only mode + const { server, port } = await setupTestServer(tempDir); + + // Set server to read-only mode + server.spaceServer.readOnly = true; + + try { + // Create HTTP space primitives for making requests + const httpSpacePrimitives = new HttpSpacePrimitives( + `http://localhost:${port}`, + ); + + // Test command in read-only mode + const response = await httpSpacePrimitives.authenticatedFetch( + `${httpSpacePrimitives.url}/.shell`, + { + method: "POST", + body: JSON.stringify({ + cmd: "echo", + args: ["Hello, World!"], + }), + }, + ); + + // Verify response status is 405 Method Not Allowed + assertEquals( + response.status, + 405, + "Status should be 405 Method Not Allowed", + ); + + const text = await response.text(); + assert( + text.includes("Read only mode"), + "Response should mention read-only mode", + ); + } finally { + // Clean up server + server.stop(); + } + } finally { + // Clean up temporary directory + await Deno.remove(tempDir, { recursive: true }); + } + } +}); diff --git a/server/shell_backend.ts b/server/shell_backend.ts index 755bebb8..4c353a3e 100644 --- a/server/shell_backend.ts +++ b/server/shell_backend.ts @@ -1,5 +1,6 @@ import type { ShellRequest, ShellResponse } from "../type/rpc.ts"; import type { ServerOptions } from "./http_server.ts"; +import { timeout } from "../lib/async.ts"; /** * Configuration via environment variables: @@ -35,6 +36,214 @@ export class NotSupportedShell implements ShellBackend { } } +export class StreamingShell { + private process: Deno.ChildProcess | null = null; + private outputHandlers: Array< + (type: string, data: string) => void + > = []; + private textDecoder = new TextDecoder(); + private textEncoder = new TextEncoder(); + + constructor(private cwd: string) {} + + start(cmd: string, args: string[]): void { + console.log("Starting streaming shell:", cmd, args); + + this.process = new Deno.Command(cmd, { + cwd: this.cwd, + args: args, + stdin: "piped", + stdout: "piped", + stderr: "piped", + }).spawn(); + + // Handle stdout + this.readStream(this.process.stdout, "stdout"); + + // Handle stderr + this.readStream(this.process.stderr, "stderr"); + + // Handle process exit + this.process.status.then((status) => { + this.notifyOutput( + "exit", + "json", + JSON.stringify({ code: status.code || 0 }), + ); + + // Process is already exited, just set to null + this.process = null; + }); + } + + private async readStream( + stream: ReadableStream, + type: string, + ): Promise { + const reader = stream.getReader(); + + try { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + + // Handle as text data (UTF-8) + const text = this.textDecoder.decode(value, { stream: true }); + if (text.trim()) { + console.log(`${type.toUpperCase()}: ${text}`); + this.notifyOutput(type, "text", text); + } + } + } catch (e) { + const error = e as Error; + console.error(`Error reading ${type}:`, error); + this.notifyOutput("error", "text", `Stream error: ${error.message}`); + } finally { + reader.releaseLock(); + } + } + + async writeToStdin(data: string): Promise { + if (!this.process || !this.process.stdin) { + throw new Error("Process not started or stdin not available"); + } + + const writer = this.process.stdin.getWriter(); + try { + // Add newline for text data + await writer.write(this.textEncoder.encode(data + "\n")); + } finally { + writer.releaseLock(); + } + } + + onOutput( + handler: (type: string, data: string) => void, + ): void { + this.outputHandlers.push(handler); + } + + private notifyOutput(type: string, _format: string, data: string): void { + for (const handler of this.outputHandlers) { + handler(type, data); + } + } + + /** + * Send a signal to the process + * @param signal The signal to send (e.g., "SIGTERM", "SIGINT", "SIGKILL") + * @returns True if the signal was sent successfully, false otherwise + */ + sendSignal(signal: string): boolean { + if (!this.process) { + console.error("Cannot send signal: Process not started"); + return false; + } + + try { + // Convert signal string to Deno.Signal enum + let denoSignal: Deno.Signal; + switch (signal.toUpperCase()) { + case "SIGINT": + denoSignal = "SIGINT"; + break; + case "SIGTERM": + denoSignal = "SIGTERM"; + break; + case "SIGKILL": + denoSignal = "SIGKILL"; + break; + case "SIGHUP": + denoSignal = "SIGHUP"; + break; + default: + console.error(`Unsupported signal: ${signal}`); + return false; + } + + this.process.kill(denoSignal); + console.log(`Sent ${signal} to process`); + return true; + } catch (e) { + console.error(`Error sending ${signal} to process:`, e); + return false; + } + } + + async close(): Promise { + if (this.process) { + try { + // First try to send SIGTERM to allow graceful shutdown + try { + console.log("Sending SIGTERM to process"); + this.process.kill("SIGTERM"); + } catch (e) { + console.error("Error sending SIGTERM to process:", e); + } + + // Close stdin if it's available + if (this.process.stdin) { + try { + // Create a writer to check if the stream is writable + const writer = this.process.stdin.getWriter(); + writer.releaseLock(); // Release immediately to avoid locking + + // If we got here, the stream is still open, so close it + this.process.stdin.close(); + } catch (e) { + // If we get an error, the stream might already be closed or errored + // Just log it and continue + console.log("Note: stdin already closed or errored:", e); + } + } + + // Wait for the process to exit with a short timeout + try { + const status = await Promise.race([ + this.process.status, + timeout(500).then(() => { + throw new Error("Process exit timeout after SIGTERM"); + }), + ]); + console.log("Process exited with code:", status.code); + this.process = null; + return; + } catch (e) { + console.log("Process did not exit after SIGTERM:", e); + } + + // If we're here, the process didn't exit after SIGTERM + // Try SIGKILL as a last resort + if (this.process) { // Check if process is still not null + try { + console.log("Sending SIGKILL to process"); + this.process.kill("SIGKILL"); + + // Wait again with a longer timeout + const status = await Promise.race([ + this.process.status, + timeout(500).then(() => { + throw new Error("Process exit timeout after SIGKILL"); + }), + ]); + console.log("Process exited with code after SIGKILL:", status.code); + } catch (e) { + console.error( + "Error waiting for process to exit after SIGKILL:", + e, + ); + } + } + } catch (e) { + console.error("Error closing process:", e); + } + + // Always set process to null to avoid memory leaks + this.process = null; + } + } +} + export class LocalShell implements ShellBackend { constructor(private cwd: string) { } diff --git a/server/shell_endpoints.ts b/server/shell_endpoints.ts new file mode 100644 index 00000000..161375c5 --- /dev/null +++ b/server/shell_endpoints.ts @@ -0,0 +1,125 @@ +import type { Context } from "hono"; +import type { ShellRequest } from "@silverbulletmd/silverbullet/type/rpc"; +import type { ShellBackend } from "./shell_backend.ts"; +import { StreamingShell } from "./shell_backend.ts"; + +/** + * Handles the /.shell endpoint for non-streaming shell commands + */ +export async function handleShellEndpoint( + c: Context, + shellBackend: ShellBackend, + readOnly: boolean, +): Promise { + const req = c.req; + const body = await req.json(); + try { + if (readOnly) { + return c.text("Read only mode, no shell commands allowed", 405); + } + const shellCommand: ShellRequest = body; + // Note: in read-only this is set to NoShellSupport, so don't worry + const shellResponse = await shellBackend.handle(shellCommand); + return c.json(shellResponse); + } catch (e: any) { + console.error("Shell error", e); + return c.text(e.message, 500); + } +} + +/** + * Handles the /.shell/stream endpoint for streaming shell commands + */ +export function handleShellStreamEndpoint( + c: Context, + cwd: string, + readOnly: boolean, +): Response { + const req = c.req; + const url = new URL(req.url); + + // Check if read-only mode is enabled + if (readOnly) { + return c.text("Read only mode, no shell commands allowed", 405); + } + + // Get command and arguments from query parameters + const cmd = url.searchParams.get("cmd"); + if (!cmd) { + return c.text("Missing cmd parameter", 400); + } + + let args: string[] = []; + const argsParam = url.searchParams.get("args"); + if (argsParam) { + try { + args = JSON.parse(argsParam); + if (!Array.isArray(args)) { + return c.text("args must be a JSON array", 400); + } + } catch (_e) { + return c.text("Invalid args parameter: must be a JSON array", 400); + } + } + + // Create a streaming shell + const shell = new StreamingShell(cwd); + + // Upgrade the connection to a WebSocket + try { + const { socket, response } = Deno.upgradeWebSocket(req.raw); + + // Set up WebSocket event handlers + socket.onopen = () => { + // Start the shell process when the WebSocket is opened + shell.start(cmd, args); + + // Forward shell output to the WebSocket + shell.onOutput((type, data) => { + socket.send(JSON.stringify({ + type, + data, + })); + }); + }; + + socket.onmessage = async (event) => { + try { + const message = JSON.parse(event.data); + if (message.type === "stdin") { + await shell.writeToStdin(message.data); + } else if (message.type === "signal") { + const success = shell.sendSignal(message.signal); + if (!success) { + socket.send(JSON.stringify({ + type: "error", + data: `Failed to send signal: ${message.signal}`, + })); + } + } + } catch (e) { + console.error("Error handling WebSocket message:", e); + socket.send(JSON.stringify({ + type: "error", + data: `Error: ${e}`, + })); + } + }; + + socket.onclose = async () => { + await shell.close(); + }; + + socket.onerror = (error) => { + console.error("WebSocket error:", error); + shell.close().catch((e) => { + console.error("Error closing shell:", e); + }); + }; + + return response; + } catch (e) { + console.error("WebSocket upgrade failed:", e); + return c.text("WebSocket upgrade failed", 400); + } +} diff --git a/web/syscalls/shell.ts b/web/syscalls/shell.ts index 45c7a833..0711a333 100644 --- a/web/syscalls/shell.ts +++ b/web/syscalls/shell.ts @@ -1,5 +1,26 @@ import type { SysCallMapping } from "../../lib/plugos/system.ts"; import type { Client } from "../client.ts"; +import { ShellStreamClient } from "../../common/shell_stream_client.ts"; + +/** + * Interface for the shell stream client + */ +export interface ShellStream { + /** + * Send data to the process stdin + */ + send(data: string): void; + + /** + * Send a signal to the process + */ + kill(signal: string): void; + + /** + * Close the connection + */ + close(): void; +} export function shellSyscalls( client: Client, @@ -29,5 +50,21 @@ export function shellSyscalls( } return { code, stderr, stdout }; }, + "shell.spawn": ( + _ctx, + cmd: string, + args: string[], + ): ShellStreamClient => { + if (!client.httpSpacePrimitives) { + throw new Error("Not supported in fully local mode"); + } + + // Create a shell stream client + return new ShellStreamClient({ + httpSpacePrimitives: client.httpSpacePrimitives, + cmd, + args: args || [], + }); + }, }; }