SSO MCP Server

"""Authentication exceptions with actionable error messages. Provides user-friendly error messages that guide users on how to resolve auth issues. """ from __future__ import annotations class AuthError(Exception): """Base class for authentication errors. Attributes: message: User-friendly error message. action: Suggested action to resolve the error. code: Error code for programmatic handling. """ def __init__( self, message: str, action: str | None = None, code: str = "AUTH_ERROR", ) -> None: """Initialize auth error. Args: message: User-friendly error message. action: Suggested action to resolve the error. code: Error code for programmatic handling. """ self.message = message self.action = action self.code = code super().__init__(message) def __str__(self) -> str: """Return formatted error message with action.""" if self.action: return f"{self.message}\n\nAction: {self.action}" return self.message class ConfigurationError(AuthError): """Error for missing or invalid configuration.""" def __init__(self, message: str, action: str | None = None) -> None: """Initialize configuration error. Args: message: Description of what configuration is missing/invalid. action: How to fix the configuration. """ super().__init__( message=message, action=action or "Check your .env file and ensure all required variables are set.", code="CONFIG_ERROR", ) class TokenExpiredError(AuthError): """Error when token has expired and cannot be refreshed.""" def __init__(self) -> None: """Initialize token expired error.""" super().__init__( message="Your session has expired and could not be renewed.", action="Please log in again using the browser-based authentication.", code="TOKEN_EXPIRED", ) class SilentAuthFailedError(AuthError): """Error when silent authentication fails.""" def __init__(self, reason: str | None = None) -> None: """Initialize silent auth failed error. Args: reason: Reason for the failure. """ msg = "Silent authentication failed." if reason: msg = f"{msg} Reason: {reason}" super().__init__( message=msg, action="Browser-based authentication will be attempted.", code="SILENT_AUTH_FAILED", ) class InteractiveAuthFailedError(AuthError): """Error when interactive browser authentication fails.""" def __init__(self, error: str | None = None, error_description: str | None = None) -> None: """Initialize interactive auth failed error. Args: error: OAuth error code. error_description: OAuth error description. """ msg = "Browser authentication failed." if error: msg = f"{msg} Error: {error}" if error_description: msg = f"{msg} ({error_description})" super().__init__( message=msg, action="Please ensure you have access to the Azure tenant and try again. " "If the problem persists, contact your administrator.", code="INTERACTIVE_AUTH_FAILED", ) class AuthNotConfiguredError(AuthError): """Error when authentication is not properly configured.""" def __init__(self) -> None: """Initialize auth not configured error.""" super().__init__( message="Authentication is not configured.", action="Ensure the server is properly initialized with Azure credentials.", code="AUTH_NOT_CONFIGURED", ) class NotAuthenticatedError(AuthError): """Error when user is not authenticated but authentication is required.""" def __init__(self) -> None: """Initialize not authenticated error.""" super().__init__( message="Authentication required.", action="Please complete the SSO login flow to continue.", code="NOT_AUTHENTICATED", ) # ============================================================================= # Cloud Mode Exceptions (OAuth 2.1 Resource Server) # ============================================================================= class CloudAuthError(AuthError): """Base class for cloud mode authentication errors. Cloud mode errors include HTTP status codes and WWW-Authenticate header content per RFC 6750. Attributes: http_status: HTTP status code for the error response. www_authenticate: Content for WWW-Authenticate header. """ def __init__( self, message: str, action: str | None = None, code: str = "CLOUD_AUTH_ERROR", http_status: int = 401, error: str = "invalid_token", error_description: str | None = None, ) -> None: """Initialize cloud auth error. Args: message: User-friendly error message. action: Suggested action to resolve the error. code: Error code for programmatic handling. http_status: HTTP status code (401 or 403). error: OAuth error code for WWW-Authenticate header. error_description: OAuth error description. """ super().__init__(message=message, action=action, code=code) self.http_status = http_status self.error = error self.error_description = error_description or message class MissingAuthorizationError(CloudAuthError): """Error when Authorization header is missing.""" def __init__(self) -> None: """Initialize missing authorization error.""" super().__init__( message="Authorization header is required.", action="Include 'Authorization: Bearer <token>' header in your request.", code="MISSING_AUTHORIZATION", http_status=401, error="invalid_request", error_description="Missing Authorization header", ) class InvalidTokenError(CloudAuthError): """Error when the token is malformed or has an invalid signature.""" def __init__(self, reason: str | None = None) -> None: """Initialize invalid token error. Args: reason: Specific reason the token is invalid. """ msg = "The access token is invalid." if reason: msg = f"{msg} {reason}" super().__init__( message=msg, action="Obtain a new access token from the authorization server.", code="INVALID_TOKEN", http_status=401, error="invalid_token", error_description=reason or "Token validation failed", ) class TokenSignatureError(CloudAuthError): """Error when the token signature verification fails.""" def __init__(self) -> None: """Initialize token signature error.""" super().__init__( message="Token signature verification failed.", action="Ensure the token was issued by a trusted authorization server.", code="INVALID_SIGNATURE", http_status=401, error="invalid_token", error_description="Signature verification failed", ) class CloudTokenExpiredError(CloudAuthError): """Error when the Bearer token has expired.""" def __init__(self) -> None: """Initialize cloud token expired error.""" super().__init__( message="The access token has expired.", action="Obtain a new access token from the authorization server.", code="TOKEN_EXPIRED", http_status=401, error="invalid_token", error_description="Token has expired", ) class InvalidAudienceError(CloudAuthError): """Error when the token audience doesn't match this server.""" def __init__(self, expected: str, actual: str | list[str]) -> None: """Initialize invalid audience error. Args: expected: Expected audience (this server's resource identifier). actual: Actual audience in the token. """ actual_str = actual if isinstance(actual, str) else ", ".join(actual) super().__init__( message=f"Token audience mismatch. Expected: {expected}, got: {actual_str}", action="Request a token with the correct audience (resource parameter).", code="INVALID_AUDIENCE", http_status=401, error="invalid_token", error_description="Token audience does not match this resource", ) class InvalidIssuerError(CloudAuthError): """Error when the token issuer is not trusted.""" def __init__(self, issuer: str) -> None: """Initialize invalid issuer error. Args: issuer: The issuer in the token. """ super().__init__( message=f"Token issuer '{issuer}' is not trusted.", action="Obtain a token from a trusted authorization server.", code="INVALID_ISSUER", http_status=401, error="invalid_token", error_description="Token issuer is not trusted", ) class InsufficientScopeError(CloudAuthError): """Error when the token lacks required scopes.""" def __init__(self, required_scopes: list[str], actual_scopes: list[str]) -> None: """Initialize insufficient scope error. Args: required_scopes: Scopes required for the operation. actual_scopes: Scopes present in the token. """ required_str = " ".join(required_scopes) actual_str = " ".join(actual_scopes) if actual_scopes else "(none)" super().__init__( message=f"Insufficient scope. Required: {required_str}, got: {actual_str}", action="Request a token with the required scopes.", code="INSUFFICIENT_SCOPE", http_status=403, error="insufficient_scope", error_description=f"Required scope: {required_str}", ) self.required_scopes = required_scopes self.actual_scopes = actual_scopes class JWKSFetchError(CloudAuthError): """Error when JWKS cannot be fetched from the authorization server.""" def __init__(self, issuer: str, reason: str | None = None) -> None: """Initialize JWKS fetch error. Args: issuer: The issuer whose JWKS couldn't be fetched. reason: Specific reason for the failure. """ msg = f"Failed to fetch signing keys from issuer: {issuer}" if reason: msg = f"{msg}. {reason}" super().__init__( message=msg, action="Ensure the authorization server is reachable and properly configured.", code="JWKS_FETCH_ERROR", http_status=401, error="invalid_token", error_description="Unable to verify token signature", )