systemprompt_api/routes/admin/
cli.rs

1//! Admin CLI gateway route: streams `systemprompt` subprocess output over SSE.
2//!
3//! Exposes a single authenticated endpoint that validates and forwards an argv
4//! to the CLI binary, propagating the caller's session/context/auth into the
5//! child's environment and relaying stdout/stderr as [`CliOutputEvent`] frames.
6//!
7//! Copyright (c) systemprompt.io — Business Source License 1.1.
8//! See <https://systemprompt.io> for licensing details.
9
10use axum::extract::Extension;
11use axum::response::IntoResponse;
12use axum::response::sse::{Event, KeepAlive, Sse};
13use axum::routing::post;
14use axum::{Json, Router};
15use futures_util::stream::Stream;
16use std::convert::Infallible;
17use std::sync::Arc;
18use std::time::Duration;
19use systemprompt_events::ToSse;
20use systemprompt_logging::sanitize::redact_argv;
21use systemprompt_models::RequestContext;
22use systemprompt_models::api::{ApiError, CliExecuteRequest, CliOutputEvent};
23use systemprompt_runtime::AppContext;
24use tokio::io::{AsyncBufReadExt, BufReader};
25use tokio::process::Command;
26
27fn cli_event_to_sse(event: &CliOutputEvent) -> Event {
28    event
29        .to_sse()
30        .unwrap_or_else(|_| Event::default().event("cli").data("{}"))
31}
32
33const MAX_TIMEOUT_SECS: u64 = 600;
34const DEFAULT_CLI_BINARY_PATH: &str = "/app/bin/systemprompt";
35const MAX_CLI_ARGS: usize = 32;
36
37#[derive(Clone, Debug)]
38pub struct CliBinaryPath(Arc<str>);
39
40impl CliBinaryPath {
41    pub fn new(path: impl AsRef<str>) -> Self {
42        Self(Arc::from(path.as_ref()))
43    }
44
45    fn as_str(&self) -> &str {
46        &self.0
47    }
48}
49
50impl Default for CliBinaryPath {
51    fn default() -> Self {
52        Self::new(DEFAULT_CLI_BINARY_PATH)
53    }
54}
55const MAX_CLI_ARG_LEN: usize = 256;
56
57fn validate_cli_args(args: &[String]) -> Result<(), Box<ApiError>> {
58    if args.is_empty() {
59        return Err(Box::new(ApiError::bad_request(
60            "cli args must not be empty",
61        )));
62    }
63    if args.len() > MAX_CLI_ARGS {
64        return Err(Box::new(ApiError::bad_request(format!(
65            "too many cli args (max {MAX_CLI_ARGS})"
66        ))));
67    }
68    for (i, arg) in args.iter().enumerate() {
69        if arg.len() > MAX_CLI_ARG_LEN {
70            return Err(Box::new(ApiError::bad_request(format!(
71                "cli arg #{i} exceeds {MAX_CLI_ARG_LEN} bytes"
72            ))));
73        }
74        if arg
75            .chars()
76            .any(|c| c.is_control() || matches!(c, '`' | '$' | '|' | ';' | '&' | '\n' | '\r'))
77        {
78            return Err(Box::new(ApiError::bad_request(format!(
79                "cli arg #{i} contains forbidden character"
80            ))));
81        }
82    }
83    let first = &args[0];
84    let first_ok = first.chars().next().is_some_and(|c| c.is_ascii_lowercase())
85        && first
86            .chars()
87            .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-');
88    if !first_ok {
89        return Err(Box::new(ApiError::bad_request(
90            "first cli arg must be a lowercase subcommand",
91        )));
92    }
93    Ok(())
94}
95
96pub(super) fn router() -> Router<AppContext> {
97    router_with_binary(CliBinaryPath::default())
98}
99
100pub fn router_with_binary(binary: CliBinaryPath) -> Router<AppContext> {
101    Router::new()
102        .route("/", post(execute_cli))
103        .layer(Extension(binary))
104}
105
106async fn execute_cli(
107    Extension(req_ctx): Extension<RequestContext>,
108    Extension(binary): Extension<CliBinaryPath>,
109    Json(request): Json<CliExecuteRequest>,
110) -> Result<impl IntoResponse, ApiError> {
111    let args = request.args;
112    validate_cli_args(&args).map_err(|e| *e)?;
113    let timeout_secs = request.timeout_secs.min(MAX_TIMEOUT_SECS);
114    let timeout = Duration::from_secs(timeout_secs);
115
116    tracing::info!(
117        user_id = %req_ctx.user_id(),
118        args = ?redact_argv(&args),
119        timeout_secs = timeout_secs,
120        "CLI gateway: executing command"
121    );
122
123    let auth_token = req_ctx.auth_token().as_str();
124    let context_id = req_ctx.context_id().to_string();
125    let session_env = SessionEnv {
126        session: req_ctx.session_id().to_string(),
127        context: context_id,
128        user: req_ctx.user_id().to_string(),
129        auth_token: if auth_token.is_empty() {
130            None
131        } else {
132            Some(auth_token.to_owned())
133        },
134    };
135
136    let stream = create_cli_stream(binary, args, timeout, session_env);
137
138    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
139}
140
141struct SessionEnv {
142    session: String,
143    context: String,
144    user: String,
145    auth_token: Option<String>,
146}
147
148fn build_cli_command(binary: &CliBinaryPath, args: &[String], session_env: &SessionEnv) -> Command {
149    let mut cmd = Command::new(binary.as_str());
150    cmd.args(args)
151        .env("SYSTEMPROMPT_CLI_REMOTE", "1")
152        .env("SYSTEMPROMPT_SESSION_ID", &session_env.session)
153        .env("SYSTEMPROMPT_CONTEXT_ID", &session_env.context)
154        .env("SYSTEMPROMPT_USER_ID", &session_env.user)
155        .stdout(std::process::Stdio::piped())
156        .stderr(std::process::Stdio::piped());
157
158    if let Some(token) = &session_env.auth_token {
159        cmd.env("SYSTEMPROMPT_AUTH_TOKEN", token);
160    }
161    cmd
162}
163
164fn spawn_line_forwarder<R>(
165    reader: R,
166    tx: tokio::sync::mpsc::Sender<CliOutputEvent>,
167    make_event: fn(String) -> CliOutputEvent,
168) where
169    R: tokio::io::AsyncRead + Unpin + Send + 'static,
170{
171    tokio::spawn(async move {
172        let mut lines = BufReader::new(reader).lines();
173        while let Ok(Some(line)) = lines.next_line().await {
174            if tx.send(make_event(format!("{line}\n"))).await.is_err() {
175                break;
176            }
177        }
178    });
179}
180
181async fn kill_on_timeout(
182    child: &mut tokio::process::Child,
183    timeout: Duration,
184) -> [CliOutputEvent; 2] {
185    tracing::warn!(timeout_secs = timeout.as_secs(), "CLI command timed out");
186    if let Err(e) = child.kill().await {
187        tracing::error!(error = %e, "Failed to kill CLI process");
188    }
189    [
190        CliOutputEvent::Error {
191            message: format!("Timeout after {}s", timeout.as_secs()),
192        },
193        CliOutputEvent::ExitCode { code: -1 },
194    ]
195}
196
197async fn wait_exit_events(mut child: tokio::process::Child) -> Vec<CliOutputEvent> {
198    match child.wait().await {
199        Ok(status) => {
200            let code = status.code().unwrap_or_else(|| {
201                tracing::debug!("Process terminated by signal");
202                -1
203            });
204            tracing::info!(exit_code = code, "CLI command completed");
205            vec![CliOutputEvent::ExitCode { code }]
206        },
207        Err(e) => {
208            tracing::error!(error = %e, "Failed to wait for CLI process");
209            vec![
210                CliOutputEvent::Error {
211                    message: e.to_string(),
212                },
213                CliOutputEvent::ExitCode { code: -1 },
214            ]
215        },
216    }
217}
218
219fn create_cli_stream(
220    binary: CliBinaryPath,
221    args: Vec<String>,
222    timeout: Duration,
223    session_env: SessionEnv,
224) -> impl Stream<Item = Result<Event, Infallible>> {
225    async_stream::stream! {
226        let mut child = match build_cli_command(&binary, &args, &session_env).spawn() {
227            Ok(c) => c,
228            Err(e) => {
229                tracing::error!(error = %e, "Failed to spawn CLI process");
230                yield Ok(cli_event_to_sse(&CliOutputEvent::Error { message: e.to_string() }));
231                yield Ok(cli_event_to_sse(&CliOutputEvent::ExitCode { code: 1 }));
232                return;
233            }
234        };
235
236        let pid = child.id().unwrap_or_else(|| {
237            tracing::debug!("Child process has no PID (already exited?)");
238            0
239        });
240        yield Ok(cli_event_to_sse(&CliOutputEvent::Started { pid }));
241
242        let (tx, mut rx) = tokio::sync::mpsc::channel::<CliOutputEvent>(100);
243        if let Some(stdout) = child.stdout.take() {
244            spawn_line_forwarder(stdout, tx.clone(), |data| CliOutputEvent::Stdout { data });
245        }
246        if let Some(stderr) = child.stderr.take() {
247            spawn_line_forwarder(stderr, tx.clone(), |data| CliOutputEvent::Stderr { data });
248        }
249        drop(tx);
250
251        let deadline = tokio::time::Instant::now() + timeout;
252
253        loop {
254            tokio::select! {
255                event = rx.recv() => {
256                    match event {
257                        Some(ref e) => yield Ok(cli_event_to_sse(e)),
258                        None => break,
259                    }
260                }
261                () = tokio::time::sleep_until(deadline) => {
262                    for event in kill_on_timeout(&mut child, timeout).await {
263                        yield Ok(cli_event_to_sse(&event));
264                    }
265                    return;
266                }
267            }
268        }
269
270        for event in wait_exit_events(child).await {
271            yield Ok(cli_event_to_sse(&event));
272        }
273    }
274}
systemprompt_api/routes/admin/cli.rs

systemprompt_api/routes/admin/
cli.rs
