Merge pull request #70 from djpugh/fix/api_auth_required_decorator-64

docs/source/advanced.rst

@@ -2,6 +2,20 @@ Using fastapi_aad_auth
 **********************
 Please see `Basic Usage <usage>`_ for information on how to configure and setup ``fastapi_aad_auth``.
 
+Customising authentication dependencies
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The :meth:`~fastapi_aad_auth.auth.Authenticator.api_auth_required` method decorator provides controls over the scope, and is
+using the ``fastapi`` dependency injection system. If you want more customisation, you can use::
+
+    router = APIRouter()
+
+    @router.get('/hello')
+    async def hello_world(auth_state: AuthenticationState = Depends(auth_provider.auth_backend.requires_auth(allow_session=True))):
+        print(auth_state)
+        return {'hello': 'world'}
+
+
 Accessing User Tokens/View
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 

docs/source/usage.rst

@@ -112,7 +112,8 @@ You can use it for fastapi routes::
     router = APIRouter()
 
     @router.get('/hello')
-    async def hello_world(auth_state: AuthenticationState = Depends(auth_provider.auth_backend.requires_auth(allow_session=True))):
+    @auth_provider.api_auth_required(allow_session=True)
+    async def hello_world(auth_state: AuthenticationState):
         print(auth_state)
         return {'hello': 'world'}
 

src/fastapi_aad_auth/_base/backend.py

@@ -7,6 +7,7 @@
 
 from fastapi_aad_auth._base.state import AuthenticationState
 from fastapi_aad_auth._base.validators import SessionValidator, TokenValidator, Validator
+from fastapi_aad_auth.errors import AuthorisationError
 from fastapi_aad_auth.mixins import LoggingMixin, NotAuthenticatedMixin
 from fastapi_aad_auth.utilities import deprecate
 
@@ -58,7 +59,7 @@ def _iter_validators(self):
         for validator in self.validators:
             yield validator
 
-    def requires_auth(self, allow_session: bool = False):
+    def requires_auth(self, scopes: str = 'authenticated', allow_session: bool = False):
         """Require authentication, use with fastapi Depends."""
         # This is a bit horrible, but is needed for fastapi to get this into OpenAPI (or similar) - it needs to be an OAuth2 object
         # We create this here "dynamically" for each endpoint, as we allow customisation on whether a session is permissible
@@ -78,6 +79,9 @@ async def __call__(self_, request: Request):
                     state = self.check(request, allow_session)
                     if state is None or not state.is_authenticated():
                         raise self.not_authenticated
+                    elif not state.check_scopes(scopes):
+                        raise AuthorisationError(f'Not authorised for this API endpoint - Requires {scopes}')
+
                     return state
 
             return OAuthValidator()

src/fastapi_aad_auth/_base/state.py

@@ -2,7 +2,7 @@
 from enum import Enum
 import importlib
 import json
-from typing import List, Optional
+from typing import List, Optional, Union
 import uuid
 
 from itsdangerous import URLSafeSerializer
@@ -164,3 +164,14 @@ def authenticate_as(cls, user, serializer, session):
     def as_unauthenticated(cls, serializer, session):
         """Store as an un-authenticated user."""
         return cls.authenticate_as(None, serializer, session)
+
+    def check_scopes(self, required_scopes: Optional[Union[List[str], str]] = None):
+        """Check if the user has the required scopes."""
+        if required_scopes is None:
+            return True
+        elif isinstance(required_scopes, str):
+            required_scopes = required_scopes.split(' ')
+        for scope in required_scopes:
+            if scope in self.credentials.scopes:
+                return True
+        return False

src/fastapi_aad_auth/auth.py

@@ -1,9 +1,10 @@
 """Authenticator Class."""
 from functools import wraps
+import inspect
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
-from fastapi import FastAPI
+from fastapi import Depends, FastAPI
 from starlette.authentication import requires
 from starlette.middleware.authentication import AuthenticationMiddleware
 from starlette.middleware.sessions import SessionMiddleware
@@ -12,6 +13,7 @@
 from starlette.routing import request_response, Route
 
 from fastapi_aad_auth._base.backend import BaseOAuthBackend
+from fastapi_aad_auth._base.state import AuthenticationState
 from fastapi_aad_auth._base.validators import SessionValidator
 from fastapi_aad_auth.config import Config
 from fastapi_aad_auth.errors import AuthenticationError, AuthorisationError, base_error_handler, ConfigurationError, json_error_handler, redirect_error_handler
@@ -168,6 +170,51 @@ async def req_wrapper(request: Request, *args, **kwargs):
 
         return wrapper
 
+    def api_auth_required(self, scopes: str = 'authenticated', allow_session: bool = True):
+        """Decorator to require specific scopes (and redirect to the login ui) for an endpoint.
+
+        This can be used for enabling authentication on an API endpoint, using the fastapi
+        dependency injection logic.
+
+        This adds the authentication state to the endpoint arguments as ``auth_state``.
+
+        Keyword Args:
+            scopes: scopes for the starlette requires decorator
+            allow_session: whether to allow session authentication or not
+        """
+        def wrapper(endpoint):
+            if self.config.enabled:
+
+                # Create the oauth endpoint
+                oauth = self.auth_backend.requires_auth(scopes=scopes, allow_session=allow_session)
+
+                # We need to do some signature hackery for fastapi
+                endpoint_signature = inspect.signature(endpoint)
+                endpoint_args = [v for v in endpoint_signature.parameters.values() if v.default is inspect._empty and v.name != 'auth_state']
+                endpoint_kwarg_params = [v for v in endpoint_signature.parameters.values() if v.default is not inspect._empty and v.name != 'auth_state']
+                new_params = [inspect.Parameter('auth_state',
+                                                inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                                                default=Depends(oauth),
+                                                annotation=AuthenticationState)]
+
+                # This is the actual dectorator
+
+                @wraps(endpoint)
+                async def require_endpoint(auth_state: AuthenticationState = Depends(oauth), *args, **kwargs):
+                    if ('auth_state' in endpoint_signature.parameters):
+                        kwargs['auth_state'] = auth_state
+                    return await endpoint(*args, **kwargs)
+
+                # We need to set the signature to have the endpoints signature with the additional auth_state params
+                require_endpoint.__signature__ = endpoint_signature.replace(parameters=endpoint_args+new_params+endpoint_kwarg_params)
+                # We also want to set the annotation correctly
+                require_endpoint.__annotations__['auth_state'] = AuthenticationState
+                return require_endpoint
+            else:
+                return endpoint
+
+        return wrapper
+
     def app_routes_add_auth(self, app: FastAPI, route_list: List[str], invert: bool = False):
         """Add authentication to specified routes in application router.
 

tests/testapp/server.py

@@ -21,6 +21,12 @@ async def hello_world(auth_state: AuthenticationState = Depends(auth_provider.au
     print(auth_state)
     return {'hello': 'world'}
 
+@router.get('/test_auth_decorator')
+@auth_provider.api_auth_required('authenticated', allow_session=False)
+async def hello_world2(auth_state: AuthenticationState, a: str = 'b'):
+    print(auth_state)
+    return {'hello': 'world', 'a': a}
+
 
 if 'untagged' in __version__ or 'unknown':
     API_VERSION = 0
@@ -38,7 +44,7 @@ async def homepage(request):
 async def test(request):
     if request.user.is_authenticated:
         return PlainTextResponse('Hello, ' + request.user.display_name)
- 
+
 routes = [
     Route("/", endpoint=homepage),
     Route("/test", endpoint=test)