feat(authz): route who_can_access and what_can_target_access through FGA cache

descope/management/authz.py

@@ -8,6 +8,11 @@
 
 
 class Authz(HTTPBase):
+
+    def __init__(self, http_client, fga_cache_url: Optional[str] = None):
+        super().__init__(http_client)
+        self._fga_cache_url = fga_cache_url
+
     def save_schema(self, schema: dict, upgrade: bool = False):
         """
         Create or update the ReBAC schema.
@@ -288,6 +293,7 @@ def who_can_access(
                 "relationDefinition": relation_definition,
                 "namespace": namespace,
             },
+            base_url=self._fga_cache_url,
         )
         return response.json()["targets"]
 
@@ -339,6 +345,7 @@ def what_can_target_access(self, target: str) -> List[dict]:
         response = self._http.post(
             MgmtV1.authz_re_target_all,
             body={"target": target},
+            base_url=self._fga_cache_url,
         )
         return response.json()["relations"]
 

descope/mgmt.py

@@ -41,7 +41,7 @@ def __init__(
         self._http = http_client
         self._access_key = AccessKey(http_client)
         self._audit = Audit(http_client)
-        self._authz = Authz(http_client)
+        self._authz = Authz(http_client, fga_cache_url=fga_cache_url)
         self._descoper = Descoper(http_client)
         self._fga = FGA(http_client, fga_cache_url=fga_cache_url)
         self._flow = Flow(http_client)

tests/management/test_authz.py

@@ -643,3 +643,67 @@ def test_get_modified(self):
                 verify=True,
                 timeout=DEFAULT_TIMEOUT_SECONDS,
             )
+
+    def test_authz_cache_url_who_can_access(self):
+        fga_cache_url = "https://my-fga-cache.example.com"
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+            fga_cache_url=fga_cache_url,
+        )
+
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            mock_post.return_value.json.return_value = {"targets": ["u1"]}
+            result = client.mgmt.authz.who_can_access("a", "b", "c")
+            mock_post.assert_called_with(
+                f"{fga_cache_url}{MgmtV1.authz_re_who}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={
+                    "resource": "a",
+                    "relationDefinition": "b",
+                    "namespace": "c",
+                },
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+            self.assertEqual(result, ["u1"])
+
+    def test_authz_cache_url_what_can_target_access(self):
+        fga_cache_url = "https://my-fga-cache.example.com"
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+            fga_cache_url=fga_cache_url,
+        )
+
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            mock_post.return_value.json.return_value = {
+                "relations": [{"resource": "r1"}]
+            }
+            result = client.mgmt.authz.what_can_target_access("a")
+            mock_post.assert_called_with(
+                f"{fga_cache_url}{MgmtV1.authz_re_target_all}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={"target": "a"},
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+            self.assertEqual(result, [{"resource": "r1"}])