| 1 | //! Committed official SDK + real Rust FetchProxy/tickets + guarded loopback |
| 2 | //! HTTP peer. No provider, credentials, SDK-side fetch or user filesystem. |
| 3 | use super::*; |
| 4 | use crate::extension_host::TestManagerGuard; |
| 5 | use crate::extension_host::tests::{FixturePlugins, node_for_tests}; |
| 6 | use crate::mcp::{McpBackend, McpConfig, McpPool}; |
| 7 | use crate::plugins::activation::TestPolicyGuard; |
| 8 | use tokio::io::{AsyncReadExt, AsyncWriteExt}; |
| 9 | use tokio::net::{TcpListener, TcpStream}; |
| 10 | |
| 11 | #[derive(Clone, Copy)] |
| 12 | enum Mode { |
| 13 | Json, |
| 14 | PostSse, |
| 15 | Legacy, |
| 16 | RequirePreflight, |
| 17 | Negotiate, |
| 18 | } |
| 19 | struct Peer { |
| 20 | url: String, |
| 21 | frames: Arc<Mutex<Vec<Value>>>, |
| 22 | headers: Arc<Mutex<Vec<String>>>, |
| 23 | unauthorized: Arc<std::sync::atomic::AtomicBool>, |
| 24 | cancel: CancellationToken, |
| 25 | task: tokio::task::JoinHandle<()>, |
| 26 | } |
| 27 | impl Drop for Peer { |
| 28 | fn drop(&mut self) { |
| 29 | self.cancel.cancel(); |
| 30 | self.task.abort(); |
| 31 | } |
| 32 | } |
| 33 | async fn request(socket: &mut TcpStream) -> Option<(String, Value)> { |
| 34 | let mut bytes = Vec::new(); |
| 35 | let mut chunk = [0; 4096]; |
| 36 | let header_end = loop { |
| 37 | let count = socket.read(&mut chunk).await.ok()?; |
| 38 | if count == 0 { |
| 39 | return None; |
| 40 | } |
| 41 | bytes.extend_from_slice(&chunk[..count]); |
| 42 | assert!(bytes.len() <= MAX_MCP_RESPONSE_BYTES + 16 * 1024); |
| 43 | if let Some(at) = bytes.windows(4).position(|window| window == b"\r\n\r\n") { |
| 44 | break at + 4; |
| 45 | } |
| 46 | }; |
| 47 | let head = String::from_utf8(bytes[..header_end].to_vec()).unwrap(); |
| 48 | let length = head |
| 49 | .lines() |
| 50 | .find_map(|line| { |
| 51 | let (name, value) = line.split_once(':')?; |
| 52 | name.eq_ignore_ascii_case("content-length") |
| 53 | .then(|| value.trim().parse::<usize>().unwrap()) |
| 54 | }) |
| 55 | .unwrap_or(0); |
| 56 | assert!(length <= MAX_MCP_RESPONSE_BYTES); |
| 57 | while bytes.len() < header_end + length { |
| 58 | let count = socket.read(&mut chunk).await.ok()?; |
| 59 | if count == 0 { |
| 60 | return None; |
| 61 | } |
| 62 | bytes.extend_from_slice(&chunk[..count]); |
| 63 | } |
| 64 | let frame = if length == 0 { |
| 65 | Value::Null |
| 66 | } else { |
| 67 | serde_json::from_slice(&bytes[header_end..header_end + length]).unwrap() |
| 68 | }; |
| 69 | Some((head, frame)) |
| 70 | } |
| 71 | async fn reply(socket: &mut TcpStream, status: &str, content: &str, body: &[u8], session: bool) { |
| 72 | let header = format!( |
| 73 | "HTTP/1.1 {status}\r\nConnection: close\r\nContent-Type: {content}\r\n{}Content-Length: {}\r\n\r\n", |
| 74 | if session { |
| 75 | "Mcp-Session-Id: observed-fixture-session\r\n" |
| 76 | } else { |
| 77 | "" |
| 78 | }, |
| 79 | body.len() |
| 80 | ); |
| 81 | socket.write_all(header.as_bytes()).await.unwrap(); |
| 82 | socket.write_all(body).await.unwrap(); |
| 83 | } |
| 84 | async fn peer(mode: Mode) -> Peer { |
| 85 | let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); |
| 86 | let addr = listener.local_addr().unwrap(); |
| 87 | let legacy = matches!(mode, Mode::Legacy | Mode::Negotiate); |
| 88 | let path = if matches!(mode, Mode::Legacy) { |
| 89 | "sse" |
| 90 | } else { |
| 91 | "mcp" |
| 92 | }; |
| 93 | let preflights = Arc::new(std::sync::atomic::AtomicUsize::new(0)); |
| 94 | let frames = Arc::new(Mutex::new(Vec::new())); |
| 95 | let headers = Arc::new(Mutex::new(Vec::new())); |
| 96 | let unauthorized = Arc::new(std::sync::atomic::AtomicBool::new(false)); |
| 97 | let cancel = CancellationToken::new(); |
| 98 | let token = cancel.clone(); |
| 99 | let received = Arc::clone(&frames); |
| 100 | let seen_headers = Arc::clone(&headers); |
| 101 | let reject = Arc::clone(&unauthorized); |
| 102 | let channel: Arc<Mutex<Option<mpsc::UnboundedSender<Value>>>> = Arc::new(Mutex::new(None)); |
| 103 | let task = tokio::spawn(async move { |
| 104 | loop { |
| 105 | let accepted = tokio::select! { biased; _ = token.cancelled() => break, accepted = listener.accept() => accepted }; |
| 106 | let Ok((mut socket, _)) = accepted else { |
| 107 | break; |
| 108 | }; |
| 109 | let token = token.clone(); |
| 110 | let received = Arc::clone(&received); |
| 111 | let headers = Arc::clone(&seen_headers); |
| 112 | let channel = Arc::clone(&channel); |
| 113 | let reject = Arc::clone(&reject); |
| 114 | let preflights = Arc::clone(&preflights); |
| 115 | tokio::spawn(async move { |
| 116 | let Some((head, frame)) = (tokio::select! { biased; _ = token.cancelled() => return, value = request(&mut socket) => value }) |
| 117 | else { |
| 118 | return; |
| 119 | }; |
| 120 | headers.lock().unwrap().push(head.clone()); |
| 121 | if head.starts_with("GET ") { |
| 122 | if matches!(mode, Mode::RequirePreflight) |
| 123 | && preflights.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 |
| 124 | { |
| 125 | reply(&mut socket, "200 OK", "application/json", b"", true).await; |
| 126 | return; |
| 127 | } |
| 128 | if !legacy { |
| 129 | reply( |
| 130 | &mut socket, |
| 131 | "405 Method Not Allowed", |
| 132 | "text/plain", |
| 133 | b"", |
| 134 | false, |
| 135 | ) |
| 136 | .await; |
| 137 | return; |
| 138 | } |
| 139 | let (send, mut rx) = mpsc::unbounded_channel(); |
| 140 | *channel.lock().unwrap() = Some(send); |
| 141 | let initial = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\nevent: endpoint\ndata: /messages?fixture=only\n\n"; |
| 142 | socket.write_all(initial.as_bytes()).await.unwrap(); |
| 143 | loop { |
| 144 | let frame = tokio::select! { biased; _ = token.cancelled() => break, next = rx.recv() => next }; |
| 145 | let Some(frame) = frame else { |
| 146 | break; |
| 147 | }; |
| 148 | if socket |
| 149 | .write_all(format!("event: message\ndata: {frame}\n\n").as_bytes()) |
| 150 | .await |
| 151 | .is_err() |
| 152 | { |
| 153 | break; |
| 154 | } |
| 155 | } |
| 156 | return; |
| 157 | } |
| 158 | assert!( |
| 159 | head.to_ascii_lowercase() |
| 160 | .contains("authorization: bearer fixture-only-bearer") |
| 161 | ); |
| 162 | if matches!(mode, Mode::Negotiate) && head.starts_with("POST /mcp ") { |
| 163 | received.lock().unwrap().push(frame.clone()); |
| 164 | reply( |
| 165 | &mut socket, |
| 166 | "405 Method Not Allowed", |
| 167 | "text/plain", |
| 168 | b"", |
| 169 | false, |
| 170 | ) |
| 171 | .await; |
| 172 | return; |
| 173 | } |
| 174 | if matches!(mode, Mode::RequirePreflight) { |
| 175 | assert!( |
| 176 | head.to_ascii_lowercase() |
| 177 | .contains("mcp-session-id: observed-fixture-session"), |
| 178 | "preflight session must accompany initialize too" |
| 179 | ); |
| 180 | } |
| 181 | assert!(head.starts_with(if legacy { |
| 182 | "POST /messages?fixture=only " |
| 183 | } else { |
| 184 | "POST /mcp " |
| 185 | })); |
| 186 | if frame.get("id").is_some() && frame.get("method").is_some() { |
| 187 | assert!( |
| 188 | frame["id"].is_string(), |
| 189 | "official SDK numeric IDs must be translated before the peer" |
| 190 | ); |
| 191 | } |
| 192 | received.lock().unwrap().push(frame.clone()); |
| 193 | let method = frame["method"].as_str().unwrap(); |
| 194 | if reject.load(std::sync::atomic::Ordering::SeqCst) && method != "initialize" { |
| 195 | reply( |
| 196 | &mut socket, |
| 197 | "401 Unauthorized", |
| 198 | "text/plain", |
| 199 | b"untrusted provider error must stay out of TS", |
| 200 | false, |
| 201 | ) |
| 202 | .await; |
| 203 | return; |
| 204 | } |
| 205 | if frame.get("id").is_none() { |
| 206 | reply(&mut socket, "202 Accepted", "text/plain", b"", false).await; |
| 207 | return; |
| 208 | } |
| 209 | let result = match method { |
| 210 | "initialize" => { |
| 211 | json!({"protocolVersion":"2025-06-18","capabilities":{"tools":{}},"serverInfo":{"name":"fetch-fixture","version":"1"}}) |
| 212 | } |
| 213 | "tools/list" => { |
| 214 | json!({"tools":[{"name":"echo","inputSchema":{"type":"object"}}]}) |
| 215 | } |
| 216 | "tools/call" => { |
| 217 | json!({"content":[{"type":"text","text":frame["params"]["arguments"].to_string()}],"isError":false}) |
| 218 | } |
| 219 | _ => panic!("unexpected fixture method {method}"), |
| 220 | }; |
| 221 | let frame = json!({"jsonrpc":"2.0","id":frame["id"],"result":result}); |
| 222 | if legacy { |
| 223 | channel |
| 224 | .lock() |
| 225 | .unwrap() |
| 226 | .as_ref() |
| 227 | .unwrap() |
| 228 | .send(frame) |
| 229 | .unwrap(); |
| 230 | reply(&mut socket, "202 Accepted", "text/plain", b"", false).await; |
| 231 | } else if matches!(mode, Mode::PostSse) { |
| 232 | reply( |
| 233 | &mut socket, |
| 234 | "200 OK", |
| 235 | "text/event-stream", |
| 236 | format!("event: message\ndata: {frame}\n\n").as_bytes(), |
| 237 | method == "initialize", |
| 238 | ) |
| 239 | .await; |
| 240 | } else { |
| 241 | reply( |
| 242 | &mut socket, |
| 243 | "200 OK", |
| 244 | "application/json", |
| 245 | frame.to_string().as_bytes(), |
| 246 | method == "initialize", |
| 247 | ) |
| 248 | .await; |
| 249 | } |
| 250 | }); |
| 251 | } |
| 252 | }); |
| 253 | Peer { |
| 254 | url: format!("http://{addr}/{path}"), |
| 255 | frames, |
| 256 | headers, |
| 257 | unauthorized, |
| 258 | cancel, |
| 259 | task, |
| 260 | } |
| 261 | } |
| 262 | async fn config(peer: &Peer) -> McpServerConfig { |
| 263 | serde_json::from_value(json!({ |
| 264 | "url": peer.url, |
| 265 | "transport": peer.url.ends_with("/sse").then_some("sse"), |
| 266 | "headers": {"Authorization":"Bearer fixture-only-bearer"}, |
| 267 | "connect_timeout": 5, |
| 268 | "read_timeout": 5, |
| 269 | })) |
| 270 | .unwrap() |
| 271 | } |
| 272 | async fn real_mode(mode: Mode, name: &str) { |
| 273 | let _env = crate::test_support::lock_test_env(); |
| 274 | let _proxies: Vec<_> = [ |
| 275 | "HTTP_PROXY", |
| 276 | "HTTPS_PROXY", |
| 277 | "ALL_PROXY", |
| 278 | "http_proxy", |
| 279 | "https_proxy", |
| 280 | "all_proxy", |
| 281 | ] |
| 282 | .into_iter() |
| 283 | .map(crate::test_support::EnvVarGuard::remove) |
| 284 | .collect(); |
| 285 | let _policy = TestPolicyGuard::extension_host(true); |
| 286 | let Some(node) = node_for_tests(name) else { |
| 287 | return; |
| 288 | }; |
| 289 | let fixture = FixturePlugins::new(&[]).await; |
| 290 | let manager = fixture.manager(node); |
| 291 | let _manager = TestManagerGuard::install(Arc::clone(&manager)); |
| 292 | let peer = peer(mode).await; |
| 293 | let mut pool = McpPool::new(McpConfig { |
| 294 | servers: [(name.to_string(), config(&peer).await)].into(), |
| 295 | ..Default::default() |
| 296 | }) |
| 297 | .with_backend(McpBackend::Host); |
| 298 | let connection = pool.get_or_connect(name).await.unwrap(); |
| 299 | assert!(connection.is_ready()); |
| 300 | assert_eq!(connection.tools().len(), 1); |
| 301 | let result = connection |
| 302 | .call_tool("echo", json!({"payload":"exact"}), 5) |
| 303 | .await |
| 304 | .unwrap(); |
| 305 | assert_eq!(result["content"][0]["text"], "{\"payload\":\"exact\"}"); |
| 306 | assert!( |
| 307 | peer.headers.lock().unwrap().iter().all(|head| head |
| 308 | .to_ascii_lowercase() |
| 309 | .contains("authorization: bearer fixture-only-bearer")), |
| 310 | "preflight, every POST and the actual SSE GET retain Rust auth" |
| 311 | ); |
| 312 | let frames = peer.frames.lock().unwrap().clone(); |
| 313 | for method in [ |
| 314 | "initialize", |
| 315 | "notifications/initialized", |
| 316 | "tools/list", |
| 317 | "tools/call", |
| 318 | ] { |
| 319 | assert_eq!( |
| 320 | frames |
| 321 | .iter() |
| 322 | .filter(|frame| frame["method"] == method) |
| 323 | .count(), |
| 324 | if method == "initialize" && matches!(mode, Mode::Negotiate) { |
| 325 | 2 |
| 326 | } else { |
| 327 | 1 |
| 328 | } |
| 329 | ); |
| 330 | } |
| 331 | assert!( |
| 332 | frames |
| 333 | .iter() |
| 334 | .filter(|frame| frame.get("id").is_some()) |
| 335 | .all(|frame| frame["id"].is_string()) |
| 336 | ); |
| 337 | for (method, expected) in [ |
| 338 | ("initialize", "1"), |
| 339 | ("tools/list", "2"), |
| 340 | ("tools/call", "3"), |
| 341 | ] { |
| 342 | assert_eq!( |
| 343 | frames |
| 344 | .iter() |
| 345 | .find(|frame| frame["method"] == method) |
| 346 | .unwrap()["id"], |
| 347 | expected, |
| 348 | "the original Rust facade ID must reach the real peer" |
| 349 | ); |
| 350 | } |
| 351 | if !matches!(mode, Mode::Legacy | Mode::Negotiate) { |
| 352 | assert!(peer.headers.lock().unwrap().iter().any(|head| { |
| 353 | head.to_ascii_lowercase() |
| 354 | .contains("mcp-session-id: observed-fixture-session") |
| 355 | })); |
| 356 | } |
| 357 | let sessions: Vec<_> = manager |
| 358 | .shared |
| 359 | .mcp_broker |
| 360 | .sessions |
| 361 | .lock() |
| 362 | .unwrap() |
| 363 | .values() |
| 364 | .cloned() |
| 365 | .collect(); |
| 366 | assert_eq!(sessions.len(), 1); |
| 367 | assert!(sessions[0].http.is_some()); |
| 368 | assert!(sessions[0].broker.lock().await.is_none()); |
| 369 | pool.shutdown_all().await; |
| 370 | manager.shutdown().await; |
| 371 | assert!( |
| 372 | manager |
| 373 | .shared |
| 374 | .mcp_broker |
| 375 | .sessions |
| 376 | .lock() |
| 377 | .unwrap() |
| 378 | .is_empty() |
| 379 | ); |
| 380 | } |
| 381 | #[tokio::test(flavor = "current_thread")] |
| 382 | async fn host_http_json_uses_real_sdk_guarded_client_and_exact_string_ids() { |
| 383 | real_mode(Mode::Json, "host_http_json").await; |
| 384 | } |
| 385 | #[tokio::test(flavor = "current_thread")] |
| 386 | async fn host_http_sse_body_uses_real_sdk_and_shared_bounded_parser() { |
| 387 | real_mode(Mode::PostSse, "host_http_sse").await; |
| 388 | } |
| 389 | #[tokio::test(flavor = "current_thread")] |
| 390 | async fn host_legacy_sse_observes_endpoint_and_uses_same_guarded_client() { |
| 391 | real_mode(Mode::Legacy, "host_legacy_sse").await; |
| 392 | } |
| 393 | |
| 394 | #[tokio::test(flavor = "current_thread")] |
| 395 | async fn host_http_preflight_session_reaches_initialize_and_subsequent_requests() { |
| 396 | real_mode(Mode::RequirePreflight, "host_http_preflight").await; |
| 397 | } |
| 398 | #[tokio::test(flavor = "current_thread")] |
| 399 | async fn host_http_incompatible_handshake_negotiates_real_sdk_sse_with_original_id() { |
| 400 | real_mode(Mode::Negotiate, "host_http_negotiate").await; |
| 401 | } |
| 402 | |
| 403 | #[tokio::test(flavor = "current_thread")] |
| 404 | async fn fetch_proxy_refuses_wire_id_mutation_before_any_network_write() { |
| 405 | let _env = crate::test_support::lock_test_env(); |
| 406 | let _policy = TestPolicyGuard::extension_host(true); |
| 407 | let Some(node) = node_for_tests("fetch_proxy_wire_binding") else { |
| 408 | return; |
| 409 | }; |
| 410 | let fixture = FixturePlugins::new(&[]).await; |
| 411 | let manager = fixture.manager(node); |
| 412 | let _manager = TestManagerGuard::install(Arc::clone(&manager)); |
| 413 | let peer = peer(Mode::Json).await; |
| 414 | let config = config(&peer).await; |
| 415 | for invalid_id in [json!("forged-id"), json!(7), Value::Null] { |
| 416 | let client = McpHttpClient::new( |
| 417 | &peer.url, |
| 418 | false, |
| 419 | false, |
| 420 | false, |
| 421 | None, |
| 422 | Duration::from_secs(5), |
| 423 | Duration::from_secs(5), |
| 424 | ) |
| 425 | .unwrap(); |
| 426 | let transport = SdkTransport::connect_with_http( |
| 427 | "fixture", |
| 428 | &config, |
| 429 | CancellationToken::new(), |
| 430 | Duration::from_secs(5), |
| 431 | Some(client), |
| 432 | ) |
| 433 | .await |
| 434 | .unwrap(); |
| 435 | let grant = transport |
| 436 | .grant( |
| 437 | "tools/list", |
| 438 | json!({}), |
| 439 | Duration::from_secs(5), |
| 440 | None, |
| 441 | Some("rust-id"), |
| 442 | ) |
| 443 | .unwrap(); |
| 444 | *transport |
| 445 | .session |
| 446 | .http |
| 447 | .as_ref() |
| 448 | .unwrap() |
| 449 | .started |
| 450 | .lock() |
| 451 | .unwrap() = true; |
| 452 | let params = NetFetchParams { |
| 453 | owner: transport.session.owner.clone(), |
| 454 | session_id: transport.session_id.clone(), |
| 455 | url: opaque(&transport.session), |
| 456 | method: "POST".into(), |
| 457 | headers: McpHttpHeaders::default(), |
| 458 | frame: Some(json!({"jsonrpc":"2.0","id":invalid_id,"method":"tools/list","params":{}})), |
| 459 | ticket: Some(grant.ticket), |
| 460 | operation_id: Some(grant.operation_id), |
| 461 | }; |
| 462 | let (cx, _, _) = HostRequestContext::for_test(20); |
| 463 | assert!( |
| 464 | manager |
| 465 | .shared |
| 466 | .mcp_broker |
| 467 | .serve( |
| 468 | &manager.shared, |
| 469 | transport.session.host_generation, |
| 470 | HostRequest::NetFetch(params), |
| 471 | cx |
| 472 | ) |
| 473 | .await |
| 474 | .is_err() |
| 475 | ); |
| 476 | assert!(peer.frames.lock().unwrap().is_empty()); |
| 477 | assert!(transport.session.cancel.is_cancelled()); |
| 478 | } |
| 479 | manager.shutdown().await; |
| 480 | } |
| 481 | #[tokio::test(flavor = "current_thread")] |
| 482 | async fn fetch_proxy_foreign_url_cannot_use_session_network_authority() { |
| 483 | let _env = crate::test_support::lock_test_env(); |
| 484 | let _policy = TestPolicyGuard::extension_host(true); |
| 485 | let Some(node) = node_for_tests("fetch_proxy_url_binding") else { |
| 486 | return; |
| 487 | }; |
| 488 | let fixture = FixturePlugins::new(&[]).await; |
| 489 | let manager = fixture.manager(node); |
| 490 | let _manager = TestManagerGuard::install(Arc::clone(&manager)); |
| 491 | let peer = peer(Mode::Json).await; |
| 492 | let config = config(&peer).await; |
| 493 | let client = McpHttpClient::new( |
| 494 | &peer.url, |
| 495 | false, |
| 496 | false, |
| 497 | false, |
| 498 | None, |
| 499 | Duration::from_secs(5), |
| 500 | Duration::from_secs(5), |
| 501 | ) |
| 502 | .unwrap(); |
| 503 | let transport = SdkTransport::connect_with_http( |
| 504 | "fixture", |
| 505 | &config, |
| 506 | CancellationToken::new(), |
| 507 | Duration::from_secs(5), |
| 508 | Some(client), |
| 509 | ) |
| 510 | .await |
| 511 | .unwrap(); |
| 512 | *transport |
| 513 | .session |
| 514 | .http |
| 515 | .as_ref() |
| 516 | .unwrap() |
| 517 | .started |
| 518 | .lock() |
| 519 | .unwrap() = true; |
| 520 | let admitted_preflight_count = peer.headers.lock().unwrap().len(); |
| 521 | let params = NetFetchParams { |
| 522 | owner: transport.session.owner.clone(), |
| 523 | session_id: transport.session_id.clone(), |
| 524 | url: peer.url.clone(), |
| 525 | method: "GET".into(), |
| 526 | headers: McpHttpHeaders::default(), |
| 527 | frame: None, |
| 528 | ticket: None, |
| 529 | operation_id: None, |
| 530 | }; |
| 531 | let (cx, _, _) = HostRequestContext::for_test(21); |
| 532 | assert!( |
| 533 | manager |
| 534 | .shared |
| 535 | .mcp_broker |
| 536 | .serve( |
| 537 | &manager.shared, |
| 538 | transport.session.host_generation, |
| 539 | HostRequest::NetFetch(params), |
| 540 | cx |
| 541 | ) |
| 542 | .await |
| 543 | .is_err() |
| 544 | ); |
| 545 | assert_eq!( |
| 546 | peer.headers.lock().unwrap().len(), |
| 547 | admitted_preflight_count, |
| 548 | "forged URL must not make any additional request" |
| 549 | ); |
| 550 | manager.shutdown().await; |
| 551 | } |
| 552 | #[test] |
| 553 | fn fetch_proxy_wire_is_builtin_only_and_carries_no_credential_or_auth_api() { |
| 554 | let owner = json!({"plugin_id":"host:mcp","generation":1,"owner_token":"fixture"}); |
| 555 | let params = json!({"owner":owner,"session_id":"s","url":"https://mcp-proxy.invalid/s","method":"GET","headers":{}}); |
| 556 | let frame = json!({"jsonrpc":"2.0","id":1,"method":"net/fetch","params":params}); |
| 557 | assert!(parse_host_message(frame.clone(), HostTier::Builtin).is_ok()); |
| 558 | assert!(parse_host_message(frame.clone(), HostTier::Plugin).is_err()); |
| 559 | let mut forbidden = frame; |
| 560 | forbidden["params"]["bearer_token"] = json!("forbidden"); |
| 561 | assert!(parse_host_message(forbidden, HostTier::Builtin).is_err()); |
| 562 | } |
| 563 | |
| 564 | #[tokio::test(flavor = "current_thread")] |
| 565 | async fn fetch_proxy_owner_revocation_releases_retained_legacy_body() { |
| 566 | let _env = crate::test_support::lock_test_env(); |
| 567 | let _proxies: Vec<_> = [ |
| 568 | "HTTP_PROXY", |
| 569 | "HTTPS_PROXY", |
| 570 | "ALL_PROXY", |
| 571 | "http_proxy", |
| 572 | "https_proxy", |
| 573 | "all_proxy", |
| 574 | ] |
| 575 | .into_iter() |
| 576 | .map(crate::test_support::EnvVarGuard::remove) |
| 577 | .collect(); |
| 578 | let _policy = TestPolicyGuard::extension_host(true); |
| 579 | let Some(node) = node_for_tests("fetch_proxy_owner_revoke") else { |
| 580 | return; |
| 581 | }; |
| 582 | let fixture = FixturePlugins::new(&[]).await; |
| 583 | let manager = fixture.manager(node); |
| 584 | let _manager = TestManagerGuard::install(Arc::clone(&manager)); |
| 585 | let peer = peer(Mode::Legacy).await; |
| 586 | let mut pool = McpPool::new(McpConfig { |
| 587 | servers: [("fixture".into(), config(&peer).await)].into(), |
| 588 | ..Default::default() |
| 589 | }) |
| 590 | .with_backend(McpBackend::Host); |
| 591 | pool.get_or_connect("fixture").await.unwrap(); |
| 592 | let session = manager |
| 593 | .shared |
| 594 | .mcp_broker |
| 595 | .sessions |
| 596 | .lock() |
| 597 | .unwrap() |
| 598 | .values() |
| 599 | .next() |
| 600 | .cloned() |
| 601 | .unwrap(); |
| 602 | let http = session.http.as_ref().unwrap(); |
| 603 | assert_eq!( |
| 604 | http.responses.lock().unwrap().len(), |
| 605 | 1, |
| 606 | "the real SDK is reading the retained legacy GET" |
| 607 | ); |
| 608 | assert_eq!(http.slots.load(std::sync::atomic::Ordering::SeqCst), 1); |
| 609 | let count = manager.shared.mcp_broker.revoke_owner("host:mcp"); |
| 610 | manager |
| 611 | .shared |
| 612 | .mcp_users |
| 613 | .fetch_sub(count, std::sync::atomic::Ordering::SeqCst); |
| 614 | assert!(session.cancel.is_cancelled()); |
| 615 | assert!(http.responses.lock().unwrap().is_empty()); |
| 616 | tokio::time::timeout(Duration::from_secs(5), async { |
| 617 | while http.slots.load(std::sync::atomic::Ordering::SeqCst) != 0 { |
| 618 | tokio::task::yield_now().await; |
| 619 | } |
| 620 | }) |
| 621 | .await |
| 622 | .unwrap(); |
| 623 | assert!( |
| 624 | session |
| 625 | .validate(&manager.shared, session.host_generation, &session.owner) |
| 626 | .is_err() |
| 627 | ); |
| 628 | assert_eq!( |
| 629 | peer.frames |
| 630 | .lock() |
| 631 | .unwrap() |
| 632 | .iter() |
| 633 | .filter(|frame| frame["method"] == "initialize") |
| 634 | .count(), |
| 635 | 1, |
| 636 | "withdrawal cannot replay or reconnect" |
| 637 | ); |
| 638 | pool.shutdown_all().await; |
| 639 | manager.shutdown().await; |
| 640 | } |
| 641 | #[tokio::test(flavor = "current_thread")] |
| 642 | async fn fetch_proxy_unauthorized_is_safe_auth_recovery_without_operation_replay() { |
| 643 | let _env = crate::test_support::lock_test_env(); |
| 644 | let _proxies: Vec<_> = [ |
| 645 | "HTTP_PROXY", |
| 646 | "HTTPS_PROXY", |
| 647 | "ALL_PROXY", |
| 648 | "http_proxy", |
| 649 | "https_proxy", |
| 650 | "all_proxy", |
| 651 | ] |
| 652 | .into_iter() |
| 653 | .map(crate::test_support::EnvVarGuard::remove) |
| 654 | .collect(); |
| 655 | let _policy = TestPolicyGuard::extension_host(true); |
| 656 | let Some(node) = node_for_tests("fetch_proxy_auth_recovery") else { |
| 657 | return; |
| 658 | }; |
| 659 | let fixture = FixturePlugins::new(&[]).await; |
| 660 | let manager = fixture.manager(node); |
| 661 | let _manager = TestManagerGuard::install(Arc::clone(&manager)); |
| 662 | let peer = peer(Mode::Json).await; |
| 663 | let mut pool = McpPool::new(McpConfig { |
| 664 | servers: [("fixture".into(), config(&peer).await)].into(), |
| 665 | ..Default::default() |
| 666 | }) |
| 667 | .with_backend(McpBackend::Host); |
| 668 | let connection = pool.get_or_connect("fixture").await.unwrap(); |
| 669 | peer.unauthorized |
| 670 | .store(true, std::sync::atomic::Ordering::SeqCst); |
| 671 | let error = connection |
| 672 | .call_tool("echo", json!({}), 5) |
| 673 | .await |
| 674 | .unwrap_err(); |
| 675 | let safe = format!("{error:#}"); |
| 676 | assert!(safe.contains("401"), "{safe}"); |
| 677 | assert!(safe.contains("bearer token"), "{safe}"); |
| 678 | assert!(!safe.contains("untrusted provider error")); |
| 679 | assert_eq!( |
| 680 | peer.frames |
| 681 | .lock() |
| 682 | .unwrap() |
| 683 | .iter() |
| 684 | .filter(|frame| frame["method"] == "tools/call") |
| 685 | .count(), |
| 686 | 1 |
| 687 | ); |
| 688 | pool.shutdown_all().await; |
| 689 | manager.shutdown().await; |
| 690 | } |
| 691 |