Add 'can_user_access_resources' function

.pre-commit-config.yaml

@@ -13,6 +13,6 @@ repos:
     -   id: no-commit-to-branch
         args: [--branch, develop, --branch, master, --branch, main, --pattern, release/.*]
 -   repo: https://github.com/psf/black
-    rev: 20.8b1
+    rev: 24.10.0
     hooks:
     -   id: black

python/src/gen3authz/client/arborist/base.py

@@ -18,6 +18,7 @@
 
 from ..arborist.errors import ArboristError, ArboristUnhealthyError
 from ..base import AuthzClient
+from ...utils import is_path_prefix_of_path
 from ... import string_types
 
 
@@ -282,16 +283,32 @@ async def healthy(self, timeout=1):
         return response.code == 200
 
     @maybe_sync
-    async def auth_mapping(self, username):
+    async def auth_mapping(self, username: str = "", jwt: str = ""):
         """
         For given user, get mapping from the resources that this user can access
         to the actions on those resources for which they are authorized.
 
+        Args:
+            jwt (str): user's valid jwt access token
+            username (str): username (exactly one of `jwt` and `username` must be provided)
+
         Return:
             dict: response JSON from arborist
         """
-        data = {"username": username}
-        response = await self.post(self._auth_url.rstrip("/") + "/mapping", json=data)
+        assert bool(username) != bool(
+            jwt
+        ), "Exactly one of 'username' or 'jwt' must be provided"
+        if username:
+            data = {"username": username}
+            response = await self.post(
+                self._auth_url.rstrip("/") + "/mapping", json=data
+            )
+        else:
+            response = await self.post(
+                self._auth_url.rstrip("/") + "/mapping",
+                json=data,
+                headers={"Authorization": f"bearer {jwt}"},
+            )
         if not response.successful:
             raise ArboristError(response.error_msg, response.code)
         return response.json
@@ -368,7 +385,6 @@ async def auth_request(self, jwt, service, methods, resources, user_id=None):
 
     @maybe_sync
     async def create_resource(self, parent_path, resource_json, create_parents=False):
-
         """
         Create a new resource in arborist (does not affect fence database or
         otherwise have any interaction with userdatamodel).
@@ -1078,3 +1094,47 @@ async def delete_client(self, client_id):
         )
         self.logger.info("deleted client {}".format(client_id))
         return response.code == 204
+
+    @maybe_sync
+    async def can_user_access_resources(
+        self,
+        service: str,
+        method: str,
+        resource_paths: list,
+        username: str = "",
+        jwt: str = "",
+    ):
+        """
+        Using the user's authz mapping, return "true" for each resource path the user has access
+        to, and "false" for each resource path they don't have access to. Take into account that
+        if a user has access to "/a", they also have access to "/a/b".
+
+        Args:
+            service (str): service name to check the access for
+            method (str): method to check the access for
+            resource_paths (list): resource paths to check the access for
+            jwt (str): user's valid jwt access token
+            username (str): username (exactly one of `jwt` and `username` must be provided)
+
+        Return:
+            dict: for each provided resource path, whether or not the user has access to the
+            provided method and service
+        """
+        mapping = self.auth_mapping(username, jwt)
+        if inspect.isawaitable(mapping):
+            mapping = await mapping
+        authorized_resource_paths = [
+            resource_path
+            for resource_path, access in mapping.items()
+            if any(
+                e["service"] in [service, "*"] and e["method"] in [method, "*"]
+                for e in access
+            )
+        ]
+        return {
+            resource_path: any(
+                is_path_prefix_of_path(authorized_resource_path, resource_path)
+                for authorized_resource_path in authorized_resource_paths
+            )
+            for resource_path in resource_paths
+        }

python/src/gen3authz/utils.py

@@ -22,3 +22,18 @@ def _wrapper(*args, **kwargs):
             return si.value
 
     return _wrapper
+
+
+def is_path_prefix_of_path(resource_prefix: str, resource_path: str) -> bool:
+    """
+    Return True if the arborist resource path "resource_prefix" is a
+    prefix of the arborist resource path "resource_path".
+    """
+    prefix_list = resource_prefix.rstrip("/").split("/")
+    path_list = resource_path.rstrip("/").split("/")
+    if len(prefix_list) > len(path_list):
+        return False
+    for i, prefix_item in enumerate(prefix_list):
+        if path_list[i] != prefix_item:
+            return False
+    return True

python/tests/arborist/test_urls.py

@@ -2,6 +2,7 @@
 Run some basic tests that the methods on ``ArboristClient`` actually try to hit
 the correct URLs on the arborist API.
 """
+
 import pytest
 import datetime
 from gen3authz.client.arborist.errors import ArboristError
@@ -205,3 +206,64 @@ async def test_update_user_raises_error(
             )
         else:
             response = arborist_client.update_user(username, new_username=new_username)
+
+
+async def test_auth_request_positive(arborist_client, mock_arborist_request, use_async):
+    mock_arborist_request({"/auth/request": {"POST": (200, {"auth": True})}})
+    if use_async:
+        assert await arborist_client.auth_request(
+            "", "fence", "file_upload", "/data_upload"
+        )
+    else:
+        assert (
+            arborist_client.auth_request("", "fence", "file_upload", "/data_upload")
+            is True
+        )
+
+
+async def test_can_user_access_resources(
+    arborist_client, mock_arborist_request, use_async
+):
+    with pytest.raises(AssertionError, match="Exactly one"):
+        await arborist_client.can_user_access_resources(
+            service="service1", method="read", resource_paths=[]
+        )
+    with pytest.raises(AssertionError, match="Exactly one"):
+        await arborist_client.can_user_access_resources(
+            username="test-user",
+            jwt="abc",
+            service="service1",
+            method="read",
+            resource_paths=[],
+        )
+
+    mock_arborist_request(
+        {
+            "/auth/mapping": {
+                "POST": (
+                    200,
+                    {
+                        "/a": [{"service": "service1", "method": "read"}],
+                        "/c": [
+                            {"service": "service2", "method": "read"},
+                            {"service": "service1", "method": "write"},
+                        ],
+                    },
+                )
+            }
+        }
+    )
+
+    res = arborist_client.can_user_access_resources(
+        username="test-user",
+        service="service1",
+        method="read",
+        resource_paths=["/a", "/a/b", "/c"],
+    )
+    if use_async:
+        res = await res
+
+    # /a: right service and method => True
+    # /a/b: nested under /a which is accessible => True
+    # /c: right service, wrong method and right method, wrong service => False
+    assert res == {"/a": True, "/a/b": True, "/c": False}