Merge pull request #3 from brighthive/HIVE-1800-graphql-authorization

Data Catalog API (GraphQL) Authorization (HIVE-1800)

bhjwt/__init__.py

@@ -1,6 +1,6 @@
 from bhjwt.main import create_asserter, PublicKeys
 from bhjwt.assertions import AssertJwt
-from bhjwt.exceptions import FailedToDecodeJwt
+from bhjwt.exceptions import FailedToDecodeJwt, AuthorizationError
 from bhjwt.config import AuthLibConfiguration
 from bhjwt.providers import (
     BrightHiveProvider,

bhjwt/assertions/main.py

@@ -3,45 +3,119 @@
 
 class AdminAssertions:
     def is_super_admin(self):
-        assertion = "brighthive-super-admin" in self.claims and self.claims["brighthive-super-admin"] == True
+        assertion = (
+            "brighthive-super-admin" in self.claims
+            and self.claims["brighthive-super-admin"] == True
+        )
         self.assertions.append(assertion)
+        return assertion
 
 
 class DataResourceAssertions:
-    def _is_data_resource_role_of(self, data_resource_id: str, role: str) -> None:
-        assertion = "brighthive-data-resource-claims" in self.claims and self.claims[
-            "brighthive-data-resource-claims"].get(data_resource_id) is not None and \
-            self.claims["brighthive-data-resource-claims"][data_resource_id].get("role",
-                                                                                 "").lower() == role.lower()
+    def _is_data_resource_role_of(self, data_resource_id: str, role: str) -> bool:
+        assertion = (
+            "brighthive-data-resource-claims" in self.claims
+            and self.claims["brighthive-data-resource-claims"].get(data_resource_id)
+            is not None
+            and self.claims["brighthive-data-resource-claims"][data_resource_id]
+            .get("role", "")
+            .lower()
+            == role.lower()
+        )
         self.assertions.append(assertion)
+        return assertion
 
     def is_data_resource_admin_of(self, data_resource_id):
-        self._is_data_resource_role_of(data_resource_id, "admin")
+        assertion = self._is_data_resource_role_of(data_resource_id, "admin")
+        return assertion
 
     def is_data_resource_user_of(self, data_resource_id):
-        self._is_data_resource_role_of(data_resource_id, "user")
+        assertion = self._is_data_resource_role_of(data_resource_id, "user")
+        return assertion
+
+    def _get_data_resources_with_role_access(self, role: str) -> set:
+        data_resources = set()
+        if "brighthive-data-resource-claims" in self.claims:
+            data_resource_claims = self.claims["brighthive-data-resource-claims"]
+            for data_resource_id, claim in data_resource_claims.items():
+                if role.lower() == claim.get("role"):
+                    data_resources.add(data_resource_id)
+        return data_resources
+
+    def get_data_resources_with_admin_access(self) -> set:
+        data_resources = self._get_data_resources_with_role_access("admin")
+        return data_resources
 
-    def _has_permission_to_data_resource(self, data_resource_id: str, permission: str) -> None:
-        data_perm = f"data:{permission.lower()}"
+    def _has_permission_to_data_resource(
+        self, data_resource_id: str, permission: str
+    ) -> bool:
         assertion = (
             "brighthive-data-resource-claims" in self.claims
             and self.claims["brighthive-data-resource-claims"].get(data_resource_id)
             is not None
-            and data_perm
+            and permission
             in self.claims["brighthive-data-resource-claims"]
             .get(data_resource_id)
             .get("permissions", [])
         )
         self.assertions.append(assertion)
+        return assertion
 
-    def has_edit_to_data_resource(self, data_resource_id):
-        self._has_permission_to_data_resource(data_resource_id, "edit")
+    def has_edit_to_data_resource(self, data_resource_id) -> bool:
 
-    def has_view_to_data_resource(self, data_resource_id):
-        self._has_permission_to_data_resource(data_resource_id, "view")
+        assertion = self._has_permission_to_data_resource(data_resource_id, "data:edit")
+        return assertion
 
-    def has_download_to_data_resource(self, data_resource_id):
-        self._has_permission_to_data_resource(data_resource_id, "download")
+    def has_view_to_data_resource(self, data_resource_id) -> bool:
+        assertion = self._has_permission_to_data_resource(data_resource_id, "data:view")
+        return assertion
+
+    def has_download_to_data_resource(self, data_resource_id) -> bool:
+        assertion = self._has_permission_to_data_resource(
+            data_resource_id, "data:download"
+        )
+        return assertion
+
+    def has_edit_to_data_dictionary(self, data_resource_id) -> bool:
+
+        assertion = self._has_permission_to_data_resource(
+            data_resource_id, "data-dict:edit"
+        )
+        return assertion
+
+    def has_view_to_data_dictionary(self, data_resource_id) -> bool:
+        assertion = self._has_permission_to_data_resource(
+            data_resource_id, "data-dict:view"
+        )
+        return assertion
+
+    def has_download_to_data_dictionary(self, data_resource_id) -> bool:
+        assertion = self._has_permission_to_data_resource(
+            data_resource_id, "data-dict:download"
+        )
+        return assertion
+
+    def _get_data_resources_with_permission(self, permission: str) -> set:
+        data_resources = set()
+        if "brighthive-data-resource-claims" in self.claims:
+            data_resource_claims = self.claims["brighthive-data-resource-claims"]
+            data_perm = f"data:{permission.lower()}"
+            for data_resource_id, claim in data_resource_claims.items():
+                if data_perm in claim.get("permissions", []):
+                    data_resources.add(data_resource_id)
+        return data_resources
+
+    def get_data_resources_with_view_permission(self) -> set:
+        data_resources = self._get_data_resources_with_permission("view")
+        return data_resources
+
+    def get_data_resources_with_edit_permission(self) -> set:
+        data_resources = self._get_data_resources_with_permission("edit")
+        return data_resources
+
+    def get_data_resources_with_download_permission(self) -> set:
+        data_resources = self._get_data_resources_with_permission("download")
+        return data_resources
 
 
 class AssertJwt(AdminAssertions, DataResourceAssertions):
@@ -56,5 +130,6 @@ def claims(self):
     def fail_if_none(self):
         if not any(self.assertions):
             raise AuthorizationError(
-                http_status_code=403, error_message="JWT does not have the necessary claims to perform this action.")
-
+                http_status_code=403,
+                error_message="JWT does not have the necessary claims to perform this action.",
+            )

bhjwt/decorators/token_required_decorator.py

@@ -14,10 +14,11 @@ def wrap(f):
         def wrapped_f(*args, **kwargs):
             # if provider.validate_token(scopes=scopes):
             #     return f(*args, **kwargs)
+            # NOTE: 'scopes' is not currently being utilized. Deprecate?
             asserter = provider.validate_token(scopes=scopes)
             if asserter:
                 if api_type == "graphql":
-                    kwargs["info"].context.jwt_claims = asserter.claims
+                    args[1].context.asserter = asserter
                     return f(*args, **kwargs)
 
         return wrapped_f

bhjwt/main.py

@@ -52,7 +52,8 @@ def _decode_jwt(encoded_jwt, public_keys: list) -> dict:
                     encoded_jwt,
                     key,
                     issuer="brighthive-authserver",
-                    audience="brighthive-permissions-service",  # TODO: get value from entry point
+                    # TODO: get value from entry point
+                    audience="brighthive-permissions-service",
                     algorithms=["RS256"],
                 )