Fix type errors in oauth_proxy and test_auth_integration

- Add null checks for client_id before using in OAuthTransaction, AuthorizationCode, AccessToken, RefreshToken
- Add null check for redirect_uris before len() call
- Import AuthorizeError from mcp.server.auth.provider
This commit is contained in:
Jeremiah Lowin 2025-11-15 12:06:33 -05:00
commit d40f22b3e6
3 changed files with 34 additions and 3 deletions

View file

@ -44,6 +44,7 @@ from mcp.server.auth.provider import (
AccessToken,
AuthorizationCode,
AuthorizationParams,
AuthorizeError,
RefreshToken,
TokenError,
)
@ -941,6 +942,8 @@ class OAuthProxy(OAuthProvider):
"""
# Create a ProxyDCRClient with configured redirect URI validation
if client_info.client_id is None:
raise ValueError("client_id is required for client registration")
proxy_client: ProxyDCRClient = ProxyDCRClient(
client_id=client_info.client_id,
client_secret=client_info.client_secret,
@ -970,7 +973,7 @@ class OAuthProxy(OAuthProvider):
logger.debug(
"Registered client %s with %d redirect URIs",
client_info.client_id,
len(proxy_client.redirect_uris),
len(proxy_client.redirect_uris) if proxy_client.redirect_uris else 0,
)
# -------------------------------------------------------------------------
@ -1007,6 +1010,10 @@ class OAuthProxy(OAuthProvider):
)
# Store transaction data for IdP callback processing
if client.client_id is None:
raise AuthorizeError(
error="invalid_client", error_description="Client ID is required"
)
transaction = OAuthTransaction(
txn_id=txn_id,
client_id=client.client_id,
@ -1085,6 +1092,10 @@ class OAuthProxy(OAuthProvider):
return None
# Create authorization code object with PKCE challenge
if client.client_id is None:
raise AuthorizeError(
error="invalid_client", error_description="Client ID is required"
)
return AuthorizationCode(
code=authorization_code,
client_id=client.client_id,
@ -1170,7 +1181,7 @@ class OAuthProxy(OAuthProvider):
expires_at=time.time() + expires_in,
token_type=idp_tokens.get("token_type", "Bearer"),
scope=" ".join(authorization_code.scopes),
client_id=client.client_id,
client_id=client.client_id or "",
created_at=time.time(),
raw_token_data=idp_tokens,
)
@ -1182,6 +1193,8 @@ class OAuthProxy(OAuthProvider):
logger.debug("Stored encrypted upstream tokens (jti=%s)", access_jti[:8])
# Issue minimal FastMCP access token (just a reference via JTI)
if client.client_id is None:
raise TokenError("invalid_client", "Client ID is required")
fastmcp_access_token = self._jwt_issuer.issue_access_token(
client_id=client.client_id,
scopes=authorization_code.scopes,
@ -1377,6 +1390,8 @@ class OAuthProxy(OAuthProvider):
)
# Issue new minimal FastMCP access token (just a reference via JTI)
if client.client_id is None:
raise TokenError("invalid_client", "Client ID is required")
new_access_jti = secrets.token_urlsafe(32)
new_fastmcp_access = self._jwt_issuer.issue_access_token(
client_id=client.client_id,

View file

@ -80,6 +80,8 @@ class InMemoryOAuthProvider(OAuthProvider):
f"Requested scopes are not valid: {', '.join(invalid_scopes)}"
)
if client_info.client_id is None:
raise ValueError("client_id is required for client registration")
if client_info.client_id in self.clients:
# As per RFC 7591, if client_id is already known, it's an update.
# For this simple provider, we'll treat it as re-registration.
@ -105,7 +107,7 @@ class InMemoryOAuthProvider(OAuthProvider):
# OAuthClientInformationFull should have a method like validate_redirect_uri
# For this test provider, we assume it's valid if it matches one in client_info
# The AuthorizationHandler already does robust validation using client.validate_redirect_uri
if params.redirect_uri not in client.redirect_uris:
if client.redirect_uris and params.redirect_uri not in client.redirect_uris:
# This check might be too simplistic if redirect_uris can be patterns
# or if params.redirect_uri is None and client has a default.
# However, the AuthorizationHandler handles the primary validation.
@ -124,6 +126,10 @@ class InMemoryOAuthProvider(OAuthProvider):
client_allowed_scopes = set(client.scope.split())
scopes_list = [s for s in scopes_list if s in client_allowed_scopes]
if client.client_id is None:
raise AuthorizeError(
error="invalid_client", error_description="Client ID is required"
)
auth_code = AuthorizationCode(
code=auth_code_value,
client_id=client.client_id,
@ -180,6 +186,8 @@ class InMemoryOAuthProvider(OAuthProvider):
time.time() + DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS
)
if client.client_id is None:
raise TokenError("invalid_client", "Client ID is required")
self.access_tokens[access_token_value] = AccessToken(
token=access_token_value,
client_id=client.client_id,
@ -250,6 +258,8 @@ class InMemoryOAuthProvider(OAuthProvider):
time.time() + DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS
)
if client.client_id is None:
raise TokenError("invalid_client", "Client ID is required")
self.access_tokens[new_access_token_value] = AccessToken(
token=new_access_token_value,
client_id=client.client_id,

View file

@ -49,6 +49,8 @@ class MockOAuthProvider(OAuthAuthorizationServerProvider):
) -> str:
# toy authorize implementation which just immediately generates an authorization
# code and completes the redirect
if client.client_id is None:
raise ValueError("client_id is required")
code = AuthorizationCode(
code=f"code_{int(time.time())}",
client_id=client.client_id,
@ -79,6 +81,8 @@ class MockOAuthProvider(OAuthAuthorizationServerProvider):
refresh_token = f"refresh_{secrets.token_hex(32)}"
# Store the tokens
if client.client_id is None:
raise ValueError("client_id is required")
self.tokens[access_token] = AccessToken(
token=access_token,
client_id=client.client_id,
@ -142,6 +146,8 @@ class MockOAuthProvider(OAuthAuthorizationServerProvider):
new_refresh_token = f"refresh_{secrets.token_hex(32)}"
# Store the new tokens
if client.client_id is None:
raise ValueError("client_id is required")
self.tokens[new_access_token] = AccessToken(
token=new_access_token,
client_id=client.client_id,