mirror of
https://github.com/pjh456/OxideJS.git
synced 2026-10-23 18:21:13 +08:00
feat: server 连接加共享 token 认证门
This commit is contained in:
Generated
+20
-20
@@ -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",
|
||||
|
||||
+1
-1
@@ -24,7 +24,7 @@ members = [
|
||||
]
|
||||
|
||||
[workspace.package]
|
||||
version = "0.15.11"
|
||||
version = "0.15.12"
|
||||
edition = "2021"
|
||||
license = "MIT"
|
||||
rust-version = "1.80"
|
||||
|
||||
@@ -307,6 +307,7 @@ mod tests {
|
||||
starttime,
|
||||
port: 0,
|
||||
listen_addr: String::new(),
|
||||
token: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -52,16 +52,18 @@ pub fn send_control_request(request: &ServerRequest) -> Result<ServerResponse, C
|
||||
send_control_request_from(&sidecar::well_known_sidecar_path(), request)
|
||||
}
|
||||
|
||||
/// 注入 sidecar 路径的入口(测试与复用):读 sidecar 取 socket 路径,
|
||||
/// 连接、发请求、读响应,三分支错误映射。
|
||||
/// 注入 sidecar 路径的入口(测试与复用):读 sidecar 取 socket 路径与 token,
|
||||
/// 连接、发认证帧、读 AuthOk、发请求、读响应,三分支错误映射。
|
||||
///
|
||||
/// # 步骤
|
||||
/// 1. 读 sidecar;缺失或损坏(解析失败)同归连接失败分支。
|
||||
/// 2. 连接 socket(5 秒超时安全网)。
|
||||
/// 3. 设读超时与写超时各 5 秒。
|
||||
/// 4. 写请求帧。
|
||||
/// 5. 读响应帧;io 错误或 server 未应答即关闭(EOF)判连接失败。
|
||||
/// 6. 解析响应帧;畸形判协议错误,Error 帧判 server 错误,其余返回。
|
||||
/// 4. 发认证帧(token 取 sidecar),读回 `AuthOk`;非 `AuthOk` 判 server
|
||||
/// 错误(server 版本过旧需重启)。
|
||||
/// 5. 写请求帧。
|
||||
/// 6. 读响应帧;io 错误或 server 未应答即关闭(EOF)判连接失败。
|
||||
/// 7. 解析响应帧;畸形判协议错误,Error 帧判 server 错误,其余返回。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - sidecar 路径由调用方注入,生产入口用 well-known 路径。
|
||||
@@ -80,19 +82,42 @@ pub fn send_control_request_from(sidecar_path: &Path, request: &ServerRequest) -
|
||||
|
||||
// 连接:5 秒超时是安全网,本地连接立即完成或立即失败。endpoint 取 sidecar
|
||||
// 记录的文本形态(host:port 或路径),撤掉 Unix 过渡包装。
|
||||
let mut stream = connect_with_timeout(&sidecar::parse_endpoint(&identity.socket_path), CONNECT_TIMEOUT)?;
|
||||
let stream = connect_with_timeout(&sidecar::parse_endpoint(&identity.socket_path), CONNECT_TIMEOUT)?;
|
||||
|
||||
// 读写超时各 5 秒:server 挂起(不回应)时不无限阻塞。
|
||||
let _ = stream.set_read_timeout(Some(CONNECT_TIMEOUT));
|
||||
let _ = stream.set_write_timeout(Some(CONNECT_TIMEOUT));
|
||||
|
||||
// 克隆写端:原流交给帧读取器,写端独立用于帧写回。
|
||||
let mut writer = stream
|
||||
.try_clone()
|
||||
.map_err(|e| ClientError::Connect(format!("克隆流失败:{e}")))?;
|
||||
|
||||
// 发认证帧:token 取 sidecar,server 侧逐字节异或判等。
|
||||
writer
|
||||
.write_all(protocol::encode_request(&ServerRequest::Auth { token: identity.token }).as_bytes())
|
||||
.map_err(|e| ClientError::Connect(format!("写认证帧失败:{e}")))?;
|
||||
|
||||
let mut reader = FrameReader::new(BufReader::new(stream));
|
||||
|
||||
// 读认证确认帧:只有 AuthOk 表示 server 已受理认证。
|
||||
let frame = reader
|
||||
.read_frame()
|
||||
.map_err(|e| ClientError::Connect(format!("读认证确认帧失败:{e}")))?
|
||||
.ok_or_else(|| ClientError::Connect("server 未应答即关闭连接".into()))?;
|
||||
let auth_response = protocol::parse_response(&frame).map_err(ClientError::Protocol)?;
|
||||
if !matches!(auth_response, ServerResponse::AuthOk) {
|
||||
return Err(ClientError::Server(format!(
|
||||
"server 未受理认证(server 版本可能过旧,需重启):{auth_response:?}"
|
||||
)));
|
||||
}
|
||||
|
||||
// 写请求帧(含结尾换行)。
|
||||
stream
|
||||
writer
|
||||
.write_all(protocol::encode_request(request).as_bytes())
|
||||
.map_err(|e| ClientError::Connect(format!("写请求帧失败:{e}")))?;
|
||||
|
||||
// 读响应帧:io 错误判连接失败,EOF(server 未应答即关闭)判连接失败。
|
||||
let mut reader = FrameReader::new(BufReader::new(stream));
|
||||
let frame = reader
|
||||
.read_frame()
|
||||
.map_err(|e| ClientError::Connect(format!("读响应帧失败:{e}")))?
|
||||
@@ -180,9 +205,9 @@ mod tests {
|
||||
});
|
||||
}
|
||||
|
||||
/// 写一个指向给定 socket 路径的合法 sidecar。
|
||||
/// 写一个指向给定 socket 路径的合法 sidecar(含 token)。
|
||||
fn write_sidecar(sidecar: &Path, socket: &Path) {
|
||||
let identity = sidecar::ServerIdentity::new(&Endpoint::Unix(socket.to_path_buf()), "0.0.0");
|
||||
let identity = sidecar::ServerIdentity::new(&Endpoint::Unix(socket.to_path_buf()), "0.0.0", "test-token");
|
||||
sidecar::write_exclusive(&identity, sidecar).expect("写 sidecar 应成功");
|
||||
}
|
||||
|
||||
@@ -232,7 +257,7 @@ mod tests {
|
||||
assert!(matches!(err, ClientError::Protocol(_)), "应判协议错误:{err:?}");
|
||||
}
|
||||
|
||||
/// 假 server 回 Error 帧:判 server 错误分支且 message 原样带出。
|
||||
/// 假 server 回 Error 帧(认证确认阶段):判 server 错误分支。
|
||||
#[test]
|
||||
fn error_frame_is_server_branch() {
|
||||
let dir = TestDir::new();
|
||||
@@ -248,7 +273,7 @@ mod tests {
|
||||
let _ = stream.write_all(frame.as_bytes());
|
||||
});
|
||||
let err = send_control_request_from(&sidecar, &ServerRequest::Version).unwrap_err();
|
||||
assert_eq!(err, ClientError::Server("server 内部错误".into()), "message 应原样带出:{err:?}");
|
||||
assert!(matches!(err, ClientError::Server(_)), "应判 server 错误分支:{err:?}");
|
||||
}
|
||||
|
||||
/// 假 server 接受后不回应:5 秒读超时后判连接失败分支。
|
||||
@@ -270,4 +295,26 @@ mod tests {
|
||||
assert!(matches!(err, ClientError::Connect(_)), "应判连接失败:{err:?}");
|
||||
assert!(elapsed >= 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:?}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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<ServerContext>, router: &Arc<Worke
|
||||
}
|
||||
}
|
||||
|
||||
/// 每连接处理:帧读取 → 分派 → 响应写回,循环至对端关闭。
|
||||
/// 共享 token 判等:逐字节异或累加,避免逐字节短路带来的时序侧信道。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - 长度不等直接判不等(token 长度固定 64 位,生产路径长度恒等)。
|
||||
fn tokens_match(expected: &str, given: &str) -> 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 文本形态");
|
||||
|
||||
@@ -310,6 +310,7 @@ mod tests {
|
||||
starttime: 0,
|
||||
port: 0,
|
||||
listen_addr: String::new(),
|
||||
token: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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::<u16>();
|
||||
let (token_tx, token_rx) = std::sync::mpsc::channel::<String>();
|
||||
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 {
|
||||
|
||||
Reference in New Issue
Block a user