Improved the reusability of the middleware by passing all headers (#3)

* Improved the reusability of the middleware by passing all headers instead of Authorization

* fix tests

* fix signatures

Co-authored-by: Yannic Schröer <yannicschroer@Yannics-MBP.fritz.box>

Author: yannicschroeer

docs/docs/examples/multiple.md

@@ -7,7 +7,7 @@ elegant solution to this. But the current solution is to mount multiple apps ins
 **multiple.py**
 
 ```python
-from typing import Tuple, List
+from typing import Tuple, List, Dict
 import uvicorn
 from fastapi import FastAPI
 from starlette.requests import Request
@@ -16,14 +16,14 @@ from fastapi_auth_middleware import AuthMiddleware, FastAPIUser
 
 
 # The method you have to provide
-def verify_authorization_header(auth_header: str) -> Tuple[List[str], FastAPIUser]:
+def verify_header(headers: Dict) -> Tuple[List[str], FastAPIUser]:
     user = FastAPIUser(first_name="Code", last_name="Specialist", user_id=1)  # Usually you would decode the JWT here and verify its signature to extract the 'sub'
     scopes = ["admin"]  # You could for instance use the scopes provided in the JWT or request them by looking up the scopes with the 'sub' somewhere
     return scopes, user
 
 
 users_app = FastAPI()
-users_app.add_middleware(AuthMiddleware, verify_authorization_header=verify_authorization_header)  # Add the middleware with your verification method to the whole application
+users_app.add_middleware(AuthMiddleware, verify_header=verify_header)  # Add the middleware with your verification method to the whole application
 
 
 @users_app.get('/')  # Sample endpoint (secured)

docs/docs/examples/simple_with_scopes.md

@@ -1,7 +1,7 @@
 # Basic Example with Scopes
 
 ```python
-from typing import Tuple, List
+from typing import Tuple, List, Dict
 import uvicorn
 from fastapi import FastAPI
 from starlette.authentication import requires
@@ -11,14 +11,14 @@ from fastapi_auth_middleware import AuthMiddleware, FastAPIUser
 
 
 # The method you have to provide
-def verify_authorization_header(auth_header: str) -> Tuple[List[str], FastAPIUser]:
+def verify_header(headers: Dict) -> Tuple[List[str], FastAPIUser]:
     user = FastAPIUser(first_name="Code", last_name="Specialist", user_id=1)  # Usually you would decode the JWT here and verify its signature to extract the 'sub'
     scopes = ["admin"]  # You could for instance use the scopes provided in the JWT or request them by looking up the scopes with the 'sub' somewhere
     return scopes, user
 
 
 app = FastAPI()
-app.add_middleware(AuthMiddleware, verify_authorization_header=verify_authorization_header)  # Add the middleware with your verification method to the whole application
+app.add_middleware(AuthMiddleware, verify_header=verify_header)  # Add the middleware with your verification method to the whole application
 
 
 @app.get('/home')  # Sample endpoint (secured)

docs/docs/index.md

@@ -38,7 +38,7 @@ from starlette.authentication import BaseUser
 
 ...
 # Takes a string that will look like 'Bearer eyJhbGc...'
-def verify_authorization_header(auth_header: str) -> Tuple[List[str], BaseUser]: # Returns a Tuple of a List of scopes (string) and a BaseUser
+def verify_header(headers: List[str]) -> Tuple[List[str], BaseUser]: # Returns a Tuple of a List of scopes (string) and a BaseUser
     user = FastAPIUser(first_name="Code", last_name="Specialist", user_id=1)  # Usually you would decode the JWT here and verify its signature to extract the 'sub'
     scopes = []  # You could for instance use the scopes provided in the JWT or request them by looking up the scopes with the 'sub' somewhere
     return scopes, user
@@ -53,7 +53,7 @@ from fastapi_auth_middleware import AuthMiddleware
 ...
 
 app = FastAPI()
-app.add_middleware(AuthMiddleware, verify_authorization_header=verify_authorization_header)
+app.add_middleware(AuthMiddleware, verify_header=verify_header)
 ```
 
 After adding this middleware, all requests will pass the `verify_authorization_header` function and contain the **scopes** as well as the **user object** as injected dependencies.

examples/simple.py

@@ -1,4 +1,4 @@
-from typing import Tuple, List
+from typing import Tuple, List, Dict
 import uvicorn
 from fastapi import FastAPI
 from starlette.requests import Request
@@ -7,14 +7,14 @@
 
 
 # The method you have to provide
-def verify_authorization_header(auth_header: str) -> Tuple[List[str], FastAPIUser]:
+def verify_header(headers: Dict) -> Tuple[List[str], FastAPIUser]:
     user = FastAPIUser(first_name="Code", last_name="Specialist", user_id=1)  # Usually you would decode the JWT here and verify its signature to extract the 'sub'
     scopes = []  # You could for instance use the scopes provided in the JWT or request them by looking up the scopes with the 'sub' somewhere
     return scopes, user
 
 
 app = FastAPI()
-app.add_middleware(AuthMiddleware, verify_authorization_header=verify_authorization_header)  # Add the middleware with your verification method to the whole application
+app.add_middleware(AuthMiddleware, verify_header=verify_header)  # Add the middleware with your verification method to the whole application
 
 
 @app.get('/')  # Sample endpoint (secured)

examples/simple_with_scopes.py

@@ -1,4 +1,4 @@
-from typing import Tuple, List
+from typing import Tuple, List, Dict
 import uvicorn
 from fastapi import FastAPI
 from starlette.authentication import requires
@@ -8,14 +8,14 @@
 
 
 # The method you have to provide
-def verify_authorization_header(auth_header: str) -> Tuple[List[str], FastAPIUser]:
+def verify_header(headers: Dict) -> Tuple[List[str], FastAPIUser]:
     user = FastAPIUser(first_name="Code", last_name="Specialist", user_id=1)  # Usually you would decode the JWT here and verify its signature to extract the 'sub'
     scopes = ["admin"]  # You could for instance use the scopes provided in the JWT or request them by looking up the scopes with the 'sub' somewhere
     return scopes, user
 
 
 app = FastAPI()
-app.add_middleware(AuthMiddleware, verify_authorization_header=verify_authorization_header)  # Add the middleware with your verification method to the whole application
+app.add_middleware(AuthMiddleware, verify_header=verify_header)  # Add the middleware with your verification method to the whole application
 
 
 @app.get('/home')  # Sample endpoint (secured)

fastapi_auth_middleware/middleware.py

@@ -1,5 +1,5 @@
 import inspect
-from typing import Tuple, Callable, List
+from typing import Tuple, Callable, List, Dict
 
 from fastapi import FastAPI
 from starlette.authentication import AuthenticationBackend, AuthCredentials, AuthenticationError, BaseUser
@@ -46,13 +46,13 @@ def identity(self) -> str:
 class FastAPIAuthBackend(AuthenticationBackend):
     """ Auth Backend for FastAPI """
 
-    def __init__(self, verify_authorization_header: Callable[[str], Tuple[List[str], BaseUser]]):
+    def __init__(self, verify_header: Callable[[Dict], Tuple[List[str], BaseUser]]):
         """ Auth Backend constructor. Part of an AuthenticationMiddleware as backend.
 
         Args:
-            verify_authorization_header (callable): A function handle that returns a list of scopes and a BaseUser
+            verify_header (callable): A function handle that returns a list of scopes and a BaseUser
         """
-        self.verify_authorization_header = verify_authorization_header
+        self.verify_header = verify_header
 
     async def authenticate(self, conn: HTTPConnection) -> Tuple[AuthCredentials, BaseUser]:
         """ The 'magic' happens here. The authenticate method is invoked each time a route is called that the middleware is applied to.
@@ -67,11 +67,10 @@ async def authenticate(self, conn: HTTPConnection) -> Tuple[AuthCredentials, Bas
             raise AuthenticationError("Authorization header missing")
 
         try:
-            authorization_header: str = conn.headers["Authorization"]
-            if inspect.iscoroutinefunction(self.verify_authorization_header):
-                scopes, user = await self.verify_authorization_header(authorization_header)
+            if inspect.iscoroutinefunction(self.verify_header):
+                scopes, user = await self.verify_header(conn.headers)
             else:
-                scopes, user = self.verify_authorization_header(authorization_header)
+                scopes, user = self.verify_header(conn.headers)
 
         except Exception as exception:
             raise AuthenticationError(exception) from None
@@ -82,15 +81,15 @@ async def authenticate(self, conn: HTTPConnection) -> Tuple[AuthCredentials, Bas
 # noinspection PyPep8Naming
 def AuthMiddleware(
         app: FastAPI,
-        verify_authorization_header: Callable[[str], Tuple[List[str], BaseUser]],
+        verify_header: Callable[[str], Tuple[List[str], BaseUser]],
         auth_error_handler: Callable[[Request, AuthenticationError], JSONResponse] = None
 ):
     """ Factory method, returning an AuthenticationMiddleware
     Intentionally not named with lower snake case convention as this is a factory method returning a class. Should feel like a class.
 
     Args:
         app (FastAPI): The FastAPI instance the middleware should be applied to. The `add_middleware` function of FastAPI adds the app as first argument by default.
-        verify_authorization_header (Callable[[str], Tuple[List[str], BaseUser]]): A function handle that returns a list of scopes and a BaseUser
+        verify_header (Callable[[str], Tuple[List[str], BaseUser]]): A function handle that returns a list of scopes and a BaseUser
         auth_error_handler (Callable[[Request, Exception], JSONResponse]): Optional error handler for creating responses when an exception was raised in verify_authorization_header
 
     Examples:
@@ -104,4 +103,4 @@ def verify_authorization_header(auth_header: str) -> Tuple[List[str], FastAPIUse
         app.add_middleware(AuthMiddleware, verify_authorization_header=verify_authorization_header)
         ```
     """
-    return AuthenticationMiddleware(app, backend=FastAPIAuthBackend(verify_authorization_header), on_error=auth_error_handler)
+    return AuthenticationMiddleware(app, backend=FastAPIAuthBackend(verify_header), on_error=auth_error_handler)

tests/test_middleware.py

@@ -1,4 +1,4 @@
-from typing import Callable
+from typing import Callable, List, Dict
 
 from _pytest.fixtures import fixture
 from fastapi import FastAPI
@@ -11,13 +11,13 @@
 
 
 # Sample verification function, does nothing of effect
-def verify_authorization_header_basic(auth_header: str):
+def verify_header(headers: Dict):
     user = FastAPIUser(first_name="Code", last_name="Specialist", user_id=1)
     scopes = ["authenticated"]
     return scopes, user
 
 
-async def verify_authorization_header_basic_admin_scope(auth_header: str):
+async def verify_header_basic_admin_scope(headers: List[str]):
     user = FastAPIUser(first_name="Code", last_name="Specialist", user_id=1)
     scopes = ["admin"]
     return scopes, user
@@ -28,9 +28,9 @@ def raise_exception_in_verify_authorization_header(_):
 
 
 #  Sample app with simple routes, takes a verify_authorization_header callable that is applied to the middleware
-def fastapi_app(verify_authorization_header: Callable, auth_error_handler: Callable = None):
+def fastapi_app(verify_header: Callable, auth_error_handler: Callable = None):
     app = FastAPI()
-    app.add_middleware(AuthMiddleware, verify_authorization_header=verify_authorization_header, auth_error_handler=auth_error_handler)
+    app.add_middleware(AuthMiddleware, verify_header=verify_header, auth_error_handler=auth_error_handler)
 
     @app.get("/")
     def home():
@@ -57,12 +57,12 @@ class TestBasicBehaviour:
 
     @fixture
     def client(self) -> TestClient:
-        app = fastapi_app(verify_authorization_header_basic)
+        app = fastapi_app(verify_header)
         return TestClient(app)
 
     @fixture
     def client_with_scopes(self) -> TestClient:
-        app = fastapi_app(verify_authorization_header_basic_admin_scope)
+        app = fastapi_app(verify_header_basic_admin_scope)
         return TestClient(app)
 
     def test_home_fail_no_header(self, client: TestClient):
@@ -81,7 +81,7 @@ def test_scopes(self, client: TestClient, client_with_scopes: TestClient):
         assert client_with_scopes.get("/admin-scope", headers={"Authorization": "ey.."}).status_code == 200  # Contains the requested scope
 
     def test_fail_auth_error(self):
-        app = fastapi_app(verify_authorization_header=raise_exception_in_verify_authorization_header)
+        app = fastapi_app(verify_header=raise_exception_in_verify_authorization_header)
         client_with_auth_error = TestClient(app=app)
 
         response = client_with_auth_error.get('/', headers={"Authorization": "ey.."})
@@ -92,7 +92,7 @@ def handle_auth_error(request: Request, exception: AuthenticationError):
             assert isinstance(exception, AuthenticationError)
             return JSONResponse(content={'message': str(exception)}, status_code=401)
 
-        app = fastapi_app(verify_authorization_header=raise_exception_in_verify_authorization_header, auth_error_handler=handle_auth_error)
+        app = fastapi_app(verify_header=raise_exception_in_verify_authorization_header, auth_error_handler=handle_auth_error)
         client_with_auth_error = TestClient(app=app)
 
         response = client_with_auth_error.get('/', headers={"Authorization": "ey.."})