Shell streaming support (Lua)
This commit is contained in:
Zef Hemel
@@ -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<void> {
|
||||
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<void>((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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<string, string> = {},
|
||||
): Promise<WebSocket> {
|
||||
// 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<ShellStreamClient> {
|
||||
return syscall("shell.spawn", cmd, args);
|
||||
}
|
||||
|
||||
+19
-15
@@ -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
|
||||
|
||||
@@ -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<string> {
|
||||
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<string> {
|
||||
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<string> {
|
||||
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<string> {
|
||||
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 });
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -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<Uint8Array>,
|
||||
type: string,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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) {
|
||||
}
|
||||
|
||||
@@ -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<Response> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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 || [],
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user