a3s_box_runtime/local_execution/
port.rs

1use std::num::NonZeroU16;
2use std::time::Duration;
3
4use a3s_box_core::{
5    ExecutionBackend, ExecutionGeneration, ExecutionId, ExecutionManagerError,
6    ExecutionManagerResult, ExecutionPortConnector, ExecutionPortStream,
7};
8use async_trait::async_trait;
9
10use super::LocalExecutionManager;
11use crate::BoxRecord;
12
13#[async_trait]
14impl ExecutionPortConnector for LocalExecutionManager {
15    async fn connect_port(
16        &self,
17        execution_id: &ExecutionId,
18        generation: ExecutionGeneration,
19        port: NonZeroU16,
20        timeout: Duration,
21    ) -> ExecutionManagerResult<ExecutionPortStream> {
22        if timeout.is_zero() {
23            return Err(ExecutionManagerError::InvalidRequest(
24                "port connection timeout must be non-zero".to_string(),
25            ));
26        }
27
28        #[cfg(target_os = "linux")]
29        {
30            let record = self.require_connectable(execution_id, generation).await?;
31            let pid = record
32                .pid
33                .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
34            let pid_start_time = record.pid_start_time;
35            if !crate::process::is_process_alive_with_identity(pid, pid_start_time) {
36                return Err(ExecutionManagerError::NotFound(execution_id.clone()));
37            }
38
39            let stream = connect_in_network_namespace(
40                execution_id.clone(),
41                pid,
42                pid_start_time,
43                port,
44                timeout,
45            )
46            .await?;
47
48            // The lifecycle may have advanced while the blocking connect was in
49            // flight. Re-read the canonical record before publishing the stream.
50            let current = self.require_connectable(execution_id, generation).await?;
51            if current.pid != Some(pid)
52                || current.pid_start_time != pid_start_time
53                || !crate::process::is_process_alive_with_identity(pid, pid_start_time)
54            {
55                return Err(ExecutionManagerError::Conflict {
56                    execution_id: execution_id.clone(),
57                    message: "runtime generation changed while connecting its data plane"
58                        .to_string(),
59                });
60            }
61            return Ok(Box::pin(stream));
62        }
63
64        #[cfg(not(target_os = "linux"))]
65        {
66            let _ = (execution_id, generation, port, timeout);
67            Err(ExecutionManagerError::Unavailable(
68                "Sandbox port connections require Linux network namespaces".to_string(),
69            ))
70        }
71    }
72}
73
74impl LocalExecutionManager {
75    async fn require_connectable(
76        &self,
77        execution_id: &ExecutionId,
78        generation: ExecutionGeneration,
79    ) -> ExecutionManagerResult<BoxRecord> {
80        let record = self
81            .require_running_record(execution_id, generation)
82            .await?;
83        let backend = record
84            .managed_execution
85            .as_ref()
86            .map(|metadata| metadata.plan.backend)
87            .ok_or_else(|| {
88                ExecutionManagerError::Internal(format!(
89                    "execution {execution_id} has no managed execution plan"
90                ))
91            })?;
92        if backend != ExecutionBackend::Crun {
93            return Err(ExecutionManagerError::Unavailable(format!(
94                "execution {execution_id} does not expose a Sandbox network namespace"
95            )));
96        }
97        Ok(record)
98    }
99}
100
101#[cfg(target_os = "linux")]
102async fn connect_in_network_namespace(
103    execution_id: ExecutionId,
104    pid: u32,
105    pid_start_time: Option<u64>,
106    port: NonZeroU16,
107    timeout: Duration,
108) -> ExecutionManagerResult<tokio::net::TcpStream> {
109    let (sender, receiver) = tokio::sync::oneshot::channel();
110    std::thread::Builder::new()
111        .name(format!("a3s-port-{pid}-{}", port.get()))
112        .spawn(move || {
113            let result = connect_in_network_namespace_blocking(
114                &execution_id,
115                pid,
116                pid_start_time,
117                port,
118                timeout,
119            );
120            let _ = sender.send(result);
121        })
122        .map_err(|error| {
123            ExecutionManagerError::Unavailable(format!(
124                "failed to start Sandbox port connector: {error}"
125            ))
126        })?;
127
128    let stream = receiver.await.map_err(|_| {
129        ExecutionManagerError::Internal(
130            "Sandbox port connector exited without a result".to_string(),
131        )
132    })??;
133    tokio::net::TcpStream::from_std(stream).map_err(|error| {
134        ExecutionManagerError::Unavailable(format!(
135            "failed to register Sandbox port stream with Tokio: {error}"
136        ))
137    })
138}
139
140#[cfg(target_os = "linux")]
141fn connect_in_network_namespace_blocking(
142    execution_id: &ExecutionId,
143    pid: u32,
144    pid_start_time: Option<u64>,
145    port: NonZeroU16,
146    timeout: Duration,
147) -> ExecutionManagerResult<std::net::TcpStream> {
148    use std::fs::File;
149    use std::os::fd::AsRawFd;
150
151    if !crate::process::is_process_alive_with_identity(pid, pid_start_time) {
152        return Err(ExecutionManagerError::NotFound(execution_id.clone()));
153    }
154    let namespace_path = format!("/proc/{pid}/ns/net");
155    let namespace = File::open(&namespace_path).map_err(|error| {
156        ExecutionManagerError::Unavailable(format!(
157            "failed to open Sandbox network namespace {namespace_path}: {error}"
158        ))
159    })?;
160    let result = unsafe { libc::setns(namespace.as_raw_fd(), libc::CLONE_NEWNET) };
161    if result != 0 {
162        return Err(ExecutionManagerError::Unavailable(format!(
163            "failed to enter Sandbox network namespace for PID {pid}: {}",
164            std::io::Error::last_os_error()
165        )));
166    }
167    if !crate::process::is_process_alive_with_identity(pid, pid_start_time) {
168        return Err(ExecutionManagerError::Unavailable(
169            "Sandbox runtime exited while entering its network namespace".to_string(),
170        ));
171    }
172
173    let address = std::net::SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, port.get()));
174    let stream = std::net::TcpStream::connect_timeout(&address, timeout).map_err(|error| {
175        ExecutionManagerError::Unavailable(format!(
176            "failed to connect to Sandbox loopback port {}: {error}",
177            port.get()
178        ))
179    })?;
180    stream.set_nonblocking(true).map_err(|error| {
181        ExecutionManagerError::Unavailable(format!(
182            "failed to configure Sandbox port stream: {error}"
183        ))
184    })?;
185    Ok(stream)
186}
a3s_box_runtime/local_execution/port.rs

a3s_box_runtime/local_execution/
port.rs
