add excluded_paths to JWTMiddleware

docs/source/jwt/middleware.rst

@@ -62,6 +62,67 @@ actions when an error occurs before rejecting the request.
 
 -------------------------------------------------------------------------------
 
+excluded_paths
+~~~~~~~~~~~~~~
+
+By default, if the JWT token is invalid then the HTTP request is rejected.
+However, by setting ``excluded_paths`` will allow the request
+to continue on the endpoints specified in ``excluded_paths`` instead.
+
+This is useful when using Swagger docs as they can be viewed in a browser,
+but they are still token protected. If we want to communicate with endpoints, 
+we need to set `FastAPI APIKeyHeader <https://github.com/tiangolo/fastapi/tree/master/fastapi/security>`_ as a dependency. After that we 
+can authorize the user with a valid jwt token as in the example below.
+
+.. code-block:: python
+
+    # An example usage of excluded_paths.
+
+    from fastapi import Depends, FastAPI
+    from fastapi.security.api_key import APIKeyHeader
+    from home.tables import Movie  # An example Table
+    from piccolo_api.jwt_auth.endpoints import jwt_login
+    from piccolo_api.jwt_auth.middleware import JWTMiddleware
+    from starlette.routing import Route
+
+    public_app = FastAPI(
+        routes=[
+            Route(
+                path="/login/",
+                endpoint=jwt_login(
+                    secret="mysecret123",
+                    expiry=timedelta(minutes=60), 
+                ),
+            ),
+        ],
+    )
+
+
+    auth_header = APIKeyHeader(name="Authorization")
+    private_app = FastAPI(dependencies=[Depends(auth_header)])
+
+    protected_app = JWTMiddleware(
+        private_app,
+        auth_table=BaseUser,
+        secret="mysecret123",
+        excluded_paths=["/docs", "/openapi.json"],
+    )
+
+
+    FastAPIWrapper(
+        "/movies/",
+        fastapi_app=private_app,
+        piccolo_crud=PiccoloCRUD(Movie, read_only=False),
+        fastapi_kwargs=FastAPIKwargs(
+            all_routes={"tags": ["Movie"]},
+        ),
+    )
+
+    public_app.mount("/private", protected_app)
+
+-------------------------------------------------------------------------------
+
+
 Source
 ------
 

piccolo_api/jwt_auth/middleware.py

@@ -73,6 +73,7 @@ def __init__(
         auth_table: t.Type[BaseUser] = BaseUser,
         blacklist: JWTBlacklist = JWTBlacklist(),
         allow_unauthenticated: bool = False,
+        excluded_paths: t.Optional[t.Sequence[str]] = None,
     ) -> None:
         """
         :param asgi:
@@ -88,13 +89,19 @@ def __init__(
         :param allow_unauthenticated:
             By default the middleware rejects any requests with an invalid
             token.
+        :param excluded_paths:
+            Useful for Swagger docs
+            (eg excluded_paths=["/docs", "/openapi.json"])
+            Swagger docs can be viewed in a browser, but are
+            still token protected.
 
         """
         self.asgi = asgi
         self.secret = secret
         self.auth_table = auth_table
         self.blacklist = blacklist
         self.allow_unauthenticated = allow_unauthenticated
+        self.excluded_paths = excluded_paths or []
 
     def get_token(self, headers: dict) -> t.Optional[str]:
         """
@@ -130,6 +137,20 @@ async def __call__(self, scope, receive, send):
         """
         allow_unauthenticated = self.allow_unauthenticated
 
+        for excluded_path in self.excluded_paths:
+            if excluded_path.endswith("*"):
+                if (
+                    scope["raw_path"]
+                    .decode("utf-8")
+                    .startswith(excluded_path.rstrip("*"))
+                ):
+                    await self.asgi(scope, receive, send)
+                    return
+            else:
+                if scope["path"] == excluded_path:
+                    await self.asgi(scope, receive, send)
+                    return
+
         headers = dict(scope["headers"])
         token = self.get_token(headers)
         if not token:

tests/jwt_auth/test_jwt_middleware.py

@@ -2,6 +2,7 @@
 from unittest import TestCase
 
 import jwt
+from fastapi import FastAPI
 from piccolo.apps.user.tables import BaseUser
 from starlette.endpoints import HTTPEndpoint
 from starlette.exceptions import HTTPException
@@ -24,11 +25,35 @@ def get(self, request: Request):
         )
 
 
+fastapi_app = FastAPI(title="Test excluded paths")
+
+fastapi_app_wildcard = FastAPI()
+
+
+@fastapi_app_wildcard.get("/")
+def home_root():
+    return "Root"
+
+
+@fastapi_app_wildcard.get("/path/a/")
+def sub_root():
+    return "Sub route"
+
+
 ECHO_APP = Router([Route("/", EchoEndpoint)])
 APP = JWTMiddleware(asgi=ECHO_APP, secret="SECRET")
 APP_UNAUTH = JWTMiddleware(
     asgi=ECHO_APP, secret="SECRET", allow_unauthenticated=True
 )
+APP_EXCLUDED_PATHS = JWTMiddleware(
+    asgi=fastapi_app, secret="SECRET", excluded_paths=["/docs"]
+)
+
+APP_EXCLUDED_PATHS_WILDCARD = JWTMiddleware(
+    asgi=fastapi_app_wildcard,
+    secret="SECRET",
+    excluded_paths=["/path/*"],
+)
 
 
 class TestJWTMiddleware(TestCase):
@@ -199,3 +224,25 @@ def test_token_without_user_id(self):
             response.json(),
             {"user_id": None, "jwt_error": JWTError.user_not_found.value},
         )
+
+    def test_excluded_paths(self):
+        client = TestClient(APP_EXCLUDED_PATHS)
+
+        response = client.get("/docs")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(
+            b"<title>Test excluded paths - Swagger UI</title>",
+            response.content,
+        )
+
+    def test_excluded_paths_wildcard(self):
+        client = TestClient(APP_EXCLUDED_PATHS_WILDCARD)
+
+        with self.assertRaises(HTTPException):
+            response = client.get("/")
+            # Requires a token
+            self.assertEqual(response.status_code, 403)
+
+        # Is an excluded path, so doesn't need a token
+        response = client.get("/path/a/")
+        self.assertEqual(response.status_code, 200)