refactor: extract OAuth helper functions and simplify provider state (#1586)

src/mcp/client/auth/utils.py

@@ -0,0 +1,250 @@
+import logging
+import re
+from urllib.parse import urljoin, urlparse
+
+from httpx import Request, Response
+from pydantic import ValidationError
+
+from mcp.client.auth import OAuthRegistrationError, OAuthTokenError
+from mcp.client.streamable_http import MCP_PROTOCOL_VERSION
+from mcp.shared.auth import (
+    OAuthClientInformationFull,
+    OAuthClientMetadata,
+    OAuthMetadata,
+    OAuthToken,
+    ProtectedResourceMetadata,
+)
+from mcp.types import LATEST_PROTOCOL_VERSION
+
+logger = logging.getLogger(__name__)
+
+
+def extract_field_from_www_auth(response: Response, field_name: str) -> str | None:
+    """
+    Extract field from WWW-Authenticate header.
+
+    Returns:
+        Field value if found in WWW-Authenticate header, None otherwise
+    """
+    www_auth_header = response.headers.get("WWW-Authenticate")
+    if not www_auth_header:
+        return None
+
+    # Pattern matches: field_name="value" or field_name=value (unquoted)
+    pattern = rf'{field_name}=(?:"([^"]+)"|([^\s,]+))'
+    match = re.search(pattern, www_auth_header)
+
+    if match:
+        # Return quoted value if present, otherwise unquoted value
+        return match.group(1) or match.group(2)
+
+    return None
+
+
+def extract_scope_from_www_auth(response: Response) -> str | None:
+    """
+    Extract scope parameter from WWW-Authenticate header as per RFC6750.
+
+    Returns:
+        Scope string if found in WWW-Authenticate header, None otherwise
+    """
+    return extract_field_from_www_auth(response, "scope")
+
+
+def extract_resource_metadata_from_www_auth(response: Response) -> str | None:
+    """
+    Extract protected resource metadata URL from WWW-Authenticate header as per RFC9728.
+
+    Returns:
+        Resource metadata URL if found in WWW-Authenticate header, None otherwise
+    """
+    if not response or response.status_code != 401:
+        return None  # pragma: no cover
+
+    return extract_field_from_www_auth(response, "resource_metadata")
+
+
+def build_protected_resource_discovery_urls(www_auth_url: str | None, server_url: str) -> list[str]:
+    """
+    Build ordered list of URLs to try for protected resource metadata discovery.
+
+    Per SEP-985, the client MUST:
+    1. Try resource_metadata from WWW-Authenticate header (if present)
+    2. Fall back to path-based well-known URI: /.well-known/oauth-protected-resource/{path}
+    3. Fall back to root-based well-known URI: /.well-known/oauth-protected-resource
+
+    Args:
+        www_auth_url: optional resource_metadata url extracted from the WWW-Authenticate header
+        server_url: server url
+
+    Returns:
+        Ordered list of URLs to try for discovery
+    """
+    urls: list[str] = []
+
+    # Priority 1: WWW-Authenticate header with resource_metadata parameter
+    if www_auth_url:
+        urls.append(www_auth_url)
+
+    # Priority 2-3: Well-known URIs (RFC 9728)
+    parsed = urlparse(server_url)
+    base_url = f"{parsed.scheme}://{parsed.netloc}"
+
+    # Priority 2: Path-based well-known URI (if server has a path component)
+    if parsed.path and parsed.path != "/":
+        path_based_url = urljoin(base_url, f"/.well-known/oauth-protected-resource{parsed.path}")
+        urls.append(path_based_url)
+
+    # Priority 3: Root-based well-known URI
+    root_based_url = urljoin(base_url, "/.well-known/oauth-protected-resource")
+    urls.append(root_based_url)
+
+    return urls
+
+
+def get_client_metadata_scopes(
+    www_authenticate_scope: str | None,
+    protected_resource_metadata: ProtectedResourceMetadata | None,
+    authorization_server_metadata: OAuthMetadata | None = None,
+) -> str | None:
+    """Select scopes as outlined in the 'Scope Selection Strategy' in the MCP spec."""
+    # Per MCP spec, scope selection priority order:
+    # 1. Use scope from WWW-Authenticate header (if provided)
+    # 2. Use all scopes from PRM scopes_supported (if available)
+    # 3. Omit scope parameter if neither is available
+
+    if www_authenticate_scope is not None:
+        # Priority 1: WWW-Authenticate header scope
+        return www_authenticate_scope
+    elif protected_resource_metadata is not None and protected_resource_metadata.scopes_supported is not None:
+        # Priority 2: PRM scopes_supported
+        return " ".join(protected_resource_metadata.scopes_supported)
+    elif authorization_server_metadata is not None and authorization_server_metadata.scopes_supported is not None:
+        return " ".join(authorization_server_metadata.scopes_supported)  # pragma: no cover
+    else:
+        # Priority 3: Omit scope parameter
+        return None
+
+
+def get_discovery_urls(auth_server_url: str) -> list[str]:
+    """Generate ordered list of (url, type) tuples for discovery attempts."""
+    urls: list[str] = []
+    parsed = urlparse(auth_server_url)
+    base_url = f"{parsed.scheme}://{parsed.netloc}"
+
+    # RFC 8414: Path-aware OAuth discovery
+    if parsed.path and parsed.path != "/":
+        oauth_path = f"/.well-known/oauth-authorization-server{parsed.path.rstrip('/')}"
+        urls.append(urljoin(base_url, oauth_path))
+
+    # OAuth root fallback
+    urls.append(urljoin(base_url, "/.well-known/oauth-authorization-server"))
+
+    # RFC 8414 section 5: Path-aware OIDC discovery
+    # See https://www.rfc-editor.org/rfc/rfc8414.html#section-5
+    if parsed.path and parsed.path != "/":
+        oidc_path = f"/.well-known/openid-configuration{parsed.path.rstrip('/')}"
+        urls.append(urljoin(base_url, oidc_path))
+
+    # OIDC 1.0 fallback (appends to full URL per OIDC spec)
+    oidc_fallback = f"{auth_server_url.rstrip('/')}/.well-known/openid-configuration"
+    urls.append(oidc_fallback)
+
+    return urls
+
+
+async def handle_protected_resource_response(
+    response: Response,
+) -> ProtectedResourceMetadata | None:
+    """
+    Handle protected resource metadata discovery response.
+
+    Per SEP-985, supports fallback when discovery fails at one URL.
+
+    Returns:
+        True if metadata was successfully discovered, False if we should try next URL
+    """
+    if response.status_code == 200:
+        try:
+            content = await response.aread()
+            metadata = ProtectedResourceMetadata.model_validate_json(content)
+            return metadata
+
+        except ValidationError:  # pragma: no cover
+            # Invalid metadata - try next URL
+            return None
+    else:
+        # Not found - try next URL in fallback chain
+        return None
+
+
+async def handle_auth_metadata_response(response: Response) -> tuple[bool, OAuthMetadata | None]:
+    if response.status_code == 200:
+        try:
+            content = await response.aread()
+            asm = OAuthMetadata.model_validate_json(content)
+            return True, asm
+        except ValidationError:  # pragma: no cover
+            return True, None
+    elif response.status_code < 400 or response.status_code >= 500:
+        return False, None  # Non-4XX error, stop trying
+    return True, None
+
+
+def create_oauth_metadata_request(url: str) -> Request:
+    return Request("GET", url, headers={MCP_PROTOCOL_VERSION: LATEST_PROTOCOL_VERSION})
+
+
+def create_client_registration_request(
+    auth_server_metadata: OAuthMetadata | None, client_metadata: OAuthClientMetadata, auth_base_url: str
+) -> Request:
+    """Build registration request or skip if already registered."""
+
+    if auth_server_metadata and auth_server_metadata.registration_endpoint:
+        registration_url = str(auth_server_metadata.registration_endpoint)
+    else:
+        registration_url = urljoin(auth_base_url, "/register")
+
+    registration_data = client_metadata.model_dump(by_alias=True, mode="json", exclude_none=True)
+
+    return Request("POST", registration_url, json=registration_data, headers={"Content-Type": "application/json"})
+
+
+async def handle_registration_response(response: Response) -> OAuthClientInformationFull:
+    """Handle registration response."""
+    if response.status_code not in (200, 201):
+        await response.aread()
+        raise OAuthRegistrationError(f"Registration failed: {response.status_code} {response.text}")
+
+    try:
+        content = await response.aread()
+        client_info = OAuthClientInformationFull.model_validate_json(content)
+        return client_info
+        # self.context.client_info = client_info
+        # await self.context.storage.set_client_info(client_info)
+    except ValidationError as e:  # pragma: no cover
+        raise OAuthRegistrationError(f"Invalid registration response: {e}")
+
+
+async def handle_token_response_scopes(
+    response: Response,
+) -> OAuthToken:
+    """Parse and validate token response with optional scope validation.
+
+    Parses token response JSON. Callers should check response.status_code before calling.
+
+    Args:
+        response: HTTP response from token endpoint (status already checked by caller)
+
+    Returns:
+        Validated OAuthToken model
+
+    Raises:
+        OAuthTokenError: If response JSON is invalid
+    """
+    try:
+        content = await response.aread()
+        token_response = OAuthToken.model_validate_json(content)
+        return token_response
+    except ValidationError as e:  # pragma: no cover
+        raise OAuthTokenError(f"Invalid token response: {e}")