Skip to content

feat(plugin): stream real-time stdout from WASM plugins using Pipe #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
maheshbhatiya73 merged 1 commit into from
Jul 30, 2025
Merged

feat(plugin): stream real-time stdout from WASM plugins using Pipe #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions sandcrate-backend/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ name = "sandcrate-backend"
edition = "2021"

[dependencies]
axum = "0.7"
axum = { version = "0.7", features = ["ws"] }
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
Expand All @@ -14,4 +14,7 @@ jsonwebtoken = "9"
chrono = { version = "0.4", features = ["serde"] }
tower = "0.4"
tower-http = { version = "0.5", features = ["cors"] }
axum-extra = { version = "0.9", features = ["typed-header"] }
axum-extra = { version = "0.9", features = ["typed-header"] }
tokio-tungstenite = "0.21"
futures-util = { version = "0.3", features = ["sink"] }
uuid = { version = "1.0", features = ["v4"] }
6 changes: 3 additions & 3 deletions sandcrate-backend/src/api.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ struct ErrorResponse {
async fn get_plugins(
State(_config): State<Arc<AuthConfig>>,
) -> Json<ApiResponse<PluginList>> {
let plugins_dir = FsPath::new("../assets/plugins");
let plugins_dir = FsPath::new("assets/plugins");
let mut plugins = Vec::new();

if let Ok(entries) = fs::read_dir(plugins_dir) {
Expand Down Expand Up @@ -113,7 +113,7 @@ async fn get_plugin(
State(_config): State<Arc<AuthConfig>>,
Path(plugin_id): Path<String>,
) -> Response {
let plugins_dir = FsPath::new("../assets/plugins");
let plugins_dir = FsPath::new("assets/plugins");
let plugin_path = plugins_dir.join(format!("{}.wasm", plugin_id));

if !plugin_path.exists() {
Expand Down Expand Up @@ -184,7 +184,7 @@ async fn execute_plugin(
let start_time = std::time::Instant::now();

// Find the plugin file
let plugins_dir = FsPath::new("../assets/plugins");
let plugins_dir = FsPath::new("assets/plugins");
let plugin_path = plugins_dir.join(format!("{}.wasm", plugin_id));

if !plugin_path.exists() {
Expand Down
3 changes: 0 additions & 3 deletions sandcrate-backend/src/auth.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,14 +61,11 @@ impl AuthConfig {
}
}

// Check if user has sudo/root privileges
fn check_user_privileges(username: &str) -> (bool, String) {
// Check if user is root
if username == "root" {
return (true, "root".to_string());
}

// Check if user has sudo privileges
let output = std::process::Command::new("sudo")
.args(["-l", "-U", username])
.output();
Expand Down
76 changes: 43 additions & 33 deletions sandcrate-backend/src/bin/execute_plugin.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,47 +1,57 @@
use std::env;
use std::fs;
use std::path::Path;
use sandcrate_backend::plugin;
use wasmtime::*;
use wasmtime_wasi::WasiCtxBuilder;
use std::env;

fn main() -> Result<(), Box<dyn std::error::Error>> {
let args: Vec<String> = env::args().collect();

if args.len() != 2 {
println!("Usage: {} <plugin_name>", args[0]);
println!("Example: {} sandcrate-plugin", args[0]);
return Ok(());
eprintln!("Usage: {} <plugin_path>", args[0]);
std::process::exit(1);
}

let plugin_name = &args[1];
let plugin_path = format!("assets/plugins/{}.wasm", plugin_name);

if !Path::new(&plugin_path).exists() {
println!("❌ Plugin '{}' not found at {}", plugin_name, plugin_path);
println!("Available plugins:");

let plugins = plugin::list_plugins();
if plugins.is_empty() {
println!(" No plugins found in assets/plugins directory");
} else {
for plugin in plugins {
println!(" - {}", plugin);
}
}
return Ok(());
}
let plugin_path = &args[1];

println!("🚀 Executing plugin: {}", plugin_name);
println!("📁 Path: {}", plugin_path);
println!("---");
// Create a WASM engine
let engine = Engine::default();

match plugin::run_plugin(&plugin_path) {
Ok(result) => {
println!("✅ Plugin executed successfully!");
println!("📋 Result: {}", result);
}
Err(e) => {
println!("❌ Plugin execution failed: {}", e);
// Create a store with WASI context
let wasi = WasiCtxBuilder::new()
.inherit_stdio()
.inherit_args()?
.build();

let mut store = Store::new(&engine, wasi);

// Read the WASM module
let wasm_bytes = fs::read(plugin_path)?;
let module = Module::new(&engine, &wasm_bytes)?;

// Create a linker and add WASI
let mut linker = Linker::new(&engine);
wasmtime_wasi::add_to_linker(&mut linker, |s| s)?;

// Instantiate the module
let instance = linker.instantiate(&mut store, &module)?;

// Try to find and call a suitable function
let function_names = ["_start", "start", "main", "run"];

let mut executed = false;

for func_name in &function_names {
if let Ok(func) = instance.get_typed_func::<(), ()>(&mut store, func_name) {
func.call(&mut store, ())?;
executed = true;
break;
}
}

if !executed {
eprintln!("No suitable entry function found in WASM module");
std::process::exit(1);
}

Ok(())
}
20 changes: 15 additions & 5 deletions sandcrate-backend/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,31 +1,41 @@
mod api;
mod auth;
pub mod plugin;
mod websocket;

use std::net::SocketAddr;
use std::sync::Arc;
use tokio::net::TcpListener;
use tower_http::cors::CorsLayer;
use axum::Router;
use axum::{Router, routing::get};

// Re-export plugin runner
pub use plugin::run_plugin;
pub use websocket::{WebSocketManager, PluginExecutionSession};

#[tokio::main]
pub async fn run_backend() {
let auth_config = Arc::new(auth::AuthConfig::new());
let ws_manager = Arc::new(websocket::WebSocketManager::new());

let api_router = api::routes().with_state(auth_config.clone());
let auth_router = auth::auth_routes().with_state(auth_config.clone());
let ws_router = Router::new()
.route("/plugins", get(websocket::plugin_execution_websocket))
.with_state((auth_config, ws_manager));

let app = Router::new()
.nest("/api", api::routes())
.nest("/auth", auth::auth_routes())
.with_state(auth_config)
.nest("/api", api_router)
.nest("/auth", auth_router)
.nest("/ws", ws_router)
.layer(CorsLayer::permissive());

let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
println!("Backend running at http://{}", addr);
println!("API endpoints:");
println!(" GET http://{}/api/plugins", addr);
println!(" POST http://{}/auth/login", addr);
println!("WebSocket endpoints:");
println!(" WS ws://{}/ws/plugins", addr);

let listener = TcpListener::bind(addr).await.unwrap();
axum::serve(listener, app.into_make_service())
Expand Down
94 changes: 84 additions & 10 deletions sandcrate-backend/src/plugin.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,14 @@
use std::fs;
use std::path::Path;
use std::time::Duration;
use wasmtime::*;
use wasmtime_wasi::WasiCtxBuilder;
use serde_json::Value;
use tokio::sync::broadcast;
use std::sync::Arc;
use tokio::sync::Mutex;
use std::io::{Read, Write};



pub fn list_plugins() -> Vec<String> {
let plugins_dir = Path::new("assets/plugins");
Expand Down Expand Up @@ -36,30 +41,23 @@ pub fn run_plugin_with_params(
parameters: Option<Value>,
_timeout: Option<u64>
) -> Result<String, Box<dyn std::error::Error>> {
// Create a WASM engine
let engine = Engine::default();

// Create a store with WASI context
let wasi = WasiCtxBuilder::new()
.inherit_stdio()
.inherit_args()?
.build();

let mut store = Store::new(&engine, wasi);

// Read the WASM module
let wasm_bytes = fs::read(plugin_path)?;
let module = Module::new(&engine, &wasm_bytes)?;

// Create a linker and add WASI
let mut linker = Linker::new(&engine);
wasmtime_wasi::add_to_linker(&mut linker, |s| s)?;

// Instantiate the module
let instance = linker.instantiate(&mut store, &module)?;

// Try to find and call a suitable function
// Common function names for WASM modules: _start, start, main, run
let function_names = ["_start", "start", "main", "run"];

let mut executed = false;
Expand All @@ -81,6 +79,84 @@ pub fn run_plugin_with_params(
Ok(result)
}

pub async fn run_plugin_with_realtime_output(
plugin_path: &str,
_parameters: Option<Value>,
_timeout: Option<u64>,
ws_tx: broadcast::Sender<crate::websocket::PluginExecutionSession>,
session_id: &str,
) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
let session_id = session_id.to_string();
let plugin_id = Path::new(plugin_path)
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("unknown")
.to_string();

let _ = ws_tx.send(crate::websocket::PluginExecutionSession {
id: session_id.clone(),
plugin_id: plugin_id.clone(),
status: "starting".to_string(),
output: "Plugin execution started".to_string(),
});

let output = tokio::process::Command::new("cargo")
.args(&["run", "--bin", "execute_plugin", "--quiet"])
.arg(plugin_path)
.output()
.await?;

let _ = ws_tx.send(crate::websocket::PluginExecutionSession {
id: session_id.clone(),
plugin_id: plugin_id.clone(),
status: "running".to_string(),
output: "Executing plugin...".to_string(),
});

if !output.stdout.is_empty() {
let stdout_str = String::from_utf8_lossy(&output.stdout);
for line in stdout_str.lines() {
if !line.trim().is_empty() {
let _ = ws_tx.send(crate::websocket::PluginExecutionSession {
id: session_id.clone(),
plugin_id: plugin_id.clone(),
status: "running".to_string(),
output: line.trim().to_string(),
});
}
}
}

if !output.stderr.is_empty() {
let stderr_str = String::from_utf8_lossy(&output.stderr);
for line in stderr_str.lines() {
if !line.trim().is_empty() {
let _ = ws_tx.send(crate::websocket::PluginExecutionSession {
id: session_id.clone(),
plugin_id: plugin_id.clone(),
status: "running".to_string(),
output: format!("ERROR: {}", line.trim()),
});
}
}
}

let result = if output.status.success() {
"Plugin executed successfully".to_string()
} else {
format!("Plugin execution failed with exit code: {}", output.status)
};

let _ = ws_tx.send(crate::websocket::PluginExecutionSession {
id: session_id.clone(),
plugin_id: plugin_id.clone(),
status: "completed".to_string(),
output: format!("Plugin completed: {}", result),
});

Ok(result)
}

pub fn get_plugin_info(plugin_path: &str) -> Result<PluginInfo, Box<dyn std::error::Error>> {
let path = Path::new(plugin_path);

Expand All @@ -91,12 +167,10 @@ pub fn get_plugin_info(plugin_path: &str) -> Result<PluginInfo, Box<dyn std::err
let metadata = fs::metadata(path)?;
let file_size = metadata.len();

// Try to read WASM module info
let wasm_bytes = fs::read(path)?;
let engine = Engine::default();
let module = Module::new(&engine, &wasm_bytes)?;

// Get exported functions from module
let exports: Vec<String> = module
.exports()
.map(|export| export.name().to_string())
Expand Down
Loading
Loading