Check for permissions in rel add/delete mutations

opsmill · Feb 17, 2025 · 1c504e3 · 1c504e3
1 parent ca89edc
commit 1c504e3
Show file tree

Hide file tree

Showing 3 changed files with 57 additions and 2 deletions.
diff --git a/backend/infrahub/graphql/mutations/relationship.py b/backend/infrahub/graphql/mutations/relationship.py
@@ -5,7 +5,8 @@
 from graphene import Boolean, InputField, InputObjectType, List, Mutation, String
 from infrahub_sdk.utils import compare_lists
 
-from infrahub.core.constants import InfrahubKind, RelationshipCardinality
+from infrahub.core.account import GlobalPermission, ObjectPermission
+from infrahub.core.constants import InfrahubKind, PermissionAction, PermissionDecision, RelationshipCardinality
 from infrahub.core.manager import NodeManager
 from infrahub.core.query.relationship import (
     RelationshipGetPeerQuery,
@@ -14,6 +15,7 @@
 from infrahub.core.relationship import Relationship
 from infrahub.database import retry_db_transaction
 from infrahub.exceptions import NodeNotFoundError, ValidationError
+from infrahub.permissions import get_global_permission_for_kind
 
 from ..types import RelatedNodeInput
 
@@ -76,6 +78,32 @@ async def mutate(
             db=context.db, ids=node_ids, fields={"display_label": None}, branch=context.branch
         )
 
+        if context.account_session:
+            impacted_schemas = {node.get_schema() for node in [source] + list(nodes.values())}
+            required_permissions: list[GlobalPermission | ObjectPermission] = []
+            decision = (
+                PermissionDecision.ALLOW_DEFAULT.value
+                if context.branch.is_default
+                else PermissionDecision.ALLOW_OTHER.value
+            )
+
+            for impacted_schema in impacted_schemas:
+                global_action = get_global_permission_for_kind(schema=impacted_schema)
+
+                if global_action:
+                    required_permissions.append(GlobalPermission(action=global_action, decision=decision))
+                else:
+                    required_permissions.append(
+                        ObjectPermission(
+                            namespace=impacted_schema.namespace,
+                            name=impacted_schema.name,
+                            action=PermissionAction.UPDATE.value,
+                            decision=decision,
+                        )
+                    )
+
+            context.active_permissions.raise_for_permissions(permissions=required_permissions)
+
         _, _, in_list2 = compare_lists(list1=list(nodes.keys()), list2=node_ids)
         if in_list2:
             for node_id in in_list2:

diff --git a/backend/infrahub/permissions/__init__.py b/backend/infrahub/permissions/__init__.py
@@ -2,12 +2,13 @@
 from infrahub.permissions.local_backend import LocalPermissionBackend
 from infrahub.permissions.manager import PermissionManager
 from infrahub.permissions.report import report_schema_permissions
-from infrahub.permissions.types import AssignedPermissions
+from infrahub.permissions.types import AssignedPermissions, get_global_permission_for_kind
 
 __all__ = [
     "AssignedPermissions",
     "LocalPermissionBackend",
     "PermissionBackend",
     "PermissionManager",
+    "get_global_permission_for_kind",
     "report_schema_permissions",
 ]
diff --git a/backend/infrahub/permissions/types.py b/backend/infrahub/permissions/types.py
@@ -2,8 +2,12 @@
 
 from typing import TYPE_CHECKING, TypedDict
 
+from infrahub.core.constants import GlobalPermissions, InfrahubKind
+from infrahub.core.schema import NodeSchema
+
 if TYPE_CHECKING:
     from infrahub.core.account import GlobalPermission, ObjectPermission
+    from infrahub.core.schema import MainSchemaTypes
     from infrahub.permissions.constants import BranchRelativePermissionDecision
 
 
@@ -18,3 +22,25 @@ class KindPermissions(TypedDict):
     delete: BranchRelativePermissionDecision
     update: BranchRelativePermissionDecision
     view: BranchRelativePermissionDecision
+
+
+def get_global_permission_for_kind(schema: MainSchemaTypes) -> GlobalPermissions | None:
+    kind_permission_map = {
+        InfrahubKind.GENERICACCOUNT: GlobalPermissions.MANAGE_ACCOUNTS,
+        InfrahubKind.ACCOUNTGROUP: GlobalPermissions.MANAGE_ACCOUNTS,
+        InfrahubKind.ACCOUNTROLE: GlobalPermissions.MANAGE_ACCOUNTS,
+        InfrahubKind.BASEPERMISSION: GlobalPermissions.MANAGE_PERMISSIONS,
+        InfrahubKind.GENERICREPOSITORY: GlobalPermissions.MANAGE_REPOSITORIES,
+    }
+
+    if schema.kind in kind_permission_map:
+        return kind_permission_map[schema.kind]
+
+    if isinstance(schema, NodeSchema):
+        for base in schema.inherit_from:
+            try:
+                return kind_permission_map[base]
+            except KeyError:
+                continue
+
+    return None