1use std::collections::BTreeMap;
9
10use async_trait::async_trait;
11use base64::engine::general_purpose::STANDARD as BASE64;
12use base64::Engine as _;
13use futures::stream::{self, BoxStream};
14use futures::StreamExt;
15use serde::Deserialize;
16use serde_json::json;
17
18use crate::error::{ErrorData, Result};
19use crate::providers::sandbox::{guard_for, Bounded, DeadlineReport};
20use crate::traits::{
21 Binding, CommandOutput, CreateSessionRequest, JobPoll, JobStart, PreviewCapability,
22 RunCommandRequest, Sandbox, SandboxSession, SandboxSessionState,
23};
24use alien_core::bindings::LocalSandboxBinding;
25use alien_core::{Platform, SandboxCapabilities};
26use alien_error::{AlienError, Context, ContextError, IntoAlienError};
27
28#[derive(Debug, Deserialize)]
29#[serde(rename_all = "camelCase")]
30struct SessionBody {
31 session_id: String,
32 #[allow(dead_code)]
33 container_id: String,
34}
35
36#[derive(Debug, Deserialize)]
37#[serde(rename_all = "camelCase", tag = "stream", content = "dataBase64")]
38enum OutputFrame {
39 Stdout(String),
40 Stderr(String),
41}
42
43#[derive(Debug, Deserialize)]
44#[serde(rename_all = "camelCase")]
45struct ExecResponse {
46 output: Vec<OutputFrame>,
47 exit_code: i64,
48}
49
50#[derive(Debug, Deserialize)]
51#[serde(rename_all = "camelCase")]
52struct ReadFileResponse {
53 contents_base64: String,
54}
55
56#[derive(Debug, Deserialize)]
57#[serde(rename_all = "camelCase")]
58struct PreviewResponse {
59 endpoint: String,
60 allowed_ports: Vec<u16>,
61}
62
63#[derive(Debug)]
65pub struct LocalSandbox {
66 client: reqwest::Client,
67 base_url: String,
68 token: String,
69}
70
71impl LocalSandbox {
72 pub async fn new(binding_name: &str, binding: &LocalSandboxBinding) -> Result<Self> {
77 let base_url = binding
78 .manager_url
79 .clone()
80 .into_value(binding_name, "managerUrl")
81 .map_err(|error| {
82 AlienError::new(ErrorData::BindingConfigInvalid {
83 binding_name: binding_name.to_string(),
84 env_var: alien_core::bindings::binding_env_var_name(binding_name),
85 reason: error.to_string(),
86 })
87 })?;
88
89 let token_path = binding
90 .token_path
91 .clone()
92 .into_value(binding_name, "tokenPath")
93 .map_err(|error| {
94 AlienError::new(ErrorData::BindingConfigInvalid {
95 binding_name: binding_name.to_string(),
96 env_var: alien_core::bindings::binding_env_var_name(binding_name),
97 reason: error.to_string(),
98 })
99 })?;
100
101 let token = tokio::fs::read_to_string(&token_path)
102 .await
103 .into_alien_error()
104 .context(ErrorData::BindingConfigInvalid {
105 binding_name: binding_name.to_string(),
106 env_var: alien_core::bindings::binding_env_var_name(binding_name),
107 reason: format!("could not read the sandbox route token at '{token_path}'"),
108 })?;
109
110 Ok(Self {
111 client: reqwest::Client::new(),
112 base_url: base_url.trim_end_matches('/').to_string(),
113 token: token.trim().to_string(),
114 })
115 }
116
117 fn url(&self, path: &str) -> String {
118 format!("{}{path}", self.base_url)
119 }
120
121 async fn send<T: for<'de> Deserialize<'de>>(
122 &self,
123 request: reqwest::RequestBuilder,
124 operation: &str,
125 ) -> Result<T> {
126 let response = self
127 .request(request, operation)
128 .await?
129 .json::<T>()
130 .await
131 .into_alien_error()
132 .context(ErrorData::UnexpectedResponseFormat {
133 provider: "local-sandbox".to_string(),
134 binding_name: operation.to_string(),
135 field: "body".to_string(),
136 response_json: "the sandbox route returned a body this provider cannot parse"
137 .to_string(),
138 })?;
139
140 Ok(response)
141 }
142
143 async fn request(
144 &self,
145 request: reqwest::RequestBuilder,
146 operation: &str,
147 ) -> Result<reqwest::Response> {
148 let response = request
149 .bearer_auth(&self.token)
150 .send()
151 .await
152 .into_alien_error()
153 .context(ErrorData::OperationNotSupported {
154 operation: operation.to_string(),
155 reason: "the local sandbox route is unreachable".to_string(),
156 })?;
157
158 if response.status().is_success() {
159 return Ok(response);
160 }
161
162 let status = response.status();
163 let body = response.text().await.unwrap_or_default();
166 Err(AlienError::new(ErrorData::OperationNotSupported {
167 operation: operation.to_string(),
168 reason: format!("the local sandbox route returned {status}: {body}"),
169 }))
170 }
171
172 fn unsupported(&self, capability: &str) -> AlienError<ErrorData> {
173 AlienError::new(ErrorData::OperationNotSupported {
174 operation: capability.to_string(),
175 reason: "not supported on local".to_string(),
176 })
177 }
178
179 async fn exec_within(
187 &self,
188 session_id: &str,
189 command: &[String],
190 request: &RunCommandRequest,
191 ) -> Result<ExecResponse> {
192 match tokio::time::timeout(
193 guard_for(request.deadline)?,
194 self.send(
195 self.client
196 .post(self.url(&format!("/v1/sessions/{session_id}/exec")))
197 .json(&json!({ "command": command })),
198 "sandbox.runCommand",
199 ),
200 )
201 .await
202 {
203 Ok(inner) => inner,
204 Err(_) => {
205 let outcome = ErrorData::SandboxOutcomeUnknown {
211 operation: "sandbox.runCommand".to_string(),
212 reason: format!(
213 "the command exceeded its {}s deadline and the session could not end it",
214 request.deadline.as_secs()
215 ),
216 };
217 Err(match self.terminate(session_id).await {
218 Ok(()) => AlienError::new(outcome),
219 Err(error) => error.context(outcome),
220 })
221 }
222 }
223 }
224}
225
226impl Binding for LocalSandbox {}
227
228#[async_trait]
229impl Sandbox for LocalSandbox {
230 fn capabilities(&self) -> SandboxCapabilities {
231 SandboxCapabilities::for_platform(Platform::Local).expect("Local has a sandbox backend")
232 }
233
234 async fn create(&self, request: CreateSessionRequest) -> Result<SandboxSession> {
235 let session_id = request
236 .session_id
237 .unwrap_or_else(|| uuid::Uuid::new_v4().simple().to_string());
238
239 let created: SessionBody = self
242 .send(
243 self.client
244 .post(self.url("/v1/sessions"))
245 .json(&json!({ "sessionId": session_id })),
246 "sandbox.create",
247 )
248 .await?;
249
250 Ok(SandboxSession {
251 session_id: created.session_id,
252 state: SandboxSessionState::Running,
253 generation: 1,
254 })
255 }
256
257 async fn get(&self, session_id: &str) -> Result<Option<SandboxSession>> {
258 Ok(self
259 .list()
260 .await?
261 .into_iter()
262 .find(|session| session.session_id == session_id))
263 }
264
265 async fn get_or_create(&self, request: CreateSessionRequest) -> Result<SandboxSession> {
266 if let Some(id) = request.session_id.as_deref() {
267 if let Some(existing) = self.get(id).await? {
268 return Ok(existing);
269 }
270 }
271
272 self.create(request).await
273 }
274
275 async fn list(&self) -> Result<Vec<SandboxSession>> {
276 let sessions: Vec<SessionBody> = self
277 .send(self.client.get(self.url("/v1/sessions")), "sandbox.list")
278 .await?;
279
280 Ok(sessions
281 .into_iter()
282 .map(|session| SandboxSession {
283 session_id: session.session_id,
284 state: SandboxSessionState::Running,
285 generation: 1,
286 })
287 .collect())
288 }
289
290 async fn run_command(
291 &self,
292 session_id: &str,
293 request: RunCommandRequest,
294 ) -> Result<BoxStream<'static, Result<CommandOutput>>> {
295 if request.deadline.is_zero() {
296 return Err(AlienError::new(ErrorData::OperationNotSupported {
297 operation: "sandbox.runCommand".to_string(),
298 reason: "a command must carry a non-zero deadline".to_string(),
299 }));
300 }
301
302 let mut argv = vec![
310 "sh".to_string(),
311 "-c".to_string(),
312 DeadlineReport::bounded_program(request.deadline),
313 "sh".to_string(),
314 ];
315 argv.extend(request.command.iter().cloned());
316 let response = self.exec_within(session_id, &argv, &request).await?;
317
318 let mut stdout = Vec::new();
322 let mut stderr = String::new();
323 for frame in response.output {
324 let (data, is_stdout) = match &frame {
325 OutputFrame::Stdout(data) => (data, true),
326 OutputFrame::Stderr(data) => (data, false),
327 };
328 let decoded = BASE64.decode(data).into_alien_error().context(
329 ErrorData::UnexpectedResponseFormat {
330 provider: "local-sandbox".to_string(),
331 binding_name: "sandbox.runCommand".to_string(),
332 field: "output".to_string(),
333 response_json: "an output frame was not valid base64".to_string(),
334 },
335 )?;
336 if is_stdout {
337 stdout.extend_from_slice(&decoded);
338 } else {
339 stderr.push_str(&String::from_utf8_lossy(&decoded));
340 }
341 }
342 let (deadline_exceeded, stderr) =
343 match DeadlineReport::read(i32::try_from(response.exit_code).ok(), &stderr) {
344 Bounded::Ran { killed, stderr } => (killed, stderr),
345 Bounded::NotRun { reason } => {
346 return Err(AlienError::new(ErrorData::SandboxCommandFailed {
347 failure: "commandNotBounded".to_string(),
348 reason,
349 }))
350 }
351 };
352
353 let mut frames: Vec<Result<CommandOutput>> = Vec::new();
354 if !stdout.is_empty() {
355 frames.push(Ok(CommandOutput::Stdout {
356 seq: 0,
357 data: stdout,
358 }));
359 }
360 if !stderr.is_empty() {
361 frames.push(Ok(CommandOutput::Stderr {
362 seq: frames.len() as u64,
363 data: stderr.into_bytes(),
364 }));
365 }
366
367 if deadline_exceeded {
368 frames.push(Err(AlienError::new(ErrorData::SandboxCommandFailed {
371 failure: "deadlineExceeded".to_string(),
372 reason: format!(
373 "the command exceeded its {}s deadline and was killed; the session is still usable",
374 request.deadline.as_secs()
375 ),
376 })));
377 } else {
378 frames.push(Ok(CommandOutput::Exit {
379 code: response.exit_code as i32,
380 truncated: false,
381 }));
382 }
383
384 Ok(Box::pin(stream::iter(frames)))
385 }
386
387 async fn read_file(&self, session_id: &str, path: &str) -> Result<Vec<u8>> {
388 let response: ReadFileResponse = self
389 .send(
390 self.client
391 .get(self.url(&format!("/v1/sessions/{session_id}/files")))
392 .query(&[("path", path)]),
393 "sandbox.readFile",
394 )
395 .await?;
396
397 BASE64
398 .decode(response.contents_base64)
399 .into_alien_error()
400 .context(ErrorData::UnexpectedResponseFormat {
401 provider: "local-sandbox".to_string(),
402 binding_name: "sandbox.readFile".to_string(),
403 field: "contentsBase64".to_string(),
404 response_json: "file contents were not valid base64".to_string(),
405 })
406 }
407
408 async fn write_files(&self, session_id: &str, files: BTreeMap<String, Vec<u8>>) -> Result<()> {
409 for (path, contents) in files {
410 self.request(
411 self.client
412 .put(self.url(&format!("/v1/sessions/{session_id}/files")))
413 .json(&json!({
414 "path": path,
415 "contentsBase64": BASE64.encode(contents),
416 })),
417 "sandbox.writeFiles",
418 )
419 .await?;
420 }
421
422 Ok(())
423 }
424
425 async fn mkdir(&self, session_id: &str, path: &str) -> Result<()> {
426 let request = RunCommandRequest {
427 command: vec!["/bin/mkdir".to_string(), "-p".to_string(), path.to_string()],
428 working_directory: None,
429 env: BTreeMap::new(),
430 deadline: std::time::Duration::from_secs(30),
431 };
432
433 let mut frames = self.run_command(session_id, request).await?;
437 while let Some(frame) = frames.next().await {
438 if let CommandOutput::Exit { code, .. } = frame? {
439 if code != 0 {
440 return Err(AlienError::new(ErrorData::OperationNotSupported {
441 operation: "sandbox.mkdir".to_string(),
442 reason: format!("mkdir '{path}' exited with {code}"),
443 }));
444 }
445 }
446 }
447
448 Ok(())
449 }
450
451 async fn preview(&self, session_id: &str, port: u16) -> Result<PreviewCapability> {
452 let response: PreviewResponse = self
453 .send(
454 self.client
455 .get(self.url(&format!("/v1/sessions/{session_id}/preview")))
456 .query(&[("port", port.to_string())]),
457 "sandbox.preview",
458 )
459 .await?;
460
461 Ok(PreviewCapability {
462 endpoint: response.endpoint,
463 headers: BTreeMap::new(),
466 allowed_ports: response.allowed_ports,
467 expires_in_seconds: 0,
468 })
469 }
470
471 async fn suspend(&self, _session_id: &str) -> Result<()> {
472 Err(self.unsupported("suspendResume"))
473 }
474
475 async fn resume(&self, _session_id: &str) -> Result<()> {
476 Err(self.unsupported("suspendResume"))
477 }
478
479 async fn snapshot(&self, _session_id: &str) -> Result<String> {
480 Err(self.unsupported("snapshot"))
481 }
482
483 async fn start_job(&self, _session_id: &str, _request: RunCommandRequest) -> Result<JobStart> {
484 Err(self.unsupported("jobs"))
485 }
486
487 async fn poll_job(
488 &self,
489 _session_id: &str,
490 _job_id: &str,
491 _since_seq: Option<u64>,
492 ) -> Result<JobPoll> {
493 Err(self.unsupported("jobs"))
494 }
495
496 async fn cancel_job(&self, _session_id: &str, _job_id: &str) -> Result<()> {
497 Err(self.unsupported("jobs"))
498 }
499
500 async fn terminate(&self, session_id: &str) -> Result<()> {
501 self.request(
502 self.client
503 .delete(self.url(&format!("/v1/sessions/{session_id}"))),
504 "sandbox.terminate",
505 )
506 .await?;
507
508 Ok(())
509 }
510
511 fn as_any(&self) -> &dyn std::any::Any {
512 self
513 }
514}
515
516#[cfg(test)]
517mod tests {
518 use super::*;
519 use axum::extract::{Path, State};
520 use axum::routing::post;
521 use axum::{Json, Router};
522 use std::net::SocketAddr;
523 use std::sync::{Arc, Mutex};
524
525 #[derive(Default)]
528 struct Route {
529 execs: Mutex<std::collections::VecDeque<serde_json::Value>>,
530 commands: Mutex<Vec<Vec<String>>>,
531 deleted: Mutex<Vec<String>>,
532 delete_refuses: std::sync::atomic::AtomicBool,
535 }
536
537 const DEADLINE_PLACEHOLDER: &str = "<deadline>";
539 const SESSION_NONCE: &str = "a1b2c3d4a1b2c3d4a1b2c3d4a1b2c3d4";
545
546 fn as_session_stderr(stderr: &str) -> String {
548 match stderr {
549 DEADLINE_PLACEHOLDER => format!("{SESSION_NONCE}\npartial-err{SESSION_NONCE}"),
550 other => format!("{SESSION_NONCE}\n{other}"),
551 }
552 }
553
554 fn exec_response(exit_code: i64, stdout: &str, stderr: &str) -> serde_json::Value {
555 let mut output = Vec::new();
556 if !stdout.is_empty() {
557 output.push(json!({ "stream": "stdout", "dataBase64": BASE64.encode(stdout) }));
558 }
559 output.push(json!({
560 "stream": "stderr",
561 "dataBase64": BASE64.encode(as_session_stderr(stderr)),
562 }));
563 json!({ "output": output, "exitCode": exit_code })
564 }
565
566 async fn serve(route: Arc<Route>) -> LocalSandbox {
567 async fn exec(
568 State(route): State<Arc<Route>>,
569 Path(_session): Path<String>,
570 Json(body): Json<serde_json::Value>,
571 ) -> Json<serde_json::Value> {
572 let command: Vec<String> =
573 serde_json::from_value(body["command"].clone()).expect("argv");
574 route.commands.lock().expect("commands").push(command);
575 let next = route.execs.lock().expect("execs").pop_front();
576 match next {
577 Some(response) => Json(response),
578 None => std::future::pending().await,
579 }
580 }
581 async fn delete(
582 State(route): State<Arc<Route>>,
583 Path(session): Path<String>,
584 ) -> std::result::Result<Json<serde_json::Value>, axum::http::StatusCode> {
585 if route
586 .delete_refuses
587 .load(std::sync::atomic::Ordering::SeqCst)
588 {
589 return Err(axum::http::StatusCode::INTERNAL_SERVER_ERROR);
590 }
591 route.deleted.lock().expect("deleted").push(session);
592 Ok(Json(json!({})))
593 }
594
595 let router = Router::new()
596 .route("/v1/sessions/{session}/exec", post(exec))
597 .route("/v1/sessions/{session}", axum::routing::delete(delete))
598 .with_state(route);
599 let listener = tokio::net::TcpListener::bind::<SocketAddr>("127.0.0.1:0".parse().unwrap())
600 .await
601 .expect("bind");
602 let address = listener.local_addr().expect("address");
603 tokio::spawn(async move {
604 axum::serve(listener, router).await.expect("serve");
605 });
606
607 LocalSandbox {
608 client: reqwest::Client::new(),
609 base_url: format!("http://{address}"),
610 token: "test-token".to_string(),
611 }
612 }
613
614 fn command(deadline_secs: u64) -> RunCommandRequest {
615 RunCommandRequest {
616 command: vec!["sleep".to_string(), "forever".to_string()],
617 working_directory: None,
618 env: BTreeMap::new(),
619 deadline: std::time::Duration::from_secs(deadline_secs),
620 }
621 }
622
623 #[tokio::test]
630 async fn a_command_past_its_deadline_is_killed_in_place_and_the_session_survives() {
631 let route = Arc::new(Route::default());
632 route.execs.lock().expect("execs").push_back(exec_response(
633 137,
634 "partial\n",
635 DEADLINE_PLACEHOLDER,
636 ));
637 let sandbox = serve(route.clone()).await;
638
639 let frames: Vec<Result<CommandOutput>> = sandbox
640 .run_command("s1", command(30))
641 .await
642 .expect("the deadline is reported in the stream")
643 .collect()
644 .await;
645
646 assert!(
647 matches!(&frames[0], Ok(CommandOutput::Stdout { data, .. }) if data == b"partial\n"),
648 "{frames:?}"
649 );
650 let terminal = frames
651 .last()
652 .expect("frames")
653 .as_ref()
654 .expect_err("the stream must end in the deadline error, not an exit frame");
655 assert!(
656 terminal.to_string().contains("deadlineExceeded"),
657 "{terminal}"
658 );
659 assert!(
660 route.deleted.lock().expect("deleted").is_empty(),
661 "the session survives an in-session kill"
662 );
663 let sent = route.commands.lock().expect("commands").clone();
664 assert_eq!(sent.len(), 1, "one command: {sent:?}");
665 assert_eq!(sent[0][0], "sh");
666 assert_eq!(sent[0][1], "-c");
667 assert!(sent[0][2].contains("sleep 30"), "{:?}", sent[0]);
668 assert_eq!(
669 &sent[0][3..],
670 &["sh".to_string(), "sleep".to_string(), "forever".to_string()],
671 "the command is passed as arguments, not pasted into the program"
672 );
673 }
674
675 #[tokio::test]
679 async fn a_deadline_split_across_stderr_frames_is_still_read() {
680 let route = Arc::new(Route::default());
681 route.execs.lock().expect("execs").push_back(json!({
682 "output": [
683 { "stream": "stderr", "dataBase64": BASE64.encode(format!("{SESSION_NONCE}\n")) },
684 { "stream": "stderr", "dataBase64": BASE64.encode("partial-err") },
685 { "stream": "stderr", "dataBase64": BASE64.encode(SESSION_NONCE) },
686 ],
687 "exitCode": 137,
688 }));
689 let sandbox = serve(route.clone()).await;
690
691 let frames: Vec<Result<CommandOutput>> = sandbox
692 .run_command("s1", command(30))
693 .await
694 .expect("the deadline is reported in the stream")
695 .collect()
696 .await;
697
698 assert!(
699 matches!(&frames[0], Ok(CommandOutput::Stderr { data, .. }) if data == b"partial-err"),
700 "the caller keeps its output, without the protocol bytes: {frames:?}"
701 );
702 let terminal = frames
703 .last()
704 .expect("frames")
705 .as_ref()
706 .expect_err("a killed command ends in the deadline error");
707 assert!(
708 terminal.to_string().contains("deadlineExceeded"),
709 "{terminal}"
710 );
711 }
712
713 #[tokio::test]
716 async fn a_command_exiting_124_of_its_own_accord_is_an_exit_not_a_deadline() {
717 let route = Arc::new(Route::default());
718 route
719 .execs
720 .lock()
721 .expect("execs")
722 .push_back(exec_response(124, "done\n", ""));
723 let sandbox = serve(route.clone()).await;
724
725 let frames: Vec<Result<CommandOutput>> = sandbox
726 .run_command("s1", command(300))
727 .await
728 .expect("runs")
729 .collect()
730 .await;
731
732 assert!(matches!(
733 frames.last().expect("frames"),
734 Ok(CommandOutput::Exit { code: 124, .. })
735 ));
736 }
737
738 #[tokio::test(start_paused = true)]
742 async fn a_terminate_that_fails_does_not_hide_the_unknown_outcome() {
743 let route = Arc::new(Route::default());
744 route
745 .delete_refuses
746 .store(true, std::sync::atomic::Ordering::SeqCst);
747 let sandbox = serve(route.clone()).await;
748
749 let error = sandbox
750 .run_command("s1", command(30))
751 .await
752 .err()
753 .expect("a command that outran its deadline has not succeeded");
754
755 assert_eq!(error.code, "SANDBOX_OUTCOME_UNKNOWN", "{error}");
756 assert!(
757 error.to_string().contains("could not end it"),
758 "the deadline stays the headline: {error}"
759 );
760 }
761
762 #[tokio::test(start_paused = true)]
766 async fn a_command_the_session_cannot_end_takes_the_session_with_it() {
767 let route = Arc::new(Route::default());
768 let sandbox = serve(route.clone()).await;
769
770 let error = sandbox
771 .run_command("s1", command(30))
772 .await
773 .err()
774 .expect("a command that outran its deadline has not succeeded");
775
776 assert_eq!(error.code, "SANDBOX_OUTCOME_UNKNOWN", "{error}");
777 assert!(
778 !error.retryable,
779 "the command may have run before the kill landed: {error}"
780 );
781 assert_eq!(
782 route.deleted.lock().expect("deleted").clone(),
783 vec!["s1".to_string()],
784 "the session must actually be removed, not merely reported as terminated"
785 );
786 }
787}