fix: mypy issue ony incompatible types in assignment

Author: davidhuser

demo_project/api/dependencies.py

@@ -11,7 +11,7 @@
     MultiTenantAzureAuthorizationCodeBearer,
     SingleTenantAzureAuthorizationCodeBearer,
 )
-from fastapi_azure_auth.exceptions import InvalidAuthHttp
+from fastapi_azure_auth.exceptions import ForbiddenHttp, UnauthorizedHttp
 from fastapi_azure_auth.user import User
 
 log = logging.getLogger(__name__)
@@ -30,7 +30,7 @@ async def validate_is_admin_user(user: User = Depends(azure_scheme)) -> None:
     Raises a 401 authentication error if not.
     """
     if 'AdminUser' not in user.roles:
-        raise InvalidAuthHttp('User is not an AdminUser')
+        raise ForbiddenHttp('User is not an AdminUser')
 
 
 class IssuerFetcher:
@@ -44,7 +44,7 @@ def __init__(self) -> None:
     async def __call__(self, tid: str) -> str:
         """
         Check if memory cache needs to be updated or not, and then returns an issuer for a given tenant
-        :raises InvalidAuth when it's not a valid tenant
+        :raises Unauthorized when it's not a valid tenant
         """
         refresh_time = datetime.now() - timedelta(hours=1)
         if not self._config_timestamp or self._config_timestamp < refresh_time:
@@ -58,7 +58,7 @@ async def __call__(self, tid: str) -> str:
             return self.tid_to_iss[tid]
         except Exception as error:
             log.exception('`iss` not found for `tid` %s. Error %s', tid, error)
-            raise InvalidAuthHttp('You must be an Intility customer to access this resource')
+            raise UnauthorizedHttp('You must be an Intility customer to access this resource')
 
 
 issuer_fetcher = IssuerFetcher()
@@ -101,7 +101,7 @@ async def multi_auth(
         return azure_auth
     if api_key == 'JonasIsCool':
         return api_key
-    raise InvalidAuthHttp('You must either provide a valid bearer token or API key')
+    raise UnauthorizedHttp('You must either provide a valid bearer token or API key')
 
 
 async def multi_auth_b2c(
@@ -115,4 +115,4 @@ async def multi_auth_b2c(
         return azure_auth
     if api_key == 'JonasIsCool':
         return api_key
-    raise InvalidAuthHttp('You must either provide a valid bearer token or API key')
+    raise UnauthorizedHttp('You must either provide a valid bearer token or API key')

demo_project/core/config.py

@@ -13,9 +13,9 @@ class AzureActiveDirectory(BaseSettings):  # type: ignore[misc, valid-type]
     OPENAPI_CLIENT_ID: str = Field(default='')
     TENANT_ID: str = Field(default='')
     APP_CLIENT_ID: str = Field(default='')
-    AUTH_URL: AnyHttpUrl = Field(default='https://dummy.com/')
-    CONFIG_URL: AnyHttpUrl = Field(default='https://dummy.com/')
-    TOKEN_URL: AnyHttpUrl = Field(default='https://dummy.com/')
+    AUTH_URL: AnyHttpUrl = Field(default=AnyHttpUrl('https://dummy.com/'))
+    CONFIG_URL: AnyHttpUrl = Field(default=AnyHttpUrl('https://dummy.com/'))
+    TOKEN_URL: AnyHttpUrl = Field(default=AnyHttpUrl('https://dummy.com/'))
     GRAPH_SECRET: str = Field(default='')
     CLIENT_SECRET: str = Field(default='')
 

docs/docs/multi-tenant/accept_specific_tenants_only.mdx

@@ -15,7 +15,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from pydantic import AnyHttpUrl
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from fastapi_azure_auth import MultiTenantAzureAuthorizationCodeBearer
-from fastapi_azure_auth.exceptions import InvalidAuth
+from fastapi_azure_auth.exceptions import Unauthorized
 
 
 class Settings(BaseSettings):
@@ -56,7 +56,7 @@ async def check_if_valid_tenant(tid: str) -> str:
     try:
         return tid_to_iss_mapping[tid]
     except KeyError:
-        raise InvalidAuth('Tenant not allowed')
+        raise Unauthorized('Tenant not allowed')
 
 azure_scheme = MultiTenantAzureAuthorizationCodeBearer(
     app_client_id=settings.APP_CLIENT_ID,
@@ -86,7 +86,7 @@ if __name__ == '__main__':
 ```
 
 We're first creating an `async function`, which takes a `tid` as an argument, and returns the tenant ID's `iss` if it's a valid tenant.
-If it's not a valid tenant, it has to raise an `InvalidAuth()` exception.
+If it's not a valid tenant, it has to raise an `Unauthorized()` exception.
 
 ## More sophisticated callable
 If you want to cache these results in memory, you can do so by creating a more sophisticated callable:
@@ -103,7 +103,7 @@ class IssuerFetcher:
     async def __call__(self, tid: str) -> str:
         """
         Check if memory cache needs to be updated or not, and then returns an issuer for a given tenant
-        :raises InvalidAuth when it's not a valid tenant
+        :raises Unauthorized when it's not a valid tenant
         """
         refresh_time = datetime.now() - timedelta(hours=1)
         if not self._config_timestamp or self._config_timestamp < refresh_time:
@@ -117,7 +117,7 @@ class IssuerFetcher:
             return self.tid_to_iss[tid]
         except Exception as error:
             log.exception('`iss` not found for `tid` %s. Error %s', tid, error)
-            raise InvalidAuth('You must be an Intility customer to access this resource')
+            raise Unauthorized('You must be an Intility customer to access this resource')
 
 
 issuer_fetcher = IssuerFetcher()

docs/docs/usage-and-faq/guest_users.mdx

@@ -39,15 +39,15 @@ would like to lock down specific endpoints.
 
 ```python title="security.py"
 from fastapi import Depends
-from fastapi_azure_auth.exceptions import InvalidAuth
+from fastapi_azure_auth.exceptions import Unauthorized
 from fastapi_azure_auth.user import User
 
 async def deny_guest_users(user: User = Depends(azure_scheme)) -> None:
     """
     Deny guest users
     """
     if user.is_guest:
-        raise InvalidAuth('Guest user not allowed')
+        raise Unauthorized('Guest user not allowed')
 ```
 
 
@@ -57,15 +57,15 @@ Alternatively, after [FastAPI 0.95.0](https://github.com/tiangolo/fastapi/releas
 ```python title="security.py"
 from typing import Annotated
 from fastapi import Depends
-from fastapi_azure_auth.exceptions import InvalidAuth
+from fastapi_azure_auth.exceptions import Unauthorized
 from fastapi_azure_auth.user import User
 
 async def deny_guest_users(user: User = Depends(azure_scheme)) -> None:
     """
     Deny guest users
     """
     if user.is_guest:
-        raise InvalidAuth('Guest user not allowed')
+        raise Unauthorized('Guest user not allowed')
 
 NonGuestUser = Annotated[User, Depends(deny_guest_users)]
 ```

docs/docs/usage-and-faq/locking_down_on_roles.mdx

@@ -30,7 +30,7 @@ You can lock down on roles by creating your own wrapper dependency:
 
 ```python title="dependencies.py"
 from fastapi import Depends
-from fastapi_azure_auth.exceptions import InvalidAuth
+from fastapi_azure_auth.exceptions import Unauthorized
 from fastapi_azure_auth.user import User
 
 async def validate_is_admin_user(user: User = Depends(azure_scheme)) -> None:
@@ -39,7 +39,7 @@ async def validate_is_admin_user(user: User = Depends(azure_scheme)) -> None:
     Raises a 401 authentication error if not.
     """
     if 'AdminUser' not in user.roles:
-        raise InvalidAuth('User is not an AdminUser')
+        raise Unauthorized('User is not an AdminUser')
 ```
 
 and then use this dependency over `azure_scheme`.
@@ -51,7 +51,7 @@ Alternatively, after [FastAPI 0.95.0](https://github.com/tiangolo/fastapi/releas
 ```python title="security.py"
 from typing import Annotated
 from fastapi import Depends
-from fastapi_azure_auth.exceptions import InvalidAuth
+from fastapi_azure_auth.exceptions import Unauthorized
 from fastapi_azure_auth.user import User
 
 async def validate_is_admin_user(user: User = Depends(azure_scheme)) -> None:
@@ -60,7 +60,7 @@ async def validate_is_admin_user(user: User = Depends(azure_scheme)) -> None:
     Raises a 401 authentication error if not.
     """
     if 'AdminUser' not in user.roles:
-        raise InvalidAuth('User is not an AdminUser')
+        raise Unauthorized('User is not an AdminUser')
 
 AdminUser = Annotated[User, Depends(validate_is_admin_user)]
 ```

fastapi_azure_auth/auth.py

@@ -17,7 +17,16 @@
 )
 from starlette.requests import HTTPConnection
 
-from fastapi_azure_auth.exceptions import InvalidAuth, InvalidAuthHttp, InvalidAuthWebSocket
+from fastapi_azure_auth.exceptions import (
+    Forbidden,
+    ForbiddenHttp,
+    ForbiddenWebSocket,
+    InvalidRequest,
+    InvalidRequestHttp,
+    Unauthorized,
+    UnauthorizedHttp,
+    UnauthorizedWebSocket,
+)
 from fastapi_azure_auth.openid_config import OpenIdConfig
 from fastapi_azure_auth.user import User
 from fastapi_azure_auth.utils import get_unverified_claims, get_unverified_header, is_guest
@@ -148,28 +157,28 @@ async def __call__(self, request: HTTPConnection, security_scopes: SecurityScope
             access_token = await self.extract_access_token(request)
             try:
                 if access_token is None:
-                    raise InvalidAuth('No access token provided', request=request)
+                    raise InvalidRequest('No access token provided', request=request)
                 # Extract header information of the token.
                 header: dict[str, Any] = get_unverified_header(access_token)
                 claims: dict[str, Any] = get_unverified_claims(access_token)
             except Exception as error:
                 log.warning('Malformed token received. %s. Error: %s', access_token, error, exc_info=True)
-                raise InvalidAuth(detail='Invalid token format', request=request) from error
+                raise Unauthorized(detail='Invalid token format', request=request) from error
 
             user_is_guest: bool = is_guest(claims=claims)
             if not self.allow_guest_users and user_is_guest:
                 log.info('User denied, is a guest user', claims)
-                raise InvalidAuth(detail='Guest users not allowed', request=request)
+                raise Forbidden(detail='Guest users not allowed', request=request)
 
             for scope in security_scopes.scopes:
                 token_scope_string = claims.get('scp', '')
                 log.debug('Scopes: %s', token_scope_string)
                 if not isinstance(token_scope_string, str):
-                    raise InvalidAuth('Token contains invalid formatted scopes', request=request)
+                    raise Forbidden('Token contains invalid formatted scopes', request=request)
 
                 token_scopes = token_scope_string.split(' ')
                 if scope not in token_scopes:
-                    raise InvalidAuth('Required scope missing', request=request)
+                    raise Forbidden('Required scope missing', request=request)
             # Load new config if old
             await self.openid_config.load_config()
 
@@ -211,27 +220,34 @@ async def __call__(self, request: HTTPConnection, security_scopes: SecurityScope
                 MissingRequiredClaimError,
             ) as error:
                 log.info('Token contains invalid claims. %s', error)
-                raise InvalidAuth(detail='Token contains invalid claims', request=request) from error
+                raise Unauthorized(detail='Token contains invalid claims', request=request) from error
             except ExpiredSignatureError as error:
                 log.info('Token signature has expired. %s', error)
-                raise InvalidAuth(detail='Token signature has expired', request=request) from error
+                raise Unauthorized(detail='Token signature has expired', request=request) from error
             except InvalidTokenError as error:
                 log.warning('Invalid token. Error: %s', error, exc_info=True)
-                raise InvalidAuth(detail='Unable to validate token', request=request) from error
+                raise Unauthorized(detail='Unable to validate token', request=request) from error
             except Exception as error:
                 # Extra failsafe in case of a bug in a future version of the jwt library
                 log.exception('Unable to process jwt token. Uncaught error: %s', error)
-                raise InvalidAuth(detail='Unable to process token', request=request) from error
+                raise Unauthorized(detail='Unable to process token', request=request) from error
             log.warning('Unable to verify token. No signing keys found')
-            raise InvalidAuth(detail='Unable to verify token, no signing keys found', request=request)
-        except (InvalidAuthHttp, InvalidAuthWebSocket, HTTPException):
+            raise Unauthorized(detail='Unable to verify token, no signing keys found', request=request)
+        except (
+            InvalidRequestHttp,
+            UnauthorizedHttp,
+            UnauthorizedWebSocket,
+            ForbiddenHttp,
+            ForbiddenWebSocket,
+            HTTPException,
+        ):
             if not self.auto_error:
                 return None
             raise
         except Exception as error:
             if not self.auto_error:
                 return None
-            raise InvalidAuth(detail='Unable to validate token', request=request) from error
+            raise InvalidRequest(detail='Unable to validate token', request=request) from error
 
     async def extract_access_token(self, request: HTTPConnection) -> Optional[str]:
         """

fastapi_azure_auth/exceptions.py

@@ -4,33 +4,80 @@
 from starlette.requests import HTTPConnection
 
 
-class InvalidAuthHttp(HTTPException):
-    """
-    Exception raised when the user is not authorized over HTTP
-    """
+class InvalidRequestHttp(HTTPException):
+    """HTTP exception for malformed/invalid requests"""
 
     def __init__(self, detail: str) -> None:
         super().__init__(
-            status_code=status.HTTP_401_UNAUTHORIZED, detail=detail, headers={'WWW-Authenticate': 'Bearer'}
+            status_code=status.HTTP_400_BAD_REQUEST, detail={"error": "invalid_request", "message": detail}
         )
 
 
-class InvalidAuthWebSocket(WebSocketException):
-    """
-    Exception raised when the user is not authorized over WebSockets
-    """
+class InvalidRequestWebSocket(WebSocketException):
+    """WebSocket exception for malformed/invalid requests"""
 
     def __init__(self, detail: str) -> None:
         super().__init__(
-            code=status.WS_1008_POLICY_VIOLATION,
-            reason=detail,
+            code=status.WS_1008_POLICY_VIOLATION, reason=str({"error": "invalid_request", "message": detail})
         )
 
 
-def InvalidAuth(detail: str, request: HTTPConnection) -> InvalidAuthHttp | InvalidAuthWebSocket:
-    """
-    Returns the correct exception based on the connection type
-    """
+class UnauthorizedHttp(HTTPException):
+    """HTTP exception for authentication failures"""
+
+    def __init__(self, detail: str) -> None:
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail={"error": "invalid_token", "message": detail},
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+
+class UnauthorizedWebSocket(WebSocketException):
+    """WebSocket exception for authentication failures"""
+
+    def __init__(self, detail: str) -> None:
+        super().__init__(
+            code=status.WS_1008_POLICY_VIOLATION, reason=str({"error": "invalid_token", "message": detail})
+        )
+
+
+class ForbiddenHttp(HTTPException):
+    """HTTP exception for insufficient permissions"""
+
+    def __init__(self, detail: str) -> None:
+        super().__init__(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail={"error": "insufficient_scope", "message": detail},
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+
+class ForbiddenWebSocket(WebSocketException):
+    """WebSocket exception for insufficient permissions"""
+
+    def __init__(self, detail: str) -> None:
+        super().__init__(
+            code=status.WS_1008_POLICY_VIOLATION, reason=str({"error": "insufficient_scope", "message": detail})
+        )
+
+
+def InvalidRequest(detail: str, request: HTTPConnection) -> InvalidRequestHttp | InvalidRequestWebSocket:
+    """Factory function for invalid request exceptions (HTTP only, as request validation happens pre-connection)"""
     if request.scope['type'] == 'http':
-        return InvalidAuthHttp(detail)
-    return InvalidAuthWebSocket(detail)
+        return InvalidRequestHttp(detail)
+    return InvalidRequestWebSocket(detail)
+
+
+def Unauthorized(detail: str, request: HTTPConnection) -> UnauthorizedHttp | UnauthorizedWebSocket:
+    """Factory function for unauthorized exceptions"""
+    if request.scope["type"] == "http":
+        return UnauthorizedHttp(detail)
+    return UnauthorizedWebSocket(detail)
+
+
+def Forbidden(detail: str, request: HTTPConnection) -> ForbiddenHttp | ForbiddenWebSocket:
+    """Factory function for forbidden exceptions"""
+    if request.scope["type"] == "http":
+        return ForbiddenHttp(detail)
+    return ForbiddenWebSocket(detail)