oauth.rs - CodeWhale

oauth.rs

1	use super::http_client::McpHttpClient;
2	use crate::network_policy::NetworkPolicyDecider;
3	use std::collections::HashMap;
4	use std::sync::Arc;
5	use std::time::{Duration, SystemTime, UNIX_EPOCH};
6
7	use anyhow::{Context, Result, anyhow, bail};
8	use base64::Engine as _;
9	use base64::engine::general_purpose::URL_SAFE_NO_PAD;
10	use oauth2::TokenResponse;
11	use reqwest::Url;
12	use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue};
13	use rmcp::transport::AuthorizationManager;
14	use rmcp::transport::AuthorizationSession;
15	use rmcp::transport::auth::{
16	AuthError, AuthorizationRequest, OAuthClientConfig, OAuthHttpClient, OAuthHttpClientError,
17	OAuthHttpClientFuture, OAuthHttpRedirectPolicy, OAuthHttpRequest, OAuthState,
18	OAuthTokenResponse,
19	};
20	use serde::{Deserialize, Serialize};
21	use sha2::{Digest, Sha256};
22	use tokio::io::{AsyncReadExt, AsyncWriteExt};
23	use tokio::net::TcpListener;
24	use tokio::sync::{Mutex, oneshot};
25	use tokio::time::timeout;
26	use tokio_util::sync::CancellationToken;
27	use urlencoding::decode;
28
29	use super::McpServerConfig;
30
31	const REFRESH_SKEW_MILLIS: u64 = 30_000;
32
33	#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
34	#[serde(rename_all = "snake_case")]
35	pub enum McpAuthStatus {
36	Unsupported,
37	NotLoggedIn,
38	BearerToken,
39	OAuth,
40	}
41
42	impl std::fmt::Display for McpAuthStatus {
43	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
44	let text = match self {
45	Self::Unsupported => "Unsupported",
46	Self::NotLoggedIn => "Not logged in",
47	Self::BearerToken => "Bearer token",
48	Self::OAuth => "OAuth",
49	};
50	f.write_str(text)
51	}
52	}
53
54	/// Context for a failed token refresh. An auth-required failure already
55	/// flips the server to `◆ auth required` and offers the login tool; any
56	/// other failure (a token endpoint answering something the client could
57	/// not parse, a transport error) names the same remedy in words, because
58	/// the operator otherwise sees only the provider's parse error (#5926).
59	/// When the token endpoint did answer, its receipt (status line,
60	/// content-type, masked excerpt) rides along so a provider outage — an HTML
61	/// 502 page — reads differently from a parser defect on JSON it should
62	/// have accepted.
63	fn refresh_failure_context(
64	server_name: &str,
65	names_remedy: bool,
66	receipt: Option<&TokenEndpointReceipt>,
67	) -> String {
68	if names_remedy {
69	let answered =
70	receipt.map_or_else(String::new, \|receipt\| format!(" (it answered {receipt})"));
71	format!(
72	"refreshing MCP OAuth token for server {server_name}: the token endpoint did not answer the way the client expects{answered}; \
73	if this persists, run `codewhale mcp login {server_name}` (or `/mcp login {server_name}`) to re-authorize"
74	)
75	} else {
76	format!("refreshing MCP OAuth token for server {server_name}")
77	}
78	}
79
80	/// Longest excerpt of a token-endpoint body a refresh failure keeps.
81	const TOKEN_RECEIPT_EXCERPT_BYTES: usize = 200;
82
83	/// rmcp's own cap on an OAuth response body, mirrored so the recording
84	/// client refuses the same oversized answers the stock one does.
85	const MAX_OAUTH_HTTP_RESPONSE_BODY_BYTES: usize = 1024 * 1024;
86
87	/// Response fields whose values are credentials. Their values are masked
88	/// before any body excerpt is kept; the field names themselves are not
89	/// secrets and stay so the operator can see which fields the answer had.
90	const OAUTH_SECRET_FIELDS: &[&str] = &[
91	"access_token",
92	"refresh_token",
93	"client_secret",
94	"id_token",
95	"authorization",
96	];
97
98	/// What the token endpoint actually answered. rmcp collapses an
99	/// unparseable answer to `Failed to parse server response` and drops the
100	/// body; this is the receipt it drops, with every credential-shaped value
101	/// masked and the body cut to its first [`TOKEN_RECEIPT_EXCERPT_BYTES`].
102	#[derive(Debug, Clone, PartialEq, Eq)]
103	pub(crate) struct TokenEndpointReceipt {
104	status: u16,
105	reason: Option<&'static str>,
106	content_type: Option<String>,
107	excerpt: String,
108	}
109
110	impl TokenEndpointReceipt {
111	fn from_response(status: reqwest::StatusCode, headers: &HeaderMap, body: &[u8]) -> Self {
112	let content_type = headers
113	.get(CONTENT_TYPE)
114	.and_then(\|value\| value.to_str().ok())
115	.map(str::trim)
116	.filter(\|value\| !value.is_empty())
117	.map(str::to_string);
118	Self {
119	status: status.as_u16(),
120	reason: status.canonical_reason(),
121	content_type,
122	excerpt: token_response_excerpt(body),
123	}
124	}
125	}
126
127	impl std::fmt::Display for TokenEndpointReceipt {
128	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
129	write!(f, "HTTP {}", self.status)?;
130	if let Some(reason) = self.reason {
131	write!(f, " {reason}")?;
132	}
133	match self.content_type.as_deref() {
134	Some(content_type) => write!(f, " ({content_type})")?,
135	None => f.write_str(" (no content-type)")?,
136	}
137	if self.excerpt.is_empty() {
138	f.write_str(" with an empty body")
139	} else {
140	write!(f, ": {}", self.excerpt)
141	}
142	}
143	}
144
145	/// Masked, whitespace-collapsed, byte-capped excerpt of a token-endpoint
146	/// body. Masking runs before the cut so a truncated credential never leaks
147	/// its prefix.
148	fn token_response_excerpt(body: &[u8]) -> String {
149	let masked = mask_oauth_secrets(&String::from_utf8_lossy(body));
150	let collapsed = masked.split_whitespace().collect::<Vec<_>>().join(" ");
151	if collapsed.len() <= TOKEN_RECEIPT_EXCERPT_BYTES {
152	return collapsed;
153	}
154	let mut end = TOKEN_RECEIPT_EXCERPT_BYTES;
155	while !collapsed.is_char_boundary(end) {
156	end -= 1;
157	}
158	format!("{}…", &collapsed[..end])
159	}
160
161	fn is_word_byte(byte: u8) -> bool {
162	byte.is_ascii_alphanumeric() \|\| byte == b'_'
163	}
164
165	/// Replace every credential-shaped value in `text` with `***`: JSON
166	/// members (`"access_token": "…"`), form/query pairs (`refresh_token=…`),
167	/// and bearer schemes (`Bearer …`). Field names, separators and everything
168	/// else survive so the shape of the answer stays readable.
169	pub(crate) fn mask_oauth_secrets(text: &str) -> String {
170	let lower = text.to_ascii_lowercase();
171	let bytes = text.as_bytes();
172	let mut out = String::with_capacity(text.len());
173	let mut index = 0;
174	while index < text.len() {
175	let at_word_start = index == 0 \|\| !is_word_byte(bytes[index - 1]);
176	if at_word_start
177	&& let Some((value_start, value_end)) = secret_value_span(text, &lower, index)
178	{
179	out.push_str(&text[index..value_start]);
180	out.push_str("***");
181	index = value_end;
182	continue;
183	}
184	let ch = text[index..]
185	.chars()
186	.next()
187	.expect("index sits on a char boundary");
188	out.push(ch);
189	index += ch.len_utf8();
190	}
191	out
192	}
193
194	/// The byte span of the secret value that starts at `start`, if a secret
195	/// field or bearer scheme begins there. Every scan step consumes ASCII
196	/// bytes only, so both ends land on char boundaries.
197	fn secret_value_span(text: &str, lower: &str, start: usize) -> Option<(usize, usize)> {
198	let bytes = text.as_bytes();
199	let skip_spaces = \|mut cursor: usize\| {
200	while bytes
201	.get(cursor)
202	.is_some_and(\|byte\| byte == b' ' \|\| byte == b'\t')
203	{
204	cursor += 1;
205	}
206	cursor
207	};
208	let unquoted_end = \|mut cursor: usize\| {
209	while bytes.get(cursor).is_some_and(\|byte\| {
210	!matches!(byte, b'&' \| b',' \| b';' \| b'}' \| b'"' \| b'\'') && !byte.is_ascii_whitespace()
211	}) {
212	cursor += 1;
213	}
214	cursor
215	};
216	if lower[start..].starts_with("bearer ") {
217	let value_start = skip_spaces(start + "bearer".len());
218	let value_end = unquoted_end(value_start);
219	return (value_end > value_start).then_some((value_start, value_end));
220	}
221	for field in OAUTH_SECRET_FIELDS {
222	if !lower[start..].starts_with(field) {
223	continue;
224	}
225	let mut cursor = start + field.len();
226	if bytes.get(cursor).is_some_and(\|byte\| is_word_byte(*byte)) {
227	continue;
228	}
229	if bytes.get(cursor) == Some(&b'"') {
230	cursor += 1;
231	}
232	cursor = skip_spaces(cursor);
233	match bytes.get(cursor) {
234	Some(b':' \| b'=') => cursor += 1,
235	_ => continue,
236	}
237	cursor = skip_spaces(cursor);
238	if bytes.get(cursor) == Some(&b'"') {
239	let value_start = cursor + 1;
240	let mut value_end = value_start;
241	while let Some(byte) = bytes.get(value_end) {
242	match byte {
243	b'\\' => value_end += 2,
244	b'"' => break,
245	_ => value_end += 1,
246	}
247	}
248	return Some((value_start, value_end.min(text.len())));
249	}
250	// An unquoted `Authorization: Bearer <token>` carries its scheme in
251	// front of the credential; the whole value is the secret.
252	let value_start = cursor;
253	let mut value_end = unquoted_end(cursor);
254	if matches!(lower[value_start..value_end].as_ref(), "bearer" \| "basic")
255	&& bytes.get(value_end) == Some(&b' ')
256	{
257	value_end = unquoted_end(skip_spaces(value_end));
258	}
259	return Some((value_start, value_end));
260	}
261	None
262	}
263
264	/// Shared guarded HTTP client for discovery, login and stored credentials.
265	/// It honors each OAuth operation's redirect policy, caps response bodies and keeps
266	/// the receipt of the latest token-endpoint answer (every token request is
267	/// a POST; discovery is GET) so a failed refresh can say what came back.
268	pub(crate) struct RecordingOAuthHttpClient {
269	client: McpHttpClient,
270	last_token_response: std::sync::Mutex<Option<TokenEndpointReceipt>>,
271	}
272
273	impl RecordingOAuthHttpClient {
274	fn new(client: McpHttpClient) -> Self {
275	Self {
276	client,
277	last_token_response: std::sync::Mutex::new(None),
278	}
279	}
280
281	fn take_token_endpoint_receipt(&self) -> Option<TokenEndpointReceipt> {
282	self.last_token_response
283	.lock()
284	.unwrap_or_else(std::sync::PoisonError::into_inner)
285	.take()
286	}
287	}
288
289	impl OAuthHttpClient for RecordingOAuthHttpClient {
290	fn execute(&self, request: OAuthHttpRequest) -> OAuthHttpClientFuture<'_> {
291	Box::pin(async move {
292	let OAuthHttpRequest {
293	request,
294	timeout,
295	redirect_policy,
296	..
297	} = request;
298	let is_token_request = request.method() == reqwest::Method::POST;
299	let mut request = reqwest::Request::try_from(request)
300	.map_err(\|error\| Box::new(error) as OAuthHttpClientError)?;
301	if let Some(timeout) = timeout {
302	*request.timeout_mut() = Some(timeout);
303	}
304	let mut response = self
305	.client
306	.execute(
307	request,
308	matches!(redirect_policy, OAuthHttpRedirectPolicy::Follow),
309	)
310	.await
311	.map_err(\|error\| -> OAuthHttpClientError { error.into() })?;
312	let status = response.status();
313	let version = response.version();
314	let headers = response.headers().clone();
315	let mut body = Vec::new();
316	while let Some(chunk) = response
317	.chunk()
318	.await
319	.map_err(\|error\| Box::new(error) as OAuthHttpClientError)?
320	{
321	if chunk.len() > MAX_OAUTH_HTTP_RESPONSE_BODY_BYTES - body.len() {
322	return Err(anyhow!(
323	"OAuth HTTP response body exceeds {MAX_OAUTH_HTTP_RESPONSE_BODY_BYTES} bytes"
324	)
325	.into());
326	}
327	body.extend_from_slice(&chunk);
328	}
329	if is_token_request {
330	*self
331	.last_token_response
332	.lock()
333	.unwrap_or_else(std::sync::PoisonError::into_inner) =
334	Some(TokenEndpointReceipt::from_response(status, &headers, &body));
335	}
336	let mut builder = oauth2::http::Response::builder()
337	.status(status)
338	.version(version);
339	for (name, value) in &headers {
340	builder = builder.header(name, value);
341	}
342	builder
343	.body(body)
344	.map_err(\|error\| Box::new(error) as OAuthHttpClientError)
345	})
346	}
347	}
348
349	pub fn error_looks_auth_required(error: &anyhow::Error) -> bool {
350	error_text_looks_auth_required(&format!("{error:#}"))
351	}
352
353	/// Whether the error chain carries the OAuth `invalid_grant` code: the
354	/// authorization server definitively rejected the presented grant (typically
355	/// a stale or already-rotated refresh token).
356	fn error_is_invalid_grant(error: &anyhow::Error) -> bool {
357	format!("{error:#}")
358	.to_ascii_lowercase()
359	.contains("invalid_grant")
360	}
361
362	/// The one auth-required classifier every surface consults: the pool's
363	/// `◆ auth required` state, the session-boot row, the `/mcp` manager
364	/// recovery verb, and the synthetic `mcp_<server>_authenticate` tool all
365	/// derive from this predicate so a failure is never "needs login" on one
366	/// surface and "failed" on another. `invalid_grant` belongs here because the
367	/// authorization server has definitively rejected the stored grant — only a
368	/// fresh login recovers it.
369	pub fn error_text_looks_auth_required(text: &str) -> bool {
370	let text = text.to_ascii_lowercase();
371	// `auth required` and `requires oauth` are anchored to the shapes this
372	// product and the Codex-compatible managers actually emit (`◆ auth
373	// required`, `requires OAuth login/authentication/reauthentication`) —
374	// bare substrings would misclassify incidental server errors like
375	// "auth required parameter is missing".
376	text.contains("401")
377	\|\| text.contains("unauthorized")
378	\|\| text.contains("authentication_required")
379	\|\| text.contains("invalid_grant")
380	// rmcp 3.2 collapsed several unrecoverable refresh outcomes onto
381	// `AuthError::AuthorizationRequired` (Display: "OAuth authorization
382	// required") — a stored credential with no usable refresh grant, and
383	// every refresh the server definitively rejected. In 2.2 those
384	// arrived as `TokenRefreshFailed("No refresh token available")`, which
385	// no surface recognised. The full phrase is matched so it stays
386	// anchored to rmcp's own wording.
387	\|\| text.contains("oauth authorization required")
388	\|\| text.contains("◆ auth required")
389	\|\| text.contains("requires oauth login")
390	\|\| text.contains("requires oauth authentication")
391	\|\| text.contains("requires oauth reauthentication")
392	\|\| text.contains("not logged in")
393	\|\| text.contains("not-logged-in")
394	\|\| text.contains("re-authorize")
395	\|\| text.contains("/mcp login")
396	\|\| text.contains("mcp login")
397	}
398
399	pub fn auth_required_login_hint(server_name: &str) -> String {
400	format!(
401	"MCP server '{server_name}' requires OAuth authentication. Run `codewhale mcp login {server_name}` to authenticate."
402	)
403	}
404
405	/// The one recovery sentence for a server in the `◆ auth required` state,
406	/// chosen by how that server is allowed to authenticate. OAuth-servable
407	/// servers get the login command; plugin-contributed servers (OAuth is
408	/// disabled for them by review policy) and servers with a manual
409	/// Authorization configuration are told which environment-backed
410	/// credential to supply instead, so `/mcp login` is never named for a
411	/// server it would refuse. Environment variable names are not secrets;
412	/// their values never appear here.
413	pub(crate) fn auth_required_recovery_hint(server_name: &str, server: &McpServerConfig) -> String {
414	let mut env_vars: Vec<&str> = server
415	.env_headers
416	.values()
417	.map(String::as_str)
418	.chain(server.bearer_token_env_var.as_deref())
419	.collect();
420	env_vars.sort_unstable();
421	env_vars.dedup();
422	let credential_source = if env_vars.is_empty() {
423	"its configured Authorization header".to_string()
424	} else {
425	format!(
426	"the environment variable{} {}",
427	if env_vars.len() == 1 { "" } else { "s" },
428	env_vars.join(", ")
429	)
430	};
431	if let Some(source) = server.reviewed_plugin.as_ref() {
432	return format!(
433	"MCP server '{server_name}' is contributed by plugin '{}' and its credential comes from {credential_source} (OAuth login is disabled for plugin-contributed servers). Set the credential, then run `/mcp reload`.",
434	source.authority.plugin_name
435	);
436	}
437	if server_has_manual_authorization(server) {
438	return format!(
439	"MCP server '{server_name}' authenticates with {credential_source}; the server rejected that credential. Correct it, then run `/mcp reload`."
440	);
441	}
442	auth_required_login_hint(server_name)
443	}
444
445	/// TUI recovery for a stale Streamable HTTP OAuth session. `/mcp auth` is not a
446	/// command; login is `/mcp login <name>` (CLI: `codewhale mcp login <name>`).
447	pub fn tui_reauth_hint() -> &'static str {
448	"Re-authorize this server (/mcp login <name>) to continue."
449	}
450
451	pub fn tui_reauth_refresh_failed_hint() -> &'static str {
452	"Re-authorize this server (/mcp login <name>) or configure a fresh bearer token."
453	}
454
455	#[derive(Debug, Clone, Serialize, Deserialize)]
456	pub struct StoredMcpOAuthTokens {
457	pub server_name: String,
458	pub url: String,
459	pub client_id: String,
460	pub token_response: WrappedOAuthTokenResponse,
461	#[serde(default)]
462	pub expires_at: Option<u64>,
463	}
464
465	impl PartialEq for StoredMcpOAuthTokens {
466	fn eq(&self, other: &Self) -> bool {
467	if self.server_name != other.server_name
468	\|\| self.url != other.url
469	\|\| self.client_id != other.client_id
470	\|\| self.expires_at != other.expires_at
471	{
472	return false;
473	}
474	if self.expires_at.is_none() {
475	return self.token_response == other.token_response;
476	}
477	// Loading a credential derives a decreasing expires_in from the
478	// durable expires_at. That countdown is not a peer token rotation:
479	// comparing it would adopt the same rejected grant after one second
480	// instead of invalidating it. Preserve every other response field.
481	let mut left = self.token_response.clone();
482	let mut right = other.token_response.clone();
483	left.0.set_expires_in(None);
484	right.0.set_expires_in(None);
485	left == right
486	}
487	}
488
489	#[derive(Debug, Clone, Serialize, Deserialize)]
490	pub struct WrappedOAuthTokenResponse(pub OAuthTokenResponse);
491
492	impl PartialEq for WrappedOAuthTokenResponse {
493	fn eq(&self, other: &Self) -> bool {
494	match (serde_json::to_string(self), serde_json::to_string(other)) {
495	(Ok(left), Ok(right)) => left == right,
496	_ => false,
497	}
498	}
499	}
500
501	#[derive(Clone)]
502	pub struct McpOAuthRuntime {
503	inner: Arc<McpOAuthRuntimeInner>,
504	}
505
506	struct McpOAuthRuntimeInner {
507	server_name: String,
508	url: String,
509	manager: Arc<Mutex<AuthorizationManager>>,
510	last_tokens: Mutex<Option<StoredMcpOAuthTokens>>,
511	/// Why the held credential was invalidated (the provider's error code,
512	/// e.g. `invalid_grant`), so every later failure names the cause even
513	/// though the rejected grant is never replayed. `None` while a
514	/// credential is held.
515	rejection: Mutex<Option<String>>,
516	/// The HTTP client the runtime was built with, shared with the manager
517	/// so an adopted on-disk rotation rebuilds it with identical HTTP shape
518	/// and a failed refresh can read the token endpoint's receipt.
519	http_client: Arc<RecordingOAuthHttpClient>,
520	}
521
522	#[derive(Debug, Clone, PartialEq, Eq)]
523	pub struct McpOAuthDiscovery {
524	pub scopes_supported: Option<Vec<String>>,
525	}
526
527	#[derive(Debug, Clone, PartialEq, Eq)]
528	pub struct ResolvedMcpOAuthScopes {
529	pub scopes: Vec<String>,
530	pub source: McpOAuthScopesSource,
531	}
532
533	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
534	pub enum McpOAuthScopesSource {
535	Explicit,
536	Configured,
537	Discovered,
538	Empty,
539	}
540
541	#[derive(Debug, Clone, PartialEq, Eq)]
542	pub struct OAuthProviderError {
543	error: Option<String>,
544	error_description: Option<String>,
545	}
546
547	impl OAuthProviderError {
548	fn new(error: Option<String>, error_description: Option<String>) -> Self {
549	Self {
550	error,
551	error_description,
552	}
553	}
554	}
555
556	impl std::fmt::Display for OAuthProviderError {
557	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
558	match (self.error.as_deref(), self.error_description.as_deref()) {
559	(Some(error), Some(description)) => {
560	write!(f, "OAuth provider returned `{error}`: {description}")
561	}
562	(Some(error), None) => write!(f, "OAuth provider returned `{error}`"),
563	(None, Some(description)) => write!(f, "OAuth error: {description}"),
564	(None, None) => write!(f, "OAuth provider returned an error"),
565	}
566	}
567	}
568
569	impl std::error::Error for OAuthProviderError {}
570
571	/// Build an `AuthorizationManager` preloaded with stored credentials, the
572	/// shared construction step for initial load and for adopting a credential
573	/// that another process rotated on disk.
574	async fn manager_from_stored_tokens(
575	url: &str,
576	tokens: &StoredMcpOAuthTokens,
577	http_client: &Arc<RecordingOAuthHttpClient>,
578	) -> Result<AuthorizationManager> {
579	let client = Arc::clone(http_client) as Arc<dyn OAuthHttpClient>;
580	let mut state = OAuthState::new_with_oauth_http_client(url.to_string(), client).await?;
581	state
582	.set_credentials(&tokens.client_id, tokens.token_response.0.clone())
583	.await
584	.context("installing stored MCP OAuth credentials")?;
585
586	match state {
587	OAuthState::Authorized(manager) \| OAuthState::Unauthorized(manager) => Ok(manager),
588	_ => bail!("unexpected MCP OAuth state while preparing stored credentials"),
589	}
590	}
591
592	impl McpOAuthRuntime {
593	#[cfg(test)]
594	pub(super) async fn from_server_config(
595	server_name: &str,
596	server: &McpServerConfig,
597	default_headers: HeaderMap,
598	) -> Result<Option<Self>> {
599	if server.reviewed_plugin.is_some() \|\| server_has_manual_authorization(server) {
600	return Ok(None);
601	}
602	let Some(url) = server.url.as_deref() else {
603	return Ok(None);
604	};
605	let client = oauth_http_client(server, url, None)?;
606	Self::from_server_config_with_client(server_name, server, default_headers, client).await
607	}
608
609	pub(super) async fn from_server_config_with_client(
610	server_name: &str,
611	server: &McpServerConfig,
612	default_headers: HeaderMap,
613	client: McpHttpClient,
614	) -> Result<Option<Self>> {
615	if server.reviewed_plugin.is_some() {
616	return Ok(None);
617	}
618	let Some(url) = server.url.as_deref() else {
619	return Ok(None);
620	};
621	if server_has_manual_authorization(server) {
622	return Ok(None);
623	}
624	let Some(tokens) = load_oauth_tokens(server_name, url)? else {
625	return Ok(None);
626	};
627	Self::from_stored_tokens(
628	server_name,
629	url,
630	tokens,
631	client.with_default_headers(default_headers),
632	)
633	.await
634	.map(Some)
635	}
636
637	async fn from_stored_tokens(
638	server_name: &str,
639	url: &str,
640	mut tokens: StoredMcpOAuthTokens,
641	client: McpHttpClient,
642	) -> Result<Self> {
643	refresh_expires_in_from_timestamp(&mut tokens);
644	let http_client = Arc::new(RecordingOAuthHttpClient::new(client));
645	let manager = manager_from_stored_tokens(url, &tokens, &http_client).await?;
646
647	Ok(Self {
648	inner: Arc::new(McpOAuthRuntimeInner {
649	server_name: server_name.to_string(),
650	url: url.to_string(),
651	manager: Arc::new(Mutex::new(manager)),
652	last_tokens: Mutex::new(Some(tokens)),
653	rejection: Mutex::new(None),
654	http_client,
655	}),
656	})
657	}
658
659	pub async fn authorization_header(&self) -> Result<Option<String>> {
660	self.refresh_if_needed().await?;
661	// Never send a credential the provider already rejected; the request
662	// goes out unauthenticated and the server's 401 drives the reactive
663	// refresh, which adopts a peer's login or reports auth-required.
664	if self.is_invalidated().await {
665	return Ok(None);
666	}
667	let credentials = {
668	let guard = self.inner.manager.lock().await;
669	let (_client_id, credentials) = guard
670	.get_credentials()
671	.await
672	.context("reading MCP OAuth credentials")?;
673	credentials
674	};
675	let Some(credentials) = credentials else {
676	return Ok(None);
677	};
678	let token = credentials.access_token().secret().trim();
679	if token.is_empty() {
680	Ok(None)
681	} else {
682	Ok(Some(format!("Bearer {token}")))
683	}
684	}
685
686	async fn refresh_if_needed(&self) -> Result<()> {
687	let expires_at = {
688	let guard = self.inner.last_tokens.lock().await;
689	guard.as_ref().and_then(\|tokens\| tokens.expires_at)
690	};
691	if !token_needs_refresh(expires_at) {
692	return Ok(());
693	}
694	self.refresh_and_persist().await
695	}
696
697	/// Force a token refresh regardless of the local expiry clock (T4): a
698	/// 401/403 means the server no longer accepts the token — clock skew,
699	/// server-side revocation, or rotation — so the expiry-based gate must
700	/// not decide alone.
701	pub(crate) async fn force_refresh(&self) -> Result<()> {
702	self.refresh_and_persist().await
703	}
704
705	/// Whether this runtime's credential was definitively rejected by the
706	/// provider and invalidated. `last_tokens` is `Some` from construction
707	/// and after every persisted refresh; only [`Self::clear_stored_tokens`]
708	/// empties it.
709	async fn is_invalidated(&self) -> bool {
710	self.inner.last_tokens.lock().await.is_none()
711	}
712
713	async fn refresh_and_persist(&self) -> Result<()> {
714	// A credential the provider definitively rejected is never replayed:
715	// the `AuthorizationManager` still holds it, but every later refresh
716	// with that grant is a guaranteed `invalid_grant`. The only way back
717	// is a credential another process stored since (a completed login),
718	// so adopt that when present and otherwise report auth-required
719	// without touching the token endpoint.
720	if self.is_invalidated().await {
721	if !self.adopt_rotated_on_disk_tokens().await? {
722	let reason = self
723	.inner
724	.rejection
725	.lock()
726	.await
727	.clone()
728	.unwrap_or_else(\|\| "unauthorized".to_string());
729	bail!(
730	"stored MCP OAuth credential for server {} was rejected by the provider ({reason}) and removed; the server requires OAuth login again",
731	self.inner.server_name
732	);
733	}
734	let adopted_needs_refresh = {
735	let last = self.inner.last_tokens.lock().await;
736	token_needs_refresh(last.as_ref().and_then(\|tokens\| tokens.expires_at))
737	};
738	if !adopted_needs_refresh {
739	return Ok(());
740	}
741	}
742	// Only this refresh's answer may explain this refresh's failure.
743	self.inner.http_client.take_token_endpoint_receipt();
744	let mut err = match self.try_refresh_and_persist().await {
745	Ok(()) => return Ok(()),
746	Err(err) => err,
747	};
748	// Refresh-race tolerance: another codewhale process sharing this token
749	// store (a concurrent `mcp login`, or a peer session's refresh) may
750	// have rotated the credential after this runtime loaded its copy, and
751	// single-use refresh tokens then fail here with `invalid_grant`.
752	// Re-read the store once; when the on-disk credential changed, adopt
753	// it — using it directly while fresh, or retrying the refresh exactly
754	// once with the rotated grant — before surfacing failure. When the
755	// store is unchanged the grant is simply dead: the auth-required
756	// branch below invalidates it so the server flips to `◆ auth
757	// required` and the self-serve login tool appears, instead of every
758	// later connect replaying the same rejected refresh.
759	if error_is_invalid_grant(&err) && self.adopt_rotated_on_disk_tokens().await? {
760	let adopted_needs_refresh = {
761	let last = self.inner.last_tokens.lock().await;
762	token_needs_refresh(last.as_ref().and_then(\|tokens\| tokens.expires_at))
763	};
764	if !adopted_needs_refresh {
765	return Ok(());
766	}
767	match self.try_refresh_and_persist().await {
768	Ok(()) => return Ok(()),
769	Err(retry_err) => err = retry_err,
770	}
771	}
772	if error_looks_auth_required(&err) {
773	let reason = if error_is_invalid_grant(&err) {
774	"invalid_grant"
775	} else {
776	"unauthorized"
777	};
778	self.clear_stored_tokens(reason).await?;
779	}
780	let server_name = self.inner.server_name.clone();
781	let names_remedy = !error_looks_auth_required(&err);
782	let receipt = self.inner.http_client.take_token_endpoint_receipt();
783	Err(err)
784	.with_context(\|\| refresh_failure_context(&server_name, names_remedy, receipt.as_ref()))
785	}
786
787	async fn try_refresh_and_persist(&self) -> Result<()> {
788	let refresh_result = {
789	let guard = self.inner.manager.lock().await;
790	guard.refresh_token().await
791	};
792	refresh_result.map_err(\|err\| anyhow!(err))?;
793	self.persist_if_needed().await
794	}
795
796	/// Re-read the on-disk credential after an `invalid_grant` refresh
797	/// failure and, when another process rotated it, rebuild the manager
798	/// around the rotated token exactly like initial construction. Returns
799	/// `true` only when the stored credential actually changed; an unchanged
800	/// store means the failure is ours to report.
801	async fn adopt_rotated_on_disk_tokens(&self) -> Result<bool> {
802	let Some(stored) = load_oauth_tokens(&self.inner.server_name, &self.inner.url)? else {
803	return Ok(false);
804	};
805	let changed = {
806	let last = self.inner.last_tokens.lock().await;
807	last.as_ref() != Some(&stored)
808	};
809	if !changed {
810	return Ok(false);
811	}
812	let manager =
813	manager_from_stored_tokens(&self.inner.url, &stored, &self.inner.http_client).await?;
814	*self.inner.manager.lock().await = manager;
815	*self.inner.last_tokens.lock().await = Some(stored);
816	*self.inner.rejection.lock().await = None;
817	Ok(true)
818	}
819
820	/// Invalidate the credential this runtime holds after the provider
821	/// definitively rejected it. Never deletes a newer durable winner: when
822	/// the on-disk credential no longer matches the one we hold, another
823	/// process rotated it after our copy loaded, and that credential — not
824	/// ours — is the one the next connect must try. The manager is rebuilt
825	/// around that winner exactly like initial construction; remembering
826	/// the rotated token while keeping our dead grant would make the next
827	/// `invalid_grant` compare an "unchanged" store and delete the newer
828	/// valid credential.
829	async fn clear_stored_tokens(&self, reason: &str) -> Result<()> {
830	let held = { self.inner.last_tokens.lock().await.take() };
831	let Some(held) = held else {
832	return Ok(());
833	};
834	match load_oauth_tokens(&self.inner.server_name, &self.inner.url)? {
835	Some(stored) if stored != held => {
836	tracing::debug!(
837	target: "mcp",
838	server = %self.inner.server_name,
839	"MCP OAuth credential was rotated by another process; keeping the on-disk winner"
840	);
841	let manager =
842	manager_from_stored_tokens(&self.inner.url, &stored, &self.inner.http_client)
843	.await?;
844	*self.inner.manager.lock().await = manager;
845	*self.inner.last_tokens.lock().await = Some(stored);
846	*self.inner.rejection.lock().await = None;
847	}
848	_ => {
849	delete_oauth_tokens(&self.inner.server_name, &self.inner.url)?;
850	*self.inner.rejection.lock().await = Some(reason.to_string());
851	}
852	}
853	Ok(())
854	}
855
856	async fn persist_if_needed(&self) -> Result<()> {
857	let (client_id, credentials) = {
858	let guard = self.inner.manager.lock().await;
859	guard
860	.get_credentials()
861	.await
862	.context("reading refreshed MCP OAuth credentials")?
863	};
864	let Some(credentials) = credentials else {
865	let mut last = self.inner.last_tokens.lock().await;
866	if last.take().is_some() {
867	delete_oauth_tokens(&self.inner.server_name, &self.inner.url)?;
868	}
869	return Ok(());
870	};
871
872	let new_response = WrappedOAuthTokenResponse(credentials.clone());
873	let mut last = self.inner.last_tokens.lock().await;
874	let same_token = last
875	.as_ref()
876	.map(\|previous\| previous.token_response == new_response)
877	.unwrap_or(false);
878	let expires_at = if same_token {
879	last.as_ref().and_then(\|previous\| previous.expires_at)
880	} else {
881	compute_expires_at_millis(&credentials)
882	};
883	let stored = StoredMcpOAuthTokens {
884	server_name: self.inner.server_name.clone(),
885	url: self.inner.url.clone(),
886	client_id,
887	token_response: new_response,
888	expires_at,
889	};
890	if last.as_ref() != Some(&stored) {
891	save_oauth_tokens(&stored)?;
892	*last = Some(stored);
893	}
894	Ok(())
895	}
896	}
897
898	pub async fn auth_status_for_server(
899	name: &str,
900	server: &McpServerConfig,
901	network_policy: Option<&NetworkPolicyDecider>,
902	) -> McpAuthStatus {
903	if server.reviewed_plugin.is_some() \|\| !server.is_enabled() \|\| server.url.is_none() {
904	return McpAuthStatus::Unsupported;
905	}
906	if server_has_manual_authorization(server) {
907	return McpAuthStatus::BearerToken;
908	}
909	let Some(url) = server.url.as_deref() else {
910	return McpAuthStatus::Unsupported;
911	};
912	match load_oauth_tokens(name, url) {
913	Ok(Some(tokens)) if oauth_tokens_are_usable(&tokens) => return McpAuthStatus::OAuth,
914	Ok(Some(_)) => return McpAuthStatus::NotLoggedIn,
915	Ok(None) => {}
916	Err(err) => {
917	tracing::warn!(target: "mcp", server = %name, error = %err, "failed to read MCP OAuth tokens");
918	}
919	}
920
921	let headers = match build_default_headers(&server.headers, &server.env_headers) {
922	Ok(headers) => headers,
923	Err(err) => {
924	tracing::warn!(target: "mcp", server = %name, error = %err, "failed to build MCP OAuth discovery headers");
925	return McpAuthStatus::Unsupported;
926	}
927	};
928	match discover_streamable_http_oauth_for_server(server, url, headers, network_policy).await {
929	Ok(Some(_)) => McpAuthStatus::NotLoggedIn,
930	Ok(None) => McpAuthStatus::Unsupported,
931	Err(err) => {
932	tracing::debug!(target: "mcp", server = %name, error = %err, "MCP OAuth discovery failed");
933	McpAuthStatus::Unsupported
934	}
935	}
936	}
937
938	pub async fn oauth_login_support(
939	server: &McpServerConfig,
940	network_policy: Option<&NetworkPolicyDecider>,
941	) -> Result<Option<McpOAuthDiscovery>> {
942	if server.reviewed_plugin.is_some() {
943	return Ok(None);
944	}
945	let Some(url) = server.url.as_deref() else {
946	return Ok(None);
947	};
948	if server_has_manual_authorization(server) {
949	return Ok(None);
950	}
951	let headers = build_default_headers(&server.headers, &server.env_headers)?;
952	discover_streamable_http_oauth_for_server(server, url, headers, network_policy).await
953	}
954
955	fn oauth_http_client(
956	server: &McpServerConfig,
957	url: &str,
958	network_policy: Option<&NetworkPolicyDecider>,
959	) -> Result<McpHttpClient> {
960	let timeouts = super::McpTimeouts::default();
961	McpHttpClient::new(
962	url,
963	server.runtime_added,
964	server.reviewed_plugin.is_some(),
965	server.allow_private_network,
966	network_policy,
967	Duration::from_secs(server.effective_connect_timeout(&timeouts)),
968	Duration::from_secs(server.effective_read_timeout(&timeouts)),
969	)
970	}
971
972	fn oauth_login_client(
973	server: &McpServerConfig,
974	url: &str,
975	network_policy: Option<&NetworkPolicyDecider>,
976	) -> Result<McpHttpClient> {
977	let headers = build_default_headers(&server.headers, &server.env_headers)?;
978	Ok(oauth_http_client(server, url, network_policy)?.with_default_headers(headers))
979	}
980
981	async fn discover_streamable_http_oauth_for_server(
982	server: &McpServerConfig,
983	url: &str,
984	default_headers: HeaderMap,
985	network_policy: Option<&NetworkPolicyDecider>,
986	) -> Result<Option<McpOAuthDiscovery>> {
987	let client =
988	oauth_http_client(server, url, network_policy)?.with_default_headers(default_headers);
989	discover_streamable_http_oauth_with_client(url, client).await
990	}
991
992	async fn discover_streamable_http_oauth_with_client(
993	url: &str,
994	client: McpHttpClient,
995	) -> Result<Option<McpOAuthDiscovery>> {
996	let client = Arc::new(RecordingOAuthHttpClient::new(client));
997	let manager = AuthorizationManager::new_with_oauth_http_client(url, client).await?;
998	match tokio::time::timeout(Duration::from_secs(5), manager.resolve_metadata()).await? {
999	Ok(resolution) => Ok(Some(McpOAuthDiscovery {
1000	scopes_supported: normalize_scopes(resolution.metadata.scopes_supported),
1001	})),
1002	Err(AuthError::NoAuthorizationSupport) => Ok(None),
1003	Err(err) => Err(err.into()),
1004	}
1005	}
1006
1007	pub fn resolve_oauth_scopes(
1008	explicit_scopes: Option<Vec<String>>,
1009	configured_scopes: Vec<String>,
1010	discovered_scopes: Option<Vec<String>>,
1011	) -> ResolvedMcpOAuthScopes {
1012	if let Some(scopes) = explicit_scopes {
1013	return ResolvedMcpOAuthScopes {
1014	scopes,
1015	source: McpOAuthScopesSource::Explicit,
1016	};
1017	}
1018	if !configured_scopes.is_empty() {
1019	return ResolvedMcpOAuthScopes {
1020	scopes: configured_scopes,
1021	source: McpOAuthScopesSource::Configured,
1022	};
1023	}
1024	if let Some(scopes) = discovered_scopes
1025	&& !scopes.is_empty()
1026	{
1027	return ResolvedMcpOAuthScopes {
1028	scopes,
1029	source: McpOAuthScopesSource::Discovered,
1030	};
1031	}
1032	ResolvedMcpOAuthScopes {
1033	scopes: Vec::new(),
1034	source: McpOAuthScopesSource::Empty,
1035	}
1036	}
1037
1038	pub async fn perform_oauth_login_for_server(
1039	name: &str,
1040	server: &McpServerConfig,
1041	explicit_scopes: Option<Vec<String>>,
1042	callback_port: Option<u16>,
1043	callback_url: Option<&str>,
1044	network_policy: Option<&NetworkPolicyDecider>,
1045	) -> Result<()> {
1046	perform_oauth_login_for_server_with_cancel(
1047	name,
1048	server,
1049	explicit_scopes,
1050	callback_port,
1051	callback_url,
1052	CancellationToken::new(),
1053	network_policy,
1054	)
1055	.await
1056	}
1057
1058	/// Run an MCP OAuth login that can be stopped by the caller.
1059	///
1060	/// Cancellation drops the in-flight OAuth future before this function returns,
1061	/// which also closes its callback listener. A caller that replaces one login
1062	/// with another should await the cancelled call before starting the replacement.
1063	pub async fn perform_oauth_login_for_server_with_cancel(
1064	name: &str,
1065	server: &McpServerConfig,
1066	explicit_scopes: Option<Vec<String>>,
1067	callback_port: Option<u16>,
1068	callback_url: Option<&str>,
1069	cancellation_token: CancellationToken,
1070	network_policy: Option<&NetworkPolicyDecider>,
1071	) -> Result<()> {
1072	if server.reviewed_plugin.is_some() {
1073	bail!(
1074	"OAuth is disabled for plugin-contributed MCP servers; use a reviewed environment-backed header or bearer token"
1075	);
1076	}
1077	run_cancellable_oauth(
1078	&cancellation_token,
1079	perform_oauth_login_for_server_inner(
1080	name,
1081	server,
1082	explicit_scopes,
1083	callback_port,
1084	callback_url,
1085	network_policy,
1086	),
1087	)
1088	.await
1089	}
1090
1091	async fn run_cancellable_oauth<F, T>(cancellation_token: &CancellationToken, future: F) -> Result<T>
1092	where
1093	F: std::future::Future<Output = Result<T>>,
1094	{
1095	tokio::select! {
1096	biased;
1097	_ = cancellation_token.cancelled() => bail!("OAuth login was cancelled"),
1098	result = future => result,
1099	}
1100	}
1101
1102	/// Shared gate + scope resolution for `/mcp login` and the model-driven
1103	/// authenticate tool: URL-based servers only, no manual Authorization config,
1104	/// scopes from explicit argument, config, or discovery (in that order).
1105	async fn resolve_oauth_login(
1106	name: &str,
1107	server: &McpServerConfig,
1108	explicit_scopes: Option<Vec<String>>,
1109	network_policy: Option<&NetworkPolicyDecider>,
1110	) -> Result<(String, ResolvedMcpOAuthScopes)> {
1111	let Some(url) = server.url.as_deref() else {
1112	bail!("OAuth login is only supported for URL-based MCP servers");
1113	};
1114	if server_has_manual_authorization(server) {
1115	bail!("MCP server '{name}' already has bearer/static Authorization configured");
1116	}
1117
1118	let discovery = if explicit_scopes.is_none() && server.scopes.is_empty() {
1119	oauth_login_support(server, network_policy).await?
1120	} else {
1121	None
1122	};
1123	let resolved_scopes = resolve_oauth_scopes(
1124	explicit_scopes,
1125	server.scopes.clone(),
1126	discovery.and_then(\|discovery\| discovery.scopes_supported),
1127	);
1128	Ok((url.to_string(), resolved_scopes))
1129	}
1130
1131	async fn perform_oauth_login_for_server_inner(
1132	name: &str,
1133	server: &McpServerConfig,
1134	explicit_scopes: Option<Vec<String>>,
1135	callback_port: Option<u16>,
1136	callback_url: Option<&str>,
1137	network_policy: Option<&NetworkPolicyDecider>,
1138	) -> Result<()> {
1139	let (url, resolved_scopes) =
1140	resolve_oauth_login(name, server, explicit_scopes, network_policy).await?;
1141
1142	match perform_oauth_login(
1143	name,
1144	&url,
1145	oauth_login_client(server, &url, network_policy)?,
1146	&resolved_scopes.scopes,
1147	server.oauth_client_id(),
1148	server.oauth_resource.as_deref(),
1149	callback_port,
1150	callback_url,
1151	)
1152	.await
1153	{
1154	Ok(()) => Ok(()),
1155	Err(err)
1156	if resolved_scopes.source == McpOAuthScopesSource::Discovered
1157	&& err.downcast_ref::<OAuthProviderError>().is_some() =>
1158	{
1159	println!("OAuth provider rejected discovered scopes. Retrying without scopes...");
1160	perform_oauth_login(
1161	name,
1162	&url,
1163	oauth_login_client(server, &url, network_policy)?,
1164	&[],
1165	server.oauth_client_id(),
1166	server.oauth_resource.as_deref(),
1167	callback_port,
1168	callback_url,
1169	)
1170	.await
1171	}
1172	Err(err) => Err(err),
1173	}
1174	}
1175
1176	#[allow(clippy::too_many_arguments)]
1177	async fn perform_oauth_login(
1178	server_name: &str,
1179	server_url: &str,
1180	client: McpHttpClient,
1181	scopes: &[String],
1182	oauth_client_id: Option<&str>,
1183	oauth_resource: Option<&str>,
1184	callback_port: Option<u16>,
1185	callback_url: Option<&str>,
1186	) -> Result<()> {
1187	OauthLoginFlow::new(
1188	server_name,
1189	server_url,
1190	client,
1191	scopes,
1192	oauth_client_id,
1193	oauth_resource,
1194	callback_port,
1195	callback_url,
1196	)
1197	.await?
1198	.finish()
1199	.await
1200	}
1201
1202	/// How an OAuth login announces its authorization URL.
1203	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1204	enum OAuthLoginAnnounce {
1205	/// `/mcp login` in a terminal: print the URL and open a browser.
1206	Terminal,
1207	/// Model-driven `mcp_<server>_authenticate` tool: never write to stdout
1208	/// (a tool call inside a running session is not a terminal); the result
1209	/// carries the URL for the model to relay to the user verbatim.
1210	Tool { open_browser: bool },
1211	}
1212
1213	/// An in-flight OAuth login started by the model-driven authenticate tool.
1214	///
1215	/// The authorization URL is available immediately so the model can relay it
1216	/// to the user verbatim; [`McpOAuthToolLogin::finish`] then blocks on the
1217	/// loopback callback (up to 5 minutes, same as `/mcp login`) and persists
1218	/// the issued tokens to the shared store on success.
1219	pub struct McpOAuthToolLogin {
1220	server_name: String,
1221	server: McpServerConfig,
1222	scopes_source: McpOAuthScopesSource,
1223	network_policy: Option<NetworkPolicyDecider>,
1224	flow: OauthLoginFlow,
1225	open_browser: bool,
1226	}
1227
1228	impl McpOAuthToolLogin {
1229	/// The exact authorization URL the user must visit. Treat it as
1230	/// sensitive: never modify it or strip query parameters.
1231	#[must_use]
1232	pub fn authorization_url(&self) -> &str {
1233	&self.flow.auth_url
1234	}
1235
1236	/// Block on the browser callback and persist the issued tokens. Mirrors
1237	/// the `/mcp login` retry: when the provider rejects scopes that came
1238	/// from discovery (rather than explicit config), restart once without
1239	/// scopes.
1240	pub async fn finish(self) -> Result<()> {
1241	let announce = OAuthLoginAnnounce::Tool {
1242	open_browser: self.open_browser,
1243	};
1244	let retry_without_scopes = self.scopes_source == McpOAuthScopesSource::Discovered;
1245	match self.flow.finish_with_announce(announce).await {
1246	Ok(()) => Ok(()),
1247	Err(err)
1248	if retry_without_scopes && err.downcast_ref::<OAuthProviderError>().is_some() =>
1249	{
1250	let server = &self.server;
1251	let url = server
1252	.url
1253	.as_deref()
1254	.expect("tool login is gated to URL-based servers at begin");
1255	OauthLoginFlow::new(
1256	&self.server_name,
1257	url,
1258	oauth_login_client(server, url, self.network_policy.as_ref())?,
1259	&[],
1260	server.oauth_client_id(),
1261	server.oauth_resource.as_deref(),
1262	None,
1263	None,
1264	)
1265	.await?
1266	.finish_with_announce(announce)
1267	.await
1268	}
1269	Err(err) => Err(err),
1270	}
1271	}
1272	}
1273
1274	/// Begin the same OAuth login flow `/mcp login` runs, for the model-driven
1275	/// `mcp_<server>_authenticate` tool. The callback listener binds an
1276	/// ephemeral loopback port; callers needing a pre-registered redirect URI
1277	/// keep the terminal `/mcp login <name>` path, which honors the configured
1278	/// callback overrides.
1279	pub async fn begin_oauth_login_for_server_tool(
1280	name: &str,
1281	server: &McpServerConfig,
1282	explicit_scopes: Option<Vec<String>>,
1283	callback_port: Option<u16>,
1284	callback_url: Option<&str>,
1285	network_policy: Option<&NetworkPolicyDecider>,
1286	) -> Result<McpOAuthToolLogin> {
1287	if server.reviewed_plugin.is_some() {
1288	bail!(
1289	"OAuth is disabled for plugin-contributed MCP servers; use a reviewed environment-backed header or bearer token"
1290	);
1291	}
1292	let (url, resolved_scopes) =
1293	resolve_oauth_login(name, server, explicit_scopes, network_policy).await?;
1294	let flow = OauthLoginFlow::new(
1295	name,
1296	&url,
1297	oauth_login_client(server, &url, network_policy)?,
1298	&resolved_scopes.scopes,
1299	server.oauth_client_id(),
1300	server.oauth_resource.as_deref(),
1301	callback_port,
1302	callback_url,
1303	)
1304	.await?;
1305	Ok(McpOAuthToolLogin {
1306	server_name: name.to_string(),
1307	server: server.clone(),
1308	scopes_source: resolved_scopes.source,
1309	network_policy: network_policy.cloned(),
1310	flow,
1311	// The test build drives the loopback callback itself; a real browser
1312	// launch from a unit test would hijack the developer's desktop.
1313	open_browser: !cfg!(test),
1314	})
1315	}
1316
1317	/// Whether the self-serve OAuth login flow can run for this server at all:
1318	/// URL-based, not plugin-contributed (plugin servers authenticate through
1319	/// reviewed environment-backed headers, and OAuth storage is disabled for
1320	/// them), and without a manual Authorization configuration that an OAuth
1321	/// login would conflict with.
1322	pub(crate) fn server_supports_oauth_login(server: &McpServerConfig) -> bool {
1323	server.reviewed_plugin.is_none()
1324	&& server.url.is_some()
1325	&& !server_has_manual_authorization(server)
1326	}
1327
1328	/// Whether the shared token store already holds a usable credential for this
1329	/// server — i.e. a login completed in another process since the caller last
1330	/// checked. Used by the authenticate tool's already-authorized branch.
1331	pub(crate) fn has_usable_stored_tokens(name: &str, server: &McpServerConfig) -> bool {
1332	let Some(url) = server.url.as_deref() else {
1333	return false;
1334	};
1335	load_oauth_tokens(name, url)
1336	.ok()
1337	.flatten()
1338	.is_some_and(\|tokens\| oauth_tokens_are_usable(&tokens))
1339	}
1340
1341	/// Model-facing description for the synthetic `mcp_<server>_authenticate`
1342	/// tool. The coaching contract (show the URL verbatim, the call blocks, real
1343	/// tools replace this one on success) is pinned by tests.
1344	pub(crate) fn authenticate_tool_description(server_name: &str) -> String {
1345	format!(
1346	"Authenticate with MCP server \"{server_name}\" via OAuth.\n\n\
1347	This server requires an OAuth login that has not yet been completed, so its \
1348	real tools are currently unavailable. Calling this tool starts the \
1349	authorization flow:\n\n\
1350	1. A browser window is opened for the user to sign in and approve the \
1351	Codewhale client, and the exact authorization URL is shown to the user in \
1352	the session status while this call waits. The same URL is returned in this \
1353	call's result; if the user reports the browser did not open, show that URL \
1354	to the user verbatim and ask them to complete the sign-in there.\n\
1355	2. The call blocks (up to 5 minutes) until the browser flow completes on the \
1356	local callback listener, is declined, or times out. Do not assume success \
1357	before the call returns.\n\
1358	3. On success the server reconnects and its real MCP tools replace this \
1359	synthetic authenticate tool, becoming callable from the next model request \
1360	in this session.\n\n\
1361	Treat the URL as sensitive — do not modify it or strip query parameters. If \
1362	the flow is declined, cancelled, or times out, the call returns an error; \
1363	relay it truthfully and suggest `/mcp login {server_name}` in the TUI or \
1364	`codewhale mcp login {server_name}` from a terminal."
1365	)
1366	}
1367
1368	pub fn delete_oauth_tokens_for_server(name: &str, server: &McpServerConfig) -> Result<bool> {
1369	if server.reviewed_plugin.is_some() {
1370	bail!("OAuth storage is disabled for plugin-contributed MCP servers");
1371	}
1372	let Some(url) = server.url.as_deref() else {
1373	bail!("OAuth logout is only supported for URL-based MCP servers");
1374	};
1375	delete_oauth_tokens(name, url)
1376	}
1377
1378	pub(crate) fn server_has_manual_authorization(server: &McpServerConfig) -> bool {
1379	server.bearer_token_env_var.is_some()
1380	\|\| contains_authorization_header(&server.headers)
1381	\|\| contains_authorization_header(&server.env_headers)
1382	}
1383
1384	pub fn build_default_headers(
1385	http_headers: &HashMap<String, String>,
1386	env_headers: &HashMap<String, String>,
1387	) -> Result<HeaderMap> {
1388	let mut headers = HeaderMap::new();
1389	for (name, value) in http_headers {
1390	insert_header(&mut headers, name, value)?;
1391	}
1392	for (name, env_var) in env_headers {
1393	if let Ok(value) = std::env::var(env_var)
1394	&& !value.trim().is_empty()
1395	{
1396	insert_header(&mut headers, name, &value)?;
1397	}
1398	}
1399	Ok(headers)
1400	}
1401
1402	fn insert_header(headers: &mut HeaderMap, name: &str, value: &str) -> Result<()> {
1403	if !super::headers::is_safe_custom_header(name, value) {
1404	bail!("unsafe MCP HTTP header '{name}'");
1405	}
1406	let name = HeaderName::from_bytes(name.as_bytes())
1407	.with_context(\|\| format!("invalid MCP HTTP header name '{name}'"))?;
1408	let value = HeaderValue::from_str(value).with_context(\|\| "invalid MCP HTTP header value")?;
1409	headers.insert(name, value);
1410	Ok(())
1411	}
1412
1413	fn contains_authorization_header(headers: &HashMap<String, String>) -> bool {
1414	headers
1415	.keys()
1416	.any(\|key\| key.trim().eq_ignore_ascii_case("authorization"))
1417	}
1418
1419	fn normalize_scopes(scopes_supported: Option<Vec<String>>) -> Option<Vec<String>> {
1420	let scopes_supported = scopes_supported?;
1421	let mut normalized = Vec::new();
1422	for scope in scopes_supported {
1423	let scope = scope.trim();
1424	if scope.is_empty() {
1425	continue;
1426	}
1427	let scope = scope.to_string();
1428	if !normalized.contains(&scope) {
1429	normalized.push(scope);
1430	}
1431	}
1432	(!normalized.is_empty()).then_some(normalized)
1433	}
1434
1435	pub(crate) fn load_oauth_tokens(
1436	server_name: &str,
1437	url: &str,
1438	) -> Result<Option<StoredMcpOAuthTokens>> {
1439	let secrets = codewhale_secrets::Secrets::auto_detect();
1440	let key = store_key(server_name, url);
1441	let Some(serialized) = secrets
1442	.get(&key)
1443	.with_context(\|\| format!("reading MCP OAuth token for '{server_name}'"))?
1444	else {
1445	return Ok(None);
1446	};
1447	let mut tokens = parse_stored_oauth_tokens(&serialized, server_name)?;
1448	refresh_expires_in_from_timestamp(&mut tokens);
1449	Ok(Some(tokens))
1450	}
1451
1452	fn parse_stored_oauth_tokens(serialized: &str, server_name: &str) -> Result<StoredMcpOAuthTokens> {
1453	serde_json::from_str(serialized).map_err(\|_\| {
1454	anyhow!(
1455	"stored MCP OAuth token for '{server_name}' is not valid credential JSON; contents were omitted"
1456	)
1457	})
1458	}
1459
1460	pub(crate) fn save_oauth_tokens(tokens: &StoredMcpOAuthTokens) -> Result<()> {
1461	let secrets = codewhale_secrets::Secrets::auto_detect();
1462	let key = store_key(&tokens.server_name, &tokens.url);
1463	let serialized = serde_json::to_string(tokens).context("serializing MCP OAuth token")?;
1464	secrets
1465	.set(&key, &serialized)
1466	.with_context(\|\| format!("saving MCP OAuth token for '{}'", tokens.server_name))
1467	}
1468
1469	fn delete_oauth_tokens(server_name: &str, url: &str) -> Result<bool> {
1470	let secrets = codewhale_secrets::Secrets::auto_detect();
1471	let key = store_key(server_name, url);
1472	let existed = secrets
1473	.get(&key)
1474	.with_context(\|\| format!("reading MCP OAuth token for '{server_name}'"))?
1475	.is_some();
1476	secrets
1477	.delete(&key)
1478	.with_context(\|\| format!("deleting MCP OAuth token for '{server_name}'"))?;
1479	Ok(existed)
1480	}
1481
1482	fn store_key(server_name: &str, url: &str) -> String {
1483	let mut payload = Vec::with_capacity(server_name.len() + url.len() + 1);
1484	payload.extend_from_slice(server_name.as_bytes());
1485	payload.push(0);
1486	payload.extend_from_slice(url.as_bytes());
1487	let digest = Sha256::digest(&payload);
1488	format!("mcp_oauth_{}", URL_SAFE_NO_PAD.encode(digest))
1489	}
1490
1491	fn oauth_tokens_are_usable(tokens: &StoredMcpOAuthTokens) -> bool {
1492	if tokens.client_id.trim().is_empty() {
1493	return false;
1494	}
1495	let response = &tokens.token_response.0;
1496	if token_needs_refresh(tokens.expires_at) {
1497	return response
1498	.refresh_token()
1499	.is_some_and(\|token\| !token.secret().trim().is_empty());
1500	}
1501	!response.access_token().secret().trim().is_empty()
1502	}
1503
1504	fn refresh_expires_in_from_timestamp(tokens: &mut StoredMcpOAuthTokens) {
1505	let Some(expires_at) = tokens.expires_at else {
1506	return;
1507	};
1508	match expires_in_from_timestamp(expires_at) {
1509	Some(seconds) => {
1510	let duration = Duration::from_secs(seconds);
1511	tokens.token_response.0.set_expires_in(Some(&duration));
1512	}
1513	None => {
1514	tokens
1515	.token_response
1516	.0
1517	.set_expires_in(Some(&Duration::ZERO));
1518	}
1519	}
1520	}
1521
1522	fn compute_expires_at_millis(response: &OAuthTokenResponse) -> Option<u64> {
1523	let expires = response.expires_in()?;
1524	let now = SystemTime::now()
1525	.duration_since(UNIX_EPOCH)
1526	.ok()?
1527	.as_millis() as u64;
1528	Some(now.saturating_add(expires.as_millis() as u64))
1529	}
1530
1531	fn expires_in_from_timestamp(expires_at: u64) -> Option<u64> {
1532	let now = SystemTime::now()
1533	.duration_since(UNIX_EPOCH)
1534	.ok()?
1535	.as_millis() as u64;
1536	if expires_at <= now {
1537	return None;
1538	}
1539	Some((expires_at - now) / 1000)
1540	}
1541
1542	fn token_needs_refresh(expires_at: Option<u64>) -> bool {
1543	let Some(expires_at) = expires_at else {
1544	return false;
1545	};
1546	let now = SystemTime::now()
1547	.duration_since(UNIX_EPOCH)
1548	.map(\|duration\| duration.as_millis() as u64)
1549	.unwrap_or(0);
1550	now.saturating_add(REFRESH_SKEW_MILLIS) >= expires_at
1551	}
1552
1553	struct CallbackServerGuard {
1554	accept_task: tokio::task::JoinHandle<()>,
1555	}
1556
1557	impl Drop for CallbackServerGuard {
1558	fn drop(&mut self) {
1559	// Aborting drops the accept future and its owned listener instead of
1560	// leaving a detached task holding a fixed callback port indefinitely.
1561	self.accept_task.abort();
1562	}
1563	}
1564
1565	struct OauthLoginFlow {
1566	auth_url: String,
1567	oauth_state: OAuthState,
1568	rx: oneshot::Receiver<CallbackResult>,
1569	guard: CallbackServerGuard,
1570	server_name: String,
1571	server_url: String,
1572	}
1573
1574	impl OauthLoginFlow {
1575	#[allow(clippy::too_many_arguments)]
1576	async fn new(
1577	server_name: &str,
1578	server_url: &str,
1579	client: McpHttpClient,
1580	scopes: &[String],
1581	oauth_client_id: Option<&str>,
1582	oauth_resource: Option<&str>,
1583	callback_port: Option<u16>,
1584	callback_url: Option<&str>,
1585	) -> Result<Self> {
1586	let bind_host = callback_bind_host(callback_url);
1587	let bind_addr = match callback_port {
1588	Some(0) => bail!("invalid MCP OAuth callback port 0"),
1589	Some(port) => format!("{bind_host}:{port}"),
1590	None => format!("{bind_host}:0"),
1591	};
1592	let listener = TcpListener::bind(&bind_addr)
1593	.await
1594	.map_err(\|err\| anyhow!(err))?;
1595	let redirect_uri = resolve_redirect_uri(&listener, callback_url)?;
1596	let callback_id = callback_id_from_server_url(server_url)?;
1597	let redirect_uri = append_callback_id_to_redirect_uri(&redirect_uri, &callback_id)?;
1598	let callback_path = callback_path_from_redirect_uri(&redirect_uri)?;
1599
1600	let (tx, rx) = oneshot::channel();
1601	let guard = CallbackServerGuard {
1602	accept_task: spawn_callback_server(listener, tx, callback_path),
1603	};
1604
1605	let scope_refs: Vec<&str> = scopes.iter().map(String::as_str).collect();
1606	let oauth_state = start_authorization(
1607	server_url,
1608	client,
1609	&scope_refs,
1610	&redirect_uri,
1611	oauth_client_id,
1612	)
1613	.await?;
1614	let auth_url = append_query_param(
1615	&oauth_state.get_authorization_url().await?,
1616	"resource",
1617	oauth_resource,
1618	);
1619	// #6040: logout clears this machine's token only — the provider keeps
1620	// its standing grant. Without a forced prompt the next login silently
1621	// re-grants it (same account/workspace, no picker ever shown), so an
1622	// explicit login could never change the authorized workspace.
1623	let auth_url = append_query_param(&auth_url, "prompt", Some("consent"));
1624
1625	Ok(Self {
1626	auth_url,
1627	oauth_state,
1628	rx,
1629	guard,
1630	server_name: server_name.to_string(),
1631	server_url: server_url.to_string(),
1632	})
1633	}
1634
1635	async fn finish(self) -> Result<()> {
1636	self.finish_with_announce(OAuthLoginAnnounce::Terminal)
1637	.await
1638	}
1639
1640	async fn finish_with_announce(mut self, announce: OAuthLoginAnnounce) -> Result<()> {
1641	match announce {
1642	OAuthLoginAnnounce::Terminal => {
1643	println!(
1644	"Authorize `{}` by opening this URL in your browser:\n{}\n",
1645	self.server_name, self.auth_url
1646	);
1647	if webbrowser::open(&self.auth_url).is_err() {
1648	eprintln!("Browser launch failed; copy the URL above manually.");
1649	}
1650	println!(
1651	"Waiting for browser authorization for MCP server '{}'...",
1652	self.server_name
1653	);
1654	}
1655	OAuthLoginAnnounce::Tool { open_browser } => {
1656	// A tool call is not a terminal: nothing goes to stdout. The
1657	// tool result carries the URL for the model to relay; the
1658	// browser open is a best-effort convenience on top.
1659	if open_browser {
1660	let _ = webbrowser::open(&self.auth_url);
1661	}
1662	}
1663	}
1664
1665	let result = async {
1666	let callback = timeout(Duration::from_secs(300), &mut self.rx)
1667	.await
1668	.with_context(\|\| {
1669	let retry_hint = match announce {
1670	OAuthLoginAnnounce::Terminal => "Retry from a terminal, or use task_shell_start/background shell if an agent is running the login flow.".to_string(),
1671	OAuthLoginAnnounce::Tool { .. } => format!(
1672	"The user can complete the sign-in directly via `/mcp login {}` or `codewhale mcp login {}`, then this tool can be called again.",
1673	self.server_name, self.server_name
1674	),
1675	};
1676	format!(
1677	"timed out waiting for OAuth callback for MCP server '{}'. {retry_hint}",
1678	self.server_name
1679	)
1680	})?
1681	.context("OAuth callback was cancelled")?;
1682	let OauthCallbackResult {
1683	code,
1684	state,
1685	issuer,
1686	} = match callback {
1687	CallbackResult::Success(callback) => callback,
1688	CallbackResult::Error(error) => return Err(anyhow!(error)),
1689	};
1690
1691	// RFC 9207: servers that advertise
1692	// `authorization_response_iss_parameter_supported` send `iss` on the
1693	// redirect and rmcp requires it back; forward it so the callback binds
1694	// to the discovered issuer instead of failing as "missing".
1695	self.oauth_state
1696	.handle_callback_with_issuer(&code, &state, issuer.as_deref())
1697	.await
1698	.context("handling MCP OAuth callback")?;
1699
1700	let (client_id, credentials) = self
1701	.oauth_state
1702	.get_credentials()
1703	.await
1704	.context("reading MCP OAuth credentials")?;
1705	let credentials =
1706	credentials.ok_or_else(\|\| anyhow!("OAuth provider did not return credentials"))?;
1707	let stored = StoredMcpOAuthTokens {
1708	server_name: self.server_name.clone(),
1709	url: self.server_url.clone(),
1710	client_id,
1711	expires_at: compute_expires_at_millis(&credentials),
1712	token_response: WrappedOAuthTokenResponse(credentials),
1713	};
1714	save_oauth_tokens(&stored)
1715	}
1716	.await;
1717
1718	drop(self.guard);
1719	result
1720	}
1721	}
1722
1723	async fn start_authorization(
1724	server_url: &str,
1725	client: McpHttpClient,
1726	scopes: &[&str],
1727	redirect_uri: &str,
1728	oauth_client_id: Option<&str>,
1729	) -> Result<OAuthState> {
1730	let Some(client_id) = oauth_client_id.filter(\|client_id\| !client_id.trim().is_empty()) else {
1731	let mut attempt_scopes: Vec<String> =
1732	scopes.iter().map(\|scope\| (*scope).to_string()).collect();
1733	// Dynamic registration may reject part of the scope list the server
1734	// itself advertised (Supabase validates registration scopes against a
1735	// narrower allow-list than its `scopes_supported`). Drop exactly the
1736	// scopes the server named invalid and retry once; if it named none,
1737	// register without scopes so the server applies its defaults.
1738	for retried in [false, true] {
1739	let mut oauth_state = OAuthState::new_with_oauth_http_client(
1740	server_url,
1741	Arc::new(RecordingOAuthHttpClient::new(client.clone())),
1742	)
1743	.await?;
1744	let started = oauth_state
1745	.start_authorization(
1746	AuthorizationRequest::new(redirect_uri)
1747	.with_scopes(attempt_scopes.iter().map(String::as_str))
1748	.with_client_name("Codewhale"),
1749	)
1750	.await;
1751	match started {
1752	Ok(()) => return Ok(oauth_state),
1753	Err(error) if !retried && !attempt_scopes.is_empty() => {
1754	let message = error.to_string();
1755	let Some(narrowed) =
1756	scopes_after_registration_rejection(&attempt_scopes, &message)
1757	else {
1758	return Err(error.into());
1759	};
1760	tracing::warn!(
1761	target: "mcp::oauth",
1762	server_url,
1763	dropped = attempt_scopes.len() - narrowed.len(),
1764	"OAuth client registration rejected part of the requested scope list; retrying with the accepted scopes"
1765	);
1766	attempt_scopes = narrowed;
1767	}
1768	Err(error) => return Err(error.into()),
1769	}
1770	}
1771	unreachable!("registration retry loop returns on success or error");
1772	};
1773
1774	let mut manager = AuthorizationManager::new_with_oauth_http_client(
1775	server_url,
1776	Arc::new(RecordingOAuthHttpClient::new(client)),
1777	)
1778	.await?;
1779	let metadata = manager.resolve_metadata().await?.metadata;
1780	manager.set_metadata(metadata);
1781	manager.configure_client(
1782	OAuthClientConfig::new(client_id, redirect_uri)
1783	.with_scopes(scopes.iter().map(\|scope\| (*scope).to_string()).collect()),
1784	)?;
1785	let auth_url = manager.get_authorization_url(scopes).await?;
1786	Ok(OAuthState::Session(
1787	AuthorizationSession::for_scope_upgrade(manager, auth_url, redirect_uri),
1788	))
1789	}
1790
1791	/// Given a registration failure message, return the scopes to retry with, or
1792	/// `None` when the failure is not about scopes. Servers that validate the
1793	/// `scope` field report positions like `scope.3: Invalid option`; those exact
1794	/// entries are dropped. A scope error without positions retries with no
1795	/// scopes at all, letting the server grant its defaults.
1796	fn scopes_after_registration_rejection(scopes: &[String], message: &str) -> Option<Vec<String>> {
1797	let lower = message.to_ascii_lowercase();
1798	if !(lower.contains("registration") && lower.contains("scope")) {
1799	return None;
1800	}
1801	let mut rejected = std::collections::BTreeSet::new();
1802	for (start, _) in message.match_indices("scope.") {
1803	let digits: String = message[start + "scope.".len()..]
1804	.chars()
1805	.take_while(char::is_ascii_digit)
1806	.collect();
1807	if let Ok(index) = digits.parse::<usize>() {
1808	rejected.insert(index);
1809	}
1810	}
1811	let narrowed: Vec<String> = scopes
1812	.iter()
1813	.enumerate()
1814	.filter(\|(index, _)\| !rejected.contains(index))
1815	.map(\|(_, scope)\| scope.clone())
1816	.collect();
1817	if narrowed.len() == scopes.len() {
1818	// The server complained about scopes without naming any position:
1819	// the only safe retry is to omit the field.
1820	return Some(Vec::new());
1821	}
1822	Some(narrowed)
1823	}
1824
1825	fn spawn_callback_server(
1826	listener: TcpListener,
1827	tx: oneshot::Sender<CallbackResult>,
1828	expected_callback_path: String,
1829	) -> tokio::task::JoinHandle<()> {
1830	tokio::spawn(async move {
1831	// The sender is wrapped in Option so we can take it on success/error
1832	let mut tx_opt = Some(tx);
1833	loop {
1834	let (mut stream, _) = match listener.accept().await {
1835	Ok(pair) => pair,
1836	Err(_) => break,
1837	};
1838	let path = match read_http_path(&mut stream).await {
1839	Some(p) => p,
1840	None => {
1841	let _ = write_http_response(&mut stream, 400, "Invalid OAuth callback").await;
1842	continue;
1843	}
1844	};
1845	match parse_oauth_callback(&path, &expected_callback_path) {
1846	CallbackOutcome::Success(callback) => {
1847	let _ = write_http_response(
1848	&mut stream,
1849	200,
1850	"Authentication complete. You may close this window.",
1851	)
1852	.await;
1853	if let Some(tx) = tx_opt.take() {
1854	let _ = tx.send(CallbackResult::Success(callback));
1855	}
1856	break;
1857	}
1858	CallbackOutcome::Error(error) => {
1859	let msg = error.to_string();
1860	let _ = write_http_response(&mut stream, 400, &msg).await;
1861	if let Some(tx) = tx_opt.take() {
1862	let _ = tx.send(CallbackResult::Error(error));
1863	}
1864	break;
1865	}
1866	CallbackOutcome::Invalid => {
1867	let _ = write_http_response(&mut stream, 400, "Invalid OAuth callback").await;
1868	}
1869	}
1870	}
1871	})
1872	}
1873
1874	async fn read_http_path(stream: &mut tokio::net::TcpStream) -> Option<String> {
1875	let mut buf = Vec::new();
1876	let mut tmp = [0u8; 1024];
1877	// Read until we have \r\n\r\n or exceed limit
1878	loop {
1879	match stream.read(&mut tmp).await {
1880	Ok(0) => break,
1881	Ok(n) => {
1882	buf.extend_from_slice(&tmp[..n]);
1883	if buf.windows(4).any(\|w\| w == b"\r\n\r\n") {
1884	break;
1885	}
1886	if buf.len() > 8192 {
1887	break;
1888	}
1889	}
1890	Err(_) => return None,
1891	}
1892	}
1893	let request = String::from_utf8_lossy(&buf);
1894	let first_line = request.lines().next()?;
1895	// Expected: GET /callback?code=... HTTP/1.1
1896	let mut parts = first_line.split_whitespace();
1897	let _method = parts.next()?;
1898	let path = parts.next()?.to_string();
1899	Some(path)
1900	}
1901
1902	async fn write_http_response(
1903	stream: &mut tokio::net::TcpStream,
1904	status: u16,
1905	body: &str,
1906	) -> std::io::Result<()> {
1907	let status_text = match status {
1908	200 => "OK",
1909	400 => "Bad Request",
1910	_ => "OK",
1911	};
1912	let response = format!(
1913	"HTTP/1.1 {status} {status_text}\r\nContent-Length: {}\r\nContent-Type: text/plain\r\nConnection: close\r\n\r\n{body}",
1914	body.len()
1915	);
1916	stream.write_all(response.as_bytes()).await?;
1917	stream.flush().await?;
1918	Ok(())
1919	}
1920
1921	#[derive(Debug, Clone, PartialEq, Eq)]
1922	struct OauthCallbackResult {
1923	code: String,
1924	state: String,
1925	/// RFC 9207 `iss` from the redirect, when the authorization server sends it.
1926	issuer: Option<String>,
1927	}
1928
1929	enum CallbackResult {
1930	Success(OauthCallbackResult),
1931	Error(OAuthProviderError),
1932	}
1933
1934	#[derive(Debug, Clone, PartialEq, Eq)]
1935	enum CallbackOutcome {
1936	Success(OauthCallbackResult),
1937	Error(OAuthProviderError),
1938	Invalid,
1939	}
1940
1941	fn parse_oauth_callback(path: &str, expected_callback_path: &str) -> CallbackOutcome {
1942	let Some((route, query)) = path.split_once('?') else {
1943	return CallbackOutcome::Invalid;
1944	};
1945	if route != expected_callback_path {
1946	return CallbackOutcome::Invalid;
1947	}
1948
1949	let mut code = None;
1950	let mut state = None;
1951	let mut issuer = None;
1952	let mut error = None;
1953	let mut error_description = None;
1954	for pair in query.split('&') {
1955	let Some((key, value)) = pair.split_once('=') else {
1956	continue;
1957	};
1958	let Ok(decoded) = decode(value) else {
1959	continue;
1960	};
1961	let decoded = decoded.into_owned();
1962	match key {
1963	"code" => code = Some(decoded),
1964	"state" => state = Some(decoded),
1965	"iss" => issuer = Some(decoded),
1966	"error" => error = Some(decoded),
1967	"error_description" => error_description = Some(decoded),
1968	_ => {}
1969	}
1970	}
1971
1972	if let (Some(code), Some(state)) = (code, state) {
1973	return CallbackOutcome::Success(OauthCallbackResult {
1974	code,
1975	state,
1976	issuer,
1977	});
1978	}
1979	if error.is_some() \|\| error_description.is_some() {
1980	return CallbackOutcome::Error(OAuthProviderError::new(error, error_description));
1981	}
1982	CallbackOutcome::Invalid
1983	}
1984
1985	fn local_redirect_uri(listener: &TcpListener) -> Result<String> {
1986	let addr = listener.local_addr()?;
1987	match addr {
1988	std::net::SocketAddr::V4(v4) => Ok(format!("http://{}:{}/callback", v4.ip(), v4.port())),
1989	std::net::SocketAddr::V6(v6) => Ok(format!("http://[{}]:{}/callback", v6.ip(), v6.port())),
1990	}
1991	}
1992
1993	fn resolve_redirect_uri(listener: &TcpListener, callback_url: Option<&str>) -> Result<String> {
1994	let Some(callback_url) = callback_url else {
1995	return local_redirect_uri(listener);
1996	};
1997	Url::parse(callback_url)
1998	.with_context(\|\| format!("invalid MCP OAuth callback URL '{callback_url}'"))?;
1999	Ok(callback_url.to_string())
2000	}
2001
2002	fn callback_bind_host(callback_url: Option<&str>) -> &'static str {
2003	let Some(callback_url) = callback_url else {
2004	return "127.0.0.1";
2005	};
2006	let Ok(parsed) = Url::parse(callback_url) else {
2007	return "127.0.0.1";
2008	};
2009	match parsed.host_str() {
2010	Some("localhost" \| "127.0.0.1" \| "::1") \| None => "127.0.0.1",
2011	Some(_) => "0.0.0.0",
2012	}
2013	}
2014
2015	fn callback_id_from_server_url(server_url: &str) -> Result<String> {
2016	let mut parsed =
2017	Url::parse(server_url).with_context(\|\| format!("invalid MCP server URL '{server_url}'"))?;
2018	parsed
2019	.host_str()
2020	.ok_or_else(\|\| anyhow!("MCP server URL '{server_url}' must include a host"))?;
2021	parsed.set_fragment(None);
2022	let digest = Sha256::digest(parsed.as_str().as_bytes());
2023	Ok(URL_SAFE_NO_PAD.encode(&digest[..9]))
2024	}
2025
2026	fn append_callback_id_to_redirect_uri(redirect_uri: &str, callback_id: &str) -> Result<String> {
2027	let mut parsed = Url::parse(redirect_uri)
2028	.with_context(\|\| format!("invalid redirect URI '{redirect_uri}'"))?;
2029	let path = parsed.path();
2030	let new_path = if path.ends_with('/') {
2031	format!("{path}{callback_id}")
2032	} else {
2033	format!("{path}/{callback_id}")
2034	};
2035	parsed.set_path(&new_path);
2036	Ok(parsed.to_string())
2037	}
2038
2039	fn callback_path_from_redirect_uri(redirect_uri: &str) -> Result<String> {
2040	let parsed = Url::parse(redirect_uri)
2041	.with_context(\|\| format!("invalid redirect URI '{redirect_uri}'"))?;
2042	Ok(parsed.path().to_string())
2043	}
2044
2045	fn append_query_param(url: &str, key: &str, value: Option<&str>) -> String {
2046	let Some(value) = value else {
2047	return url.to_string();
2048	};
2049	let value = value.trim();
2050	if value.is_empty() {
2051	return url.to_string();
2052	}
2053	if let Ok(mut parsed) = Url::parse(url) {
2054	parsed.query_pairs_mut().append_pair(key, value);
2055	return parsed.to_string();
2056	}
2057	let separator = if url.contains('?') { "&" } else { "?" };
2058	format!("{url}{separator}{key}={}", urlencoding::encode(value))
2059	}
2060
2061	impl McpServerConfig {
2062	pub fn oauth_client_id(&self) -> Option<&str> {
2063	self.oauth
2064	.as_ref()
2065	.and_then(\|oauth\| oauth.client_id.as_deref())
2066	}
2067	}
2068
2069	#[cfg(test)]
2070	mod tests {
2071	#[test]
2072	fn registration_rejection_drops_exactly_the_named_scopes() {
2073	let scopes: Vec<String> = ["a", "b", "c", "d"].iter().map(\|s\| s.to_string()).collect();
2074	let message = concat!(
2075	"Registration failed: Dynamic registration failed: HTTP 400 Bad Request: ",
2076	"{\"message\":\"scope.1: Invalid option: expected one of \\\"a\\\"\|\\\"c\\\",",
2077	"scope.3: Invalid option\"}"
2078	);
2079	assert_eq!(
2080	super::scopes_after_registration_rejection(&scopes, message),
2081	Some(vec!["a".to_string(), "c".to_string()])
2082	);
2083	// A scope complaint without positions retries without scopes.
2084	assert_eq!(
2085	super::scopes_after_registration_rejection(
2086	&scopes,
2087	"Registration failed: invalid scope"
2088	),
2089	Some(Vec::new())
2090	);
2091	// Unrelated registration failures are not retried.
2092	assert_eq!(
2093	super::scopes_after_registration_rejection(&scopes, "Registration failed: HTTP 500"),
2094	None
2095	);
2096	assert_eq!(
2097	super::scopes_after_registration_rejection(&scopes, "network unreachable"),
2098	None
2099	);
2100	}
2101
2102	#[test]
2103	fn a_refresh_parse_failure_names_the_login_remedy_and_the_server() {
2104	let text = super::refresh_failure_context("supabase", true, None);
2105	assert!(text.contains("server supabase"));
2106	assert!(text.contains("codewhale mcp login supabase"), "{text}");
2107	assert!(text.contains("/mcp login supabase"), "{text}");
2108	assert!(!text.contains("it answered"), "{text}");
2109	// An auth-required failure keeps the plain context: the typed state
2110	// and the login tool already carry the remedy.
2111	let plain = super::refresh_failure_context("supabase", false, None);
2112	assert_eq!(plain, "refreshing MCP OAuth token for server supabase");
2113	}
2114
2115	#[test]
2116	fn a_refresh_parse_failure_keeps_the_token_endpoints_receipt() {
2117	// The supabase receipt (#5926): rmcp said only "Failed to parse
2118	// server response". With the status line and a masked excerpt the
2119	// operator can tell a provider's HTML 502 from our parser.
2120	let mut headers = HeaderMap::new();
2121	headers.insert(
2122	CONTENT_TYPE,
2123	HeaderValue::from_static("text/html; charset=utf-8"),
2124	);
2125	let receipt = TokenEndpointReceipt::from_response(
2126	reqwest::StatusCode::BAD_GATEWAY,
2127	&headers,
2128	b"<html>\n <body>502 Bad Gateway</body>\n</html>",
2129	);
2130	let text = super::refresh_failure_context("supabase", true, Some(&receipt));
2131	assert!(
2132	text.contains(
2133	"it answered HTTP 502 Bad Gateway (text/html; charset=utf-8): <html> <body>502 Bad Gateway</body> </html>"
2134	),
2135	"{text}"
2136	);
2137	assert!(text.contains("codewhale mcp login supabase"), "{text}");
2138	}
2139
2140	#[test]
2141	fn a_token_receipt_masks_credentials_before_it_cuts_the_body() {
2142	let secret = "sk-live-0123456789abcdef";
2143	let long_tail = "x".repeat(400);
2144	let body = format!(
2145	"{{\"token_type\":\"Bearer\",\"access_token\":\"{secret}\",\"refresh_token\": \"{secret}-r\",\"id_token\":\"{secret}-id\",\"expires_in\":\"soon\",\"note\":\"{long_tail}\"}}"
2146	);
2147	let mut headers = HeaderMap::new();
2148	headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
2149	let receipt =
2150	TokenEndpointReceipt::from_response(reqwest::StatusCode::OK, &headers, body.as_bytes());
2151	let text = receipt.to_string();
2152	assert!(!text.contains(secret), "{text}");
2153	assert!(text.contains("\"access_token\":\"***\""), "{text}");
2154	assert!(text.contains("\"refresh_token\": \"***\""), "{text}");
2155	assert!(text.contains("\"id_token\":\"***\""), "{text}");
2156	// Non-secret members survive so the shape of the answer is readable.
2157	assert!(text.contains("\"expires_in\":\"soon\""), "{text}");
2158	assert!(text.contains("\"token_type\":\"Bearer\""), "{text}");
2159	assert!(text.ends_with('…'), "{text}");
2160	assert!(receipt.excerpt.len() <= TOKEN_RECEIPT_EXCERPT_BYTES + '…'.len_utf8());
2161	}
2162
2163	#[test]
2164	fn oauth_secret_masking_covers_form_pairs_bearer_schemes_and_case() {
2165	assert_eq!(
2166	mask_oauth_secrets("client_secret=abc123&grant_type=refresh_token&refresh_token=zzz"),
2167	"client_secret=*&grant_type=refresh_token&refresh_token=*"
2168	);
2169	assert_eq!(
2170	mask_oauth_secrets("Authorization: Bearer eyJhbGciOi.payload.sig, retry"),
2171	"Authorization: ***, retry"
2172	);
2173	assert_eq!(
2174	mask_oauth_secrets("{\"Access_Token\": \"quoted \\\" inside\", \"scope\": \"read\"}"),
2175	"{\"Access_Token\": \"***\", \"scope\": \"read\"}"
2176	);
2177	// A field name that merely contains a secret name is not a secret.
2178	assert_eq!(
2179	mask_oauth_secrets("{\"error_code\":\"invalid_request\",\"my_access_token_count\":3}"),
2180	"{\"error_code\":\"invalid_request\",\"my_access_token_count\":3}"
2181	);
2182	// Multi-byte text around a secret stays intact.
2183	assert_eq!(
2184	mask_oauth_secrets("トークン access_token=秘密終わり"),
2185	"トークン access_token=*** 終わり"
2186	);
2187	}
2188
2189	#[test]
2190	fn a_token_receipt_names_an_empty_body_and_a_missing_content_type() {
2191	let receipt = TokenEndpointReceipt::from_response(
2192	reqwest::StatusCode::SERVICE_UNAVAILABLE,
2193	&HeaderMap::new(),
2194	b"",
2195	);
2196	assert_eq!(
2197	receipt.to_string(),
2198	"HTTP 503 Service Unavailable (no content-type) with an empty body"
2199	);
2200	}
2201
2202	use super::*;
2203	use std::sync::atomic::{AtomicBool, Ordering};
2204
2205	#[test]
2206	fn stored_credential_identity_ignores_only_a_derived_expiry_countdown() {
2207	let original = serde_json::json!({
2208	"server_name": "fixture", "url": "https://example.invalid/mcp",
2209	"client_id": "fixture-client", "expires_at": 9_999_999_999_999_u64,
2210	"token_response": {
2211	"access_token": "fixture-access", "refresh_token": "fixture-refresh",
2212	"token_type": "Bearer", "expires_in": 3600, "scope": "read"
2213	}
2214	});
2215	let held: StoredMcpOAuthTokens = serde_json::from_value(original.clone()).unwrap();
2216	let mut aged = original.clone();
2217	aged["token_response"]["expires_in"] = serde_json::json!(3598);
2218	let loaded: StoredMcpOAuthTokens = serde_json::from_value(aged.clone()).unwrap();
2219	assert!(
2220	held == loaded,
2221	"elapsed time alone is not credential rotation"
2222	);
2223	for (field, value) in [
2224	("access_token", "new-access"),
2225	("refresh_token", "new-refresh"),
2226	("scope", "read write"),
2227	("token_type", "Mac"),
2228	] {
2229	let mut rotated = aged.clone();
2230	rotated["token_response"][field] = serde_json::json!(value);
2231	let rotated: StoredMcpOAuthTokens = serde_json::from_value(rotated).unwrap();
2232	assert!(
2233	held != rotated,
2234	"a changed {field} remains a distinct credential"
2235	);
2236	}
2237	for (field, value) in [
2238	("server_name", serde_json::json!("other")),
2239	("client_id", serde_json::json!("other-client")),
2240	("url", serde_json::json!("https://other.invalid/mcp")),
2241	("expires_at", serde_json::json!(9_999_999_999_998_u64)),
2242	] {
2243	let mut rotated = aged.clone();
2244	rotated[field] = value;
2245	let rotated: StoredMcpOAuthTokens = serde_json::from_value(rotated).unwrap();
2246	assert!(
2247	held != rotated,
2248	"a changed {field} remains a distinct credential"
2249	);
2250	}
2251	let mut legacy = held.clone();
2252	legacy.expires_at = None;
2253	let mut legacy_aged = loaded;
2254	legacy_aged.expires_at = None;
2255	assert!(
2256	legacy != legacy_aged,
2257	"without a durable deadline the stored lifetime is meaningful"
2258	);
2259	}
2260
2261	#[test]
2262	fn resolve_oauth_scopes_prefers_explicit() {
2263	let resolved = resolve_oauth_scopes(
2264	Some(vec!["explicit".to_string()]),
2265	vec!["configured".to_string()],
2266	Some(vec!["discovered".to_string()]),
2267	);
2268	assert_eq!(resolved.source, McpOAuthScopesSource::Explicit);
2269	assert_eq!(resolved.scopes, vec!["explicit"]);
2270	}
2271
2272	#[test]
2273	fn parse_oauth_callback_accepts_success() {
2274	let parsed = parse_oauth_callback("/callback/id?code=abc&state=xyz", "/callback/id");
2275	assert_eq!(
2276	parsed,
2277	CallbackOutcome::Success(OauthCallbackResult {
2278	code: "abc".to_string(),
2279	state: "xyz".to_string(),
2280	issuer: None,
2281	})
2282	);
2283	}
2284
2285	#[test]
2286	fn parse_oauth_callback_keeps_rfc9207_issuer() {
2287	// Cloudflare's MCP authorization server advertises
2288	// authorization_response_iss_parameter_supported and sends `iss` back;
2289	// dropping it makes rmcp reject the callback as missing a required issuer.
2290	let parsed = parse_oauth_callback(
2291	"/callback/id?code=abc&state=xyz&iss=https%3A%2F%2Fmcp.cloudflare.com",
2292	"/callback/id",
2293	);
2294	assert_eq!(
2295	parsed,
2296	CallbackOutcome::Success(OauthCallbackResult {
2297	code: "abc".to_string(),
2298	state: "xyz".to_string(),
2299	issuer: Some("https://mcp.cloudflare.com".to_string()),
2300	})
2301	);
2302	}
2303
2304	#[test]
2305	fn parse_oauth_callback_accepts_provider_error() {
2306	let parsed = parse_oauth_callback(
2307	"/callback/id?error=invalid_scope&error_description=nope",
2308	"/callback/id",
2309	);
2310	assert!(matches!(parsed, CallbackOutcome::Error(_)));
2311	}
2312
2313	#[test]
2314	fn store_key_does_not_include_raw_url_or_name() {
2315	let key = store_key("github", "https://example.com/mcp");
2316	assert!(key.starts_with("mcp_oauth_"));
2317	assert!(!key.contains("github"));
2318	assert!(!key.contains("example.com"));
2319	}
2320
2321	#[test]
2322	fn malformed_stored_oauth_diagnostic_omits_secret_contents_and_keys() {
2323	let secret = "cw-secret-mcp-oauth-4507";
2324	let serialized =
2325	format!(r#"{{"token_response":{{"access_token":"{secret}"}} trailing-junk}}"#);
2326	let error = parse_stored_oauth_tokens(&serialized, "private")
2327	.expect_err("malformed credential JSON must fail");
2328	let diagnostic = format!("{error:#}");
2329	assert!(!diagnostic.contains(secret), "{diagnostic}");
2330	assert!(!diagnostic.contains("access_token"), "{diagnostic}");
2331	assert!(diagnostic.contains("contents were omitted"), "{diagnostic}");
2332	}
2333
2334	#[test]
2335	fn auth_required_classifier_matches_http_401_shapes() {
2336	let err = anyhow!("MCP Streamable HTTP rejected status=401 Unauthorized");
2337	assert!(error_looks_auth_required(&err));
2338
2339	let err = anyhow!("authentication_required for remote server");
2340	assert!(error_looks_auth_required(&err));
2341
2342	let err = anyhow!("connection refused");
2343	assert!(!error_looks_auth_required(&err));
2344	}
2345
2346	#[test]
2347	fn auth_required_classifier_treats_rejected_grants_as_auth_required() {
2348	// A definitively rejected refresh grant is recoverable only by a
2349	// fresh login, so it must classify like a 401 on every surface.
2350	let err = anyhow!("refreshing MCP OAuth token for server wiki")
2351	.context("Server returned error response: invalid_grant: stale grant");
2352	assert!(error_looks_auth_required(&err));
2353	assert!(error_text_looks_auth_required(
2354	"wiki requires OAuth — run /mcp login wiki"
2355	));
2356	assert!(error_text_looks_auth_required("wiki: ◆ auth required"));
2357	assert!(!error_text_looks_auth_required(
2358	"invalid_request: missing parameter"
2359	));
2360	// rmcp's own `AuthError::AuthorizationRequired` wording: a stored
2361	// credential that can no longer be refreshed is a login, not a
2362	// transport failure.
2363	assert!(error_text_looks_auth_required(
2364	"refreshing MCP OAuth token for server wiki: OAuth authorization required"
2365	));
2366	assert!(!error_text_looks_auth_required(
2367	"authorization required for the requested file"
2368	));
2369	}
2370
2371	#[test]
2372	fn auth_required_login_hint_names_server() {
2373	let hint = auth_required_login_hint("nordic-mcp");
2374	assert!(hint.contains("nordic-mcp"));
2375	assert!(hint.contains("codewhale mcp login nordic-mcp"));
2376	assert!(!hint.contains("/mcp auth"));
2377	}
2378
2379	#[test]
2380	fn tui_reauth_hints_name_the_login_command() {
2381	for hint in [tui_reauth_hint(), tui_reauth_refresh_failed_hint()] {
2382	assert!(
2383	hint.contains("/mcp login <name>"),
2384	"OAuth recovery must name the implemented command"
2385	);
2386	assert!(
2387	!hint.contains("/mcp auth"),
2388	"OAuth recovery must not advertise a missing /mcp auth command"
2389	);
2390	}
2391	assert!(error_text_looks_auth_required(
2392	"MCP server rejected the request with 401 Unauthorized"
2393	));
2394	assert!(!error_text_looks_auth_required("connection refused"));
2395	}
2396
2397	#[tokio::test]
2398	async fn cancellable_oauth_drops_in_flight_flow_before_returning() {
2399	struct DropFlag(Arc<AtomicBool>);
2400	impl Drop for DropFlag {
2401	fn drop(&mut self) {
2402	self.0.store(true, Ordering::SeqCst);
2403	}
2404	}
2405
2406	let cancellation_token = CancellationToken::new();
2407	let cancel_from_task = cancellation_token.clone();
2408	let dropped = Arc::new(AtomicBool::new(false));
2409	let flow_dropped = Arc::clone(&dropped);
2410	let pending_flow = async move {
2411	let _guard = DropFlag(flow_dropped);
2412	std::future::pending::<Result<()>>().await
2413	};
2414	tokio::spawn(async move {
2415	tokio::task::yield_now().await;
2416	cancel_from_task.cancel();
2417	});
2418
2419	let error = run_cancellable_oauth(&cancellation_token, pending_flow)
2420	.await
2421	.expect_err("cancellation should stop the pending OAuth flow");
2422
2423	assert!(error.to_string().contains("OAuth login was cancelled"));
2424	assert!(
2425	dropped.load(Ordering::SeqCst),
2426	"the callback-server guard must be dropped before cancellation returns"
2427	);
2428	}
2429
2430	#[tokio::test]
2431	async fn callback_guard_aborts_accept_task_and_releases_fixed_port() -> Result<()> {
2432	let listener = TcpListener::bind(("127.0.0.1", 0)).await?;
2433	let addr = listener.local_addr()?;
2434	let (tx, _rx) = oneshot::channel();
2435	let guard = CallbackServerGuard {
2436	accept_task: spawn_callback_server(listener, tx, "/callback/test".to_string()),
2437	};
2438
2439	drop(guard);
2440
2441	let rebound = timeout(Duration::from_secs(1), async {
2442	loop {
2443	match TcpListener::bind(addr).await {
2444	Ok(listener) => break Ok(listener),
2445	Err(err) if err.kind() == std::io::ErrorKind::AddrInUse => {
2446	tokio::task::yield_now().await;
2447	}
2448	Err(err) => break Err(err),
2449	}
2450	}
2451	})
2452	.await
2453	.context("callback listener did not release its fixed port")??;
2454	drop(rebound);
2455	Ok(())
2456	}
2457
2458	async fn guarded_oauth_fixture(
2459	token_target: Option<String>,
2460	redirect_token: bool,
2461	) -> (
2462	String,
2463	Arc<std::sync::atomic::AtomicUsize>,
2464	tokio::task::JoinHandle<()>,
2465	) {
2466	use std::sync::atomic::{AtomicUsize, Ordering};
2467	let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2468	let addr = listener.local_addr().unwrap();
2469	let captured = Arc::new(AtomicUsize::new(0));
2470	let seen = Arc::clone(&captured);
2471	let task = tokio::spawn(async move {
2472	loop {
2473	let Ok((mut socket, _)) = listener.accept().await else {
2474	break;
2475	};
2476	let mut bytes = Vec::new();
2477	let mut buffer = [0u8; 2048];
2478	loop {
2479	let n = socket.read(&mut buffer).await.unwrap();
2480	if n == 0 {
2481	break;
2482	}
2483	bytes.extend_from_slice(&buffer[..n]);
2484	if bytes.windows(4).any(\|part\| part == b"\r\n\r\n") {
2485	break;
2486	}
2487	}
2488	let request = String::from_utf8_lossy(&bytes);
2489	let path = request.split_whitespace().nth(1).unwrap_or("");
2490	let (status, extra, body) = if path == "/.well-known/oauth-authorization-server" {
2491	("200 OK", String::new(), serde_json::json!({
2492	"issuer": format!("http://{addr}"),
2493	"authorization_endpoint": format!("http://{addr}/authorize"),
2494	"token_endpoint": token_target.clone().unwrap_or_else(\|\| format!("http://{addr}/token")),
2495	"registration_endpoint": format!("http://{addr}/register"),
2496	"response_types_supported": ["code"]
2497	}).to_string())
2498	} else if path == "/token" && redirect_token {
2499	(
2500	"307 Redirect",
2501	"Location: /capture\r\n".to_string(),
2502	String::new(),
2503	)
2504	} else if path == "/token" \|\| path == "/capture" {
2505	seen.fetch_add(1, Ordering::SeqCst);
2506	("200 OK", String::new(), r#"{"access_token":"new-fixture","token_type":"Bearer","refresh_token":"fixture-refresh"}"#.to_string())
2507	} else if path == "/register" {
2508	(
2509	"200 OK",
2510	String::new(),
2511	r#"{"client_id":"fixture-client","redirect_uris":[]}"#.to_string(),
2512	)
2513	} else {
2514	("404 Not Found", String::new(), String::new())
2515	};
2516	let response = format!(
2517	"HTTP/1.1 {status}\r\n{extra}Content-Type: application/json\r\nConnection: close\r\nContent-Length: {}\r\n\r\n{body}",
2518	body.len()
2519	);
2520	let _ = socket.write_all(response.as_bytes()).await;
2521	}
2522	});
2523	(format!("http://{addr}/mcp"), captured, task)
2524	}
2525
2526	async fn guarded_oauth_state(
2527	url: &str,
2528	network_policy: Option<&NetworkPolicyDecider>,
2529	) -> OAuthState {
2530	let client = McpHttpClient::new(
2531	url,
2532	false,
2533	false,
2534	false,
2535	network_policy,
2536	Duration::from_secs(1),
2537	Duration::from_secs(3),
2538	)
2539	.unwrap();
2540	let mut state = OAuthState::new_with_oauth_http_client(
2541	url,
2542	Arc::new(RecordingOAuthHttpClient::new(client)),
2543	)
2544	.await
2545	.unwrap();
2546	let tokens: OAuthTokenResponse = serde_json::from_value(serde_json::json!({
2547	"access_token":"fixture-access", "token_type":"Bearer", "refresh_token":"fixture-refresh"
2548	})).unwrap();
2549	state
2550	.set_credentials("fixture-client", tokens)
2551	.await
2552	.unwrap();
2553	state
2554	}
2555
2556	#[tokio::test]
2557	async fn guarded_oauth_refresh_honors_stop_and_preserves_normal_local_refresh() {
2558	use std::sync::atomic::Ordering;
2559	let _env = crate::test_support::lock_test_env();
2560	let _proxy = crate::test_support::EnvVarGuard::set("NO_PROXY", "*");
2561	for redirect in [true, false] {
2562	let (url, captured, task) = guarded_oauth_fixture(None, redirect).await;
2563	let state = guarded_oauth_state(&url, None).await;
2564	let result = state.refresh_token().await;
2565	if redirect {
2566	assert!(result.is_err(), "redirected refresh must not be followed");
2567	assert_eq!(captured.load(Ordering::SeqCst), 0);
2568	} else {
2569	result.unwrap();
2570	assert_eq!(captured.load(Ordering::SeqCst), 1);
2571	}
2572	task.abort();
2573	}
2574	}
2575
2576	#[tokio::test]
2577	async fn guarded_oauth_discovered_private_token_endpoint_never_receives_credentials() {
2578	let _env = crate::test_support::lock_test_env();
2579	let _proxy = crate::test_support::EnvVarGuard::set("NO_PROXY", "*");
2580	let destination = TcpListener::bind("127.0.0.1:0").await.unwrap();
2581	let target = format!("http://{}/token", destination.local_addr().unwrap());
2582	let (url, _, task) = guarded_oauth_fixture(Some(target), false).await;
2583	let state = guarded_oauth_state(&url, None).await;
2584	let error = tokio::time::timeout(Duration::from_secs(1), state.refresh_token())
2585	.await
2586	.expect("the destination guard rejects before attempting a network request")
2587	.unwrap_err();
2588	// rmcp intentionally wraps HTTP client failures as `Request failed`.
2589	// The observable invariant is an immediate failed refresh and no socket
2590	// at the private destination, rather than an SDK-specific error string.
2591	assert!(
2592	matches!(error, AuthError::TokenRefreshFailed(_)),
2593	"{error:#}"
2594	);
2595	assert!(
2596	tokio::time::timeout(Duration::from_millis(30), destination.accept())
2597	.await
2598	.is_err()
2599	);
2600	task.abort();
2601	}
2602
2603	#[tokio::test]
2604	async fn guarded_oauth_network_deny_applies_to_standalone_and_synthetic_login() {
2605	use crate::mcp::{AuthenticateToolStart, McpConfig, McpPool};
2606	use crate::network_policy::{DecisionToml, NetworkPolicy};
2607	let _env = crate::test_support::lock_test_env();
2608	let dir = tempfile::tempdir().unwrap();
2609	let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", dir.path());
2610	let _backend = crate::test_support::EnvVarGuard::set("CODEWHALE_SECRET_BACKEND", "file");
2611	let _proxy = crate::test_support::EnvVarGuard::set("NO_PROXY", "*");
2612	let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2613	let url = format!("http://{}/mcp", listener.local_addr().unwrap());
2614	let server: McpServerConfig =
2615	serde_json::from_value(serde_json::json!({"url":url})).unwrap();
2616	let denied = NetworkPolicyDecider::new(
2617	NetworkPolicy {
2618	default: DecisionToml::Deny,
2619	..NetworkPolicy::default()
2620	},
2621	None,
2622	);
2623	let mut config = McpConfig::default();
2624	config
2625	.servers
2626	.insert("network-guard".to_string(), server.clone());
2627	let pool = McpPool::new(config).with_network_policy(denied.clone());
2628	let error = match pool.begin_authenticate_tool("network-guard").await {
2629	Err(error) => error,
2630	Ok(_) => panic!("a configured denied origin must not start synthetic authentication"),
2631	};
2632	assert!(error.to_string().contains("network policy"), "{error:#}");
2633	assert!(oauth_login_support(&server, Some(&denied)).await.is_err());
2634	assert_eq!(
2635	auth_status_for_server("network-guard", &server, Some(&denied)).await,
2636	McpAuthStatus::Unsupported
2637	);
2638	assert!(
2639	perform_oauth_login_for_server(
2640	"network-guard",
2641	&server,
2642	Some(vec!["explicit".to_string()]),
2643	None,
2644	None,
2645	Some(&denied)
2646	)
2647	.await
2648	.is_err()
2649	);
2650	assert!(
2651	tokio::time::timeout(Duration::from_millis(30), listener.accept())
2652	.await
2653	.is_err()
2654	);
2655
2656	let (url, _, task) = guarded_oauth_fixture(None, false).await;
2657	let server: McpServerConfig =
2658	serde_json::from_value(serde_json::json!({"url":url})).unwrap();
2659	let allowed = NetworkPolicyDecider::new(
2660	NetworkPolicy {
2661	default: DecisionToml::Allow,
2662	..NetworkPolicy::default()
2663	},
2664	None,
2665	);
2666	let mut config = McpConfig::default();
2667	config.servers.insert("network-control".to_string(), server);
2668	let pool = McpPool::new(config).with_network_policy(allowed.clone());
2669	let AuthenticateToolStart::Login(login) = pool
2670	.begin_authenticate_tool("network-control")
2671	.await
2672	.unwrap()
2673	else {
2674	panic!("the configured local control must start a fresh login");
2675	};
2676	assert!(login.authorization_url().contains("/authorize"));
2677	// A later retry carries the same shared session ceiling.
2678	allowed.deny_session("127.0.0.1", "mcp");
2679	assert_eq!(
2680	login
2681	.network_policy
2682	.as_ref()
2683	.unwrap()
2684	.evaluate("127.0.0.1", "mcp"),
2685	crate::network_policy::Decision::Deny
2686	);
2687	drop(login);
2688	task.abort();
2689	}
2690
2691	#[tokio::test]
2692	async fn interactive_login_forces_the_consent_screen() {
2693	use crate::mcp::{AuthenticateToolStart, McpConfig, McpPool};
2694	use crate::network_policy::{DecisionToml, NetworkPolicy};
2695	let _env = crate::test_support::lock_test_env();
2696	let dir = tempfile::tempdir().unwrap();
2697	let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", dir.path());
2698	let _backend = crate::test_support::EnvVarGuard::set("CODEWHALE_SECRET_BACKEND", "file");
2699	let _proxy = crate::test_support::EnvVarGuard::set("NO_PROXY", "*");
2700	let (url, _, task) = guarded_oauth_fixture(None, false).await;
2701	let server: McpServerConfig =
2702	serde_json::from_value(serde_json::json!({"url":url})).unwrap();
2703	let allowed = NetworkPolicyDecider::new(
2704	NetworkPolicy {
2705	default: DecisionToml::Allow,
2706	..NetworkPolicy::default()
2707	},
2708	None,
2709	);
2710	let mut config = McpConfig::default();
2711	config.servers.insert("consent-probe".to_string(), server);
2712	let pool = McpPool::new(config).with_network_policy(allowed);
2713	let AuthenticateToolStart::Login(login) =
2714	pool.begin_authenticate_tool("consent-probe").await.unwrap()
2715	else {
2716	panic!("a fresh server must start an interactive login");
2717	};
2718
2719	// #6040: logout only clears this machine's token; the provider keeps
2720	// its standing grant, so the login URL must force the consent screen
2721	// or the same account/workspace is silently re-granted.
2722	// The URL carries the PKCE challenge and state, so the assertion
2723	// message reports only the fact that is being checked, never the URL.
2724	let forces_consent = login.authorization_url().contains("prompt=consent");
2725	assert!(
2726	forces_consent,
2727	"an interactive login must force consent (prompt=consent is missing from the authorization URL)"
2728	);
2729
2730	drop(login);
2731	task.abort();
2732	}
2733
2734	#[tokio::test]
2735	async fn guarded_oauth_refresh_keeps_live_session_network_denials() {
2736	use crate::network_policy::{DecisionToml, NetworkPolicy};
2737	let _env = crate::test_support::lock_test_env();
2738	let _proxy = crate::test_support::EnvVarGuard::set("NO_PROXY", "*");
2739	let (url, captured, task) = guarded_oauth_fixture(None, false).await;
2740	let policy = NetworkPolicyDecider::new(
2741	NetworkPolicy {
2742	default: DecisionToml::Allow,
2743	..NetworkPolicy::default()
2744	},
2745	None,
2746	);
2747	let state = guarded_oauth_state(&url, Some(&policy)).await;
2748	state.refresh_token().await.unwrap();
2749	assert_eq!(captured.load(Ordering::SeqCst), 1);
2750	policy.deny_session("127.0.0.1", "mcp");
2751	assert!(state.refresh_token().await.is_err());
2752	assert_eq!(
2753	captured.load(Ordering::SeqCst),
2754	1,
2755	"no refresh request after session denial"
2756	);
2757	task.abort();
2758	}
2759	}
2760

2760 lines RUST