Cache permissions during request (#657)

Author: Dreamsorcerer

aiohttp_admin/backends/abc.py

@@ -7,7 +7,7 @@
 from typing import Any, Literal, Optional, TypedDict, Union
 
 from aiohttp import web
-from aiohttp_security import authorized_userid, check_permission, permits
+from aiohttp_security import check_permission, permits
 from pydantic import Json, parse_obj_as
 
 from ..security import permissions_as_dict
@@ -87,7 +87,7 @@ async def filter_by_permissions(self, request: web.Request, perm_type: str,
         """Return a filtered record containing permissible fields only."""
         return {k: v for k, v in record.items()
                 if await permits(request, f"admin.{self.name}.{k}.{perm_type}",
-                                 context=original or record)}
+                                 context=(request, original or record))}
 
     @abstractmethod
     async def get_list(self, params: GetListParams) -> tuple[list[Record], int]:
@@ -120,71 +120,69 @@ async def delete_many(self, params: DeleteManyParams) -> list[Union[int, str]]:
     # https://marmelab.com/react-admin/DataProviderWriting.html
 
     async def _get_list(self, request: web.Request) -> web.Response:
-        await check_permission(request, f"admin.{self.name}.view")
+        await check_permission(request, f"admin.{self.name}.view", context=(request, None))
         query = parse_obj_as(GetListParams, request.query)
 
         # Add filters from advanced permissions.
-        if request.app["identity_callback"]:
-            identity = await authorized_userid(request)
-            user_details = await request.app["identity_callback"](identity)
-            permissions = permissions_as_dict(user_details["permissions"])
-            filters = permissions.get(f"admin.{self.name}.view",
-                                      permissions.get(f"admin.{self.name}.*", {}))
-            for k, v in filters.items():
-                query["filter"][k] = v
+        # The permissions will be cached on the request from a previous permissions check.
+        permissions = permissions_as_dict(request["aiohttpadmin_permissions"])
+        filters = permissions.get(f"admin.{self.name}.view",
+                                  permissions.get(f"admin.{self.name}.*", {}))
+        for k, v in filters.items():
+            query["filter"][k] = v
 
         results, total = await self.get_list(query)
         results = [await self.filter_by_permissions(request, "view", r) for r in results]
         results = [r for r in results if await permits(request, f"admin.{self.name}.view",
-                                                       context=r)]
+                                                       context=(request, r))]
         return json_response({"data": results, "total": total})
 
     async def _get_one(self, request: web.Request) -> web.Response:
-        await check_permission(request, f"admin.{self.name}.view")
+        await check_permission(request, f"admin.{self.name}.view", context=(request, None))
         query = parse_obj_as(GetOneParams, request.query)
 
         result = await self.get_one(query)
-        if not await permits(request, f"admin.{self.name}.view", context=result):
+        if not await permits(request, f"admin.{self.name}.view", context=(request, result)):
             raise web.HTTPForbidden()
         result = await self.filter_by_permissions(request, "view", result)
         return json_response({"data": result})
 
     async def _get_many(self, request: web.Request) -> web.Response:
-        await check_permission(request, f"admin.{self.name}.view")
+        await check_permission(request, f"admin.{self.name}.view", context=(request, None))
         query = parse_obj_as(GetManyParams, request.query)
 
         results = await self.get_many(query)
         results = [await self.filter_by_permissions(request, "view", r) for r in results
-                   if await permits(request, f"admin.{self.name}.view", context=r)]
+                   if await permits(request, f"admin.{self.name}.view", context=(request, r))]
         return json_response({"data": results})
 
     async def _create(self, request: web.Request) -> web.Response:
         query = parse_obj_as(CreateParams, request.query)
-        await check_permission(request, f"admin.{self.name}.add", context=query["data"])
+        await check_permission(request, f"admin.{self.name}.add", context=(request, query["data"]))
         for k, v in query["data"].items():
             if v is not None:
                 await check_permission(request, f"admin.{self.name}.{k}.add",
-                                       context=query["data"])
+                                       context=(request, query["data"]))
 
         result = await self.create(query)
         result = await self.filter_by_permissions(request, "view", result)
         return json_response({"data": result})
 
     async def _update(self, request: web.Request) -> web.Response:
-        await check_permission(request, f"admin.{self.name}.edit")
+        await check_permission(request, f"admin.{self.name}.edit", context=(request, None))
         query = parse_obj_as(UpdateParams, request.query)
 
         # Check original record is allowed by permission filters.
         original = await self.get_one({"id": query["id"]})
-        if not await permits(request, f"admin.{self.name}.edit", context=original):
+        if not await permits(request, f"admin.{self.name}.edit", context=(request, original)):
             raise web.HTTPForbidden()
 
         # Filter rather than forbid because react-admin still sends fields without an
         # input component. The query may not be the complete dict though, so we must
         # pass original for testing.
         query["data"] = await self.filter_by_permissions(request, "edit", query["data"], original)
         # Check new values are allowed by permission filters.
-        if not await permits(request, f"admin.{self.name}.edit", context=query["data"]):
+        if not await permits(request, f"admin.{self.name}.edit", context=(request, query["data"])):
             raise web.HTTPForbidden()
 
         if not query["data"]:
@@ -195,24 +193,24 @@ async def _update(self, request: web.Request) -> web.Response:
         return json_response({"data": result})
 
     async def _delete(self, request: web.Request) -> web.Response:
-        await check_permission(request, f"admin.{self.name}.delete")
+        await check_permission(request, f"admin.{self.name}.delete", context=(request, None))
         query = parse_obj_as(DeleteParams, request.query)
 
         original = await self.get_one({"id": query["id"]})
-        if not await permits(request, f"admin.{self.name}.delete", context=original):
+        if not await permits(request, f"admin.{self.name}.delete", context=(request, original)):
             raise web.HTTPForbidden()
 
         result = await self.delete(query)
         result = await self.filter_by_permissions(request, "view", result)
         return json_response({"data": result})
 
     async def _delete_many(self, request: web.Request) -> web.Response:
-        await check_permission(request, f"admin.{self.name}.delete")
+        await check_permission(request, f"admin.{self.name}.delete", context=(request, None))
         query = parse_obj_as(DeleteManyParams, request.query)
 
         originals = await self.get_many(query)
         allowed = await asyncio.gather(*(permits(request, f"admin.{self.name}.delete",
-                                                 context=r) for r in originals))
+                                                 context=(request, r)) for r in originals))
         if not all(allowed):
             raise web.HTTPForbidden()
 

aiohttp_admin/security.py

@@ -66,15 +66,25 @@ async def authorized_userid(self, identity: str) -> str:
         return identity
 
     async def permits(self, identity: Optional[str], permission: Union[str, Enum],
-                      context: Optional[Mapping[str, object]] = None) -> bool:
+                      context: tuple[web.Request, Optional[Mapping[str, object]]]) -> bool:
         if identity is None:
             return False
-        if self._identity_callback is None:
-            permissions: Collection[str] = tuple(Permissions)
-        else:
-            user = await self._identity_callback(identity)
-            permissions = user["permissions"]
-        return has_permission(permission, permissions_as_dict(permissions), context)
+
+        try:
+            request, record = context
+        except (TypeError, ValueError):
+            raise TypeError("Context must be `(request, record)` or `(request, None)`")
+
+        permissions: Optional[Collection[str]] = request.get("aiohttpadmin_permissions")
+        if permissions is None:
+            if self._identity_callback is None:
+                permissions = tuple(Permissions)
+            else:
+                user = await self._identity_callback(identity)
+                permissions = user["permissions"]
+            # Cache permissions per request to avoid potentially dozens of DB calls.
+            request["aiohttpadmin_permissions"] = permissions
+        return has_permission(permission, permissions_as_dict(permissions), record)
 
 
 class TokenIdentityPolicy(SessionIdentityPolicy):  # type: ignore[misc,no-any-unimported]

tests/test_security.py

@@ -1,5 +1,6 @@
 import json
 from typing import Awaitable, Callable, Optional
+from unittest import mock
 
 from aiohttp.test_utils import TestClient
 from aiohttp_security import AbstractAuthorizationPolicy
@@ -333,6 +334,25 @@ async def identity_callback(identity: Optional[str]) -> UserDetails:
         assert await resp.json() == {"data": {"id": 1}}
 
 
+async def test_permissions_cached(create_admin_client: _CreateClient,  # type: ignore[no-any-unimported] # noqa: B950
+                                  login: _Login) -> None:
+    identity_callback = mock.AsyncMock(spec_set=(), return_value={"permissions": {"admin.*"}})
+    admin_client = await create_admin_client(identity_callback)
+
+    assert admin_client.app
+    url = admin_client.app["admin"].router["dummy2_get_list"].url_for()
+    h = await login(admin_client)
+    identity_callback.assert_called_once()
+    identity_callback.reset_mock()
+
+    p = {"pagination": json.dumps({"page": 1, "perPage": 10}),
+         "sort": json.dumps({"field": "id", "order": "DESC"}), "filter": "{}"}
+    async with admin_client.get(url, params=p, headers=h) as resp:
+        assert resp.status == 200
+
+    identity_callback.assert_called_once()
+
+
 async def test_permission_filter_list(create_admin_client: _CreateClient,  # type: ignore[no-any-unimported] # noqa: B950
                                       login: _Login) -> None:
     async def identity_callback(identity: Optional[str]) -> UserDetails: