use axum::{ extract::{Path, Query, State}, http::{header::AUTHORIZATION, HeaderMap, StatusCode}, response::{ sse::{Event, KeepAlive, Sse}, IntoResponse, Response, }, routing::{get, post}, Json, Router, }; use serde::Deserialize; use serde_json::{json, Value}; use std::{ collections::{HashMap, HashSet}, convert::Infallible, net::SocketAddr, sync::{Arc, OnceLock, RwLock}, }; use tokio::sync::{mpsc, Mutex}; use tokio_stream::{wrappers::ReceiverStream, StreamExt as _}; use crate::db; struct SessionEntry { sender: mpsc::Sender<(String, String)>, project_scope: Option, } /// SSE 会话表：session_id → 发送端与连接作用域 type Sessions = Arc>>; const MCP_LOOPBACK_HOST: &str = "127.0.0.1"; const DEV_VARIABLE_READ_TOOL: &str = "read_project_dev_variable"; const TEST_ACCOUNT_READ_TOOL: &str = "read_project_test_account"; const MCP_SERVER_INSTRUCTIONS: &str = "先读取 get_diagnostics、verify_onboarding 或项目状态，再执行后续操作。项目级工具必须使用当前项目的 canonical project_id，不得跨项目猜测或改写身份。测试需要登录时先查询 list_project_dev_variables，已有需求就复用；未登记时可登记不含秘密的开发需求元数据，账号与密码只由用户在项目 UI 填写。其他写操作仅在用户明确确认目标与影响后调用；状态 unavailable/unknown 不得当作成功。"; static INITIALIZED_PROJECT_CLIENTS: OnceLock>> = OnceLock::new(); fn initialized_project_clients() -> &'static RwLock> { INITIALIZED_PROJECT_CLIENTS.get_or_init(|| RwLock::new(HashSet::new())) } fn record_project_client_initialize(req: &RpcRequest, project_scope: Option<&str>) { let Some(project_id) = project_scope else { return; }; let client_name = req .params .as_ref() .and_then(|params| params.get("clientInfo")) .and_then(|info| info.get("name")) .and_then(Value::as_str) .unwrap_or_default(); if client_name == "lian-jing-readiness" { return; } if let Ok(mut clients) = initialized_project_clients().write() { clients.insert(project_id.to_string()); } } pub(crate) fn has_initialized_project_client(project_id: &str) -> bool { initialized_project_clients() .read() .map(|clients| clients.contains(project_id)) .unwrap_or(false) } /// MCP 仅监听 IPv4 loopback，避免把本机 agent 控制面暴露到局域网。 fn mcp_bind_addr(port: u16) -> SocketAddr { SocketAddr::from(([127, 0, 0, 1], port)) } /// 所有 MCP 对外 URL 的单一基址事实。 fn mcp_base_url(port: u16) -> String { format!("http://{MCP_LOOPBACK_HOST}:{port}") } pub(crate) fn mcp_health_url(port: u16) -> String { format!("{}/health", mcp_base_url(port)) } fn sse_messages_url(port: u16, session_id: &str) -> String { format!( "{}/mcp/messages?session_id={session_id}", mcp_base_url(port) ) } pub(crate) fn scoped_mcp_url(port: u16, project_id: &str) -> String { format!( "{}/mcp/projects/{}", mcp_base_url(port), urlencoding::encode(project_id) ) } fn scoped_sse_messages_url(port: u16, project_id: &str, session_id: &str) -> String { format!( "{}/messages?session_id={session_id}", scoped_mcp_url(port, project_id) ) } /// 从 settings 表读取 MCP 端口，默认 27190 pub(crate) fn read_port_from_conn(conn: &rusqlite::Connection) -> u16 { conn.query_row( "SELECT value FROM settings WHERE key = 'mcp_port'", [], |row| row.get::<_, String>(0), ) .ok() .and_then(|value| value.parse().ok()) .unwrap_or(27190) } pub fn read_port() -> u16 { db::pool() .get() .ok() .map(|conn| read_port_from_conn(&conn)) .unwrap_or(27190) } /// 启动 MCP HTTP 服务（同时支持 Streamable HTTP 和 SSE 两种传输） pub async fn start(port: u16) { let sessions: Sessions = Arc::new(Mutex::new(HashMap::new())); let app = mcp_router(sessions); let addr = mcp_bind_addr(port); match tokio::net::TcpListener::bind(addr).await { Ok(listener) => { db::log_event( "mcp_server", "info", &format!("MCP Server 已启动，端口 {port}（HTTP + SSE）"), None, ); if let Err(e) = axum::serve(listener, app).await { db::log_event( "mcp_server", "error", &format!("Server 异常退出: {e}"), None, ); } } Err(e) => db::log_event( "mcp_server", "error", &format!("绑定端口 {port} 失败: {e}"), None, ), } } fn mcp_router(sessions: Sessions) -> Router { Router::new() .route("/health", get(health)) // 单一全局服务器路由（lian-jing）：寻址不依赖 group_id，工具显式传 project_id .route("/mcp", post(handle_mcp).get(handle_sse_connect)) .route( "/mcp/projects/:project_id", post(handle_scoped_mcp).get(handle_scoped_sse_connect), ) // SSE 消息端点：客户端通过此 URL 发送 JSON-RPC，响应经 SSE 推回 .route("/mcp/messages", post(handle_sse_message)) .route( "/mcp/projects/:project_id/messages", post(handle_scoped_sse_message), ) // Gitea webhook：接收 push/pull_request 事件写入飞轮 .route( "/webhook/gitea/:project_id", post(crate::mcp_webhook::handle_gitea_webhook), ) .with_state(sessions) } async fn health() -> &'static str { "炼境 MCP Server running" } // ── JSON-RPC 类型 ───────────────────────────────────────────────────────────── #[derive(Deserialize)] struct RpcRequest { #[allow(dead_code)] jsonrpc: String, method: String, id: Option, #[serde(default)] params: Option, } #[derive(Deserialize)] struct SessionQuery { session_id: String, } /// 仅敏感单项读取要求项目级认证；旧的非敏感 MCP 工具保持原有契约。 fn authorize_sensitive_request( req: &RpcRequest, project_scope: Option<&str>, headers: &HeaderMap, store: &impl crate::credential_store::CredentialStore, ) -> Result<(), &'static str> { let requested_tool = req .params .as_ref() .and_then(|params| params.get("name")) .and_then(Value::as_str); if req.method != "tools/call" || !matches!( requested_tool, Some(DEV_VARIABLE_READ_TOOL | TEST_ACCOUNT_READ_TOOL) ) { return Ok(()); } let project_id = project_scope.ok_or("敏感工具仅允许项目级 MCP 连接")?; let requested_project = req .params .as_ref() .and_then(|params| params.get("arguments")) .and_then(|args| args.get("project_id")) .and_then(Value::as_str); if requested_project.is_some_and(|requested| requested != project_id) { return Err("项目作用域不匹配"); } let mut auth_values = headers.get_all(AUTHORIZATION).iter(); let header = auth_values .next() .and_then(|header| header.to_str().ok()) .ok_or("缺少 MCP 客户端认证")?; if auth_values.next().is_some() { return Err("MCP 客户端认证格式无效"); } let provided = header .strip_prefix("Bearer ") .ok_or("MCP 客户端认证格式无效")?; let digest = crate::credential_store::mcp_auth_digest(project_id).map_err(|_| "MCP 项目认证不可用")?; let expected = crate::credential_store::read_mcp_auth_token_for_digest(store, &digest) .map_err(|_| "MCP 项目认证不可用")?; let equal = provided.len() == expected.len() && provided .bytes() .zip(expected.bytes()) .fold(0u8, |difference, (left, right)| difference | (left ^ right)) == 0; if !equal { return Err("MCP 客户端认证失败"); } Ok(()) } // ── 共享 RPC 处理逻辑 ───────────────────────────────────────────────────────── async fn process_rpc(req: &RpcRequest, project_scope: Option<&str>) -> Result { if req.method == "initialize" { record_project_client_initialize(req, project_scope); } match req.method.as_str() { "initialize" => Ok(json!({ "protocolVersion": "2024-11-05", "capabilities": { "tools": {} }, "serverInfo": { "name": "炼境", "version": env!("CARGO_PKG_VERSION") }, "instructions": MCP_SERVER_INSTRUCTIONS, "lianJing": match project_scope { Some(project_id) => json!({ "scope_mode": "project", "project_id": project_id }), None => json!({ "scope_mode": "legacy", "upgrade_hint": "重跑 inject_mcp，将项目配置升级到 /mcp/projects/:project_id" }) } })), "tools/list" => Ok(match project_scope { Some(project_id) => crate::mcp_tools::tools_list_result_for_project(project_id), None => crate::mcp_tools::tools_list_result(), }), "tools/call" => Ok(match project_scope { Some(project_id) => { crate::mcp_tools::tools_call_for_project(req.params.as_ref(), project_id).await } None => crate::mcp_tools::tools_call(req.params.as_ref()).await, }), _ => Err(format!("Method not found: {}", req.method)), } } // ── Streamable HTTP（POST）───────────────────────────────────────────────────── async fn handle_mcp(headers: HeaderMap, Json(req): Json) -> Response { if authorize_sensitive_request( &req, None, &headers, &crate::credential_store::WindowsCredentialStore, ) .is_err() { return StatusCode::FORBIDDEN.into_response(); } if req.id.is_none() { return StatusCode::ACCEPTED.into_response(); } let id = req.id.clone(); match process_rpc(&req, None).await { Ok(result) => Json(json!({ "jsonrpc": "2.0", "id": id, "result": result })).into_response(), Err(msg) => Json(json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32601, "message": msg } })) .into_response(), } } /// 项目专用路由只服务已登记项目。未知 project_id 一律拒绝：否则会给 agent 一套看似可用、 /// 实则每次调用都越权失败的工具集（2026-09-30 实测 /mcp/projects/<任意字符串> 返回完整 32 个工具）。 /// 数据库不可用时同样拒绝（fail closed）。 fn scoped_project_registered(project_id: &str) -> bool { crate::db::pool() .get() .ok() .and_then(|conn| { conn.query_row( "SELECT EXISTS(SELECT 1 FROM project_workspaces WHERE id = ?1)", [project_id], |row| row.get::<_, bool>(0), ) .ok() }) .unwrap_or(false) } fn unknown_project_response(project_id: &str, id: Option) -> Response { ( StatusCode::NOT_FOUND, Json(json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32001, "message": format!("项目未登记: {project_id}（检查 .lian-jing/context.json 或重跑 inject_mcp）") } })), ) .into_response() } async fn handle_scoped_mcp( Path(project_id): Path, headers: HeaderMap, Json(req): Json, ) -> Response { if !scoped_project_registered(&project_id) { return unknown_project_response(&project_id, req.id.clone()); } if authorize_sensitive_request( &req, Some(&project_id), &headers, &crate::credential_store::WindowsCredentialStore, ) .is_err() { return StatusCode::FORBIDDEN.into_response(); } if req.id.is_none() { return StatusCode::ACCEPTED.into_response(); } let id = req.id.clone(); match process_rpc(&req, Some(&project_id)).await { Ok(result) => Json(json!({ "jsonrpc": "2.0", "id": id, "result": result })).into_response(), Err(msg) => Json(json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32601, "message": msg } })) .into_response(), } } // ── SSE 传输（GET + POST /messages）───────────────────────────────────────── /// GET /mcp — Claude Code SSE 握手，首条事件告知后续 POST URL async fn handle_sse_connect( State(sessions): State, ) -> Sse>> { let session_id = uuid::Uuid::new_v4().to_string(); let (tx, rx) = mpsc::channel::<(String, String)>(32); let port = read_port(); let endpoint_uri = sse_messages_url(port, &session_id); let _ = tx.send(("endpoint".to_string(), endpoint_uri)).await; sessions.lock().await.insert( session_id, SessionEntry { sender: tx, project_scope: None, }, ); let stream = ReceiverStream::new(rx).map(|(evt_name, data)| { Ok::(Event::default().event(evt_name).data(data)) }); Sse::new(stream).keep_alive(KeepAlive::default()) } async fn handle_scoped_sse_connect( Path(project_id): Path, State(sessions): State, ) -> Response { if !scoped_project_registered(&project_id) { return unknown_project_response(&project_id, None); } let session_id = uuid::Uuid::new_v4().to_string(); let (tx, rx) = mpsc::channel::<(String, String)>(32); let endpoint_uri = scoped_sse_messages_url(read_port(), &project_id, &session_id); let _ = tx.send(("endpoint".to_string(), endpoint_uri)).await; sessions.lock().await.insert( session_id, SessionEntry { sender: tx, project_scope: Some(project_id), }, ); let stream = ReceiverStream::new(rx).map(|(evt_name, data)| { Ok::(Event::default().event(evt_name).data(data)) }); Sse::new(stream) .keep_alive(KeepAlive::default()) .into_response() } #[cfg(test)] mod tests { use super::*; use crate::credential_store::{get_or_create_mcp_auth_token, MemoryCredentialStore}; fn rpc(method: &str) -> RpcRequest { RpcRequest { jsonrpc: "2.0".to_string(), method: method.to_string(), id: Some(json!(1)), params: None, } } #[test] fn sensitive_read_requires_matching_project_bearer_on_both_transports() { let store = MemoryCredentialStore::new(); let p1 = get_or_create_mcp_auth_token(&store, "p1").unwrap(); let _p2 = get_or_create_mcp_auth_token(&store, "p2").unwrap(); let req = RpcRequest { jsonrpc: "2.0".into(), method: "tools/call".into(), id: Some(json!(1)), params: Some(json!({ "name": DEV_VARIABLE_READ_TOOL, "arguments": {"project_id": "p1", "name": "E2E_PASSWORD"} })), }; let mut headers = HeaderMap::new(); assert!(authorize_sensitive_request(&req, Some("p1"), &headers, &store).is_err()); headers.insert(AUTHORIZATION, "Bearer wrong".parse().unwrap()); assert!(authorize_sensitive_request(&req, Some("p1"), &headers, &store).is_err()); headers.insert(AUTHORIZATION, format!("Bearer {p1}").parse().unwrap()); assert!(authorize_sensitive_request(&req, Some("p1"), &headers, &store).is_ok()); assert!(authorize_sensitive_request(&req, Some("p2"), &headers, &store).is_err()); assert!(authorize_sensitive_request(&req, None, &headers, &store).is_err()); let test_account = RpcRequest { params: Some(json!({ "name": TEST_ACCOUNT_READ_TOOL, "arguments": {"project_id": "p1", "name": "E2E_LOGIN"} })), ..rpc("tools/call") }; assert!( authorize_sensitive_request(&test_account, Some("p1"), &HeaderMap::new(), &store) .is_err() ); assert!(authorize_sensitive_request(&test_account, Some("p1"), &headers, &store).is_ok()); assert!(authorize_sensitive_request(&test_account, Some("p2"), &headers, &store).is_err()); assert!(authorize_sensitive_request(&test_account, None, &headers, &store).is_err()); let mut implicit_project = req; implicit_project.params.as_mut().unwrap()["arguments"] = json!({"name": "E2E_PASSWORD"}); assert!( authorize_sensitive_request(&implicit_project, Some("p1"), &headers, &store).is_ok() ); assert!(authorize_sensitive_request( &rpc("initialize"), Some("p1"), &HeaderMap::new(), &store ) .is_ok()); let normal = RpcRequest { method: "tools/call".into(), params: Some(json!({"name": "get_diagnostics", "arguments": {}})), ..rpc("tools/call") }; assert!( authorize_sensitive_request(&normal, Some("p1"), &HeaderMap::new(), &store).is_ok() ); } /// 单一 lian-jing 服务器的握手路径：initialize / tools/list 必须可用 #[tokio::test] async fn process_rpc_handshake() { let init = process_rpc(&rpc("initialize"), None).await.unwrap(); assert_eq!(init["serverInfo"]["name"], "炼境"); let instructions = init["instructions"].as_str().unwrap(); assert!(instructions.len() <= 512); assert!(instructions.contains("先读取")); assert!(instructions.contains("项目")); assert!(instructions.contains("用户明确确认")); assert_eq!(init["lianJing"]["scope_mode"], "legacy"); assert!(init["lianJing"]["upgrade_hint"].is_string()); let tools = process_rpc(&rpc("tools/list"), None).await.unwrap(); let list = tools["tools"].as_array().unwrap(); assert!( list.len() >= 20, "工具清单不应因去组化而缺失，实际 {}", list.len() ); assert!( list.iter() .any(|tool| tool["name"] == "get_delivery_status"), "MCP 握手后的工具清单应暴露 get_delivery_status" ); } #[tokio::test] async fn scoped_handshake_binds_project_and_filters_global_tools() { let init = process_rpc(&rpc("initialize"), Some("p1")).await.unwrap(); assert_eq!(init["lianJing"]["scope_mode"], "project"); assert_eq!(init["lianJing"]["project_id"], "p1"); let tools = process_rpc(&rpc("tools/list"), Some("p1")).await.unwrap(); let list = tools["tools"].as_array().unwrap(); assert!(list .iter() .any(|tool| tool["name"] == "get_delivery_status")); assert!(!list.iter().any(|tool| tool["name"] == "unregister_project")); } #[tokio::test(flavor = "multi_thread")] async fn scoped_enterprise_fixture_reads_non_secret_resource_status_over_mcp() { db::init_test_pool(); let listener = tokio::net::TcpListener::bind(mcp_bind_addr(0)) .await .unwrap(); let port = listener.local_addr().unwrap().port(); { let conn = db::pool().get().unwrap(); conn.execute("INSERT OR IGNORE INTO project_profiles (id, name) VALUES ('enterprise-fixture-profile', 'Enterprise fixture')", []).unwrap(); conn.execute("INSERT OR IGNORE INTO project_workspaces (id, profile_id) VALUES ('enterprise-fixture', 'enterprise-fixture-profile')", []).unwrap(); conn.execute( "UPDATE settings SET value=?1 WHERE key='mcp_port'", [port.to_string()], ) .unwrap(); } let sessions: Sessions = Arc::new(Mutex::new(HashMap::new())); let server = tokio::spawn(async move { axum::serve(listener, mcp_router(sessions)).await.unwrap(); }); let url = format!("http://127.0.0.1:{port}/mcp/projects/enterprise-fixture"); let call = |method: &str, params: Value| -> Value { ureq::post(&url) .send_json(json!({"jsonrpc":"2.0","id":1,"method":method,"params":params})) .unwrap() .into_json() .unwrap() }; let init = call("initialize", json!({})); assert_eq!( init["result"]["lianJing"]["project_id"], "enterprise-fixture" ); let catalog = call("tools/list", json!({})); assert!(catalog["result"]["tools"] .as_array() .unwrap() .iter() .any(|tool| tool["name"] == "get_project_resource_status")); let result = call( "tools/call", json!({"name":"get_project_resource_status","arguments":{}}), ); assert_eq!( result["result"]["structuredContent"]["project_id"], "enterprise-fixture" ); assert_eq!( result["result"]["structuredContent"]["gitea_pr"]["state"], "not_configured" ); assert!(!result.to_string().contains("access_token")); let denied = call( "tools/call", json!({"name":"get_project_resource_status","arguments":{"project_id":"other-project"}}), ); assert_eq!( denied["result"]["structuredContent"]["code"], "scope_violation" ); use std::io::BufRead; let sse = ureq::get(&url).call().unwrap(); assert!(sse .header("Content-Type") .unwrap_or("") .contains("text/event-stream")); let mut line = String::new(); std::io::BufReader::new(sse.into_reader()) .read_line(&mut line) .unwrap(); assert!(line.contains("endpoint")); server.abort(); } #[cfg(windows)] #[tokio::test(flavor = "multi_thread")] async fn scoped_dev_variable_http_pilot_denies_missing_wrong_cross_and_global_auth() { struct Cleanup { ids: Vec<(String, String)>, } impl Drop for Cleanup { fn drop(&mut self) { let store = crate::credential_store::WindowsCredentialStore; for (project_id, profile_id) in &self.ids { if let Ok(conn) = db::pool().get() { let _ = crate::commands::dev_variables::revoke( &conn, &store, project_id, "E2E_PASSWORD", ); let _ = crate::commands::dev_variables::revoke( &conn, &store, project_id, "E2E_LOGIN", ); let _ = conn .execute("DELETE FROM project_workspaces WHERE id=?1", [project_id]); let _ = conn.execute("DELETE FROM project_profiles WHERE id=?1", [profile_id]); } let _ = crate::credential_store::delete_mcp_auth_token_for_project( &store, project_id, ); } } } db::init_test_pool(); let p1 = format!("qa-only-dev-mcp-{}", uuid::Uuid::new_v4()); let p2 = format!("qa-only-dev-mcp-{}", uuid::Uuid::new_v4()); let profile1 = format!("{p1}-profile"); let profile2 = format!("{p2}-profile"); let _cleanup = Cleanup { ids: vec![ (p1.clone(), profile1.clone()), (p2.clone(), profile2.clone()), ], }; { let conn = db::pool().get().unwrap(); for (project_id, profile_id) in [(&p1, &profile1), (&p2, &profile2)] { conn.execute( "INSERT INTO project_profiles (id,name) VALUES (?1,'QA')", [profile_id], ) .unwrap(); conn.execute( "INSERT INTO project_workspaces (id,profile_id) VALUES (?1,?2)", rusqlite::params![project_id, profile_id], ) .unwrap(); } } let store = crate::credential_store::WindowsCredentialStore; let token = crate::credential_store::get_or_create_mcp_auth_token(&store, &p1).unwrap(); let _other_token = crate::credential_store::get_or_create_mcp_auth_token(&store, &p2).unwrap(); let listener = tokio::net::TcpListener::bind(mcp_bind_addr(0)) .await .unwrap(); let port = listener.local_addr().unwrap().port(); let sessions: Sessions = Arc::new(Mutex::new(HashMap::new())); let server = tokio::spawn(async move { axum::serve(listener, mcp_router(sessions)).await.unwrap(); }); let p1_url = format!("http://127.0.0.1:{port}/mcp/projects/{p1}"); let p2_url = format!("http://127.0.0.1:{port}/mcp/projects/{p2}"); let global_url = format!("http://127.0.0.1:{port}/mcp"); let call_body = |name: &str, arguments: Value| { json!({ "jsonrpc":"2.0","id":1,"method":"tools/call", "params":{"name":name,"arguments":arguments} }) }; let scoped_catalog = ureq::post(&p1_url) .send_json(json!({"jsonrpc":"2.0","id":1,"method":"tools/list"})) .unwrap() .into_json::() .unwrap(); let tool_names: Vec<&str> = scoped_catalog["result"]["tools"] .as_array() .unwrap() .iter() .filter_map(|tool| tool["name"].as_str()) .collect(); assert!(tool_names.contains(&"read_project_dev_variable")); assert!(tool_names.contains(&"register_project_test_account")); assert!(tool_names.contains(&"read_project_test_account")); assert!(!tool_names .iter() .any(|name| name.contains("get_secret") || name.contains("export_env"))); let global_catalog = ureq::post(&global_url) .send_json(json!({"jsonrpc":"2.0","id":1,"method":"tools/list"})) .unwrap() .into_json::() .unwrap(); assert!(!global_catalog .to_string() .contains("read_project_dev_variable")); assert!(!global_catalog .to_string() .contains("read_project_test_account")); let register = ureq::post(&p1_url) .send_json(call_body( "register_project_dev_variable", json!({"name":"E2E_PASSWORD","purpose":"合成验收"}), )) .unwrap() .into_json::() .unwrap(); assert!(register["result"]["isError"].is_null()); let pending = ureq::post(&p1_url) .send_json(call_body("list_project_dev_variables", json!({}))) .unwrap() .into_json::() .unwrap(); assert!(pending["result"]["content"][0]["text"] .as_str() .unwrap() .contains("\"configured\":false")); { let conn = db::pool().get().unwrap(); crate::commands::dev_variables::save( &conn, &store, &p1, "E2E_PASSWORD", "qa-only-http-value", ) .unwrap(); } let read_call = call_body("read_project_dev_variable", json!({"name":"E2E_PASSWORD"})); assert!(matches!( ureq::post(&p1_url).send_json(read_call.clone()), Err(ureq::Error::Status(403, _)) )); assert!(matches!( ureq::post(&p1_url) .set("Authorization", "Bearer wrong") .send_json(read_call.clone()), Err(ureq::Error::Status(403, _)) )); assert!(matches!( ureq::post(&p2_url) .set("Authorization", &format!("Bearer {token}")) .send_json(read_call.clone()), Err(ureq::Error::Status(403, _)) )); let cross_argument = call_body( "read_project_dev_variable", json!({"project_id":p2,"name":"E2E_PASSWORD"}), ); assert!(matches!( ureq::post(&p1_url) .set("Authorization", &format!("Bearer {token}")) .send_json(cross_argument), Err(ureq::Error::Status(403, _)) )); assert!(matches!( ureq::post(&global_url) .set("Authorization", &format!("Bearer {token}")) .send_json(read_call.clone()), Err(ureq::Error::Status(403, _)) )); let read = ureq::post(&p1_url) .set("Authorization", &format!("Bearer {token}")) .send_json(read_call.clone()) .unwrap() .into_json::() .unwrap(); assert_eq!(read["result"]["content"][0]["text"], "qa-only-http-value"); { let conn = db::pool().get().unwrap(); crate::commands::dev_variables::revoke(&conn, &store, &p1, "E2E_PASSWORD").unwrap(); } let after_revoke = ureq::post(&p1_url) .set("Authorization", &format!("Bearer {token}")) .send_json(read_call) .unwrap() .into_json::() .unwrap(); assert_eq!(after_revoke["result"]["isError"], true); assert!(!after_revoke.to_string().contains("qa-only-http-value")); let account_register = ureq::post(&p1_url) .send_json(call_body( "register_project_test_account", json!({"name":"E2E_LOGIN","service":"enterprise-system","environment":"test","purpose":"合成登录验收"}), )) .unwrap() .into_json::() .unwrap(); assert!(account_register["result"]["isError"].is_null()); let account_pending = ureq::post(&p1_url) .send_json(call_body("list_project_dev_variables", json!({}))) .unwrap() .into_json::() .unwrap(); let pending_text = account_pending["result"]["content"][0]["text"] .as_str() .unwrap(); assert!(pending_text.contains("\"kind\":\"test_account\"")); assert!(pending_text.contains("\"configured\":false")); assert!(!pending_text.contains("qa-only-account-user")); { let conn = db::pool().get().unwrap(); crate::commands::dev_variables::save_test_account( &conn, &store, &p1, "E2E_LOGIN", "qa-only-account-user", "qa-only-account-password", ) .unwrap(); } let account_registered_again = ureq::post(&p1_url) .send_json(call_body( "register_project_test_account", json!({"name":"E2E_LOGIN","service":"enterprise-system","environment":"test","purpose":"合成登录验收"}), )) .unwrap() .into_json::() .unwrap(); assert!(account_registered_again["result"]["isError"].is_null()); let account_ready = ureq::post(&p1_url) .send_json(call_body("list_project_dev_variables", json!({}))) .unwrap() .into_json::() .unwrap(); let ready_text = account_ready["result"]["content"][0]["text"] .as_str() .unwrap(); assert_eq!(ready_text.matches("\"name\":\"E2E_LOGIN\"").count(), 1); assert!(ready_text.contains("\"configured\":true")); assert!(!ready_text.contains("qa-only-account-user")); assert!(!ready_text.contains("qa-only-account-password")); let account_read = call_body("read_project_test_account", json!({"name":"E2E_LOGIN"})); assert!(matches!( ureq::post(&p1_url).send_json(account_read.clone()), Err(ureq::Error::Status(403, _)) )); assert!(matches!( ureq::post(&p1_url) .set("Authorization", "Bearer wrong") .send_json(account_read.clone()), Err(ureq::Error::Status(403, _)) )); assert!(matches!( ureq::post(&p2_url) .set("Authorization", &format!("Bearer {token}")) .send_json(account_read.clone()), Err(ureq::Error::Status(403, _)) )); assert!(matches!( ureq::post(&global_url) .set("Authorization", &format!("Bearer {token}")) .send_json(account_read.clone()), Err(ureq::Error::Status(403, _)) )); let account = ureq::post(&p1_url) .set("Authorization", &format!("Bearer {token}")) .send_json(account_read.clone()) .unwrap() .into_json::() .unwrap(); let value: Value = serde_json::from_str(account["result"]["content"][0]["text"].as_str().unwrap()) .unwrap(); assert_eq!(value["username"], "qa-only-account-user"); assert_eq!(value["password"], "qa-only-account-password"); let scalar_read = ureq::post(&p1_url) .set("Authorization", &format!("Bearer {token}")) .send_json(call_body( "read_project_dev_variable", json!({"name":"E2E_LOGIN"}), )) .unwrap() .into_json::() .unwrap(); assert_eq!(scalar_read["result"]["isError"], true); { let conn = db::pool().get().unwrap(); crate::commands::dev_variables::revoke(&conn, &store, &p1, "E2E_LOGIN").unwrap(); } let account_after_revoke = ureq::post(&p1_url) .set("Authorization", &format!("Bearer {token}")) .send_json(account_read) .unwrap() .into_json::() .unwrap(); assert_eq!(account_after_revoke["result"]["isError"], true); assert!(!account_after_revoke .to_string() .contains("qa-only-account-password")); server.abort(); } #[cfg(windows)] #[tokio::test(flavor = "multi_thread")] async fn scoped_mcp_create_pull_request_uses_synthetic_credential() { use std::io::{BufRead, BufReader, Read, Write}; struct SyntheticInstanceGuard(Option); impl Drop for SyntheticInstanceGuard { fn drop(&mut self) { if let Some(instance_id) = self.0.take() { let _ = crate::commands::gitea::gitea_delete_instance(instance_id); } } } db::init_test_pool(); let project_id = format!("synthetic-broker-{}", uuid::Uuid::new_v4()); let profile_id = format!("{project_id}-profile"); let token = "qa-only-synthetic-mcp-token"; let remote = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let remote_address = remote.local_addr().unwrap(); let instance = crate::commands::gitea::gitea_save_instance( crate::commands::gitea::SaveGiteaInstanceInput { name: "synthetic MCP PR".into(), url: format!("http://{remote_address}"), access_token: token.into(), is_default: Some(false), }, ) .unwrap(); let mut instance_guard = SyntheticInstanceGuard(Some(instance.id.clone())); let reference = { let conn = db::pool().get().unwrap(); conn.query_row( "SELECT access_token_ref FROM gitea_instances WHERE id = ?1", [&instance.id], |row| row.get::<_, String>(0), ) .unwrap() }; { let conn = db::pool().get().unwrap(); conn.execute( "INSERT INTO project_profiles (id, name) VALUES (?1, 'Synthetic MCP PR')", [&profile_id], ) .unwrap(); conn.execute( "INSERT INTO project_workspaces (id, profile_id) VALUES (?1, ?2)", rusqlite::params![project_id, profile_id], ) .unwrap(); conn.execute( "INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name) VALUES (?1, ?2, ?3, 'owner', 'repo')", rusqlite::params![uuid::Uuid::new_v4().to_string(), instance.id, project_id], ) .unwrap(); } let listener = tokio::net::TcpListener::bind(mcp_bind_addr(0)) .await .unwrap(); let port = listener.local_addr().unwrap().port(); let sessions: Sessions = Arc::new(Mutex::new(HashMap::new())); let server = tokio::spawn(async move { axum::serve(listener, mcp_router(sessions)).await.unwrap(); }); let url = format!("http://127.0.0.1:{port}/mcp/projects/{project_id}"); let call = |method: &str, params: Value| -> Value { ureq::post(&url) .send_json(json!({"jsonrpc":"2.0","id":1,"method":method,"params":params})) .unwrap() .into_json() .unwrap() }; let init = call("initialize", json!({})); assert_eq!(init["result"]["lianJing"]["project_id"], project_id); let catalog = call("tools/list", json!({})); assert!(catalog["result"]["tools"] .as_array() .unwrap() .iter() .any(|tool| tool["name"] == "create_pull_request")); let readiness = call( "tools/call", json!({"name":"get_project_resource_status","arguments":{}}), ); assert_eq!( readiness["result"]["structuredContent"]["gitea_pr"]["state"], "configured" ); assert!(!readiness.to_string().contains(token)); let denied = call( "tools/call", json!({"name":"create_pull_request","arguments":{"project_id":"other-project","head":"feature","base":"main","title":"synthetic"}}), ); assert_eq!( denied["result"]["structuredContent"]["code"], "scope_violation" ); remote.set_nonblocking(true).unwrap(); let mock = std::thread::spawn(move || { let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let (mut stream, _) = loop { match remote.accept() { Ok(accepted) => break accepted, Err(error) if error.kind() == std::io::ErrorKind::WouldBlock && std::time::Instant::now() < deadline => { std::thread::sleep(std::time::Duration::from_millis(10)); } Err(error) => panic!("模拟 Gitea 未收到请求: {error}"), } }; stream .set_read_timeout(Some(std::time::Duration::from_secs(3))) .unwrap(); let mut reader = BufReader::new(stream.try_clone().unwrap()); let mut line = String::new(); reader.read_line(&mut line).unwrap(); assert!(line.starts_with("POST /api/v1/repos/owner/repo/pulls ")); let mut authorized = false; let mut content_length = 0; loop { line.clear(); reader.read_line(&mut line).unwrap(); if line == "\r\n" || line.is_empty() { break; } if line.eq_ignore_ascii_case("authorization: token qa-only-synthetic-mcp-token\r\n") { authorized = true; } if let Some(value) = line.to_ascii_lowercase().strip_prefix("content-length:") { content_length = value.trim().parse().unwrap(); } } assert!(authorized, "模拟 Gitea 未收到受保护存储中的合成 token"); let mut body = vec![0; content_length]; reader.read_exact(&mut body).unwrap(); let request: Value = serde_json::from_slice(&body).unwrap(); assert_eq!(request["head"], "feature"); let response = r#"{"number":42,"html_url":"https://example.invalid/pr/42","title":"synthetic"}"#; write!(stream, "HTTP/1.1 201 Created\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", response.len(), response).unwrap(); }); let created = call( "tools/call", json!({"name":"create_pull_request","arguments":{"head":"feature","base":"main","title":"synthetic"}}), ); mock.join().unwrap(); assert_ne!(created["result"]["isError"], true); let result = created["result"]["content"][0]["text"].as_str().unwrap(); assert_eq!( serde_json::from_str::(result).unwrap()["pr_number"], 42 ); assert!(!created.to_string().contains(token)); assert!(!created.to_string().contains("access_token")); server.abort(); crate::commands::gitea::gitea_delete_instance(instance.id).unwrap(); instance_guard.0 = None; assert!(crate::credential_store::CredentialStore::get( &crate::credential_store::WindowsCredentialStore, &reference, ) .unwrap() .is_none()); } /// 仅供隔离的真实客户端试点；普通 cargo test 不运行，也不启动 Tauri 应用。 #[cfg(windows)] #[tokio::test(flavor = "multi_thread")] #[ignore = "connect a disposable Codex/Claude project to the printed scoped MCP URL"] async fn isolated_scoped_mcp_client_pilot() { use std::io::{BufRead, BufReader, Read, Write}; struct PilotGuard { instance_id: String, project_dir: std::path::PathBuf, } impl Drop for PilotGuard { fn drop(&mut self) { let _ = crate::commands::gitea::gitea_delete_instance(self.instance_id.clone()); let _ = std::fs::remove_file(self.project_dir.join(".lian-jing/context.json")); let _ = std::fs::remove_file(self.project_dir.join(".codex/config.toml")); let _ = std::fs::remove_dir(self.project_dir.join(".lian-jing")); let _ = std::fs::remove_dir(self.project_dir.join(".codex")); let _ = std::fs::remove_dir(&self.project_dir); } } db::init_test_pool(); let project_id = format!("synthetic-client-pilot-{}", uuid::Uuid::new_v4()); let profile_id = format!("{project_id}-profile"); let token = format!("qa-only-pilot-{}", uuid::Uuid::new_v4()); let remote = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let remote_address = remote.local_addr().unwrap(); let instance = crate::commands::gitea::gitea_save_instance( crate::commands::gitea::SaveGiteaInstanceInput { name: "isolated client pilot".into(), url: format!("http://{remote_address}"), access_token: token.clone(), is_default: Some(false), }, ) .unwrap(); let project_dir = std::env::temp_dir().join(format!("lianjing-r11-{}", uuid::Uuid::new_v4())); let guard = PilotGuard { instance_id: instance.id.clone(), project_dir: project_dir.clone(), }; let reference: String = { let conn = db::pool().get().unwrap(); conn.query_row( "SELECT access_token_ref FROM gitea_instances WHERE id = ?1", [&instance.id], |row| row.get(0), ) .unwrap() }; { let conn = db::pool().get().unwrap(); conn.execute( "INSERT INTO project_profiles (id, name) VALUES (?1, 'Synthetic Client Pilot')", [&profile_id], ) .unwrap(); conn.execute( "INSERT INTO project_workspaces (id, profile_id) VALUES (?1, ?2)", rusqlite::params![project_id, profile_id], ) .unwrap(); conn.execute( "INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name) VALUES (?1, ?2, ?3, 'owner', 'repo')", rusqlite::params![uuid::Uuid::new_v4().to_string(), instance.id, project_id], ) .unwrap(); } let listener = tokio::net::TcpListener::bind(mcp_bind_addr(0)) .await .unwrap(); let port = listener.local_addr().unwrap().port(); let sessions: Sessions = Arc::new(Mutex::new(HashMap::new())); let server = tokio::spawn(async move { axum::serve(listener, mcp_router(sessions)).await.unwrap(); }); let url = scoped_mcp_url(port, &project_id); std::fs::create_dir(&project_dir).unwrap(); std::fs::create_dir(project_dir.join(".lian-jing")).unwrap(); std::fs::create_dir(project_dir.join(".codex")).unwrap(); std::fs::write( project_dir.join(".lian-jing/context.json"), serde_json::to_vec_pretty(&json!({"project_id": project_id, "schema_version": 1})) .unwrap(), ) .unwrap(); std::fs::write( project_dir.join(".codex/config.toml"), format!("[mcp_servers.lian-jing]\nurl = {url:?}\n"), ) .unwrap(); println!("R11_PILOT_PROJECT_DIR={}", project_dir.display()); println!("R11_PILOT_SCOPED_URL={url}"); println!("R11_PILOT_PROJECT_ID={project_id}"); std::io::stdout().flush().unwrap(); remote.set_nonblocking(true).unwrap(); let mock = std::thread::spawn(move || { let deadline = std::time::Instant::now() + std::time::Duration::from_secs(300); let (mut stream, _) = loop { match remote.accept() { Ok(accepted) => break accepted, Err(error) if error.kind() == std::io::ErrorKind::WouldBlock && std::time::Instant::now() < deadline => { std::thread::sleep(std::time::Duration::from_millis(20)); } Err(error) => panic!("5 分钟内未收到客户端的模拟 PR 请求: {error}"), } }; stream .set_read_timeout(Some(std::time::Duration::from_secs(5))) .unwrap(); let mut reader = BufReader::new(stream.try_clone().unwrap()); let mut line = String::new(); reader.read_line(&mut line).unwrap(); assert!(line.starts_with("POST /api/v1/repos/owner/repo/pulls ")); let mut authorized = false; let mut content_length = 0; loop { line.clear(); reader.read_line(&mut line).unwrap(); if line == "\r\n" || line.is_empty() { break; } if line.eq_ignore_ascii_case(&format!("authorization: token {token}\r\n")) { authorized = true; } if let Some(value) = line.to_ascii_lowercase().strip_prefix("content-length:") { content_length = value.trim().parse().unwrap(); } } assert!(authorized, "模拟 Gitea 未收到受保护存储中的合成凭证"); let mut body = vec![0; content_length]; reader.read_exact(&mut body).unwrap(); let request: Value = serde_json::from_slice(&body).unwrap(); assert_eq!(request["head"], "synthetic-head"); let response = r#"{"number":43,"html_url":"https://example.invalid/pr/43","title":"synthetic pilot"}"#; write!(stream, "HTTP/1.1 201 Created\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", response.len(), response).unwrap(); }); mock.join().unwrap(); // 让 MCP handler 把模拟 Gitea 的响应完整交付给外部客户端后再停服。 tokio::time::sleep(std::time::Duration::from_secs(2)).await; server.abort(); drop(guard); assert!(crate::credential_store::CredentialStore::get( &crate::credential_store::WindowsCredentialStore, &reference, ) .unwrap() .is_none()); assert!(!project_dir.exists()); } /// V7 隔离真实客户端试点：仅内存数据库、随机合成凭证与临时 scoped MCP，认证走真实 helper。 /// 阶段靠临时项目 `.pilot/` 标记推进：客户端登记后宿主代人填写合成账户（写 filled）； /// 见 `revoke` 标记后代人撤销（写 revoked）；见 `finish` 标记后检查改动边界并清理。 #[cfg(windows)] #[tokio::test(flavor = "multi_thread")] #[ignore = "V7: connect disposable Codex sessions to the printed scoped MCP URL"] async fn isolated_dev_variables_client_pilot() { use crate::credential_store::WindowsCredentialStore; use std::io::Write; use std::time::{Duration, Instant}; struct PilotGuard { project_id: String, project_dir: std::path::PathBuf, } impl Drop for PilotGuard { fn drop(&mut self) { if let Ok(conn) = db::pool().get() { let _ = crate::commands::dev_variables::purge_project( &conn, &WindowsCredentialStore, &self.project_id, ); } let _ = crate::credential_store::delete_mcp_auth_token_for_project( &WindowsCredentialStore, &self.project_id, ); let _ = std::fs::remove_dir_all(&self.project_dir); } } async fn wait_until( what: &str, limit: Duration, mut probe: impl FnMut() -> Option, ) -> T { let deadline = Instant::now() + limit; loop { if let Some(value) = probe() { return value; } assert!(Instant::now() < deadline, "等待超时：{what}"); tokio::time::sleep(Duration::from_millis(500)).await; } } db::init_test_pool(); let project_id = format!("synthetic-dev-variables-pilot-{}", uuid::Uuid::new_v4()); let profile_id = format!("{project_id}-profile"); let project_dir = std::env::temp_dir().join(format!("lianjing-v7-{}", uuid::Uuid::new_v4())); let pilot_dir = project_dir.join(".pilot"); let guard = PilotGuard { project_id: project_id.clone(), project_dir: project_dir.clone(), }; { let conn = db::pool().get().unwrap(); conn.execute( "INSERT INTO project_profiles (id, name) VALUES (?1, 'Synthetic Dev Variables Pilot')", [&profile_id], ) .unwrap(); conn.execute( "INSERT INTO project_workspaces (id, profile_id) VALUES (?1, ?2)", rusqlite::params![project_id, profile_id], ) .unwrap(); } let token = crate::credential_store::get_or_create_mcp_auth_token( &WindowsCredentialStore, &project_id, ) .unwrap(); let digest = crate::credential_store::mcp_auth_digest(&project_id).unwrap(); // 测试二进制位于 target/debug/deps/，helper 即同目录上一级的应用 debug 构建。 let helper_exe = std::env::current_exe() .unwrap() .parent() .and_then(std::path::Path::parent) .unwrap() .join("dev-manager-tauri.exe"); assert!( helper_exe.is_file(), "先 cargo build 生成 helper：{}", helper_exe.display() ); let helper = format!( "\"{}\" --lianjing-mcp-header {digest}", helper_exe.display() ); let listener = tokio::net::TcpListener::bind(mcp_bind_addr(0)) .await .unwrap(); let port = listener.local_addr().unwrap().port(); let sessions: Sessions = Arc::new(Mutex::new(HashMap::new())); let server = tokio::spawn(async move { axum::serve(listener, mcp_router(sessions)).await.unwrap(); }); let url = scoped_mcp_url(port, &project_id); std::fs::create_dir_all(project_dir.join(".lian-jing")).unwrap(); std::fs::create_dir_all(project_dir.join(".codex")).unwrap(); std::fs::create_dir_all(&pilot_dir).unwrap(); std::fs::write( project_dir.join(".lian-jing/context.json"), serde_json::to_vec_pretty(&json!({"project_id": project_id, "schema_version": 1})) .unwrap(), ) .unwrap(); std::fs::write( project_dir.join(".codex/config.toml"), format!( "[mcp_servers.lian-jing]\nurl = {}\nhttp_headers_helper = {}\n", serde_json::to_string(&url).unwrap(), serde_json::to_string(&helper).unwrap() ), ) .unwrap(); let username = format!("pilot-user-{}", &uuid::Uuid::new_v4().simple().to_string()[..8]); let password = format!("pilot-pass-{}", uuid::Uuid::new_v4().simple()); println!("V7_PILOT_PROJECT_DIR={}", project_dir.display()); println!("V7_PILOT_SCOPED_URL={url}"); println!("V7_PILOT_PROJECT_ID={project_id}"); println!("V7_PILOT_HELPER={helper}"); println!("V7_PILOT_EXPECTED_USERNAME={username}"); std::io::stdout().flush().unwrap(); // 阶段 1：等客户端登记测试账户需求，再代人填写合成值。 let name = wait_until("客户端登记测试账户需求", Duration::from_secs(1800), || { let conn = db::pool().get().ok()?; conn.query_row( "SELECT name FROM project_dev_variables WHERE project_id=?1 AND kind='test_account' ORDER BY name LIMIT 1", [&project_id], |row| row.get::<_, String>(0), ) .ok() }) .await; { let conn = db::pool().get().unwrap(); crate::commands::dev_variables::save_test_account( &conn, &WindowsCredentialStore, &project_id, &name, &username, &password, ) .unwrap(); } std::fs::write(pilot_dir.join("filled"), &name).unwrap(); println!("V7_PILOT_FILLED={name}"); std::io::stdout().flush().unwrap(); // 阶段 2：客户端跨会话读取后，由操作者放置 revoke 标记，宿主代人撤销。 wait_until("撤销标记 .pilot/revoke", Duration::from_secs(1800), || { pilot_dir.join("revoke").exists().then_some(()) }) .await; { let conn = db::pool().get().unwrap(); crate::commands::dev_variables::revoke(&conn, &WindowsCredentialStore, &project_id, &name) .unwrap(); } std::fs::write(pilot_dir.join("revoked"), &name).unwrap(); println!("V7_PILOT_REVOKED={name}"); std::io::stdout().flush().unwrap(); // 阶段 3：撤销后再读由客户端验证失败；finish 标记后检查改动边界。 wait_until("结束标记 .pilot/finish", Duration::from_secs(1800), || { pilot_dir.join("finish").exists().then_some(()) }) .await; server.abort(); { let conn = db::pool().get().unwrap(); let statuses = crate::commands::dev_variables::list( &conn, &WindowsCredentialStore, &project_id, ) .unwrap(); assert!(statuses.iter().all(|status| !status.configured)); } let mut pending = vec![project_dir.clone()]; while let Some(dir) = pending.pop() { for entry in std::fs::read_dir(&dir).unwrap() { let path = entry.unwrap().path(); if path.is_dir() { pending.push(path); continue; } let file_name = path.file_name().unwrap().to_string_lossy(); assert!(!file_name.starts_with(".env"), "项目目录出现 {file_name}"); let content = std::fs::read_to_string(&path).unwrap_or_default(); assert!(!content.contains(&token), "{} 含 MCP token", path.display()); assert!(!content.contains(&password), "{} 含测试密码", path.display()); } } drop(guard); assert!(crate::credential_store::read_mcp_auth_token_for_digest( &WindowsCredentialStore, &digest ) .is_err()); assert!(!project_dir.exists()); } #[tokio::test] async fn readiness_probe_does_not_satisfy_client_reload_evidence() { let project_id = "reload-evidence-fixture"; let mut readiness = rpc("initialize"); readiness.params = Some(json!({ "clientInfo": { "name": "lian-jing-readiness", "version": "test" } })); process_rpc(&readiness, Some(project_id)).await.unwrap(); assert!(!has_initialized_project_client(project_id)); let mut codex = rpc("initialize"); codex.params = Some(json!({ "clientInfo": { "name": "codex", "version": "test" } })); process_rpc(&codex, Some(project_id)).await.unwrap(); assert!(has_initialized_project_client(project_id)); } #[tokio::test] async fn scoped_route_rejects_unregistered_project() { db::init_test_pool(); let known = format!("scoped-known-{}", uuid::Uuid::new_v4()); { let conn = db::pool().get().unwrap(); conn.execute( "INSERT INTO project_profiles (id, name) VALUES (?1, 'Scoped Known')", [format!("{known}-profile")], ) .unwrap(); conn.execute( "INSERT INTO project_workspaces (id, profile_id) VALUES (?1, ?2)", rusqlite::params![known, format!("{known}-profile")], ) .unwrap(); } let unknown = handle_scoped_mcp( Path("__not_registered__".to_string()), HeaderMap::new(), Json(rpc("tools/list")), ) .await; assert_eq!(unknown.status(), StatusCode::NOT_FOUND); let registered = handle_scoped_mcp(Path(known.clone()), HeaderMap::new(), Json(rpc("tools/list"))).await; assert_eq!(registered.status(), StatusCode::OK); let sessions: Sessions = Arc::new(Mutex::new(HashMap::new())); let sse = handle_scoped_sse_connect(Path("__not_registered__".to_string()), State(sessions)).await; assert_eq!(sse.status(), StatusCode::NOT_FOUND); } #[test] fn loopback_urls_share_one_ipv4_base() { let port = 27190; let session_id = "test-session"; assert_eq!(mcp_bind_addr(port).to_string(), "127.0.0.1:27190"); assert_eq!(mcp_health_url(port), "http://127.0.0.1:27190/health"); assert_eq!( scoped_mcp_url(port, "project one"), "http://127.0.0.1:27190/mcp/projects/project%20one" ); assert_eq!( sse_messages_url(port, session_id), "http://127.0.0.1:27190/mcp/messages?session_id=test-session" ); } } /// POST /mcp/messages?session_id=xxx /// 立即返回 202，后台异步处理工具调用并经 SSE channel 推回响应 async fn handle_sse_message( State(sessions): State, Query(query): Query, headers: HeaderMap, Json(req): Json, ) -> Response { if authorize_sensitive_request( &req, None, &headers, &crate::credential_store::WindowsCredentialStore, ) .is_err() { return StatusCode::FORBIDDEN.into_response(); } if req.id.is_none() { return StatusCode::ACCEPTED.into_response(); } let session_id = query.session_id.clone(); tokio::spawn(async move { let id = req.id.clone(); let response_json = match process_rpc(&req, None).await { Ok(result) => json!({ "jsonrpc": "2.0", "id": id, "result": result }), Err(msg) => json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32601, "message": msg } }), }; let mut guard = sessions.lock().await; if let Some(entry) = guard.get(&session_id) { if entry.project_scope.is_some() { return; } if entry .sender .send(("message".to_string(), response_json.to_string())) .await .is_err() { guard.remove(&session_id); } } }); StatusCode::ACCEPTED.into_response() } async fn handle_scoped_sse_message( Path(project_id): Path, State(sessions): State, Query(query): Query, headers: HeaderMap, Json(req): Json, ) -> Response { // 握手后项目可能被删除：每条消息复核登记。只移除确实属于该项目的会话， // 不能凭请求里任意给出的 session_id 删掉别的会话。 if !scoped_project_registered(&project_id) { let mut guard = sessions.lock().await; let owned = guard .get(&query.session_id) .and_then(|entry| entry.project_scope.as_deref()) == Some(project_id.as_str()); if owned { guard.remove(&query.session_id); } return unknown_project_response(&project_id, req.id.clone()); } if authorize_sensitive_request( &req, Some(&project_id), &headers, &crate::credential_store::WindowsCredentialStore, ) .is_err() { return StatusCode::FORBIDDEN.into_response(); } if req.id.is_none() { return StatusCode::ACCEPTED.into_response(); } let session_id = query.session_id.clone(); let scope_matches = sessions .lock() .await .get(&session_id) .and_then(|entry| entry.project_scope.as_deref()) == Some(project_id.as_str()); if !scope_matches { return StatusCode::FORBIDDEN.into_response(); } tokio::spawn(async move { let id = req.id.clone(); let response_json = match process_rpc(&req, Some(&project_id)).await { Ok(result) => json!({ "jsonrpc": "2.0", "id": id, "result": result }), Err(msg) => json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32601, "message": msg } }), }; let mut guard = sessions.lock().await; if let Some(entry) = guard.get(&session_id) { if entry .sender .send(("message".to_string(), response_json.to_string())) .await .is_err() { guard.remove(&session_id); } } }); StatusCode::ACCEPTED.into_response() }