Files
Genarrative/apps/ai-game-creator-shell/src-tauri/src/process_session_bridge.rs
T
AIGameCreator App 689082901f 为持久进程增加可信启动握手
新增 PTY 外私有 bridge 与 sandbox ready/commit/exec 协议
升级 process record v3 并收紧恢复、幂等与 final/idle 门禁
完善 target 前台进程组、graceful 后代收束和审计失败处理
补齐跨平台回归测试及 Runtime 文档
2026-07-14 09:41:18 +08:00

875 lines
32 KiB
Rust

#[cfg(target_os = "linux")]
mod linux {
use crate::command_sandbox::{
command_sandbox_platform_metadata, stage_command_sandbox_launch_for_process_session,
CommandSandboxLaunch,
};
use crate::command_sandbox_trampoline::{TargetExecState, TargetTerminalState};
use crate::{ProjectCommandLaunchSpec, ProjectCommandSpec};
use serde::{Deserialize, Serialize};
use std::ffi::OsString;
use std::fs::File;
use std::io::{self, Read, Write};
use std::os::fd::AsRawFd;
use std::os::linux::net::SocketAddrExt;
use std::os::unix::ffi::{OsStrExt, OsStringExt};
use std::os::unix::net::{SocketAddr, UnixListener, UnixStream};
use std::path::PathBuf;
use std::process::{Command, Stdio};
use std::thread;
use std::time::{Duration, Instant};
pub(crate) const PROCESS_SESSION_BRIDGE_ENDPOINT_ENV: &str =
"GENARRATIVE_PROCESS_SESSION_BRIDGE_ENDPOINT";
pub(crate) const PROCESS_SESSION_BRIDGE_NONCE_ENV: &str =
"GENARRATIVE_PROCESS_SESSION_BRIDGE_NONCE";
const BRIDGE_NONCE_BYTES: usize = 32;
const BRIDGE_FRAME_MAX_BYTES: usize = 1024 * 1024;
const BRIDGE_ENDPOINT_MAX_BYTES: usize = 100;
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct ProcessSessionLaunchPlan {
sandbox_executable: Vec<u8>,
sandbox_arguments: Vec<Vec<u8>>,
sandbox_cwd: Vec<u8>,
sandbox_environment: Vec<(Vec<u8>, Vec<u8>)>,
target_executable: Vec<u8>,
target_arguments: Vec<Vec<u8>>,
}
impl ProcessSessionLaunchPlan {
fn from_launch(
launch: &ProjectCommandLaunchSpec,
spec: &ProjectCommandSpec,
) -> Result<Self, String> {
if !launch.executable.is_absolute()
|| !launch.cwd.is_absolute()
|| !spec.executable.is_absolute()
{
return Err("process session bridge launch path 必须是绝对路径".to_string());
}
Ok(Self {
sandbox_executable: launch.executable.as_os_str().as_bytes().to_vec(),
sandbox_arguments: launch
.arguments
.iter()
.map(|argument| argument.as_os_str().as_bytes().to_vec())
.collect(),
sandbox_cwd: launch.cwd.as_os_str().as_bytes().to_vec(),
sandbox_environment: launch
.environment
.iter()
.map(|(name, value)| {
(
name.as_os_str().as_bytes().to_vec(),
value.as_os_str().as_bytes().to_vec(),
)
})
.collect(),
target_executable: spec.executable.as_os_str().as_bytes().to_vec(),
target_arguments: crate::command_exec::project_command_actual_arguments(spec)
.into_iter()
.map(|argument| OsString::from(argument).into_vec())
.collect(),
})
}
fn into_staged(self) -> Result<crate::command_sandbox::StagedCommandSandboxLaunch, String> {
let sandbox_executable = PathBuf::from(OsString::from_vec(self.sandbox_executable));
let sandbox_cwd = PathBuf::from(OsString::from_vec(self.sandbox_cwd));
let target_executable = PathBuf::from(OsString::from_vec(self.target_executable));
if !sandbox_executable.is_absolute()
|| !sandbox_cwd.is_absolute()
|| !target_executable.is_absolute()
{
return Err("process session bridge launch path 必须是绝对路径".to_string());
}
let launch = CommandSandboxLaunch {
executable: sandbox_executable,
arguments: self
.sandbox_arguments
.into_iter()
.map(OsString::from_vec)
.collect(),
cwd: sandbox_cwd,
environment: self
.sandbox_environment
.into_iter()
.map(|(name, value)| (OsString::from_vec(name), OsString::from_vec(value)))
.collect(),
metadata: command_sandbox_platform_metadata(),
};
let target_arguments = self
.target_arguments
.into_iter()
.map(OsString::from_vec)
.collect::<Vec<_>>();
stage_command_sandbox_launch_for_process_session(
launch,
&target_executable,
&target_arguments,
)
.map_err(|error| error.to_string())
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
enum BridgeFrame {
Hello {
nonce: Vec<u8>,
},
Prepare {
nonce: Vec<u8>,
plan: ProcessSessionLaunchPlan,
},
SandboxReady {
nonce: Vec<u8>,
},
PrecommitFailed {
nonce: Vec<u8>,
failure_kind: String,
},
CommitExec {
nonce: Vec<u8>,
},
AbortLaunch {
nonce: Vec<u8>,
},
ExecEstablished {
nonce: Vec<u8>,
},
TargetExecFailed {
nonce: Vec<u8>,
errno: i32,
},
LaunchUnknown {
nonce: Vec<u8>,
},
TerminateTarget {
nonce: Vec<u8>,
},
TargetExited {
nonce: Vec<u8>,
code: i32,
},
TargetSignaled {
nonce: Vec<u8>,
signal: i32,
},
TerminalUnknown {
nonce: Vec<u8>,
},
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) enum ProcessSessionSandboxReadyVerdict {
Ready,
Failed { failure_kind: String },
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) enum ProcessSessionExecVerdict {
Established,
TargetExecFailed { errno: i32 },
LaunchUnknown,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) enum ProcessSessionTerminalVerdict {
Exited { code: i32 },
Signaled { signal: i32 },
Unknown,
}
pub(crate) struct ProcessSessionBridgeServer {
listener: UnixListener,
endpoint: String,
nonce: Vec<u8>,
nonce_hex: String,
}
impl ProcessSessionBridgeServer {
pub(crate) fn bind() -> Result<Self, String> {
let nonce = random_nonce()?;
let nonce_hex = encode_hex(&nonce);
let endpoint = format!("genarrative-ps-{}-{nonce_hex}", unsafe { libc::getpid() });
if endpoint.len() > BRIDGE_ENDPOINT_MAX_BYTES {
return Err("process session bridge endpoint 过长".to_string());
}
let address = SocketAddr::from_abstract_name(endpoint.as_bytes())
.map_err(|error| format!("创建 process session bridge 地址失败:{error}"))?;
let listener = UnixListener::bind_addr(&address)
.map_err(|error| format!("绑定 process session bridge 失败:{error}"))?;
set_close_on_exec(listener.as_raw_fd())
.map_err(|error| format!("保护 process session bridge listener 失败:{error}"))?;
listener
.set_nonblocking(true)
.map_err(|error| format!("配置 process session bridge 失败:{error}"))?;
Ok(Self {
listener,
endpoint,
nonce,
nonce_hex,
})
}
pub(crate) fn endpoint(&self) -> &str {
&self.endpoint
}
pub(crate) fn nonce_hex(&self) -> &str {
&self.nonce_hex
}
pub(crate) fn accept(
&self,
expected_peer_pid: u32,
timeout: Duration,
) -> Result<ProcessSessionBridge, String> {
let deadline = Instant::now() + timeout;
loop {
match self.listener.accept() {
Ok((mut stream, _)) => {
let (pid, uid) = peer_credentials(&stream)?;
if pid != expected_peer_pid || uid != unsafe { libc::geteuid() } {
continue;
}
set_close_on_exec(stream.as_raw_fd()).map_err(|error| {
format!("保护 process session bridge stream 失败:{error}")
})?;
let frame = read_frame_with_timeout(&mut stream, timeout)?;
if frame
!= (BridgeFrame::Hello {
nonce: self.nonce.clone(),
})
{
continue;
}
return Ok(ProcessSessionBridge {
stream,
nonce: self.nonce.clone(),
});
}
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {}
Err(error) => {
return Err(format!("接收 process session bridge 失败:{error}"));
}
}
if Instant::now() >= deadline {
return Err("等待 process session child bridge 超时".to_string());
}
thread::sleep(Duration::from_millis(5));
}
}
}
#[derive(Debug)]
pub(crate) struct ProcessSessionBridge {
stream: UnixStream,
nonce: Vec<u8>,
}
impl ProcessSessionBridge {
pub(crate) fn send_prepare(
&mut self,
launch: &ProjectCommandLaunchSpec,
spec: &ProjectCommandSpec,
) -> Result<(), String> {
let plan = ProcessSessionLaunchPlan::from_launch(launch, spec)?;
write_frame(
&mut self.stream,
&BridgeFrame::Prepare {
nonce: self.nonce.clone(),
plan,
},
)
.map_err(|error| format!("发送 process session launch plan 失败:{error}"))
}
pub(crate) fn wait_sandbox_ready(
&mut self,
timeout: Duration,
) -> Result<ProcessSessionSandboxReadyVerdict, String> {
match read_frame_with_timeout(&mut self.stream, timeout)? {
BridgeFrame::SandboxReady { nonce } if nonce == self.nonce => {
Ok(ProcessSessionSandboxReadyVerdict::Ready)
}
BridgeFrame::PrecommitFailed {
nonce,
failure_kind,
} if nonce == self.nonce && valid_failure_kind(&failure_kind) => {
Ok(ProcessSessionSandboxReadyVerdict::Failed { failure_kind })
}
_ => Err("process session sandbox-ready 帧无效".to_string()),
}
}
pub(crate) fn commit_exec(&mut self) -> Result<(), String> {
write_frame(
&mut self.stream,
&BridgeFrame::CommitExec {
nonce: self.nonce.clone(),
},
)
.map_err(|error| format!("发送 process session commit-exec 失败:{error}"))
}
pub(crate) fn abort_launch(&mut self) -> Result<(), String> {
write_frame(
&mut self.stream,
&BridgeFrame::AbortLaunch {
nonce: self.nonce.clone(),
},
)
.map_err(|error| format!("发送 process session abort-launch 失败:{error}"))
}
pub(crate) fn wait_exec(
&mut self,
timeout: Duration,
) -> Result<ProcessSessionExecVerdict, String> {
match read_frame_with_timeout(&mut self.stream, timeout)? {
BridgeFrame::ExecEstablished { nonce } if nonce == self.nonce => {
Ok(ProcessSessionExecVerdict::Established)
}
BridgeFrame::TargetExecFailed { nonce, errno } if nonce == self.nonce => {
Ok(ProcessSessionExecVerdict::TargetExecFailed { errno })
}
BridgeFrame::LaunchUnknown { nonce } if nonce == self.nonce => {
Ok(ProcessSessionExecVerdict::LaunchUnknown)
}
_ => Err("process session exec verdict 帧无效".to_string()),
}
}
pub(crate) fn wait_terminal(
&mut self,
timeout: Duration,
) -> Result<ProcessSessionTerminalVerdict, String> {
match read_frame_with_timeout(&mut self.stream, timeout)? {
BridgeFrame::TargetExited { nonce, code } if nonce == self.nonce => {
Ok(ProcessSessionTerminalVerdict::Exited { code })
}
BridgeFrame::TargetSignaled { nonce, signal } if nonce == self.nonce => {
Ok(ProcessSessionTerminalVerdict::Signaled { signal })
}
BridgeFrame::TerminalUnknown { nonce } if nonce == self.nonce => {
Ok(ProcessSessionTerminalVerdict::Unknown)
}
_ => Err("process session terminal 帧无效".to_string()),
}
}
pub(crate) fn terminate_target(&mut self) -> Result<(), String> {
write_frame(
&mut self.stream,
&BridgeFrame::TerminateTarget {
nonce: self.nonce.clone(),
},
)
.map_err(|error| format!("发送 process session target terminate 失败:{error}"))
}
}
pub(crate) fn run_process_session_bridge_child(
expected_parent: libc::pid_t,
) -> Result<i32, String> {
start_owner_monitor(expected_parent);
let endpoint = std::env::var(PROCESS_SESSION_BRIDGE_ENDPOINT_ENV)
.map_err(|_| "process session child 缺少 bridge endpoint".to_string())?;
let nonce_hex = std::env::var(PROCESS_SESSION_BRIDGE_NONCE_ENV)
.map_err(|_| "process session child 缺少 bridge nonce".to_string())?;
std::env::remove_var(PROCESS_SESSION_BRIDGE_ENDPOINT_ENV);
std::env::remove_var(PROCESS_SESSION_BRIDGE_NONCE_ENV);
let nonce = decode_nonce(&nonce_hex)?;
if endpoint.is_empty()
|| endpoint.len() > BRIDGE_ENDPOINT_MAX_BYTES
|| !endpoint.bytes().all(|byte| byte.is_ascii_graphic())
{
return Err("process session bridge endpoint 无效".to_string());
}
let address = SocketAddr::from_abstract_name(endpoint.as_bytes())
.map_err(|error| format!("解析 process session bridge 地址失败:{error}"))?;
let mut stream = UnixStream::connect_addr(&address)
.map_err(|error| format!("连接 process session bridge 失败:{error}"))?;
set_close_on_exec(stream.as_raw_fd())
.map_err(|error| format!("保护 process session child bridge 失败:{error}"))?;
write_frame(
&mut stream,
&BridgeFrame::Hello {
nonce: nonce.clone(),
},
)
.map_err(|error| format!("发送 process session bridge hello 失败:{error}"))?;
let plan = match read_frame_with_timeout(&mut stream, Duration::from_secs(3))? {
BridgeFrame::Prepare {
nonce: prepare_nonce,
plan,
} if prepare_nonce == nonce => plan,
_ => return Err("process session bridge prepare 帧无效".to_string()),
};
let mut staged = match plan.into_staged() {
Ok(staged) => staged,
Err(error) => {
let _ = write_frame(
&mut stream,
&BridgeFrame::PrecommitFailed {
nonce,
failure_kind: "stage-failed".to_string(),
},
);
return Err(error);
}
};
let mut command = Command::new(&staged.launch.executable);
command
.args(&staged.launch.arguments)
.current_dir(&staged.launch.cwd)
.env_clear()
.stdin(Stdio::inherit())
.stdout(Stdio::inherit())
.stderr(Stdio::inherit());
for (name, value) in &staged.launch.environment {
command.env(name, value);
}
staged
.gate
.install_on_command(&mut command)
.map_err(|error| {
let _ = write_frame(
&mut stream,
&BridgeFrame::PrecommitFailed {
nonce: nonce.clone(),
failure_kind: "gate-install-failed".to_string(),
},
);
error
})?;
let mut child = match command.spawn() {
Ok(child) => child,
Err(error) => {
staged.gate.spawn_failed();
let _ = write_frame(
&mut stream,
&BridgeFrame::PrecommitFailed {
nonce,
failure_kind: "sandbox-spawn-failed".to_string(),
},
);
return Err(format!("启动 process session sandbox 失败:{error}"));
}
};
let ready = staged
.gate
.child_created()
.and_then(|()| staged.gate.wait_sandbox_ready(Duration::from_secs(3)));
if let Err(error) = ready {
let _ = abort_staged_child(&mut staged.gate, &mut child);
let _ = write_frame(
&mut stream,
&BridgeFrame::PrecommitFailed {
nonce,
failure_kind: "sandbox-ready-failed".to_string(),
},
);
return Err(error);
}
if let Err(error) = write_frame(
&mut stream,
&BridgeFrame::SandboxReady {
nonce: nonce.clone(),
},
) {
let _ = abort_staged_child(&mut staged.gate, &mut child);
return Err(format!("发送 process session sandbox-ready 失败:{error}"));
}
match read_frame(&mut stream) {
Ok(BridgeFrame::CommitExec {
nonce: commit_nonce,
}) if commit_nonce == nonce => {}
Ok(BridgeFrame::AbortLaunch { nonce: abort_nonce }) if abort_nonce == nonce => {
abort_staged_child(&mut staged.gate, &mut child)?;
return Ok(125);
}
Ok(_) => {
let _ = abort_staged_child(&mut staged.gate, &mut child);
return Err("process session bridge commit 帧无效".to_string());
}
Err(error) => {
let _ = abort_staged_child(&mut staged.gate, &mut child);
return Err(format!("读取 process session bridge commit 失败:{error}"));
}
}
if staged.gate.commit_exec().is_err() {
let _ = abort_staged_child(&mut staged.gate, &mut child);
let _ = write_frame(
&mut stream,
&BridgeFrame::LaunchUnknown {
nonce: nonce.clone(),
},
);
return Ok(125);
}
match staged.gate.wait_target_exec(Duration::from_secs(3)) {
Ok(TargetExecState::Established) => {
write_frame(
&mut stream,
&BridgeFrame::ExecEstablished {
nonce: nonce.clone(),
},
)
.map_err(|error| {
let _ = child.kill();
let _ = child.wait();
format!("发送 process session exec-established 失败:{error}")
})?;
}
Ok(TargetExecState::Failed { errno }) => {
let _ = child.kill();
let _ = child.wait();
let _ = write_frame(&mut stream, &BridgeFrame::TargetExecFailed { nonce, errno });
return Ok(126);
}
Err(_) => {
let _ = abort_staged_child(&mut staged.gate, &mut child);
let _ = write_frame(
&mut stream,
&BridgeFrame::LaunchUnknown {
nonce: nonce.clone(),
},
);
return Ok(125);
}
}
let status = loop {
if let Some(status) = child
.try_wait()
.map_err(|error| format!("检查 process session sandbox 失败:{error}"))?
{
break status;
}
let Some(frame) = read_frame_if_available(&mut stream, Duration::from_millis(25))?
else {
continue;
};
match frame {
BridgeFrame::TerminateTarget {
nonce: terminate_nonce,
} if terminate_nonce == nonce => staged.gate.terminate_target()?,
_ => return Err("process session 运行期控制帧无效".to_string()),
}
};
let terminal = staged.gate.wait_terminal(Duration::from_secs(2));
match terminal {
Ok(TargetTerminalState::Exited { code }) => {
let _ = write_frame(
&mut stream,
&BridgeFrame::TargetExited {
nonce: nonce.clone(),
code,
},
);
}
Ok(TargetTerminalState::Signaled { signal }) => {
let _ = write_frame(
&mut stream,
&BridgeFrame::TargetSignaled {
nonce: nonce.clone(),
signal,
},
);
}
Err(_) => {
let _ = write_frame(
&mut stream,
&BridgeFrame::TerminalUnknown {
nonce: nonce.clone(),
},
);
}
}
Ok(status.code().unwrap_or(128))
}
fn start_owner_monitor(expected_parent: libc::pid_t) {
thread::spawn(move || loop {
if unsafe { libc::getppid() } != expected_parent {
unsafe {
libc::kill(0, libc::SIGKILL);
libc::_exit(125);
}
}
thread::sleep(Duration::from_millis(25));
});
}
fn abort_staged_child(
gate: &mut crate::command_sandbox_trampoline::LaunchGate,
child: &mut std::process::Child,
) -> Result<(), String> {
match gate.abort(child) {
Ok(()) => Ok(()),
Err(error) => {
let _ = child.kill();
child
.wait()
.map(|_| ())
.map_err(|wait_error| format!("{error};回收失败:{wait_error}"))
}
}
}
fn valid_failure_kind(value: &str) -> bool {
matches!(
value,
"stage-failed"
| "gate-install-failed"
| "sandbox-spawn-failed"
| "sandbox-ready-failed"
)
}
fn peer_credentials(stream: &UnixStream) -> Result<(u32, u32), String> {
let mut credentials = libc::ucred {
pid: 0,
uid: 0,
gid: 0,
};
let mut length = std::mem::size_of::<libc::ucred>() as libc::socklen_t;
let result = unsafe {
libc::getsockopt(
stream.as_raw_fd(),
libc::SOL_SOCKET,
libc::SO_PEERCRED,
&mut credentials as *mut _ as *mut libc::c_void,
&mut length,
)
};
if result != 0 || length as usize != std::mem::size_of::<libc::ucred>() {
return Err(format!(
"读取 process session bridge peer credential 失败:{}",
io::Error::last_os_error()
));
}
Ok((
u32::try_from(credentials.pid)
.map_err(|_| "process session bridge peer pid 无效".to_string())?,
credentials.uid,
))
}
fn set_close_on_exec(fd: libc::c_int) -> io::Result<()> {
let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
if flags < 0 {
return Err(io::Error::last_os_error());
}
if unsafe { libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC) } < 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
fn random_nonce() -> Result<Vec<u8>, String> {
let mut nonce = vec![0_u8; BRIDGE_NONCE_BYTES];
File::open("/dev/urandom")
.and_then(|mut file| file.read_exact(&mut nonce))
.map_err(|error| format!("读取 process session bridge nonce 失败:{error}"))?;
Ok(nonce)
}
fn encode_hex(bytes: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut encoded = String::with_capacity(bytes.len() * 2);
for byte in bytes {
encoded.push(HEX[(byte >> 4) as usize] as char);
encoded.push(HEX[(byte & 0x0f) as usize] as char);
}
encoded
}
fn decode_nonce(value: &str) -> Result<Vec<u8>, String> {
if value.len() != BRIDGE_NONCE_BYTES * 2 {
return Err("process session bridge nonce 长度无效".to_string());
}
let bytes = value.as_bytes();
let mut decoded = Vec::with_capacity(BRIDGE_NONCE_BYTES);
for index in (0..bytes.len()).step_by(2) {
let high = decode_hex_digit(bytes[index])?;
let low = decode_hex_digit(bytes[index + 1])?;
decoded.push((high << 4) | low);
}
Ok(decoded)
}
fn decode_hex_digit(value: u8) -> Result<u8, String> {
match value {
b'0'..=b'9' => Ok(value - b'0'),
b'a'..=b'f' => Ok(value - b'a' + 10),
_ => Err("process session bridge nonce 不是小写十六进制".to_string()),
}
}
fn write_frame(stream: &mut UnixStream, frame: &BridgeFrame) -> io::Result<()> {
let payload = serde_json::to_vec(frame)
.map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
if payload.is_empty() || payload.len() > BRIDGE_FRAME_MAX_BYTES {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"process session bridge 帧大小无效",
));
}
stream.write_all(&(payload.len() as u32).to_be_bytes())?;
stream.write_all(&payload)?;
stream.flush()
}
fn read_frame(stream: &mut UnixStream) -> io::Result<BridgeFrame> {
let mut length = [0_u8; 4];
stream.read_exact(&mut length)?;
let length = u32::from_be_bytes(length) as usize;
if length == 0 || length > BRIDGE_FRAME_MAX_BYTES {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"process session bridge 帧大小无效",
));
}
let mut payload = vec![0_u8; length];
stream.read_exact(&mut payload)?;
serde_json::from_slice(&payload)
.map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))
}
fn read_frame_if_available(
stream: &mut UnixStream,
timeout: Duration,
) -> Result<Option<BridgeFrame>, String> {
let timeout_ms = i32::try_from(timeout.as_millis()).unwrap_or(i32::MAX);
let mut descriptor = libc::pollfd {
fd: stream.as_raw_fd(),
events: libc::POLLIN | libc::POLLHUP,
revents: 0,
};
let result = unsafe { libc::poll(&mut descriptor, 1, timeout_ms) };
if result < 0 {
return Err(format!(
"等待 process session 运行期控制失败:{}",
io::Error::last_os_error()
));
}
if result == 0 {
return Ok(None);
}
if descriptor.revents & (libc::POLLERR | libc::POLLNVAL) != 0 {
return Err("process session 运行期控制通道失效".to_string());
}
read_frame(stream)
.map(Some)
.map_err(|error| format!("读取 process session 运行期控制失败:{error}"))
}
fn read_frame_with_timeout(
stream: &mut UnixStream,
timeout: Duration,
) -> Result<BridgeFrame, String> {
stream
.set_read_timeout(Some(timeout))
.map_err(|error| format!("配置 process session bridge timeout 失败:{error}"))?;
let result = read_frame(stream)
.map_err(|error| format!("读取 process session bridge 帧失败:{error}"));
let reset = stream.set_read_timeout(None);
match (result, reset) {
(Ok(frame), Ok(())) => Ok(frame),
(Err(error), _) => Err(error),
(Ok(_), Err(error)) => {
Err(format!("重置 process session bridge timeout 失败:{error}"))
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn send_test_frame(endpoint: String, frame: BridgeFrame) -> thread::JoinHandle<()> {
thread::spawn(move || {
let address = SocketAddr::from_abstract_name(endpoint.as_bytes())
.expect("test bridge address");
let mut stream = UnixStream::connect_addr(&address).expect("connect test bridge");
write_frame(&mut stream, &frame).expect("write test bridge frame");
})
}
#[test]
fn bridge_nonce_hex_round_trip_is_exact() {
let nonce = (0..BRIDGE_NONCE_BYTES as u8).collect::<Vec<_>>();
let encoded = encode_hex(&nonce);
assert_eq!(decode_nonce(&encoded).expect("decode nonce"), nonce);
assert!(decode_nonce(&encoded.to_uppercase()).is_err());
}
#[test]
fn bridge_failure_kinds_are_closed() {
assert!(valid_failure_kind("sandbox-ready-failed"));
assert!(!valid_failure_kind("arbitrary"));
}
#[test]
fn bridge_accepts_only_exact_peer_and_nonce_hello() {
let server = ProcessSessionBridgeServer::bind().expect("bind bridge");
let child = send_test_frame(
server.endpoint().to_string(),
BridgeFrame::Hello {
nonce: server.nonce.clone(),
},
);
server
.accept(std::process::id(), Duration::from_secs(1))
.expect("accept exact peer hello");
child.join().expect("join bridge child");
}
#[test]
fn bridge_rejects_wrong_nonce_and_out_of_order_frame() {
for frame in [
BridgeFrame::Hello {
nonce: vec![0; BRIDGE_NONCE_BYTES],
},
BridgeFrame::CommitExec {
nonce: vec![0; BRIDGE_NONCE_BYTES],
},
] {
let server = ProcessSessionBridgeServer::bind().expect("bind bridge");
let child = send_test_frame(server.endpoint().to_string(), frame);
let error = server
.accept(std::process::id(), Duration::from_millis(80))
.expect_err("reject invalid bridge hello");
assert!(error.contains("超时"), "{error}");
child.join().expect("join invalid bridge child");
}
}
#[test]
fn bridge_rejects_wrong_peer_pid() {
let server = ProcessSessionBridgeServer::bind().expect("bind bridge");
let child = send_test_frame(
server.endpoint().to_string(),
BridgeFrame::Hello {
nonce: server.nonce.clone(),
},
);
let error = server
.accept(u32::MAX, Duration::from_millis(80))
.expect_err("reject wrong peer pid");
assert!(error.contains("超时"), "{error}");
child.join().expect("join wrong peer child");
}
}
}
#[cfg(target_os = "linux")]
pub(crate) use linux::*;