IFC-768 Implement attribute level permission report (#4616)

backend/infrahub/core/attribute.py

@@ -460,6 +460,7 @@ async def to_graphql(
         fields: Optional[dict] = None,
         related_node_ids: Optional[set] = None,
         filter_sensitive: bool = False,
+        permissions: Optional[dict] = None,
     ) -> dict:
         """Generate GraphQL Payload for this attribute."""
         # pylint: disable=too-many-branches
@@ -486,6 +487,10 @@ async def to_graphql(
                 response[field_name] = self.get_kind()
                 continue
 
+            if field_name == "permissions":
+                response[field_name] = {"update_value": permissions["update"]} if permissions else None
+                continue
+
             if field_name in ["source", "owner"]:
                 node_attr_getter = getattr(self, f"get_{field_name}")
                 node_attr = await node_attr_getter(db=db)

backend/infrahub/core/constants/__init__.py

@@ -63,8 +63,8 @@ class GlobalPermissions(InfrahubStringEnum):
 
 class PermissionAction(InfrahubStringEnum):
     ANY = "any"
-    ADD = "create"
-    CHANGE = "update"
+    CREATE = "create"
+    UPDATE = "update"
     DELETE = "delete"
     VIEW = "view"
 

backend/infrahub/core/node/__init__.py

@@ -528,6 +528,7 @@ async def to_graphql(
         fields: Optional[dict] = None,
         related_node_ids: Optional[set] = None,
         filter_sensitive: bool = False,
+        permissions: Optional[dict] = None,
     ) -> dict:
         """Generate GraphQL Payload for all attributes
 
@@ -579,11 +580,11 @@ async def to_graphql(
                     fields=fields.get(field_name),
                     related_node_ids=related_node_ids,
                     filter_sensitive=filter_sensitive,
+                    permissions=permissions,
                 )
             else:
                 response[field_name] = await field.to_graphql(
-                    db=db,
-                    filter_sensitive=filter_sensitive,
+                    db=db, filter_sensitive=filter_sensitive, permissions=permissions
                 )
 
         return response

backend/infrahub/core/node/ipam.py

@@ -19,6 +19,7 @@ async def to_graphql(
         fields: Optional[dict] = None,
         related_node_ids: Optional[set] = None,
         filter_sensitive: bool = False,
+        permissions: Optional[dict] = None,
     ) -> dict:
         response = await super().to_graphql(
             db, fields=fields, related_node_ids=related_node_ids, filter_sensitive=filter_sensitive

backend/infrahub/core/node/permissions.py

@@ -15,9 +15,14 @@ async def to_graphql(
         fields: Optional[dict] = None,
         related_node_ids: Optional[set] = None,
         filter_sensitive: bool = False,
+        permissions: Optional[dict] = None,
     ) -> dict:
         response = await super().to_graphql(
-            db, fields=fields, related_node_ids=related_node_ids, filter_sensitive=filter_sensitive
+            db,
+            fields=fields,
+            related_node_ids=related_node_ids,
+            filter_sensitive=filter_sensitive,
+            permissions=permissions,
         )
 
         if fields:
@@ -34,9 +39,14 @@ async def to_graphql(
         fields: Optional[dict] = None,
         related_node_ids: Optional[set] = None,
         filter_sensitive: bool = False,
+        permissions: Optional[dict] = None,
     ) -> dict:
         response = await super().to_graphql(
-            db, fields=fields, related_node_ids=related_node_ids, filter_sensitive=filter_sensitive
+            db,
+            fields=fields,
+            related_node_ids=related_node_ids,
+            filter_sensitive=filter_sensitive,
+            permissions=permissions,
         )
 
         if fields:

backend/infrahub/core/protocols_base.py

@@ -93,6 +93,7 @@ async def to_graphql(
         fields: Optional[dict] = None,
         related_node_ids: Optional[set] = None,
         filter_sensitive: bool = False,
+        permissions: Optional[dict] = None,
     ) -> dict: ...
     async def render_display_label(self, db: Optional[InfrahubDatabase] = None) -> str: ...
     async def from_graphql(self, data: dict, db: InfrahubDatabase) -> bool: ...

backend/infrahub/core/schema/manager.py

@@ -19,6 +19,7 @@
     GenericSchema,
     MainSchemaTypes,
     NodeSchema,
+    ProfileSchema,
     RelationshipSchema,
     SchemaRoot,
 )
@@ -97,6 +98,15 @@ def get_node_schema(
 
         raise ValueError("The selected node is not of type NodeSchema")
 
+    def get_profile_schema(
+        self, name: str, branch: Optional[Union[Branch, str]] = None, duplicate: bool = True
+    ) -> ProfileSchema:
+        schema = self.get(name=name, branch=branch, duplicate=duplicate)
+        if isinstance(schema, ProfileSchema):
+            return schema
+
+        raise ValueError("The selected node is not of type ProfileSchema")
+
     def get_full(
         self, branch: Optional[Union[Branch, str]] = None, duplicate: bool = True
     ) -> dict[str, MainSchemaTypes]:

backend/infrahub/graphql/permissions.py

@@ -16,9 +16,14 @@ async def get_permissions(db: InfrahubDatabase, schema: MainSchemaTypes, context
     schema_objects = [schema]
     if isinstance(schema, GenericSchema):
         for node_name in schema.used_by:
-            schema_objects.append(
-                registry.schema.get_node_schema(name=node_name, branch=context.branch, duplicate=False)
-            )
+            schema_object: MainSchemaTypes
+            try:
+                schema_object = registry.schema.get_node_schema(name=node_name, branch=context.branch, duplicate=False)
+            except ValueError:
+                schema_object = registry.schema.get_profile_schema(
+                    name=node_name, branch=context.branch, duplicate=False
+                )
+            schema_objects.append(schema_object)
 
     response: dict[str, Any] = {"count": len(schema_objects), "edges": []}
 

backend/infrahub/graphql/resolver.py

@@ -139,9 +139,15 @@ async def default_paginated_list_resolver(
         edges = fields.get("edges", {})
         node_fields = edges.get("node", {})
 
-        permissions = fields.get("permissions")
+        permission_set: Optional[dict[str, Any]] = None
+        permissions = await get_permissions(db=db, schema=schema, context=context) if context.account_session else None
+        if fields.get("permissions"):
+            response["permissions"] = permissions
+
         if permissions:
-            response["permissions"] = await get_permissions(db=db, schema=schema, context=context)
+            for edge in permissions["edges"]:
+                if edge["node"]["kind"] == schema.kind:
+                    permission_set = edge["node"]
 
         objs = []
         if edges or "hfid" in filters:
@@ -175,7 +181,14 @@ async def default_paginated_list_resolver(
 
         if objs:
             objects = [
-                {"node": await obj.to_graphql(db=db, fields=node_fields, related_node_ids=context.related_node_ids)}
+                {
+                    "node": await obj.to_graphql(
+                        db=db,
+                        fields=node_fields,
+                        related_node_ids=context.related_node_ids,
+                        permissions=permission_set,
+                    )
+                }
                 for obj in objs
             ]
             response["edges"] = objects

backend/infrahub/graphql/types/attribute.py

@@ -55,6 +55,10 @@ class RelatedPrefixNodeInput(InputObjectType):
     _relation__source = String(required=False)
 
 
+class PermissionType(ObjectType):
+    update_value = String(required=False)
+
+
 class AttributeInterface(InfrahubInterface):
     is_default = Field(Boolean)
     is_inherited = Field(Boolean)
@@ -70,6 +74,7 @@ class AttributeInterface(InfrahubInterface):
 class BaseAttribute(ObjectType):
     id = Field(String)
     is_from_profile = Field(Boolean)
+    permissions = Field(PermissionType, required=False)
 
     @classmethod
     def __init_subclass__(cls, **kwargs: dict[str, Any]) -> None:

backend/infrahub/permissions/report.py

@@ -2,7 +2,6 @@
 
 from typing import TYPE_CHECKING
 
-from infrahub.core.account import fetch_permissions
 from infrahub.core.constants import GlobalPermissions, PermissionDecision
 from infrahub.core.registry import registry
 from infrahub.permissions.local_backend import LocalPermissionBackend
@@ -18,8 +17,8 @@
 async def report_schema_permissions(
     db: InfrahubDatabase, schemas: list[MainSchemaTypes], account_session: AccountSession, branch: Branch
 ) -> list[KindPermissions]:
-    permissions = await fetch_permissions(account_id=account_session.account_id, db=db, branch=branch)
     perm_backend = LocalPermissionBackend()
+    permissions = await perm_backend.load_permissions(db=db, account_id=account_session.account_id, branch=branch)
 
     # Check for super admin permission and handle default branch edition if account is not super admin
     is_super_admin = perm_backend.resolve_global_permission(

backend/tests/unit/graphql/queries/test_list_permissions.py

@@ -21,6 +21,7 @@
 if TYPE_CHECKING:
     from infrahub.core.branch import Branch
     from infrahub.core.protocols import CoreAccount
+    from infrahub.core.schema.schema_branch import SchemaBranch
     from infrahub.database import InfrahubDatabase
     from tests.unit.graphql.conftest import PermissionsHelper
 
@@ -68,7 +69,7 @@ class TestObjectPermissions:
     async def test_setup(
         self,
         db: InfrahubDatabase,
-        register_core_models_schema: None,
+        register_core_models_schema: SchemaBranch,
         default_branch: Branch,
         permissions_helper: PermissionsHelper,
         first_account: CoreAccount,
@@ -92,7 +93,7 @@ async def test_setup(
                 branch="*",
                 namespace="Builtin",
                 name="*",
-                action=PermissionAction.ADD.value,
+                action=PermissionAction.CREATE.value,
                 decision=PermissionDecision.ALLOW.value,
             ),
             ObjectPermission(
@@ -238,3 +239,143 @@ async def test_first_account_list_permissions_for_generics(
                 "view": "ALLOW",
             }
         } in result.data["CoreGenericRepository"]["permissions"]["edges"]
+
+
+QUERY_TAGS_ATTR = """
+query {
+  BuiltinTag {
+    count
+    edges {
+      node {
+        name {
+          value
+          permissions {
+            update_value
+          }
+        }
+      }
+    }
+  }
+}
+"""
+
+
+class TestAttributePermissions:
+    async def test_setup(
+        self,
+        db: InfrahubDatabase,
+        register_core_models_schema: SchemaBranch,
+        default_branch: Branch,
+        permissions_helper: PermissionsHelper,
+        first_account: CoreAccount,
+    ):
+        permissions_helper._first = first_account
+        permissions_helper._default_branch = default_branch
+        registry.permission_backends = [LocalPermissionBackend()]
+
+        permissions = []
+        for object_permission in [
+            ObjectPermission(
+                id="",
+                branch="*",
+                namespace="Builtin",
+                name="*",
+                action=PermissionAction.VIEW.value,
+                decision=PermissionDecision.ALLOW.value,
+            ),
+            ObjectPermission(
+                id="",
+                branch="*",
+                namespace="Builtin",
+                name="*",
+                action=PermissionAction.CREATE.value,
+                decision=PermissionDecision.ALLOW.value,
+            ),
+            ObjectPermission(
+                id="",
+                branch="*",
+                namespace="Builtin",
+                name="*",
+                action=PermissionAction.DELETE.value,
+                decision=PermissionDecision.ALLOW.value,
+            ),
+            ObjectPermission(
+                id="",
+                branch="pr-12345",
+                namespace="Builtin",
+                name="*",
+                action=PermissionAction.UPDATE.value,
+                decision=PermissionDecision.ALLOW.value,
+            ),
+        ]:
+            obj = await Node.init(db=db, schema=InfrahubKind.OBJECTPERMISSION)
+            await obj.new(
+                db=db,
+                branch=object_permission.branch,
+                namespace=object_permission.namespace,
+                name=object_permission.name,
+                action=object_permission.action,
+                decision=object_permission.decision,
+            )
+            await obj.save(db=db)
+            permissions.append(obj)
+
+        role = await Node.init(db=db, schema=InfrahubKind.ACCOUNTROLE)
+        await role.new(db=db, name="admin", permissions=permissions)
+        await role.save(db=db)
+
+        group = await Node.init(db=db, schema=InfrahubKind.ACCOUNTGROUP)
+        await group.new(db=db, name="admin", roles=[role])
+        await group.save(db=db)
+
+        await group.members.add(db=db, data={"id": first_account.id})
+        await group.members.save(db=db)
+
+        tag = await Node.init(db=db, schema=InfrahubKind.TAG)
+        await tag.new(db=db, name="Blue", description="Blue tag")
+        await tag.save(db=db)
+
+    async def test_first_account_tags_main_branch(
+        self, db: InfrahubDatabase, permissions_helper: PermissionsHelper
+    ) -> None:
+        """In the main branch the first account doesn't have the permission to make changes, so attribute cannot be changed"""
+        session = AccountSession(
+            authenticated=True,
+            account_id=permissions_helper.first.id,
+            session_id=str(uuid4()),
+            auth_type=AuthType.JWT,
+        )
+        gql_params = prepare_graphql_params(
+            db=db, include_mutation=True, branch=permissions_helper.default_branch, account_session=session
+        )
+
+        result = await graphql(schema=gql_params.schema, source=QUERY_TAGS_ATTR, context_value=gql_params.context)
+
+        assert not result.errors
+        assert result.data
+        assert result.data["BuiltinTag"]["count"] == 1
+        assert result.data["BuiltinTag"]["edges"][0]["node"]["name"]["permissions"] == {
+            "update_value": PermissionDecision.DENY
+        }
+
+    async def test_first_account_tags_non_main_branch(
+        self, db: InfrahubDatabase, permissions_helper: PermissionsHelper
+    ) -> None:
+        """In other branches the permissions for the first account is less restrictive, attribute should be updatable"""
+        branch2 = await create_branch(branch_name="pr-12345", db=db)
+        session = AccountSession(
+            authenticated=True,
+            account_id=permissions_helper.first.id,
+            session_id=str(uuid4()),
+            auth_type=AuthType.JWT,
+        )
+        gql_params = prepare_graphql_params(db=db, include_mutation=True, branch=branch2, account_session=session)
+
+        result = await graphql(schema=gql_params.schema, source=QUERY_TAGS_ATTR, context_value=gql_params.context)
+
+        assert not result.errors
+        assert result.data
+        assert result.data["BuiltinTag"]["count"] == 1
+        assert result.data["BuiltinTag"]["edges"][0]["node"]["name"]["permissions"] == {
+            "update_value": PermissionDecision.ALLOW
+        }

backend/tests/unit/permissions/test_backends.py

@@ -194,7 +194,7 @@ async def test_has_permission_object(
         branch="*",
         namespace="Builtin",
         name="Tag",
-        action=PermissionAction.ADD.value,
+        action=PermissionAction.CREATE.value,
         decision=PermissionDecision.ALLOW.value,
     )
     assert not await backend.has_permission(