From b15967ab10f15cb9577df2330b95a88fdad631c8 Mon Sep 17 00:00:00 2001 From: pjh456 <147148383@qq.com> Date: Sun, 11 Oct 2026 16:23:35 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20server=20=E8=BF=9E=E6=8E=A5=E5=8A=A0?= =?UTF-8?q?=E5=85=B1=E4=BA=AB=20token=20=E8=AE=A4=E8=AF=81=E9=97=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 40 ++-- Cargo.toml | 2 +- crates/oxide_cli/src/server/cleanup.rs | 1 + crates/oxide_cli/src/server/client.rs | 71 +++++-- crates/oxide_cli/src/server/liveness.rs | 10 +- crates/oxide_cli/src/server/protocol.rs | 25 ++- crates/oxide_cli/src/server/server.rs | 272 ++++++++++++++++++++---- crates/oxide_cli/src/server/sidecar.rs | 160 +++++++++++--- crates/oxide_cli/src/server/watchdog.rs | 1 + crates/oxide_cli/tests/server_cli.rs | 53 ++++- 10 files changed, 519 insertions(+), 116 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d3be7984..a7325531 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.15.11" +version = "0.15.12" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.15.11" +version = "0.15.12" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.15.11" +version = "0.15.12" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.15.11" +version = "0.15.12" dependencies = [ "ansi_term", "bumpalo", @@ -1228,7 +1228,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.15.11" +version = "0.15.12" dependencies = [ "lru", "oxide_bytecode", @@ -1237,7 +1237,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.15.11" +version = "0.15.12" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1257,7 +1257,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.15.11" +version = "0.15.12" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1271,7 +1271,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.15.11" +version = "0.15.12" dependencies = [ "num-bigint", "oxc_allocator", @@ -1286,7 +1286,7 @@ dependencies = [ [[package]] name = "oxide_engine" -version = "0.15.11" +version = "0.15.12" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1298,7 +1298,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.15.11" +version = "0.15.12" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1307,7 +1307,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.15.11" +version = "0.15.12" dependencies = [ "bumpalo", "dashmap", @@ -1321,7 +1321,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.15.11" +version = "0.15.12" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1333,7 +1333,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.15.11" +version = "0.15.12" dependencies = [ "tracing", "tracing-appender", @@ -1342,11 +1342,11 @@ dependencies = [ [[package]] name = "oxide_net" -version = "0.15.11" +version = "0.15.12" [[package]] name = "oxide_parser" -version = "0.15.11" +version = "0.15.12" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1361,7 +1361,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.15.11" +version = "0.15.12" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1377,7 +1377,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.15.11" +version = "0.15.12" dependencies = [ "num-bigint", "num-traits", @@ -1390,7 +1390,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.15.11" +version = "0.15.12" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1408,7 +1408,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.15.11" +version = "0.15.12" dependencies = [ "bumpalo", "dashmap", @@ -1421,7 +1421,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.15.11" +version = "0.15.12" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 0169fe91..069874ce 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ members = [ ] [workspace.package] -version = "0.15.11" +version = "0.15.12" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_cli/src/server/cleanup.rs b/crates/oxide_cli/src/server/cleanup.rs index 011dd68f..338d2ab4 100644 --- a/crates/oxide_cli/src/server/cleanup.rs +++ b/crates/oxide_cli/src/server/cleanup.rs @@ -307,6 +307,7 @@ mod tests { starttime, port: 0, listen_addr: String::new(), + token: String::new(), } } diff --git a/crates/oxide_cli/src/server/client.rs b/crates/oxide_cli/src/server/client.rs index ae445a55..6f851198 100644 --- a/crates/oxide_cli/src/server/client.rs +++ b/crates/oxide_cli/src/server/client.rs @@ -52,16 +52,18 @@ pub fn send_control_request(request: &ServerRequest) -> Result= Duration::from_secs(5), "应命中 5 秒读超时：{elapsed:?}"); } + + /// 假 server 回 AuthOk 加 Version 响应：认证通过后正常读回响应。 + #[test] + fn auth_then_request_success() { + let dir = TestDir::new(); + let sidecar = dir.path("sidecar.json"); + let socket = dir.path("server.sock"); + write_sidecar(&sidecar, &socket); + fake_server(&socket, |stream| { + let mut reader = FrameReader::new(BufReader::new(stream.try_clone().expect("克隆流应成功"))); + // 读认证帧，回 AuthOk。 + let _ = reader.read_frame(); + let auth_ok = protocol::encode_response(&ServerResponse::AuthOk); + stream.write_all(auth_ok.as_bytes()).expect("写 AuthOk 应成功"); + // 读请求帧，回 Version 响应。 + let _ = reader.read_frame(); + let version = protocol::encode_response(&ServerResponse::Version { version: "0.0.0".into() }); + stream.write_all(version.as_bytes()).expect("写 Version 应成功"); + }); + let response = send_control_request_from(&sidecar, &ServerRequest::Version).expect("版本请求应成功"); + assert!(matches!(response, ServerResponse::Version { .. }), "应得 Version 帧：{response:?}"); + } } diff --git a/crates/oxide_cli/src/server/liveness.rs b/crates/oxide_cli/src/server/liveness.rs index 01e5ba52..f05a3fff 100644 --- a/crates/oxide_cli/src/server/liveness.rs +++ b/crates/oxide_cli/src/server/liveness.rs @@ -260,6 +260,7 @@ mod tests { starttime, port: 0, listen_addr: String::new(), + token: String::new(), } } @@ -333,8 +334,11 @@ mod tests { let sidecar = dir.path("sidecar.json"); let socket = dir.path("server.sock"); let listener = UnixListener::bind(&socket).expect("绑定监听者应成功"); - sidecar::write_exclusive(&sidecar::ServerIdentity::new(&Endpoint::Unix(socket.clone()), "0.0.0"), &sidecar) - .expect("写 sidecar 应成功"); + sidecar::write_exclusive( + &sidecar::ServerIdentity::new(&Endpoint::Unix(socket.clone()), "0.0.0", "test-token"), + &sidecar, + ) + .expect("写 sidecar 应成功"); let outcome = liveness_scan(&sidecar, &Endpoint::Unix(socket.clone())); assert_eq!(outcome, LivenessOutcome::Noop, "存活 socket 应判无操作"); assert!(sidecar.exists(), "sidecar 应原样"); @@ -350,7 +354,7 @@ mod tests { let sidecar = dir.path("sidecar.json"); let socket = dir.path("server.sock"); // sidecar 记本进程 PID 加真实刻度（进程活且刻度匹配）、无 socket。 - let id = sidecar::ServerIdentity::new(&Endpoint::Unix(socket.clone()), "0.0.0"); + let id = sidecar::ServerIdentity::new(&Endpoint::Unix(socket.clone()), "0.0.0", "test-token"); sidecar::write_exclusive(&id, &sidecar).expect("写 sidecar 应成功"); // 扫描线程：sidecar 存在时进入僵尸分支，有界等待 sidecar 消失。 let sidecar_scan = sidecar.clone(); diff --git a/crates/oxide_cli/src/server/protocol.rs b/crates/oxide_cli/src/server/protocol.rs index dd0b28c6..d016c5c7 100644 --- a/crates/oxide_cli/src/server/protocol.rs +++ b/crates/oxide_cli/src/server/protocol.rs @@ -126,6 +126,11 @@ pub enum ServerRequest { Shutdown, /// 交接请求：旧 server 排空在途后退出，让出 socket 路径。 Yield, + /// 认证请求：携带共享 token，连接建立后的首帧。 + Auth { + /// 共享 token（十六进制），server 侧逐字节异或判等。 + token: String, + }, } /// server 响应帧：每类请求一个响应，外加协议错误帧。 @@ -185,6 +190,8 @@ pub enum ServerResponse { Shutdown, /// 交接确认：旧 server 已受理，开始排空退出。 Yield, + /// 认证确认：首帧 token 校验通过，客户端可继续发真实请求。 + AuthOk, /// 协议错误帧：畸形帧或未知类型，server 以该帧答复。 Error { message: String }, } @@ -208,7 +215,8 @@ impl ServerResponse { } /// 请求帧可取的 `type` 值（须与 `ServerRequest` 变体保持同步）。 -const REQUEST_TAGS: &[&str] = &["eval", "forge_query", "version", "status", "health", "info", "shutdown", "yield"]; +const REQUEST_TAGS: &[&str] = + &["eval", "forge_query", "version", "status", "health", "info", "shutdown", "yield", "auth"]; /// 响应帧可取的 `type` 值（须与 `ServerResponse` 变体保持同步）。 const RESPONSE_TAGS: &[&str] = &[ @@ -220,6 +228,7 @@ const RESPONSE_TAGS: &[&str] = &[ "info", "shutdown", "yield", + "auth_ok", "error", ]; @@ -408,7 +417,7 @@ mod tests { }); } - /// 六类控制请求 round-trip。 + /// 六类控制请求加认证请求 round-trip。 #[test] fn control_request_roundtrip() { for req in [ @@ -418,6 +427,7 @@ mod tests { ServerRequest::Info, ServerRequest::Shutdown, ServerRequest::Yield, + ServerRequest::Auth { token: "ab".into() }, ] { roundtrip_request(&req); } @@ -445,6 +455,7 @@ mod tests { }, ServerResponse::Shutdown, ServerResponse::Yield, + ServerResponse::AuthOk, ServerResponse::Error { message: "boom".into() }, ] { roundtrip_response(&resp); @@ -550,6 +561,16 @@ mod tests { assert_eq!(serde_json::to_string(&ServerRequest::Health).unwrap(), r#"{"type":"health"}"#); } + /// 序列化形态：auth 请求携带 token 字段，auth_ok 响应无载荷。 + #[test] + fn auth_frame_serialization_shape() { + assert_eq!( + serde_json::to_string(&ServerRequest::Auth { token: "ab12".into() }).unwrap(), + r#"{"type":"auth","token":"ab12"}"# + ); + assert_eq!(serde_json::to_string(&ServerResponse::AuthOk).unwrap(), r#"{"type":"auth_ok"}"#); + } + /// 序列化形态：eval_result 省略 None 字段。 #[test] fn response_serialization_shape() { diff --git a/crates/oxide_cli/src/server/server.rs b/crates/oxide_cli/src/server/server.rs index 1f61e5b1..8b5a051d 100644 --- a/crates/oxide_cli/src/server/server.rs +++ b/crates/oxide_cli/src/server/server.rs @@ -113,7 +113,7 @@ fn default_worker_count() -> usize { /// --rm 模式缺省空闲超时：30 秒。 const RM_DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(30); -/// --rm 独立模式运行配置：监听端点、worker 数、空闲超时。 +/// --rm 独立模式运行配置：监听端点、worker 数、空闲超时、共享 token。 /// /// 不含 sidecar 路径字段：「不注册 sidecar」的不变量在类型上成立， /// 不需要运行时守卫。 @@ -126,6 +126,8 @@ pub struct RmServerConfig { /// 空闲超时：绑定后该时长内无连接、或连接建立后该时长内无数据， /// 进程即干净退出。 pub idle_timeout: Duration, + /// 共享 token；空串时启动时生成（生产路径），测试注入已知值。 + pub token: String, } impl Default for RmServerConfig { @@ -134,6 +136,7 @@ impl Default for RmServerConfig { endpoint: Endpoint::Unix(sidecar::rm_socket_path()), worker_count: default_worker_count(), idle_timeout: RM_DEFAULT_IDLE_TIMEOUT, + token: String::new(), } } } @@ -156,6 +159,8 @@ struct ServerContext { // 实际监听地址（bind 后取值：TCP 臂为 host:port、Unix 臂为 socket 路径）， // Info 响应用，让控制客户端经协议层发现 server 网络端点。 listen_addr: String, + // 共享 token（十六进制）：连接首帧认证门判等用。 + token: String, } /// server 进程主体：初始化日志文件输出 → 启动时 liveness 扫描 → 认领 sidecar → @@ -222,12 +227,15 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> { liveness::LivenessOutcome::KillFailed => return Err(ServerError::ClaimRefused(ClaimResult::RefusedAmbiguous)), } + // 生成共享 token（认领之前：写入 sidecar 供控制客户端读回认证）。 + let token = sidecar::generate_token(); + // 认领 sidecar（启动顺序契约：先写 sidecar 再绑定监听器）：同版本 // 拒绝启动，异版本走交接路径，存活证据不足即返回错误。 - match sidecar::claim_sidecar(&config.sidecar_path, &config.endpoint, &config.version) { + match sidecar::claim_sidecar(&config.sidecar_path, &config.endpoint, &config.version, &token) { ClaimResult::Registered => {} ClaimResult::AlreadyRunning => return Err(ServerError::ClaimRefused(ClaimResult::AlreadyRunning)), - ClaimResult::VersionMismatch => yield_and_takeover(config)?, + ClaimResult::VersionMismatch => yield_and_takeover(config, &token)?, ClaimResult::RefusedAmbiguous => return Err(ServerError::ClaimRefused(ClaimResult::RefusedAmbiguous)), } @@ -269,6 +277,7 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> { pool_counters: counters, socket_path: config.endpoint.describe(), listen_addr, + token, }); // accept 循环：非阻塞 accept 加 10 毫秒轮询加原子关闭标志。 @@ -334,6 +343,16 @@ pub fn run_server_rm(config: &RmServerConfig) -> Result<(), ServerError> { Listener::Unix(_) => config.endpoint.describe(), }; + // 共享 token：配置注入已知值（测试）时直接用，空串时启动时生成（生产 + // 路径），向前台标准输出打印（客户端经 CLI 旗标注入，与 TCP 臂打印 + // 端点同一通道）。 + let token = if config.token.is_empty() { + sidecar::generate_token() + } else { + config.token.clone() + }; + println!("rm server token：{token}"); + // 建内核（标准配置）与共享聚合计数器（全体 worker 池的增减累加到同一聚合）。 let kernel = KernelCore::new(KernelConfig::standard()); let counters = PoolCounters::shared(); @@ -358,6 +377,7 @@ pub fn run_server_rm(config: &RmServerConfig) -> Result<(), ServerError> { // 实际地址（local_addr），Info 响应的监听地址与之一致。 socket_path: socket_path.clone(), listen_addr: socket_path, + token, }); // 空闲窗口一：等首个连接；空闲超时内无连接即干净退出。 @@ -476,23 +496,29 @@ fn finish_shutdown( Ok(()) } -/// 交接接管：连接旧 server、发 yield 请求、等旧 server 排空退出后重新认领。 +/// 交接接管：连接旧 server、发认证帧加 yield 请求、等旧 server 排空退出 +/// 后重新认领。 /// /// # 步骤 -/// 1. 连接旧 server 的 endpoint（Unix 路径或 TCP 地址）；连接失败判 +/// 1. 读旧 sidecar 取 token（旧 server 为新版本时 token 非空；旧版本或 +/// sidecar 不可读时 token 为空，跳过认证直接发 yield）。 +/// 2. 连接旧 server 的 endpoint（Unix 路径或 TCP 地址）；连接失败判 /// `YieldFailed`。 -/// 2. 设 5 秒读超时，写 yield 请求帧。 -/// 3. 读一帧应答：读超时（旧 server 挂起）或应答非 yield 确认（旧版本 -/// binary 无该变体时以错误帧答复未知类型）判 `YieldFailed`。 -/// 4. 关闭连接，轮询 sidecar 文件消失（50 毫秒间隔、10 秒上限）判旧 server +/// 3. 设 5 秒读超时；token 非空时先写认证帧并读 `AuthOk`，不匹配判 +/// `YieldFailed`。 +/// 4. 写 yield 请求帧，读一帧应答：读超时（旧 server 挂起）或应答非 yield +/// 确认（旧版本 binary 无该变体时以错误帧答复未知类型）判 `YieldFailed`。 +/// 5. 关闭连接，轮询 sidecar 文件消失（50 毫秒间隔、10 秒上限）判旧 server /// 退出；超时判 `YieldFailed`。 -/// 5. 重新认领 sidecar；得 `Registered` 即接管成功，其余态判 `YieldFailed`。 +/// 6. 重新认领 sidecar；得 `Registered` 即接管成功，其余态判 `YieldFailed`。 /// /// # 边界与前提 /// - 交接等待 10 秒长于旧 server 的排水安全超时 5 秒，正常路径必然先于 /// 等待上限完成。 /// - 两臂统一以 sidecar 消失为退出信号：旧 server 先删 socket 文件（Unix /// 臂）后删 sidecar，sidecar 消失时接管绑定无竞态。 +/// - token 为空（旧版本 server 无认证门）时跳过认证帧，直接发 yield， +/// 保证从旧版本升级的路径可用。 /// /// # 副作用 /// - 失败时不触碰旧 server 的文件（socket 与 sidecar），人工兜底归清理命令。 @@ -501,9 +527,15 @@ fn finish_shutdown( /// - 纯函数式入口：版本管理路径（版本不匹配强制交接）与显式交接路径调用 /// 同一入口。 /// - 退出信号源是 sidecar 文件消失（两臂同源，TCP server 也写 sidecar）。 -pub fn yield_and_takeover(config: &ServerConfig) -> Result<(), ServerError> { +pub fn yield_and_takeover(config: &ServerConfig, token: &str) -> Result<(), ServerError> { + // 读旧 sidecar 取 token：旧 server 为新版本时 token 非空（需认证）， + // 旧版本或 sidecar 不可读时 token 为空（跳过认证，直接发 yield）。 + let old_token = sidecar::read_identity(&config.sidecar_path) + .map(|id| id.token) + .unwrap_or_default(); + // 连接旧 server：连接成功是交接的前提，两臂按 endpoint 形态分支。 - let mut stream = match &config.endpoint { + let stream = match &config.endpoint { Endpoint::Unix(path) => Conn::Unix( UnixStream::connect(path).map_err(|e| ServerError::YieldFailed(format!("连接旧 server 失败：{e}")))?, ), @@ -512,14 +544,42 @@ pub fn yield_and_takeover(config: &ServerConfig) -> Result<(), ServerError> { ), }; - // 5 秒读超时：旧 server 挂起（不回应 yield）不无限阻塞。 + // 5 秒读超时：旧 server 挂起（不回应）不无限阻塞。 let _ = stream.set_read_timeout(Some(Duration::from_secs(5))); - stream + + // 克隆写端：原流交给帧读取器，写端独立用于请求写回。 + let mut writer = stream + .try_clone() + .map_err(|e| ServerError::YieldFailed(format!("克隆流失败：{e}")))?; + + // 认证帧：token 非空时先发 Auth 并读 AuthOk（旧 server 为新版本）。 + if !old_token.is_empty() { + writer + .write_all(protocol::encode_request(&ServerRequest::Auth { token: old_token.clone() }).as_bytes()) + .map_err(|e| ServerError::YieldFailed(format!("写认证帧失败：{e}")))?; + } + + // 写 yield 请求帧。 + writer .write_all(protocol::encode_request(&ServerRequest::Yield).as_bytes()) .map_err(|e| ServerError::YieldFailed(format!("写 yield 请求帧失败：{e}")))?; - // 读确认帧：只有 yield 确认表示旧 server 已受理交接。 let mut reader = FrameReader::new(io::BufReader::new(stream)); + + // 读认证确认帧：token 非空时首帧是 AuthOk。 + if !old_token.is_empty() { + let frame = reader + .read_frame() + .map_err(|e| ServerError::YieldFailed(format!("读认证确认帧失败：{e}")))? + .ok_or_else(|| ServerError::YieldFailed("旧 server 未应答即关闭连接".into()))?; + let response = protocol::parse_response(&frame) + .map_err(|e| ServerError::YieldFailed(format!("解析认证确认帧失败：{e}")))?; + if !matches!(response, ServerResponse::AuthOk) { + return Err(ServerError::YieldFailed(format!("旧 server 未受理认证：{response:?}"))); + } + } + + // 读 yield 确认帧：只有 yield 确认表示旧 server 已受理交接。 let frame = reader .read_frame() .map_err(|e| ServerError::YieldFailed(format!("读 yield 确认帧失败：{e}")))? @@ -543,7 +603,8 @@ pub fn yield_and_takeover(config: &ServerConfig) -> Result<(), ServerError> { } // 重新认领：sidecar 已消失，O_EXCL 创建应成功；其余态判交接失败。 - match sidecar::claim_sidecar(&config.sidecar_path, &config.endpoint, &config.version) { + // token 与上下文一致（run_server 生成），sidecar 与 ctx.token 同源。 + match sidecar::claim_sidecar(&config.sidecar_path, &config.endpoint, &config.version, token) { ClaimResult::Registered => Ok(()), other => Err(ServerError::YieldFailed(format!("交接后重新认领未通过：{other:?}"))), } @@ -682,19 +743,39 @@ fn accept_loop(listener: &Listener, ctx: &Arc, router: &Arc bool { + if expected.len() != given.len() { + return false; + } + let mut diff = 0u8; + for (x, y) in expected.bytes().zip(given.bytes()) { + diff |= x ^ y; + } + diff == 0 +} + +/// 每连接处理：认证门 → 帧读取 → 分派 → 响应写回，循环至对端关闭。 /// /// # 步骤 /// 1. 克隆写端，原流交给帧读取器，设读超时（`read_timeout` 参数）。 -/// 2. 循环读帧：EOF、读超时、非法 UTF-8 退出；帧长超限按协议约定关连接。 -/// 3. 畸形帧以 `Error` 帧答复后继续；执行请求与 forge 查询经 worker 路由 +/// 2. 认证门：读首帧，要求是 `Auth` 且 token 与上下文匹配（逐字节异或 +/// 判等，避免时序侧信道）；匹配则写回 `AuthOk` 再进主循环，不匹配 +/// 或首帧非 `Auth` 则直接断连（不回帧，避免向攻击者泄露差异）。 +/// 3. 循环读帧：EOF、读超时、非法 UTF-8 退出；帧长超限按协议约定关连接。 +/// 4. 畸形帧以 `Error` 帧答复后继续；执行请求与 forge 查询经 worker 路由 /// 并阻塞等回复；控制请求直接应答并写回。 -/// 4. 关闭或 yield 确认帧写回即退出循环。 +/// 5. 关闭或 yield 确认帧写回即退出循环。 /// /// # 边界与前提 /// - 半开连接（对端静默不关）由读超时有界化：持久 server 传 30 秒， /// --rm 独立模式传空闲超时（读超时即空闲语义）。 /// - 本线程只做 I/O 与分派，不建池（池归 worker 线程所有）。 +/// - 探活连接（只建连接不发帧）在认证门读首帧时得 EOF，直接退出， +/// 不触发任何错误路径。 /// /// # 副作用 /// - 关闭请求会置位关闭标志。 @@ -709,6 +790,25 @@ fn handle_connection(stream: Conn, ctx: &ServerContext, router: &WorkerRouter, r let _ = stream.set_read_timeout(Some(read_timeout)); let mut reader = FrameReader::new(io::BufReader::new(stream)); + + // 认证门：首帧必须是 Auth 且 token 匹配，不匹配即断连（不回帧）。 + let first_frame = match reader.read_frame() { + Ok(Some(frame)) => frame, + _ => return, + }; + let auth_ok = match protocol::parse_request(&first_frame) { + Ok(ServerRequest::Auth { token }) => tokens_match(&ctx.token, &token), + _ => false, + }; + if !auth_ok { + return; + } + // 写回 AuthOk：客户端据以继续发真实请求。 + let encoded = protocol::encode_response(&ServerResponse::AuthOk); + if writer.write_all(encoded.as_bytes()).is_err() { + return; + } + // 读帧：EOF、读超时、非法 UTF-8 均退出。 while let Ok(Some(frame)) = reader.read_frame() { // 分派：帧长超限按协议约定关连接；畸形帧以 Error 帧答复后继续； @@ -818,9 +918,11 @@ fn dispatch_control(request: &ServerRequest, ctx: &ServerContext) -> ServerRespo ServerResponse::Yield } // 防御性兜底：正常路径 eval 与 forge 查询在调用本函数前已被拦截走 - // worker 路由；若到达此臂说明上游分流缺失，以错误帧答复而非静默执行。 + // worker 路由；auth 在认证门消费；若到达此臂说明上游分流缺失， + // 以错误帧答复而非静默执行。 ServerRequest::Eval { .. } => ServerResponse::eval_err("执行请求应经 worker 路由，不应到达控制分派"), ServerRequest::ForgeQuery { .. } => ServerResponse::eval_err("forge 查询应经 worker 路由，不应到达控制分派"), + ServerRequest::Auth { .. } => ServerResponse::eval_err("认证请求应在认证门消费，不应到达控制分派"), } } @@ -882,6 +984,29 @@ mod tests { } } + /// 读 sidecar 取共享 token。 + fn read_token(config: &ServerConfig) -> String { + sidecar::read_identity(&config.sidecar_path).expect("sidecar 应可读").token + } + + /// 在已有连接上发认证帧并读回 AuthOk。 + fn authenticate(stream: &mut Conn, token: &str) { + stream + .write_all(protocol::encode_request(&ServerRequest::Auth { token: token.to_string() }).as_bytes()) + .expect("写认证帧应成功"); + let mut reader = FrameReader::new(io::BufReader::new(stream)); + let frame = reader.read_frame().expect("读认证确认应成功").expect("应读到认证确认帧"); + let response = protocol::parse_response(&frame).expect("认证确认帧解析应成功"); + assert!(matches!(response, ServerResponse::AuthOk), "应得 AuthOk 帧：{response:?}"); + } + + /// 连接 server 并完成认证（发 Auth 帧、读 AuthOk）。 + fn connect_auth(config: &ServerConfig) -> Conn { + let mut stream = connect(config); + authenticate(&mut stream, &read_token(config)); + stream + } + /// 发送请求帧并读回一帧响应。 fn send_and_recv(stream: &mut Conn, request: &ServerRequest) -> ServerResponse { stream @@ -901,7 +1026,7 @@ mod tests { } /// --rm 配置：唯一临时路径、worker 数 1、300 毫秒空闲超时（不依赖真实 - /// 超时值，断言带 5 秒裕量）。 + /// 超时值，断言带 5 秒裕量）、已知 token（测试注入）。 fn rm_config(test_name: &str) -> RmServerConfig { let dir = std::env::temp_dir().join(format!("oxide_server_rm_test_{}_{}", std::process::id(), test_name)); fs::create_dir_all(&dir).expect("测试目录创建应成功"); @@ -909,6 +1034,7 @@ mod tests { endpoint: Endpoint::Unix(dir.join("rm.sock")), worker_count: 1, idle_timeout: Duration::from_millis(300), + token: "test-rm-token".to_string(), } } @@ -933,12 +1059,13 @@ mod tests { } /// --rm TCP 配置：固定空闲端口、worker 数 1、300 毫秒空闲超时（与 Unix - /// 配置同口径，不依赖真实超时值，断言带 5 秒裕量）。 + /// 配置同口径，不依赖真实超时值，断言带 5 秒裕量）、已知 token（测试注入）。 fn rm_tcp_config() -> RmServerConfig { RmServerConfig { endpoint: Endpoint::Tcp(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), free_tcp_port())), worker_count: 1, idle_timeout: Duration::from_millis(300), + token: "test-rm-token".to_string(), } } @@ -987,7 +1114,7 @@ mod tests { fn version_response() { let config = unique_paths("version"); let handle = start_server(config.clone()); - let mut stream = connect(&config); + let mut stream = connect_auth(&config); let response = send_and_recv(&mut stream, &ServerRequest::Version); assert!( matches!(response, ServerResponse::Version { ref version } if version == env!("CARGO_PKG_VERSION")), @@ -1001,7 +1128,7 @@ mod tests { fn status_response() { let config = unique_paths("status"); let handle = start_server(config.clone()); - let mut stream = connect(&config); + let mut stream = connect_auth(&config); // worker 池异步预热：轮询至聚合总数不小于 1（预热完成）。 let deadline = Instant::now() + Duration::from_secs(5); @@ -1038,7 +1165,7 @@ mod tests { fn health_response() { let config = unique_paths("health"); let handle = start_server(config.clone()); - let mut stream = connect(&config); + let mut stream = connect_auth(&config); let response = send_and_recv(&mut stream, &ServerRequest::Health); assert!( matches!(response, ServerResponse::Health { healthy: true }), @@ -1052,7 +1179,7 @@ mod tests { fn info_response() { let config = unique_paths("info"); let handle = start_server(config.clone()); - let mut stream = connect(&config); + let mut stream = connect_auth(&config); let response = send_and_recv(&mut stream, &ServerRequest::Info); match response { ServerResponse::Info { @@ -1077,7 +1204,7 @@ mod tests { fn shutdown_response_and_cleanup() { let config = unique_paths("shutdown"); let handle = start_server(config.clone()); - let mut stream = connect(&config); + let mut stream = connect_auth(&config); stop_server(&mut stream, handle); assert!(!config.sidecar_path.exists(), "sidecar 文件应被删除"); assert!(!unix_path(&config.endpoint).exists(), "socket 文件应被删除"); @@ -1088,7 +1215,7 @@ mod tests { fn malformed_frame_error_response() { let config = unique_paths("malformed"); let handle = start_server(config.clone()); - let mut stream = connect(&config); + let mut stream = connect_auth(&config); // 发非 JSON 帧：应得 Error 帧。 stream.write_all(b"not json at all\n").expect("写畸形帧应成功"); @@ -1111,7 +1238,7 @@ mod tests { fn eval_request_returns_value() { let config = unique_paths("eval"); let handle = start_server(config.clone()); - let mut stream = connect(&config); + let mut stream = connect_auth(&config); let response = send_and_recv( &mut stream, &ServerRequest::Eval { @@ -1133,7 +1260,7 @@ mod tests { fn multi_frame_same_connection_and_second_instance_refused() { let config = unique_paths("multi"); let handle = start_server(config.clone()); - let mut stream = connect(&config); + let mut stream = connect_auth(&config); // 连续两帧逐帧应答，顺序保持。 let first = send_and_recv(&mut stream, &ServerRequest::Status); @@ -1170,8 +1297,8 @@ mod tests { fn concurrent_evals_across_connections() { let config = unique_paths("concurrent_evals"); let handle = start_server(config.clone()); - let mut stream_a = connect(&config); - let mut stream_b = connect(&config); + let mut stream_a = connect_auth(&config); + let mut stream_b = connect_auth(&config); // 两条连接各发 3 个 eval，响应逐帧按请求顺序送达。 for code in ["1 + 1", "2 * 3", "10 - 4"] { @@ -1245,8 +1372,8 @@ mod tests { fn control_request_not_blocked_by_eval() { let config = unique_paths("control_not_blocked"); let handle = start_server(config.clone()); - let mut stream_a = connect(&config); - let mut stream_b = connect(&config); + let mut stream_a = connect_auth(&config); + let mut stream_b = connect_auth(&config); // 连接 A 发长 eval（步数上限使执行有界，占住 worker 数百毫秒）。 stream_a @@ -1287,8 +1414,8 @@ mod tests { fn shutdown_with_in_flight_eval() { let config = unique_paths("shutdown_inflight"); let handle = start_server(config.clone()); - let mut stream_a = connect(&config); - let mut stream_b = connect(&config); + let mut stream_a = connect_auth(&config); + let mut stream_b = connect_auth(&config); // 连接 A 发长 eval（在途）；连接 B 发关闭请求。 stream_a @@ -1354,7 +1481,7 @@ mod tests { assert_eq!(id.version, "0.0.2", "B 的 sidecar 版本应为 0.0.2"); // B 正常应答版本请求。 - let mut stream = connect(&config); + let mut stream = connect_auth(&config); let response = send_and_recv(&mut stream, &ServerRequest::Version); assert!(matches!(response, ServerResponse::Version { .. }), "B 应应答版本请求：{response:?}"); stop_server(&mut stream, handle_b); @@ -1372,7 +1499,7 @@ mod tests { // sidecar（PID 存活且启动时刻刻度匹配，版本不匹配，认领判 // VersionMismatch）。 let listener = UnixListener::bind(unix_path(&config.endpoint)).expect("绑定模拟旧 server 应成功"); - sidecar::write_exclusive(&sidecar::ServerIdentity::new(&config.endpoint, "0.0.0"), &config.sidecar_path) + sidecar::write_exclusive(&sidecar::ServerIdentity::new(&config.endpoint, "0.0.0", ""), &config.sidecar_path) .expect("写模拟 sidecar 应成功"); // 模拟连接处理：接受循环，逐连接读一帧后回 Error 帧。须保持监听 @@ -1402,7 +1529,7 @@ mod tests { let config = unique_paths("yield_hang"); let listener = UnixListener::bind(unix_path(&config.endpoint)).expect("绑定模拟旧 server 应成功"); - sidecar::write_exclusive(&sidecar::ServerIdentity::new(&config.endpoint, "0.0.0"), &config.sidecar_path) + sidecar::write_exclusive(&sidecar::ServerIdentity::new(&config.endpoint, "0.0.0", ""), &config.sidecar_path) .expect("写模拟 sidecar 应成功"); // 模拟连接处理：接受连接后挂住不回应（模拟挂起的旧 server）。 @@ -1445,7 +1572,7 @@ mod tests { version: env!("CARGO_PKG_VERSION").to_string(), bind_addr: IpAddr::V4(Ipv4Addr::LOCALHOST), }; - yield_and_takeover(&config).expect("TCP 臂交接应成功"); + yield_and_takeover(&config, "test-token").expect("TCP 臂交接应成功"); } /// --rm 加断开 EOF：eval 得完成值后客户端断开，server 在有界时间内 @@ -1463,6 +1590,7 @@ mod tests { let handle = start_rm_server(config.clone()); let mut stream = Conn::Unix(UnixStream::connect(unix_path(&config.endpoint)).expect("连接 rm server 应成功")); + authenticate(&mut stream, &config.token); let response = send_and_recv( &mut stream, &ServerRequest::Eval { @@ -1541,6 +1669,7 @@ mod tests { let config = rm_config("second_refused"); let handle = start_rm_server(config.clone()); let mut stream = Conn::Unix(UnixStream::connect(unix_path(&config.endpoint)).expect("首个连接应成功")); + authenticate(&mut stream, &config.token); let response = send_and_recv(&mut stream, &ServerRequest::Health); assert!( matches!(response, ServerResponse::Health { healthy: true }), @@ -1569,6 +1698,7 @@ mod tests { let config = rm_config("shutdown_req"); let handle = start_rm_server(config.clone()); let mut stream = Conn::Unix(UnixStream::connect(unix_path(&config.endpoint)).expect("连接 rm server 应成功")); + authenticate(&mut stream, &config.token); stop_server(&mut stream, handle); assert!(!unix_path(&config.endpoint).exists(), "socket 文件应被删除"); } @@ -1584,6 +1714,7 @@ mod tests { _ => panic!("应为 TCP 端点"), }; let mut stream = Conn::Tcp(connect_rm_tcp(addr)); + authenticate(&mut stream, &config.token); let response = send_and_recv( &mut stream, &ServerRequest::Eval { @@ -1606,4 +1737,63 @@ mod tests { let result = handle.join().expect("rm server 线程应正常退出"); assert!(result.is_ok(), "rm server 应正常退出：{result:?}"); } + + /// 认证门：token 匹配时首帧 Auth 得 AuthOk，后续请求正常应答。 + #[test] + fn auth_gate_token_match_passes() { + let config = unique_paths("auth_match"); + let handle = start_server(config.clone()); + let mut stream = connect(&config); + authenticate(&mut stream, &read_token(&config)); + let response = send_and_recv(&mut stream, &ServerRequest::Health); + assert!( + matches!(response, ServerResponse::Health { healthy: true }), + "认证后应正常应答：{response:?}" + ); + stop_server(&mut stream, handle); + } + + /// 认证门：token 不匹配时直接断连（不回帧），读首帧得 EOF。 + #[test] + fn auth_gate_token_mismatch_disconnects() { + let config = unique_paths("auth_mismatch"); + let handle = start_server(config.clone()); + let mut stream = connect(&config); + // 发错误 token 的 Auth 帧。 + stream + .write_all( + protocol::encode_request(&ServerRequest::Auth { + token: "wrong-token".to_string(), + }) + .as_bytes(), + ) + .expect("写认证帧应成功"); + // server 应直接断连：读首帧得 EOF。 + let mut reader = FrameReader::new(io::BufReader::new(&mut stream)); + let frame = reader.read_frame().expect("读帧应成功（EOF 不判错）"); + assert!(frame.is_none(), "token 不匹配应直接断连（EOF）"); + drop(stream); + // 开第二条连接认证后发关闭请求停 server。 + let mut stop_stream = connect_auth(&config); + stop_server(&mut stop_stream, handle); + } + + /// 认证门：首帧非 Auth 时直接断连（不回帧），读首帧得 EOF。 + #[test] + fn auth_gate_non_auth_first_frame_disconnects() { + let config = unique_paths("auth_non_auth"); + let handle = start_server(config.clone()); + let mut stream = connect(&config); + // 首帧发 Health（非 Auth）：server 应直接断连。 + stream + .write_all(protocol::encode_request(&ServerRequest::Health).as_bytes()) + .expect("写请求帧应成功"); + let mut reader = FrameReader::new(io::BufReader::new(&mut stream)); + let frame = reader.read_frame().expect("读帧应成功（EOF 不判错）"); + assert!(frame.is_none(), "首帧非 Auth 应直接断连（EOF）"); + drop(stream); + // 开第二条连接认证后发关闭请求停 server。 + let mut stop_stream = connect_auth(&config); + stop_server(&mut stop_stream, handle); + } } diff --git a/crates/oxide_cli/src/server/sidecar.rs b/crates/oxide_cli/src/server/sidecar.rs index e606f0b5..a8ea8f4c 100644 --- a/crates/oxide_cli/src/server/sidecar.rs +++ b/crates/oxide_cli/src/server/sidecar.rs @@ -12,7 +12,7 @@ //! 同目录；有意不随退出删除（跨重启的诊断工件），不在清理命令范围内。 use std::fs::{self, File, OpenOptions}; -use std::io::{self, Write}; +use std::io::{self, Read, Write}; use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpStream}; use std::os::unix::io::AsRawFd; use std::os::unix::net::UnixStream; @@ -38,6 +38,8 @@ use serde::{Deserialize, Serialize}; /// - `listen_addr`：监听地址（网络身份——server 可达的网络接口，安全边界 /// 关注点：loopback-only 与 all-interfaces 的区分）；旧 sidecar 无该字段 /// 时读回为空串（serde 缺省）。 +/// - `token`：共享 token（十六进制），控制客户端认证帧用；旧 sidecar 无该 +/// 字段时读回为空串（serde 缺省）。 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct ServerIdentity { pub pid: u32, @@ -54,6 +56,10 @@ pub struct ServerIdentity { /// 缺省），探活裁决仍走 socket_path 文本，行为不变。 #[serde(default)] pub listen_addr: String, + /// 共享 token（十六进制）：控制客户端连接后首帧认证用；旧 sidecar + /// 无该字段时读回为空串（serde 缺省）。 + #[serde(default)] + pub token: String, } impl ServerIdentity { @@ -67,9 +73,10 @@ impl ServerIdentity { /// - `port` 取端点的 TCP 端口（Unix 臂无端口，记 0）。 /// - `listen_addr` 取端点的文本形态（`endpoint.describe()`），与 /// `socket_path` 同源同时机（认领时、bind 之前），记配置监听地址。 + /// - `token` 由调用方注入（server 启动时生成），本函数不生成。 /// - 非 Linux 宿主（`/proc` 缺位）时 `starttime` 填 0，交叉核对退化， /// liveness 由 endpoint 探活裁决。 - pub fn new(endpoint: &Endpoint, version: &str) -> Self { + pub fn new(endpoint: &Endpoint, version: &str, token: &str) -> Self { let pid = std::process::id(); let started_at = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0); let port = match endpoint { @@ -85,10 +92,42 @@ impl ServerIdentity { starttime: process_starttime(pid).unwrap_or(0), port, listen_addr: socket_path, + token: token.to_string(), } } } +/// 生成共享 token：读 `/dev/urandom` 32 字节转十六进制（64 位小写）。 +/// +/// # 边界与前提 +/// - 非 Linux 宿主（`/dev/urandom` 缺位）或读取失败时回落时间加 PID 熵 +/// 混合（本宿主恒为 Linux，回落臂仅保编译）。 +pub fn generate_token() -> String { + let mut bytes = [0u8; 32]; + let ok = fs::File::open("/dev/urandom") + .and_then(|mut file| file.read_exact(&mut bytes)) + .is_ok(); + if !ok { + bytes = fallback_entropy(); + } + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +/// 非 Linux 熵回落：纳秒时间加 PID 经 xorshift 混合出 32 字节。 +fn fallback_entropy() -> [u8; 32] { + let ns = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos() as u64; + let pid = std::process::id() as u64; + let mut seed = ns ^ (pid << 32); + let mut out = [0u8; 32]; + for byte in out.iter_mut() { + seed ^= seed << 13; + seed ^= seed >> 7; + seed ^= seed << 17; + *byte = (seed & 0xff) as u8; + } + out +} + /// 读取当前进程 uid（`/proc/self/status` 的 `Uid:` 行首值）。 /// /// # 边界与前提 @@ -208,12 +247,22 @@ impl WellKnownLock { /// # 边界与前提 /// - 文件已存在时返回 `AlreadyExists`，调用方据此转入读回裁决。 /// +/// # 副作用 +/// - Unix 臂写入后显式置 0600（sidecar 含共享 token，仅属主可读写）； +/// 非 Unix 平台该位语义为空操作。 +/// /// # 注意事项 /// - 写入不同步磁盘：崩溃残留的半写文件按损坏处理，由 liveness 检查兜底。 pub fn write_exclusive(identity: &ServerIdentity, path: &Path) -> io::Result<()> { let json = serde_json::to_string(identity).expect("身份序列化不会失败"); let mut file = OpenOptions::new().create_new(true).write(true).open(path)?; file.write_all(json.as_bytes())?; + // sidecar 含共享 token：显式 0600，仅属主可读写。 + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + fs::set_permissions(path, fs::Permissions::from_mode(0o600))?; + } Ok(()) } @@ -303,9 +352,9 @@ pub enum ClaimResult { /// # 注意事项 /// - 意外 io 错误（权限等）保守判 `RefusedAmbiguous`；方向安全（拒绝注册 /// 不会双注册）。 -pub fn claim_sidecar(sidecar_path: &Path, endpoint: &Endpoint, new_version: &str) -> ClaimResult { +pub fn claim_sidecar(sidecar_path: &Path, endpoint: &Endpoint, new_version: &str, token: &str) -> ClaimResult { // 先尝试 O_EXCL 创建：sidecar 不存在即成功。 - match write_exclusive(&ServerIdentity::new(endpoint, new_version), sidecar_path) { + match write_exclusive(&ServerIdentity::new(endpoint, new_version, token), sidecar_path) { Ok(()) => return ClaimResult::Registered, Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {} Err(_) => return ClaimResult::RefusedAmbiguous, @@ -330,22 +379,24 @@ pub fn claim_sidecar(sidecar_path: &Path, endpoint: &Endpoint, new_version: &str // endpoint 死：按 sidecar 可读性分支。 match recorded { - Some(id) => claim_dead_socket(&id, sidecar_path, endpoint, new_version), + Some(id) => claim_dead_socket(&id, sidecar_path, endpoint, new_version, token), None => { // 损坏文件：删 sidecar 与 socket 文件（Unix 臂）后重建（endpoint // 探活已失败，文件无监听者，不会误删存活 server 的 socket）。 remove_stale(sidecar_path, endpoint); - re_register(sidecar_path, endpoint, new_version) + re_register(sidecar_path, endpoint, new_version, token) } } } /// endpoint 死且 sidecar 可解析：按 PID 与启动时刻刻度裁决陈旧性。 -fn claim_dead_socket(id: &ServerIdentity, sidecar_path: &Path, endpoint: &Endpoint, new_version: &str) -> ClaimResult { +fn claim_dead_socket( + id: &ServerIdentity, sidecar_path: &Path, endpoint: &Endpoint, new_version: &str, token: &str, +) -> ClaimResult { // PID 死：陈旧残留，删 sidecar 与 socket 文件（Unix 臂）后重建。 if !pid_alive(id.pid) { remove_stale(sidecar_path, endpoint); - return re_register(sidecar_path, endpoint, new_version); + return re_register(sidecar_path, endpoint, new_version, token); } // PID 活：交叉核对启动时刻刻度裁决 PID 复用。 @@ -356,7 +407,7 @@ fn claim_dead_socket(id: &ServerIdentity, sidecar_path: &Path, endpoint: &Endpoi // 刻度失配：PID 已被复用，sidecar 是陈旧残留。 remove_stale(sidecar_path, endpoint); - re_register(sidecar_path, endpoint, new_version) + re_register(sidecar_path, endpoint, new_version, token) } /// 删除 sidecar 与 socket 文件（Unix 臂；endpoint 探活已失败，文件无监听者， @@ -369,8 +420,8 @@ fn remove_stale(sidecar_path: &Path, endpoint: &Endpoint) { } /// 清理后重建 sidecar，重建失败保守拒绝。 -fn re_register(sidecar_path: &Path, endpoint: &Endpoint, new_version: &str) -> ClaimResult { - match write_exclusive(&ServerIdentity::new(endpoint, new_version), sidecar_path) { +fn re_register(sidecar_path: &Path, endpoint: &Endpoint, new_version: &str, token: &str) -> ClaimResult { + match write_exclusive(&ServerIdentity::new(endpoint, new_version, token), sidecar_path) { Ok(()) => ClaimResult::Registered, Err(_) => ClaimResult::RefusedAmbiguous, } @@ -437,6 +488,7 @@ mod tests { starttime: 0, port: 0, listen_addr: String::new(), + token: String::new(), } } @@ -445,18 +497,22 @@ mod tests { fn identity_roundtrip() { let dir = TestDir::new(); let path = dir.path("sidecar.json"); - let id = ServerIdentity::new(&Endpoint::Unix(PathBuf::from("/tmp/oxide-test.sock")), "0.0.0"); + let id = ServerIdentity::new(&Endpoint::Unix(PathBuf::from("/tmp/oxide-test.sock")), "0.0.0", "test-token"); write_exclusive(&id, &path).expect("写 sidecar 应成功"); let read = read_identity(&path).expect("读回 sidecar 应成功"); - assert_eq!(read, id, "六字段应逐字段一致"); + assert_eq!(read, id, "字段应逐字段一致"); } /// 端口字段：TCP 端点记端点端口（机器可读形态），Unix 臂记 0。 #[test] fn identity_port_per_arm() { - let tcp = ServerIdentity::new(&Endpoint::Tcp(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4242)), "0.0.0"); + let tcp = ServerIdentity::new( + &Endpoint::Tcp(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4242)), + "0.0.0", + "test-token", + ); assert_eq!(tcp.port, 4242, "TCP 臂应记端点端口"); - let unix = ServerIdentity::new(&Endpoint::Unix(PathBuf::from("/tmp/oxide-test.sock")), "0.0.0"); + let unix = ServerIdentity::new(&Endpoint::Unix(PathBuf::from("/tmp/oxide-test.sock")), "0.0.0", "test-token"); assert_eq!(unix.port, 0, "Unix 臂应记 0"); } @@ -466,7 +522,7 @@ mod tests { let dir = TestDir::new(); let path = dir.path("sidecar.json"); let endpoint = Endpoint::Tcp(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4242)); - let id = ServerIdentity::new(&endpoint, "0.0.0"); + let id = ServerIdentity::new(&endpoint, "0.0.0", "test-token"); write_exclusive(&id, &path).expect("写 sidecar 应成功"); let read = read_identity(&path).expect("读回 sidecar 应成功"); assert_eq!(read, id, "字段应逐字段一致"); @@ -487,6 +543,35 @@ mod tests { let id = read_identity(&path).expect("旧 sidecar 应可读回"); assert_eq!(id.port, 0, "无 port 字段应读回 0"); assert_eq!(id.listen_addr, "", "无 listen_addr 字段应读回空串"); + assert_eq!(id.token, "", "无 token 字段应读回空串"); + } + + /// token 生成：64 位小写十六进制，两次调用取值不同。 + #[test] + fn generate_token_is_64_hex_and_unique() { + let a = generate_token(); + let b = generate_token(); + assert_eq!(a.len(), 64, "token 应为 64 位十六进制"); + assert!( + a.chars().all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()), + "token 应全为小写十六进制" + ); + assert_ne!(a, b, "两次生成的 token 应不同"); + } + + /// sidecar 文件权限：写入后显式 0600（仅属主可读写）。 + #[test] + fn write_exclusive_sets_0600() { + let dir = TestDir::new(); + let path = dir.path("sidecar.json"); + let id = ServerIdentity::new(&Endpoint::Unix(PathBuf::from("/tmp/oxide-test.sock")), "0.0.0", "test-token"); + write_exclusive(&id, &path).expect("写 sidecar 应成功"); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mode = fs::metadata(&path).expect("读元数据应成功").permissions().mode() & 0o777; + assert_eq!(mode, 0o600, "sidecar 文件权限应为 0600"); + } } /// O_EXCL 排他：同路径二次 create_new 必得 AlreadyExists。 @@ -494,7 +579,7 @@ mod tests { fn write_exclusive_second_create_fails() { let dir = TestDir::new(); let path = dir.path("sidecar.json"); - let id = ServerIdentity::new(&Endpoint::Unix(PathBuf::from("/tmp/oxide-test.sock")), "0.0.0"); + let id = ServerIdentity::new(&Endpoint::Unix(PathBuf::from("/tmp/oxide-test.sock")), "0.0.0", "test-token"); write_exclusive(&id, &path).expect("首次写入应成功"); let err = write_exclusive(&id, &path).unwrap_err(); assert_eq!(err.kind(), io::ErrorKind::AlreadyExists, "二次 create_new 必得 AlreadyExists"); @@ -519,7 +604,10 @@ mod tests { let dir = TestDir::new(); let sidecar = dir.path("sidecar.json"); let socket = dir.path("server.sock"); - assert_eq!(claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0"), ClaimResult::Registered); + assert_eq!( + claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0", "test-token"), + ClaimResult::Registered + ); let id = read_identity(&sidecar).expect("sidecar 应已写入"); assert_eq!(id.pid, std::process::id()); assert_eq!(id.socket_path, socket.to_str().unwrap()); @@ -534,10 +622,10 @@ mod tests { let socket = dir.path("server.sock"); // 存活 server 必有 sidecar（启动顺序契约：先写 sidecar 再 bind endpoint）。 let listener = UnixListener::bind(&socket).expect("绑定监听者应成功"); - write_exclusive(&ServerIdentity::new(&Endpoint::Unix(socket.clone()), "0.0.0"), &sidecar) + write_exclusive(&ServerIdentity::new(&Endpoint::Unix(socket.clone()), "0.0.0", "test-token"), &sidecar) .expect("写 sidecar 应成功"); assert_eq!( - claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0"), + claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0", "test-token"), ClaimResult::AlreadyRunning ); drop(listener); @@ -552,10 +640,10 @@ mod tests { let socket = dir.path("server.sock"); let listener = UnixListener::bind(&socket).expect("绑定监听者应成功"); // 旧 sidecar 版本与本进程版本不同：比对判不匹配。 - write_exclusive(&ServerIdentity::new(&Endpoint::Unix(socket.clone()), "0.0.0"), &sidecar) + write_exclusive(&ServerIdentity::new(&Endpoint::Unix(socket.clone()), "0.0.0", "test-token"), &sidecar) .expect("写 sidecar 应成功"); assert_eq!( - claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.1"), + claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.1", "test-token"), ClaimResult::VersionMismatch ); drop(listener); @@ -568,7 +656,10 @@ mod tests { let sidecar = dir.path("sidecar.json"); let socket = dir.path("server.sock"); write_exclusive(&stale_identity(dead_pid(), &socket), &sidecar).expect("写陈旧 sidecar 应成功"); - assert_eq!(claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0"), ClaimResult::Registered); + assert_eq!( + claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0", "test-token"), + ClaimResult::Registered + ); let id = read_identity(&sidecar).expect("重建 sidecar 应可读"); assert_eq!(id.pid, std::process::id(), "重建 sidecar 应为本进程身份"); } @@ -581,7 +672,10 @@ mod tests { let socket = dir.path("server.sock"); // PID 是本进程（存活），启动时刻刻度为占位值：PID 已复用。 write_exclusive(&stale_identity(std::process::id(), &socket), &sidecar).expect("写陈旧 sidecar 应成功"); - assert_eq!(claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0"), ClaimResult::Registered); + assert_eq!( + claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0", "test-token"), + ClaimResult::Registered + ); } /// 认领分支：存活 PID 加匹配启动时刻保守拒绝。 @@ -591,10 +685,10 @@ mod tests { let sidecar = dir.path("sidecar.json"); let socket = dir.path("server.sock"); // PID 与启动时刻刻度都是本进程的：进程活且刻度匹配。 - let id = ServerIdentity::new(&Endpoint::Unix(socket.clone()), "0.0.0"); + let id = ServerIdentity::new(&Endpoint::Unix(socket.clone()), "0.0.0", "test-token"); write_exclusive(&id, &sidecar).expect("写 sidecar 应成功"); assert_eq!( - claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0"), + claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0", "test-token"), ClaimResult::RefusedAmbiguous ); } @@ -607,7 +701,10 @@ mod tests { let socket = dir.path("server.sock"); // 损坏文件：半写 JSON。 fs::write(&sidecar, r#"{"pid":1,"socket_path":""#).expect("写损坏文件应成功"); - assert_eq!(claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0"), ClaimResult::Registered); + assert_eq!( + claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0", "test-token"), + ClaimResult::Registered + ); let id = read_identity(&sidecar).expect("重建 sidecar 应可读"); assert_eq!(id.pid, std::process::id()); } @@ -622,7 +719,10 @@ mod tests { fs::write(&sidecar, "not json at all").expect("写损坏文件应成功"); // 陈旧 socket 文件：文件存在但无监听者。 fs::write(&socket, b"stale").expect("写陈旧 socket 文件应成功"); - assert_eq!(claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0"), ClaimResult::Registered); + assert_eq!( + claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0", "test-token"), + ClaimResult::Registered + ); let id = read_identity(&sidecar).expect("重建 sidecar 应可读"); assert_eq!(id.pid, std::process::id(), "重建 sidecar 应为本进程身份"); assert!(!socket.exists(), "陈旧 socket 文件应被删除"); @@ -638,7 +738,7 @@ mod tests { let listener = UnixListener::bind(&socket).expect("绑定监听者应成功"); fs::write(&sidecar, "not json at all").expect("写损坏文件应成功"); assert_eq!( - claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0"), + claim_sidecar(&sidecar, &Endpoint::Unix(socket.clone()), "0.0.0", "test-token"), ClaimResult::VersionMismatch ); drop(listener); @@ -650,7 +750,7 @@ mod tests { let dir = TestDir::new(); let sidecar = dir.path("sidecar.json"); let endpoint = Endpoint::Tcp(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)); - assert_eq!(claim_sidecar(&sidecar, &endpoint, "0.0.0"), ClaimResult::Registered); + assert_eq!(claim_sidecar(&sidecar, &endpoint, "0.0.0", "test-token"), ClaimResult::Registered); let id = read_identity(&sidecar).expect("sidecar 应已写入"); assert_eq!(id.pid, std::process::id()); assert_eq!(id.socket_path, endpoint.describe(), "sidecar 应记 host:port 文本形态"); diff --git a/crates/oxide_cli/src/server/watchdog.rs b/crates/oxide_cli/src/server/watchdog.rs index 2cbfd8c6..d9cafb60 100644 --- a/crates/oxide_cli/src/server/watchdog.rs +++ b/crates/oxide_cli/src/server/watchdog.rs @@ -310,6 +310,7 @@ mod tests { starttime: 0, port: 0, listen_addr: String::new(), + token: String::new(), } } diff --git a/crates/oxide_cli/tests/server_cli.rs b/crates/oxide_cli/tests/server_cli.rs index 73d54ccb..3fe27ed5 100644 --- a/crates/oxide_cli/tests/server_cli.rs +++ b/crates/oxide_cli/tests/server_cli.rs @@ -95,13 +95,24 @@ fn send_and_recv(stream: &mut Conn, request: &ServerRequest) -> ServerResponse { protocol::parse_response(&frame).expect("响应帧解析应成功") } -/// 读 well-known sidecar 取 endpoint 并连接 server（两臂按形态分支）。 +/// 读 well-known sidecar 取 endpoint 与 token 并连接 server（两臂按形态分支）， +/// 完成认证（发 Auth 帧、读 AuthOk）。 fn connect_well_known() -> Conn { let id = sidecar::read_identity(&sidecar::well_known_sidecar_path()).expect("读 sidecar 应成功"); - match sidecar::parse_endpoint(&id.socket_path) { + let stream = match sidecar::parse_endpoint(&id.socket_path) { Endpoint::Unix(path) => Conn::Unix(UnixStream::connect(&path).expect("连接 server 应成功")), Endpoint::Tcp(addr) => Conn::Tcp(TcpStream::connect(addr).expect("连接 server 应成功")), - } + }; + let mut stream = stream; + // 发认证帧并读回 AuthOk。 + stream + .write_all(protocol::encode_request(&ServerRequest::Auth { token: id.token }).as_bytes()) + .expect("写认证帧应成功"); + let mut reader = FrameReader::new(BufReader::new(&mut stream)); + let frame = reader.read_frame().expect("读认证确认应成功").expect("应读到认证确认帧"); + let response = protocol::parse_response(&frame).expect("认证确认帧解析应成功"); + assert!(matches!(response, ServerResponse::AuthOk), "应得 AuthOk 帧：{response:?}"); + stream } /// 探测空闲端口：临时 `TcpListener` 绑 0 端口取操作系统分配的端口后释放。 @@ -116,9 +127,25 @@ fn free_port() -> u16 { } /// 按端口直连 server：TCP 回环连接包成 `Conn`。 +/// 按端口直连 server：TCP 回环连接包成 `Conn`，完成认证（发 Auth 帧、读 AuthOk）。 +/// token 从 well-known sidecar 读回（非 --rm 形态）。 fn connect_tcp(port: u16) -> Conn { + let id = sidecar::read_identity(&sidecar::well_known_sidecar_path()).expect("读 sidecar 应成功"); + connect_tcp_with_token(port, &id.token) +} + +/// 按端口直连 server 并以给定 token 认证（--rm 形态无 sidecar，token 经 stdout 传递）。 +fn connect_tcp_with_token(port: u16, token: &str) -> Conn { let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port); - Conn::Tcp(TcpStream::connect(addr).expect("连接 server 应成功")) + let mut stream = Conn::Tcp(TcpStream::connect(addr).expect("连接 server 应成功")); + stream + .write_all(protocol::encode_request(&ServerRequest::Auth { token: token.to_string() }).as_bytes()) + .expect("写认证帧应成功"); + let mut reader = FrameReader::new(BufReader::new(&mut stream)); + let frame = reader.read_frame().expect("读认证确认应成功").expect("应读到认证确认帧"); + let response = protocol::parse_response(&frame).expect("认证确认帧解析应成功"); + assert!(matches!(response, ServerResponse::AuthOk), "应得 AuthOk 帧：{response:?}"); + stream } /// help 列全部十一个子命令。 @@ -882,10 +909,11 @@ fn server_rm_dynamic_port_prints() { .spawn() .expect("oxide server start --rm 应可启动"); - // 后台线程消费 stdout：见到端点打印行即把端口发主线程，收集全部行至 EOF。 - // 端口解析取稳健形态：只认 `127.0.0.1:` 前缀与紧随数字，不断言完整句子。 + // 后台线程消费 stdout：见到端点打印行即把端口发主线程，见到 token 打印行 + // 即把 token 发主线程，收集全部行至 EOF。 let stdout_pipe = child.stdout.take().expect("stdout 管道应可取"); let (port_tx, port_rx) = std::sync::mpsc::channel::(); + let (token_tx, token_rx) = std::sync::mpsc::channel::(); let out_handle = std::thread::spawn(move || { let mut reader = BufReader::new(stdout_pipe); let mut lines = Vec::new(); @@ -903,6 +931,12 @@ fn server_rm_dynamic_port_prints() { let _ = port_tx.send(port); } } + if let Some(idx) = line.find("rm server token：") { + let token = line[idx + "rm server token：".len()..].trim().to_string(); + if !token.is_empty() { + let _ = token_tx.send(token); + } + } } lines }); @@ -912,9 +946,14 @@ fn server_rm_dynamic_port_prints() { .recv_timeout(Duration::from_secs(30)) .expect("--rm 进程应在有界时间内打印实际端口"); assert_ne!(port, 0, "操作系统应分配非 0 端口"); + // token 行与端口行同批打印：有界时间内应得非空 token。 + let token = token_rx + .recv_timeout(Duration::from_secs(30)) + .expect("--rm 进程应在有界时间内打印 token"); + assert!(!token.is_empty(), "token 应非空"); // 连接该端口发一枚 eval 得完成值。 - let mut stream = connect_tcp(port); + let mut stream = connect_tcp_with_token(port, &token); let response = send_and_recv( &mut stream, &ServerRequest::Eval {