use crate::{
config::McpOAuthConfig,
mcp::{McpError, McpResult},
};
use base64::Engine;
use chrono::Utc;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::{
collections::BTreeMap,
fmt, fs, io,
io::{Read, Write},
net::{TcpListener, TcpStream},
path::{Path, PathBuf},
sync::{
Arc, Mutex,
atomic::{AtomicBool, Ordering},
mpsc::Receiver,
},
time::{Duration, Instant},
};
const TOKEN_FILE_MODE: u32 = 0o600;
const HTTP_CLIENT_NAME: &str = "magi-code";
const REFRESH_SKEW_SECONDS: i64 = 60;
pub(crate) const CALLBACK_PATH: &str = "/mcp/oauth/callback";
pub(crate) const LOGIN_WAIT_TIMEOUT: Duration = Duration::from_secs(300);
#[cfg(not(test))]
const CALLBACK_STREAM_TIMEOUT: Duration = Duration::from_secs(5);
#[cfg(test)]
const CALLBACK_STREAM_TIMEOUT: Duration = Duration::from_millis(100);
mod discovery;
mod login;
mod storage;
mod tokens;
pub(crate) use discovery::{AuthorizationServerMetadata, parse_www_authenticate_resource_metadata};
pub(crate) use login::{bind_callback_listener, capture_loopback_or_manual_code, prepare_login};
pub(crate) use storage::{StoredToken, delete_token, write_token};
pub(crate) use tokens::{TokenProvider, auth_status, exchange_code, stored_token_from_response};
#[cfg(test)]
use login::McpOAuthLoginOutcome;
#[cfg(test)]
use tokens::TokenResponse;
#[cfg(test)]
use discovery::{
ProtectedResourceMetadata, RegistrationResponse, discover_authorization_server,
discover_protected_resource, register_client, select_authorization_server,
validate_authorization_server_metadata,
};
#[cfg(not(test))]
use discovery::{
discover_authorization_server, discover_protected_resource, register_client,
select_authorization_server,
};
#[cfg(test)]
use login::{
authorization_url, code_challenge, generate_code_verifier, generate_state,
parse_callback_request_line, parse_manual_fallback_input, parse_query,
};
#[cfg(test)]
pub(crate) use storage::read_token;
#[cfg(not(test))]
use storage::read_token;
#[cfg(test)]
use storage::token_file_path;
#[cfg(test)]
use tokens::{refresh_token, token_error_message};
use discovery::{sanitize_url, validate_oauth_endpoint};
use storage::{read_token_locked, validate_token_url, write_token_if_unchanged};
use tokens::read_oauth_success_text;
#[cfg(test)]
mod tests {
use super::*;
use crate::http_body::DEFAULT_BOUNDED_BODY_MAX_BYTES;
use crate::persistence::CrossProcessFileLock;
use chrono::Utc;
use std::{
io::{Read, Write},
net::{TcpListener, TcpStream},
sync::{
Arc, Mutex,
atomic::{AtomicBool, Ordering},
mpsc,
},
thread,
time::Duration,
};
fn stored_token() -> StoredToken {
StoredToken {
client_id: "client-1".to_string(),
access_token: "fake-access-token".to_string(),
refresh_token: Some("fake-refresh-token".to_string()),
expires_at: Some(1_900_000_000),
granted_scopes: vec!["search".to_string()],
client_secret: None,
authorization_server: None,
issuer: None,
token_endpoint: None,
resource: Some("https://mcp.example.test/mcp".to_string()),
server_url: "https://mcp.example.test/mcp".to_string(),
token_received_at: Utc::now().timestamp(),
}
}
#[test]
fn oauth_debug_redacts_secret_fields() {
let token = TokenResponse {
access_token: "fake-access-token".to_string(),
token_type: "Bearer".to_string(),
expires_in: Some(3600),
refresh_token: Some("fake-refresh-token".to_string()),
scope: Some("search".to_string()),
extra: BTreeMap::new(),
};
let registration = RegistrationResponse {
client_id: "client-1".to_string(),
client_secret: Some("fake-client-secret".to_string()),
extra: BTreeMap::new(),
};
let stored = stored_token();
let login = McpOAuthLoginOutcome {
authorization_url: "https://auth.example.test/authorize?code_challenge=fake-verifier&state=fake-state&client_secret=fake-client-secret".to_string(),
redirect_uri: "http://127.0.0.1:1234/mcp/oauth/callback".to_string(),
state: "fake-state".to_string(),
verifier: "fake-verifier".to_string(),
client_id: "client-1".to_string(),
client_secret: Some("fake-client-secret".to_string()),
metadata: AuthorizationServerMetadata {
issuer: "https://auth.example.test".to_string(),
authorization_endpoint: "https://auth.example.test/authorize".to_string(),
token_endpoint: "https://auth.example.test/token".to_string(),
registration_endpoint: None,
scopes_supported: None,
response_types_supported: None,
grant_types_supported: None,
code_challenge_methods_supported: None,
extra: BTreeMap::new(),
},
resource: "https://mcp.example.test/mcp".to_string(),
authorization_server: "https://auth.example.test".to_string(),
};
for debug in [
format!("{token:?}"),
format!("{registration:?}"),
format!("{stored:?}"),
format!("{login:?}"),
] {
assert!(debug.contains("[REDACTED]"), "{debug}");
assert!(!debug.contains("fake-access-token"), "{debug}");
assert!(!debug.contains("fake-refresh-token"), "{debug}");
assert!(!debug.contains("fake-client-secret"), "{debug}");
assert!(!debug.contains("fake-verifier"), "{debug}");
assert!(!debug.contains("fake-state"), "{debug}");
assert!(!debug.contains("code_challenge="), "{debug}");
}
}
#[test]
fn token_error_message_whitelists_oauth_error_codes() {
let malicious = token_error_message(
reqwest::StatusCode::BAD_REQUEST,
r#"{"error":"fake_access_token_secret_12345"}"#,
false,
);
assert_eq!(malicious, "OAuth token exchange failed");
assert!(!malicious.contains("fake_access_token_secret_12345"));
let known = token_error_message(
reqwest::StatusCode::BAD_REQUEST,
r#"{"error":"invalid_scope"}"#,
false,
);
assert!(known.contains("invalid_scope"), "{known}");
}
#[test]
fn pkce_verifier_charset_challenge_determinism_and_state_uniqueness() {
let verifier = generate_code_verifier();
assert!((43..=128).contains(&verifier.len()), "{verifier}");
assert!(verifier.bytes().all(|byte| matches!(
byte,
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~'
)));
assert_eq!(
code_challenge("dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"),
"E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM"
);
assert_ne!(generate_state(), generate_state());
}
#[test]
fn authorization_url_percent_encodes_parameters() {
let url = authorization_url(
"https://auth.example.test/authorize",
"client id",
"http://127.0.0.1:1234/mcp/oauth/callback",
"challenge",
"state",
"https://mcp.example.test/mcp?a=b",
&["search".to_string(), "offline_access".to_string()],
);
assert!(url.contains("client_id=client%20id"), "{url}");
assert!(
url.contains("redirect_uri=http%3A%2F%2F127.0.0.1%3A1234%2Fmcp%2Foauth%2Fcallback"),
"{url}"
);
assert!(
url.contains("resource=https%3A%2F%2Fmcp.example.test%2Fmcp%3Fa%3Db"),
"{url}"
);
assert!(url.contains("scope=search%20offline_access"), "{url}");
}
#[test]
fn metadata_deserialization_tolerates_unknown_fields() {
let protected: ProtectedResourceMetadata = serde_json::from_str(
r#"{"resource":"https://mcp.example.test/mcp","authorization_servers":["https://auth.example.test"],"future":true}"#,
)
.unwrap();
assert_eq!(protected.extra["future"], true);
let auth: AuthorizationServerMetadata = serde_json::from_str(
r#"{"issuer":"https://auth.example.test","authorization_endpoint":"https://auth.example.test/authorize","token_endpoint":"https://auth.example.test/token","registration_endpoint":"https://auth.example.test/register","code_challenge_methods_supported":["S256"],"future":"kept"}"#,
)
.unwrap();
assert_eq!(auth.extra["future"], "kept");
assert!(format!("{auth:?}").contains("authorization_endpoint"));
}
#[test]
fn select_authorization_server_prefers_configured_then_metadata() {
let protected = ProtectedResourceMetadata {
resource: "https://mcp.example.test/mcp".to_string(),
authorization_servers: vec!["https://auth.example.test".to_string()],
extra: BTreeMap::new(),
};
assert_eq!(
select_authorization_server(Some("https://auth.example.test"), Some(&protected))
.unwrap(),
"https://auth.example.test"
);
assert!(
select_authorization_server(Some("https://override.example.test"), Some(&protected))
.is_err()
);
assert_eq!(
select_authorization_server(None, Some(&protected)).unwrap(),
"https://auth.example.test"
);
assert!(select_authorization_server(None, None).is_err());
}
#[test]
fn www_authenticate_resource_metadata_parser_extracts_url() {
assert_eq!(
parse_www_authenticate_resource_metadata(
r#"Bearer realm="mcp", resource_metadata="https://mcp.example.test/.well-known/oauth-protected-resource""#,
)
.as_deref(),
Some("https://mcp.example.test/.well-known/oauth-protected-resource")
);
}
#[test]
fn token_storage_round_trip_permissions_delete_and_url_validation() {
let temp = tempfile::TempDir::new().unwrap();
let token = stored_token();
write_token(temp.path(), "remote", &token).unwrap();
let path = token_file_path(temp.path(), "remote").unwrap();
assert!(path.exists());
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mode = fs::metadata(&path).unwrap().permissions().mode() & 0o777;
assert_eq!(mode, 0o600);
}
let read = read_token(temp.path(), "remote").unwrap().unwrap();
assert_eq!(read, token);
assert!(validate_token_url(&read, "https://mcp.example.test/mcp"));
assert!(!validate_token_url(&read, "https://mcp.example.test/other"));
delete_token(temp.path(), "remote").unwrap();
assert!(!path.exists());
delete_token(temp.path(), "remote").unwrap();
}
#[test]
fn token_storage_nonexistent_and_corrupt_file() {
let temp = tempfile::TempDir::new().unwrap();
assert!(read_token(temp.path(), "missing").unwrap().is_none());
let path = token_file_path(temp.path(), "bad").unwrap();
fs::create_dir_all(path.parent().unwrap()).unwrap();
fs::write(&path, "not json").unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut permissions = fs::metadata(&path).unwrap().permissions();
permissions.set_mode(0o600);
fs::set_permissions(&path, permissions).unwrap();
}
let error = read_token(temp.path(), "bad").unwrap_err().to_string();
assert!(error.contains("corrupt"), "{error}");
}
#[test]
fn token_storage_rejects_path_traversal_server_names() {
let temp = tempfile::TempDir::new().unwrap();
let token = stored_token();
for malicious in ["../auth", "..", ".", "a/b", "a\\b", "a:b", "a b"] {
let err = write_token(temp.path(), malicious, &token)
.unwrap_err()
.to_string();
assert!(
err.contains("not a valid token store name"),
"{malicious}: {err}"
);
assert!(read_token(temp.path(), malicious).is_err());
assert!(delete_token(temp.path(), malicious).is_err());
}
assert!(!temp.path().join("auth.json").exists());
write_token(temp.path(), "my-server_1", &token).unwrap();
assert!(read_token(temp.path(), "my-server_1").unwrap().is_some());
}
#[cfg(unix)]
#[test]
fn token_read_rejects_symlinked_token_file() {
let temp = tempfile::TempDir::new().unwrap();
let token = stored_token();
write_token(temp.path(), "remote", &token).unwrap();
let real_path = token_file_path(temp.path(), "remote").unwrap();
use std::os::unix::fs::symlink;
let link_path = token_file_path(temp.path(), "linked").unwrap();
fs::create_dir_all(link_path.parent().unwrap()).unwrap();
symlink(&real_path, &link_path).unwrap();
let err = read_token(temp.path(), "linked").unwrap_err().to_string();
assert!(
err.contains("symlink") || err.contains("regular private file"),
"{err}"
);
assert!(!err.contains("fake-access-token"), "{err}");
}
#[test]
fn discovery_and_registration_use_loopback_http() {
let protected_body = r#"{"resource":"http://127.0.0.1/mcp","authorization_servers":["http://127.0.0.1/auth"],"future":true}"#;
let protected_url =
serve_once("/.well-known/oauth-protected-resource", protected_body, 200);
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let protected = discover_protected_resource(&format!("{protected_url}/mcp"), &client)
.unwrap()
.unwrap();
assert_eq!(protected.resource, "http://127.0.0.1/mcp");
let auth_body = r#"{"issuer":"http://127.0.0.1/auth","authorization_endpoint":"https://auth.example.test/authorize","token_endpoint":"https://auth.example.test/token","registration_endpoint":"http://127.0.0.1/register","code_challenge_methods_supported":["S256"]}"#;
let auth_url = serve_once("/.well-known/oauth-authorization-server", auth_body, 200);
let auth = discover_authorization_server(&auth_url, &client).unwrap();
assert_eq!(auth.token_endpoint, "https://auth.example.test/token");
let (registration_url, body_rx) = serve_once_with_body(
"/register",
r#"{"client_id":"registered-client","client_secret":"fake-client-secret"}"#,
201,
);
let registration = register_client(
&format!("{registration_url}/register"),
vec!["http://127.0.0.1:1234/mcp/oauth/callback".to_string()],
&client,
)
.unwrap();
assert_eq!(registration.client_id, "registered-client");
let body = body_rx.recv_timeout(Duration::from_secs(2)).unwrap();
assert!(
body.contains("\"token_endpoint_auth_method\":\"none\""),
"{body}"
);
assert!(!format!("{registration:?}").contains("fake-client-secret"));
}
#[test]
fn token_exchange_redirect_does_not_send_credentials_to_second_origin() {
let target = TcpListener::bind("127.0.0.1:0").unwrap();
target.set_nonblocking(true).unwrap();
let target_url = format!("http://{}/token", target.local_addr().unwrap());
let origin = TcpListener::bind("127.0.0.1:0").unwrap();
origin.set_nonblocking(true).unwrap();
let origin_url = format!("http://{}/token", origin.local_addr().unwrap());
let origin_thread = thread::spawn(move || {
let deadline = Instant::now() + Duration::from_secs(2);
loop {
match origin.accept() {
Ok((mut stream, _)) => {
let mut buffer = [0_u8; 4096];
let n = stream.read(&mut buffer).unwrap();
let request = String::from_utf8_lossy(&buffer[..n]);
assert!(request.contains("code=authorization-code"), "{request}");
assert!(request.contains("client_secret=client-secret"), "{request}");
loopback_write_response_with_headers(
&mut stream,
"302 Found",
"text/plain",
"redirect",
&[&format!("Location: {target_url}")],
);
break;
}
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
if Instant::now() >= deadline {
panic!("redirect origin received no request");
}
thread::sleep(Duration::from_millis(10));
}
Err(error) => panic!("redirect origin accept failed: {error}"),
}
}
});
let client = reqwest::blocking::Client::builder()
.redirect(reqwest::redirect::Policy::none())
.timeout(Duration::from_secs(1))
.build()
.unwrap();
let error = exchange_code(
&origin_url,
"authorization-code",
"http://127.0.0.1:1234/mcp/oauth/callback",
"client-1",
"verifier",
"http://127.0.0.1/mcp",
Some("client-secret"),
&client,
)
.unwrap_err()
.to_string();
assert!(error.contains("HTTP 302"), "{error}");
origin_thread.join().unwrap();
thread::sleep(Duration::from_millis(100));
assert!(matches!(target.accept(), Err(error) if error.kind() == io::ErrorKind::WouldBlock));
}
#[test]
fn token_exchange_refresh_and_callback_validation_are_redacted() {
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let (token_url, body_rx) = serve_once_with_body(
"/token",
r#"{"access_token":"fake-access-token","token_type":"Bearer","expires_in":3600,"refresh_token":"fake-refresh-token","scope":"search"}"#,
200,
);
let token = exchange_code(
&format!("{token_url}/token"),
"fake-code",
"http://127.0.0.1:1234/mcp/oauth/callback",
"client-1",
"verifier",
"http://127.0.0.1/mcp",
None,
&client,
)
.unwrap();
assert_eq!(token.access_token, "fake-access-token");
assert!(
body_rx
.recv_timeout(Duration::from_secs(2))
.unwrap()
.contains("grant_type=authorization_code")
);
let (refresh_url, refresh_rx) = serve_once_with_body(
"/token",
r#"{"access_token":"new-access-token","token_type":"Bearer","expires_in":3600}"#,
200,
);
let refreshed = refresh_token(
&format!("{refresh_url}/token"),
"fake-refresh-token",
"client-1",
"http://127.0.0.1/mcp",
None,
&client,
)
.unwrap();
assert_eq!(refreshed.access_token, "new-access-token");
assert!(
refresh_rx
.recv_timeout(Duration::from_secs(2))
.unwrap()
.contains("grant_type=refresh_token")
);
assert_eq!(
parse_callback_request_line(
"GET /mcp/oauth/callback?code=callback-code&state=expected HTTP/1.1",
"expected",
)
.unwrap(),
"callback-code"
);
let error = parse_callback_request_line(
"GET /mcp/oauth/callback?code=callback-code&state=wrong HTTP/1.1",
"expected",
)
.unwrap_err()
.to_string();
assert!(error.contains("state mismatch"), "{error}");
assert!(!error.contains("callback-code"), "{error}");
}
#[test]
fn token_exchange_rejects_oversized_success_body() {
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let body = "x".repeat(DEFAULT_BOUNDED_BODY_MAX_BYTES as usize + 1);
let body: &'static str = Box::leak(body.into_boxed_str());
let (token_url, _body_rx) = serve_once_with_body("/token", body, 200);
let error = exchange_code(
&format!("{token_url}/token"),
"fake-code",
"http://127.0.0.1:1234/mcp/oauth/callback",
"client-1",
"verifier",
"http://127.0.0.1/mcp",
None,
&client,
)
.unwrap_err()
.to_string();
assert!(
error.contains("MCP OAuth token response exceeded"),
"{error}"
);
}
#[test]
fn discovery_handles_404_as_unavailable() {
let url = serve_once("/.well-known/oauth-protected-resource", "not found", 404);
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
assert!(
discover_protected_resource(&format!("{url}/mcp"), &client)
.unwrap()
.is_none()
);
}
#[test]
fn discovery_uses_endpoint_path_protected_resource_metadata() {
let server = DiscoveryServer::start(vec![
("/.well-known/oauth-protected-resource", 404, "not found"),
(
"/api/.well-known/oauth-protected-resource",
200,
r#"{"resource":"resource-from-path","authorization_servers":["http://auth.example.test"]}"#,
),
]);
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let metadata = discover_protected_resource(&format!("{}/api/mcp", server.base), &client)
.unwrap()
.unwrap();
assert_eq!(metadata.resource, "resource-from-path");
}
#[test]
fn discovery_uses_oidc_authorization_server_fallback() {
let server = DiscoveryServer::start(vec![
("/.well-known/oauth-authorization-server", 404, "not found"),
(
"/.well-known/openid-configuration",
200,
r#"{"issuer":"http://issuer.example.test","authorization_endpoint":"https://issuer.example.test/authorize","token_endpoint":"https://issuer.example.test/token","code_challenge_methods_supported":["S256"]}"#,
),
]);
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let metadata = discover_authorization_server(&server.base, &client).unwrap();
assert_eq!(metadata.issuer, "http://issuer.example.test");
}
#[test]
fn authorization_server_metadata_rejects_non_https_endpoints() {
let metadata = AuthorizationServerMetadata {
issuer: "https://auth.example.test".to_string(),
authorization_endpoint: "http://auth.example.test/authorize".to_string(),
token_endpoint: "https://auth.example.test/token".to_string(),
registration_endpoint: None,
scopes_supported: None,
response_types_supported: None,
grant_types_supported: None,
code_challenge_methods_supported: Some(vec!["S256".to_string()]),
extra: BTreeMap::new(),
};
let error = validate_authorization_server_metadata(&metadata)
.unwrap_err()
.to_string();
assert!(
error.contains("authorization_endpoint must use https"),
"{error}"
);
}
#[test]
fn stored_token_from_response_sets_checked_expiry() {
let token = stored_token_from_response(
TokenResponse {
access_token: "access".to_string(),
token_type: "Bearer".to_string(),
expires_in: Some(3600),
refresh_token: None,
scope: None,
extra: BTreeMap::new(),
},
"client-1",
None,
"https://mcp.example.test/mcp",
None,
None,
None,
None,
)
.unwrap();
assert!(token.expires_at.unwrap() >= token.token_received_at + 3600);
}
#[test]
fn stored_token_from_response_rejects_expiry_overflow() {
match stored_token_from_response(
TokenResponse {
access_token: "access".to_string(),
token_type: "Bearer".to_string(),
expires_in: Some(u64::MAX),
refresh_token: None,
scope: None,
extra: BTreeMap::new(),
},
"client-1",
None,
"https://mcp.example.test/mcp",
None,
None,
None,
None,
) {
Err(McpError::Protocol { code, message }) => {
assert_eq!(code, -32602);
assert_eq!(message, "MCP OAuth token response expires_in is too large");
}
other => panic!("expected protocol error for expires_in overflow, got {other:?}"),
}
}
#[test]
fn concurrent_force_refreshes_keep_token_file_updates_consistent() {
let temp = tempfile::TempDir::new().unwrap();
let token_server = CountingTokenServer::start();
let mut token = stored_token();
token.expires_at = Some(Utc::now().timestamp() - 60);
token.token_endpoint = Some(format!("{}/token", token_server.base));
token.resource = None;
write_token(temp.path(), "remote", &token).unwrap();
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let oauth = McpOAuthConfig {
client_id: Some("client-1".to_string()),
scopes: Vec::new(),
authorization_server: None,
};
let provider_a = TokenProvider::new(
temp.path().to_path_buf(),
"remote".to_string(),
token.server_url.clone(),
oauth.clone(),
client.clone(),
);
let provider_b = TokenProvider::new(
temp.path().to_path_buf(),
"remote".to_string(),
token.server_url.clone(),
oauth,
client,
);
let a = thread::spawn(move || provider_a.force_refresh_access_token().unwrap());
let b = thread::spawn(move || provider_b.force_refresh_access_token().unwrap());
let first = a.join().unwrap();
let second = b.join().unwrap();
assert!(first.starts_with("new-access-token-"), "{first}");
assert!(second.starts_with("new-access-token-"), "{second}");
let refresh_count = *token_server.count.lock().unwrap();
assert!((2..=3).contains(&refresh_count), "{refresh_count}");
assert!(
read_token(temp.path(), "remote")
.unwrap()
.unwrap()
.access_token
.starts_with("new-access-token-")
);
}
#[test]
fn refresh_releases_token_file_lock_during_network_request() {
let temp = tempfile::TempDir::new().unwrap();
let (request_started_tx, request_started_rx) = mpsc::channel();
let (release_response_tx, release_response_rx) = mpsc::channel();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let server = thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut buffer = [0_u8; 4096];
let n = stream.read(&mut buffer).unwrap();
let request = String::from_utf8_lossy(&buffer[..n]);
assert!(request.contains("grant_type=refresh_token"), "{request}");
request_started_tx.send(()).unwrap();
release_response_rx
.recv_timeout(Duration::from_secs(2))
.unwrap();
loopback_write_response(
&mut stream,
"200 OK",
"application/json",
r#"{"access_token":"new-access-token","token_type":"Bearer","expires_in":3600}"#,
);
});
let mut token = stored_token();
token.expires_at = Some(Utc::now().timestamp() - 60);
token.token_endpoint = Some(format!("{base}/token"));
token.resource = None;
write_token(temp.path(), "remote", &token).unwrap();
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let provider = TokenProvider::new(
temp.path().to_path_buf(),
"remote".to_string(),
token.server_url.clone(),
McpOAuthConfig {
client_id: Some("client-1".to_string()),
scopes: Vec::new(),
authorization_server: None,
},
client,
);
let refresh = thread::spawn(move || provider.access_token().unwrap());
request_started_rx
.recv_timeout(Duration::from_secs(2))
.unwrap();
let token_path = token_file_path(temp.path(), "remote").unwrap();
let (lock_acquired_tx, lock_acquired_rx) = mpsc::channel();
let lock = thread::spawn(move || {
let lock = CrossProcessFileLock::acquire(&token_path).unwrap();
lock_acquired_tx.send(()).unwrap();
lock
});
lock_acquired_rx
.recv_timeout(Duration::from_secs(1))
.expect("token file lock should be available while refresh request waits");
let held_lock = lock.join().unwrap();
drop(held_lock);
release_response_tx.send(()).unwrap();
assert_eq!(refresh.join().unwrap(), "new-access-token");
server.join().unwrap();
}
struct DiscoveryServer {
base: String,
handle: Option<thread::JoinHandle<()>>,
}
impl DiscoveryServer {
fn start(routes: Vec<(&'static str, u16, &'static str)>) -> Self {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let handle = thread::spawn(move || {
for _ in 0..routes.len() {
let (mut stream, _) = listener.accept().unwrap();
let mut buffer = [0_u8; 4096];
let n = stream.read(&mut buffer).unwrap();
let request = String::from_utf8_lossy(&buffer[..n]);
let path = request
.lines()
.next()
.and_then(|line| line.split_whitespace().nth(1))
.unwrap_or("");
let (status, body) = routes
.iter()
.find(|(route, _, _)| *route == path)
.map(|(_, status, body)| (*status, *body))
.unwrap_or((404, "not found"));
let reason = if status == 200 { "OK" } else { "Not Found" };
let response = format!(
"HTTP/1.1 {status} {reason}
Content-Type: application/json
Content-Length: {}
Connection: close
{body}",
body.len()
);
stream.write_all(response.as_bytes()).unwrap();
}
});
Self {
base,
handle: Some(handle),
}
}
}
impl Drop for DiscoveryServer {
fn drop(&mut self) {
if let Some(handle) = self.handle.take() {
let _ = handle.join();
}
}
}
struct CountingTokenServer {
base: String,
count: Arc<Mutex<usize>>,
stop: Arc<AtomicBool>,
handle: Option<thread::JoinHandle<()>>,
}
impl CountingTokenServer {
fn start() -> Self {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let count = Arc::new(Mutex::new(0));
let thread_count = Arc::clone(&count);
let stop = Arc::new(AtomicBool::new(false));
let thread_stop = Arc::clone(&stop);
let handle = thread::spawn(move || {
while !thread_stop.load(Ordering::SeqCst) {
match listener.accept() {
Ok((mut stream, _)) => {
stream.set_nonblocking(false).unwrap();
let mut buffer = [0_u8; 4096];
let n = stream.read(&mut buffer).unwrap();
let request = String::from_utf8_lossy(&buffer[..n]);
assert!(request.contains("grant_type=refresh_token"), "{request}");
let mut count = thread_count.lock().unwrap();
*count += 1;
let body = format!(
r#"{{"access_token":"new-access-token-{}","token_type":"Bearer","expires_in":3600}}"#,
*count
);
loopback_write_response(
&mut stream,
"200 OK",
"application/json",
&body,
);
}
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
thread::sleep(Duration::from_millis(10));
}
Err(_) => break,
}
}
});
Self {
base,
count,
stop,
handle: Some(handle),
}
}
}
impl Drop for CountingTokenServer {
fn drop(&mut self) {
self.stop.store(true, Ordering::SeqCst);
if let Ok(url) = reqwest::Url::parse(&self.base)
&& let Ok(mut addrs) = url.socket_addrs(|| None)
&& let Some(addr) = addrs.pop()
{
let _ = std::net::TcpStream::connect(addr);
}
if let Some(handle) = self.handle.take() {
let _ = handle.join();
}
}
}
#[derive(Clone, Copy)]
enum LoopbackMode {
Normal,
BadState,
TokenRevoked,
}
struct LoopbackOAuthMcpServer {
url: String,
stop: Arc<AtomicBool>,
handle: Option<thread::JoinHandle<()>>,
}
impl LoopbackOAuthMcpServer {
fn start(mode: LoopbackMode) -> Self {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let addr = listener.local_addr().unwrap();
let base = format!("http://{addr}");
let stop = Arc::new(AtomicBool::new(false));
let thread_stop = Arc::clone(&stop);
let thread_base = base.clone();
let handle = thread::spawn(move || {
while !thread_stop.load(Ordering::SeqCst) {
match listener.accept() {
Ok((mut stream, _)) => {
handle_loopback_request(&mut stream, &thread_base, mode);
}
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
thread::sleep(Duration::from_millis(10));
}
Err(_) => break,
}
}
});
Self {
url: format!("{base}/mcp"),
stop,
handle: Some(handle),
}
}
}
impl Drop for LoopbackOAuthMcpServer {
fn drop(&mut self) {
self.stop.store(true, Ordering::SeqCst);
if let Ok(parsed) = reqwest::Url::parse(&self.url)
&& let Ok(mut addrs) = parsed.socket_addrs(|| None)
&& let Some(addr) = addrs.pop()
{
let _ = TcpStream::connect(addr);
}
if let Some(handle) = self.handle.take() {
let _ = handle.join();
}
}
}
fn loopback_write_response(
stream: &mut TcpStream,
status: &str,
content_type: &str,
body: &str,
) {
loopback_write_response_with_headers(stream, status, content_type, body, &[]);
}
fn loopback_write_response_with_headers(
stream: &mut TcpStream,
status: &str,
content_type: &str,
body: &str,
extra: &[&str],
) {
let mut response = format!(
"HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n",
body.len()
);
for header in extra {
response.push_str(header);
response.push_str("\r\n");
}
response.push_str("\r\n");
response.push_str(body);
stream.write_all(response.as_bytes()).unwrap();
}
fn handle_loopback_request(stream: &mut TcpStream, base: &str, mode: LoopbackMode) {
let mut buf = [0_u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
if n == 0 {
return;
}
let request = String::from_utf8_lossy(&buf[..n]);
let head = request.split("\r\n\r\n").next().unwrap_or("");
let body = request.split("\r\n\r\n").nth(1).unwrap_or("");
let first = head.lines().next().unwrap_or("");
let path = first.split_whitespace().nth(1).unwrap_or("");
match path.split('?').next().unwrap_or(path) {
"/.well-known/oauth-protected-resource" => loopback_write_response(
stream,
"200 OK",
"application/json",
&format!(r#"{{"resource":"{base}/mcp","authorization_servers":["{base}"]}}"#),
),
"/.well-known/oauth-authorization-server" => loopback_write_response(
stream,
"200 OK",
"application/json",
&format!(
r#"{{"issuer":"{base}","authorization_endpoint":"{base}/authorize","token_endpoint":"{base}/token","registration_endpoint":"{base}/register","scopes_supported":["read","tools"],"code_challenge_methods_supported":["S256"]}}"#
),
),
"/register" => loopback_write_response(
stream,
"201 Created",
"application/json",
r#"{"client_id":"loopback-client","client_secret":"fake_client_secret_abc"}"#,
),
"/authorize" => {
let params = path
.split_once('?')
.map(|(_, q)| parse_query(q).unwrap())
.unwrap_or_default();
let redirect_uri = params
.iter()
.find(|(k, _)| k == "redirect_uri")
.unwrap()
.1
.clone();
let state = params.iter().find(|(k, _)| k == "state").unwrap().1.clone();
let returned_state = if matches!(mode, LoopbackMode::BadState) {
"wrong_state"
} else {
&state
};
let location = format!(
"{redirect_uri}?code=fake_authorization_code_123&state={returned_state}"
);
loopback_write_response_with_headers(
stream,
"302 Found",
"text/plain",
"",
&[&format!("Location: {location}")],
);
}
"/token" if body.contains("grant_type=authorization_code") => loopback_write_response(
stream,
"200 OK",
"application/json",
r#"{"access_token":"fake_access_token_12345","refresh_token":"fake_refresh_abc","token_type":"Bearer","expires_in":3600,"scope":"read tools"}"#,
),
"/token"
if body.contains("grant_type=refresh_token")
&& matches!(mode, LoopbackMode::TokenRevoked) =>
{
loopback_write_response(
stream,
"400 Bad Request",
"application/json",
r#"{"error":"invalid_grant","error_description":"fake_refresh_abc revoked"}"#,
)
}
"/token" if body.contains("grant_type=refresh_token") => loopback_write_response(
stream,
"200 OK",
"application/json",
r#"{"access_token":"fake_access_token_refreshed","token_type":"Bearer","expires_in":3600}"#,
),
"/mcp" => {
if first.starts_with("DELETE ") {
loopback_write_response(stream, "204 No Content", "text/plain", "");
return;
}
if first.starts_with("GET ") {
loopback_write_response(
stream,
"405 Method Not Allowed",
"text/plain",
"no stream",
);
return;
}
let lower_head = head.to_ascii_lowercase();
if !lower_head.contains("authorization: bearer fake_access_token_12345")
&& !lower_head.contains("authorization: bearer fake_access_token_refreshed")
{
loopback_write_response_with_headers(
stream,
"401 Unauthorized",
"text/plain",
"unauthorized",
&[&format!(
"WWW-Authenticate: Bearer resource_metadata=\"{base}/.well-known/oauth-protected-resource\""
)],
);
return;
}
let value: Value = serde_json::from_str(body).unwrap();
let id = value.get("id").cloned().unwrap_or(serde_json::json!(1));
match value.get("method").and_then(Value::as_str).unwrap_or("") {
"initialize" => loopback_write_response(
stream,
"200 OK",
"application/json",
&format!(
r#"{{"jsonrpc":"2.0","id":{id},"result":{{"protocolVersion":"2025-03-26","capabilities":{{"tools":{{}}}},"serverInfo":{{"name":"loopback-oauth","version":"1"}}}}}}"#
),
),
"notifications/initialized" => {
loopback_write_response(stream, "204 No Content", "text/plain", "")
}
"tools/list" => loopback_write_response(
stream,
"200 OK",
"application/json",
&format!(
r#"{{"jsonrpc":"2.0","id":{id},"result":{{"tools":[{{"name":"echo","description":"Echo","inputSchema":{{"type":"object"}}}}]}}}}"#
),
),
"tools/call" => loopback_write_response(
stream,
"200 OK",
"application/json",
&format!(
r#"{{"jsonrpc":"2.0","id":{id},"result":{{"content":[{{"type":"text","text":"ok"}}],"isError":false}}}}"#
),
),
_ => loopback_write_response(stream, "404 Not Found", "text/plain", "missing"),
}
}
_ => loopback_write_response(stream, "404 Not Found", "text/plain", "missing"),
}
}
#[test]
fn loopback_oauth_fixture_full_login_mcp_request_logout_and_secret_non_leak() {
let server = LoopbackOAuthMcpServer::start(LoopbackMode::Normal);
let temp = tempfile::TempDir::new().unwrap();
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(5))
.redirect(reqwest::redirect::Policy::none())
.build()
.unwrap();
let redirect_uri = "http://127.0.0.1:7777/mcp/oauth/callback";
let oauth = McpOAuthConfig {
client_id: None,
scopes: vec!["read".to_string(), "tools".to_string()],
authorization_server: None,
};
let login = prepare_login(&server.url, &oauth, redirect_uri, &client).unwrap();
let redirect = client
.get(&login.authorization_url)
.send()
.unwrap()
.headers()
.get(reqwest::header::LOCATION)
.unwrap()
.to_str()
.unwrap()
.to_string();
let code = parse_manual_fallback_input(&redirect, &login.state).unwrap();
let token_response = exchange_code(
&login.metadata.token_endpoint,
&code,
&login.redirect_uri,
&login.client_id,
&login.verifier,
&login.resource,
login.client_secret.as_deref(),
&client,
)
.unwrap();
let stored = stored_token_from_response(
token_response,
&login.client_id,
login.client_secret,
&server.url,
Some(login.authorization_server),
Some(login.metadata.issuer),
Some(login.metadata.token_endpoint),
Some(login.resource),
)
.unwrap();
write_token(temp.path(), "remote", &stored).unwrap();
let config = crate::config::McpServerConfig::Http(crate::config::McpHttpServerConfig {
url: server.url.clone(),
headers: BTreeMap::new(),
oauth: Some(oauth),
enabled: true,
timeout: Some(5),
});
let mut mcp =
crate::mcp::McpClient::connect_named(Some("remote"), &config, Some(temp.path()))
.unwrap();
mcp.send_request("initialize", Some(serde_json::json!({})))
.unwrap();
let tools = mcp.list_tools().unwrap();
assert_eq!(tools[0].name, "echo");
mcp.shutdown();
delete_token(temp.path(), "remote").unwrap();
assert!(read_token(temp.path(), "remote").unwrap().is_none());
let combined = format!("{stored:?}");
for secret in [
"fake_access_token_12345",
"fake_refresh_abc",
"fake_client_secret_abc",
"fake_authorization_code_123",
] {
assert!(!combined.contains(secret), "{combined}");
}
}
#[test]
fn loopback_oauth_fixture_bad_state_and_revoked_refresh_are_actionable_and_redacted() {
let bad = LoopbackOAuthMcpServer::start(LoopbackMode::BadState);
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(5))
.redirect(reqwest::redirect::Policy::none())
.build()
.unwrap();
let oauth = McpOAuthConfig {
client_id: None,
scopes: Vec::new(),
authorization_server: None,
};
let login = prepare_login(
&bad.url,
&oauth,
"http://127.0.0.1:7777/mcp/oauth/callback",
&client,
)
.unwrap();
let redirect = client
.get(&login.authorization_url)
.send()
.unwrap()
.headers()
.get(reqwest::header::LOCATION)
.unwrap()
.to_str()
.unwrap()
.to_string();
let error = parse_manual_fallback_input(&redirect, &login.state)
.unwrap_err()
.to_string();
assert!(error.contains("state mismatch"), "{error}");
assert!(!error.contains("fake_authorization_code_123"), "{error}");
let revoked = LoopbackOAuthMcpServer::start(LoopbackMode::TokenRevoked);
let token_endpoint = format!("{}/token", revoked.url.trim_end_matches("/mcp"));
let err = refresh_token(
&token_endpoint,
"fake_refresh_abc",
"client",
&revoked.url,
None,
&client,
)
.unwrap_err()
.to_string();
assert!(err.contains("refresh token expired or revoked"), "{err}");
assert!(!err.contains("fake_refresh_abc"), "{err}");
}
fn serve_once(expected_path: &'static str, body: &'static str, status: u16) -> String {
serve_once_with_body(expected_path, body, status).0
}
fn serve_once_with_body(
expected_path: &'static str,
body: &'static str,
status: u16,
) -> (String, mpsc::Receiver<String>) {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut buffer = [0_u8; 4096];
let n = stream.read(&mut buffer).unwrap();
let request = String::from_utf8_lossy(&buffer[..n]);
let first_line = request.lines().next().unwrap_or_default();
assert!(first_line.contains(expected_path), "{first_line}");
if let Some((_, body)) = request.split_once("\r\n\r\n") {
let _ = tx.send(body.to_string());
}
let reason = if status == 200 || status == 201 {
"OK"
} else {
"Not Found"
};
let response = format!(
"HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
stream.write_all(response.as_bytes()).unwrap();
});
(format!("http://{addr}"), rx)
}
}