Split permissions query into two

backend/infrahub/auth.py

@@ -192,8 +192,10 @@ async def validate_jwt_access_token(db: InfrahubDatabase, token: str) -> Account
         account_id = payload["sub"]
         role = payload["user_claims"]["role"]
         session_id = payload["session_id"]
+        recorded_permissions = await fetch_permissions(db=db, account_id=account_id)
         permissions = AccountPermissions(
-            global_permissions=[str(p) for p in await fetch_permissions(db=db, account_id=account_id)]
+            global_permissions=[str(p) for p in recorded_permissions.get("global_permissions", [])],
+            object_permissions=[str(p) for p in recorded_permissions.get("object_permissions", [])],
         )
     except jwt.ExpiredSignatureError:
         raise AuthorizationError("Expired Signature") from None

backend/infrahub/core/account.py

@@ -14,28 +14,30 @@
 
 
 @dataclass
-class GlobalPermission:
+class Permission:
     id: str
-    name: str
     action: str
 
+
+@dataclass
+class GlobalPermission(Permission):
+    name: str
+
     def __str__(self) -> str:
         return f"global:{self.action}:allow"
 
 
 @dataclass
-class ObjectPermission:
-    id: str
+class ObjectPermission(Permission):
     namespace: str
     kind: str
-    action: str
 
     def __str__(self) -> str:
         return f"object:{self.namespace}:{self.kind}:{self.action}"
 
 
-class AccountPermissionQuery(Query):
-    name: str = "account_permissions"
+class AccountGlobalPermissionQuery(Query):
+    name: str = "account_global_permissions"
 
     def __init__(self, account_id: str, **kwargs: Any):
         self.account_id = account_id
@@ -44,37 +46,122 @@ def __init__(self, account_id: str, **kwargs: Any):
     async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:
         self.params["account_id"] = self.account_id
 
+        branch_filter, branch_params = self.branch.get_query_filter_path(
+            at=self.at.to_string(), branch_agnostic=self.branch_agnostic
+        )
+        self.params.update(branch_params)
+
         # ruff: noqa: E501
         query = """
-        MATCH (account:CoreGenericAccount { uuid: $account_id })-[]->(:Relationship {name: "usergroup__users"})<-[]-(group:CoreUserGroup)-[]->(:Relationship {name: "role__usergroups"})<-[]-(role:CoreUserRole)-[]->(:Relationship {name: "role__permissions"})<-[]-(permission:CoreBasePermission)-[:HAS_ATTRIBUTE]->(:Attribute {name: "action"})-[:HAS_VALUE]->(permission_action:AttributeValue)
-        """
+        MATCH (account:CoreGenericAccount)
+        WHERE account.uuid = $account_id
+        CALL {
+            WITH account
+            MATCH (ns)-[r:IS_PART_OF]-(root:Root)
+            WHERE %(branch_filter)s
+            RETURN account as account1, r as r1
+            ORDER BY r.branch_level DESC, r.from DESC
+            LIMIT 1
+        }
+        WITH account, r1 as r
+        WHERE r.status = "active"
+        WITH account
+        MATCH (account)-[]->(:Relationship {name: "usergroup__users"})<-[]-(:CoreUserGroup)-[]->(:Relationship {name: "role__usergroups"})<-[]-(:CoreUserRole)-[]->(:Relationship {name: "role__permissions"})<-[]-(global_permission:CoreGlobalPermission)-[:HAS_ATTRIBUTE]->(:Attribute {name: "action"})-[:HAS_VALUE]->(global_permission_action:AttributeValue)
+        """ % {"branch_filter": branch_filter}
 
         self.add_to_query(query)
 
-        self.return_labels = ["account", "permission", "permission_action"]
+        self.return_labels = ["global_permission", "global_permission_action"]
 
     def get_permissions(self) -> list[GlobalPermission]:
         permissions: list[GlobalPermission] = []
 
         for result in self.get_results():
             permissions.append(
                 GlobalPermission(
-                    id=result.get("permission").get("uuid"),
-                    name=result.get("permission_action").get("value"),
-                    action=result.get("permission_action").get("value"),
+                    id=result.get("global_permission").get("uuid"),
+                    name=result.get("global_permission_action").get("value"),
+                    action=result.get("global_permission_action").get("value"),
                 )
             )
 
         return permissions
 
 
+class AccountObjectPermissionQuery(Query):
+    name: str = "account_object_permissions"
+
+    def __init__(self, account_id: str, **kwargs: Any):
+        self.account_id = account_id
+        super().__init__(**kwargs)
+
+    async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:
+        self.params["account_id"] = self.account_id
+
+        branch_filter, branch_params = self.branch.get_query_filter_path(
+            at=self.at.to_string(), branch_agnostic=self.branch_agnostic
+        )
+        self.params.update(branch_params)
+
+        # FIXME: Need to find a way to retrieve all attribute values needed
+        # The below seems to somewhat gives what's needed
+        # ruff: noqa: E501
+        query = """
+        MATCH (account:CoreGenericAccount)
+        WHERE account.uuid = $account_id
+        CALL {
+            WITH account
+            MATCH (ns)-[r:IS_PART_OF]-(root:Root)
+            WHERE %(branch_filter)s
+            RETURN account as account1, r as r1
+            ORDER BY r.branch_level DESC, r.from DESC
+            LIMIT 1
+        }
+        WITH account, r1 as r
+        WHERE r.status = "active"
+        WITH account
+        MATCH (account)-[]->(:Relationship {name: "usergroup__users"})<-[]-(:CoreUserGroup)-[]->(:Relationship {name: "role__usergroups"})<-[]-(:CoreUserRole)-[]->(:Relationship {name: "role__permissions"})<-[]-(object_permission:CoreObjectPermission)-[:HAS_ATTRIBUTE]->(object_permission_attr:Attribute)-[:HAS_VALUE]->(object_permission_attr_value:AttributeValue)
+        WHERE object_permission_attr.name IN ["namespace", "kind", "action"]
+        WITH object_permission_attr, object_permission_value
+        WITH COLLECT([object_permission_attr, object_permission_value]) as names_values
+        UNWIND names_values as name_value
+        RETURN name_value
+        """ % {"branch_filter": branch_filter}
+
+        self.add_to_query(query)
+
+        self.return_labels = ["name_value"]
+
+    def get_permissions(self) -> list[ObjectPermission]:
+        permissions: list[ObjectPermission] = []
+        for result in self.get_results():
+            # permissions.append(
+            #     ObjectPermission(
+            #         id=result.get("object_permission").get("uuid"),
+            #         namespace="",
+            #         kind="",
+            #         action=result.get("object_permission_action").get("value"),
+            #     )
+            # )
+            ...
+
+        return permissions
+
+
 async def fetch_permissions(
     account_id: str, db: InfrahubDatabase, branch: Optional[Union[Branch, str]] = None
-) -> list[GlobalPermission]:
+) -> dict[str, list[GlobalPermission] | list[ObjectPermission]]:
     branch = await registry.get_branch(db=db, branch=branch)
-    query = await AccountPermissionQuery.init(db=db, branch=branch, account_id=account_id)
-    await query.execute(db=db)
-    return query.get_permissions()
+
+    query1 = await AccountGlobalPermissionQuery.init(db=db, branch=branch, account_id=account_id)
+    await query1.execute(db=db)
+    global_permissions = query1.get_permissions()
+
+    query2 = await AccountObjectPermissionQuery.init(db=db, branch=branch, account_id=account_id)
+    # await query2.execute(db=db)
+    object_permissions = query2.get_permissions()
+
+    return {"global_permissions": global_permissions, "object_permissions": object_permissions}
 
 
 class AccountTokenValidateQuery(Query):

backend/infrahub/graphql/queries/account.py

@@ -123,15 +123,29 @@ async def resolve_account_permissions(
     )
 
     response: dict[str, Any] = {}
+
     if "global_permissions" in fields:
-        response["global_permissions"] = {"count": len(permissions)}
+        global_list = permissions["global_permissions"]
+        response["global_permissions"] = {"count": len(global_list)}
         response["global_permissions"]["edges"] = [
-            {"node": {"id": obj.id, "name": obj.name, "action": obj.action, "identifier": str(obj)}}
-            for obj in permissions
+            {"node": {"id": obj.id, "name": obj.name, "action": obj.action, "identifier": str(obj)}}  # type: ignore[union-attr]
+            for obj in global_list
         ]
     if "object_permissions" in fields:
-        response["object_permissions"] = {"count": 0}
-        response["object_permissions"]["edges"] = []
+        object_list = permissions["object_permissions"]
+        response["object_permissions"] = {"count": len(object_list)}
+        response["object_permissions"]["edges"] = [
+            {
+                "node": {
+                    "id": obj.id,
+                    "namespace": obj.namespace,  # type: ignore[union-attr]
+                    "kind": obj.kind,  # type: ignore[union-attr]
+                    "action": obj.action,
+                    "identifier": str(obj),
+                }
+            }
+            for obj in object_list
+        ]
 
     return response