🐛 AuthenticationMiddleware continue execution on 404 or 405

Author: perdy

flama/authentication/middleware.py

@@ -2,14 +2,13 @@
 import logging
 import typing as t
 
-from flama import authentication
+from flama import authentication, exceptions
 from flama.exceptions import HTTPException
 from flama.http import APIErrorResponse, Request
 
 if t.TYPE_CHECKING:
     from flama import Flama, types
     from flama.http import Response
-    from flama.routing import BaseRoute
 
 __all__ = ["AuthenticationMiddleware"]
 
@@ -30,17 +29,19 @@ async def __call__(self, scope: "types.Scope", receive: "types.Receive", send: "
 
         await response(scope, receive, send)
 
-    def _get_permissions(self, route: "BaseRoute") -> set[str]:
-        return set(route.tags.get("permissions", []))
+    def _get_permissions(self, app: "Flama", scope: "types.Scope") -> set[str]:
+        try:
+            route, _ = app.router.resolve_route(scope)
+            permissions = set(route.tags.get("permissions", []))
+        except (exceptions.MethodNotAllowedException, exceptions.NotFoundException):
+            permissions = []
+
+        return set(permissions)
 
     async def _get_response(self, scope: "types.Scope", receive: "types.Receive") -> t.Union["Response", "Flama"]:
         app: Flama = scope["app"]
 
-        route, _ = app.router.resolve_route(scope)
-
-        required_permissions = self._get_permissions(route)
-
-        if not required_permissions:
+        if not (required_permissions := self._get_permissions(app, scope)):
             return self.app
 
         try:

flama/routing/router.py

@@ -255,6 +255,8 @@ def resolve_route(self, scope: types.Scope) -> tuple[BaseRoute, types.Scope]:
         """Look for a route that matches given ASGI scope.
 
         :param scope: ASGI scope.
+        :raise MethodNotAllowedException: If route is resolved but http method is not valid.
+        :raise NotFoundException: If route cannot be resolved.
         :return: Route and its scope.
         """
         partial = None

tests/authentication/test_middleware.py

@@ -62,22 +62,24 @@ def cookies(self, request):
             raise ValueError(f"Invalid token {request.param}")
 
     @pytest.mark.parametrize(
-        ["path", "headers", "cookies", "status_code", "result"],
+        ["path", "method", "headers", "cookies", "status_code", "result"],
         (
-            pytest.param("/auth/", "permission", None, 200, {"foo": "auth"}, id="auth_header_token_permission"),
-            pytest.param("/auth/", "role", None, 200, {"foo": "auth"}, id="auth_header_token_role"),
+            pytest.param("/auth/", "get", "permission", None, 200, {"foo": "auth"}, id="auth_header_token_permission"),
+            pytest.param("/auth/", "get", "role", None, 200, {"foo": "auth"}, id="auth_header_token_role"),
             pytest.param(
                 "/auth/",
+                "get",
                 "empty",
                 None,
                 403,
                 {"detail": "Insufficient permissions", "error": None, "status_code": 403},
                 id="auth_header_token_empty",
             ),
-            pytest.param("/auth/", None, "permission", 200, {"foo": "auth"}, id="auth_cookie_token_permission"),
-            pytest.param("/auth/", None, "role", 200, {"foo": "auth"}, id="auth_cookie_token_role"),
+            pytest.param("/auth/", "get", None, "permission", 200, {"foo": "auth"}, id="auth_cookie_token_permission"),
+            pytest.param("/auth/", "get", None, "role", 200, {"foo": "auth"}, id="auth_cookie_token_role"),
             pytest.param(
                 "/auth/",
+                "get",
                 None,
                 "empty",
                 403,
@@ -86,28 +88,47 @@ def cookies(self, request):
             ),
             pytest.param(
                 "/auth/",
+                "get",
                 None,
                 None,
                 401,
                 {"detail": "Unauthorized", "error": None, "status_code": 401},
                 id="auth_no_token",
             ),
             pytest.param(
-                "/no-auth/", "permission", None, 200, {"foo": "no-auth"}, id="no_auth_header_token_permission"
+                "/no-auth/", "get", "permission", None, 200, {"foo": "no-auth"}, id="no_auth_header_token_permission"
             ),
-            pytest.param("/no-auth/", "role", None, 200, {"foo": "no-auth"}, id="no_auth_header_token_role"),
+            pytest.param("/no-auth/", "get", "role", None, 200, {"foo": "no-auth"}, id="no_auth_header_token_role"),
             pytest.param(
-                "/no-auth/", None, "permission", 200, {"foo": "no-auth"}, id="no_auth_cookie_token_permission"
+                "/no-auth/", "get", None, "permission", 200, {"foo": "no-auth"}, id="no_auth_cookie_token_permission"
+            ),
+            pytest.param("/no-auth/", "get", None, "role", 200, {"foo": "no-auth"}, id="no_auth_cookie_token_role"),
+            pytest.param("/no-auth/", "get", None, None, 200, {"foo": "no-auth"}, id="no_auth_no_token"),
+            pytest.param(
+                "/not-found/",
+                "get",
+                None,
+                None,
+                404,
+                {"detail": "Not Found", "error": "HTTPException", "status_code": 404},
+                id="not_found",
+            ),
+            pytest.param(
+                "/auth/",
+                "post",
+                None,
+                None,
+                405,
+                {"detail": "Method Not Allowed", "error": "HTTPException", "status_code": 405},
+                id="method_not_allowed",
             ),
-            pytest.param("/no-auth/", None, "role", 200, {"foo": "no-auth"}, id="no_auth_cookie_token_role"),
-            pytest.param("/no-auth/", None, None, 200, {"foo": "no-auth"}, id="no_auth_no_token"),
         ),
         indirect=["headers", "cookies"],
     )
-    async def test_request(self, client, path, headers, cookies, status_code, result):
+    async def test_request(self, client, path, method, headers, cookies, status_code, result):
         client.headers = headers
         client.cookies = cookies
-        response = await client.request("get", path)
+        response = await client.request(method, path)
 
         assert response.status_code == status_code
         assert response.json() == result