Check perm the same way everywhere

backend/infrahub/api/artifact.py

@@ -8,7 +8,7 @@
 from infrahub.api.dependencies import BranchParams, get_branch_params, get_current_user, get_db
 from infrahub.core import registry
 from infrahub.core.account import ObjectPermission
-from infrahub.core.constants import InfrahubKind, PermissionAction
+from infrahub.core.constants import GLOBAL_BRANCH_NAME, InfrahubKind, PermissionAction
 from infrahub.core.protocols import CoreArtifactDefinition
 from infrahub.database import InfrahubDatabase  # noqa: TCH001
 from infrahub.exceptions import NodeNotFoundError, PermissionDeniedError
@@ -65,7 +65,7 @@ async def generate_artifact(
 ) -> None:
     permission_decision = (
         PermissionDecisionFlag.ALLOW_DEFAULT
-        if branch_params.branch.name == registry.default_branch
+        if branch_params.branch.name in (GLOBAL_BRANCH_NAME, registry.default_branch)
         else PermissionDecisionFlag.ALLOW_OTHER
     )
     for permission in [
@@ -74,9 +74,10 @@ async def generate_artifact(
     ]:
         has_permission = False
         for permission_backend in registry.permission_backends:
-            has_permission = await permission_backend.has_permission(
+            if has_permission := await permission_backend.has_permission(
                 db=db, account_session=account_session, permission=permission, branch=branch_params.branch
-            )
+            ):
+                break
         if not has_permission:
             raise PermissionDeniedError(f"You do not have the following permission: {permission}")
 

backend/infrahub/api/schema.py

@@ -243,32 +243,42 @@ async def load_schema(
     branch: Branch = Depends(get_branch_dep),
     account_session: AccountSession = Depends(get_current_user),
 ) -> SchemaUpdate:
+    has_permission = False
+    has_branch_permission = branch.name not in (GLOBAL_BRANCH_NAME, registry.default_branch)
     for permission_backend in registry.permission_backends:
-        if not await permission_backend.has_permission(
-            db=db,
-            account_session=account_session,
-            permission=GlobalPermission(
-                action=GlobalPermissions.MANAGE_SCHEMA.value,
-                decision=(
-                    PermissionDecision.ALLOW_DEFAULT
-                    if branch.name in (GLOBAL_BRANCH_NAME, registry.default_branch)
-                    else PermissionDecision.ALLOW_OTHER
-                ).value,
-            ),
-            branch=branch,
-        ):
-            raise PermissionDeniedError("You are not allowed to manage the schema")
-
-        if branch.name in (GLOBAL_BRANCH_NAME, registry.default_branch) and not await permission_backend.has_permission(
-            db=db,
-            account_session=account_session,
-            permission=GlobalPermission(
-                action=GlobalPermissions.EDIT_DEFAULT_BRANCH.value,
-                decision=PermissionDecision.ALLOW_DEFAULT.value,
-            ),
-            branch=branch,
-        ):
-            raise PermissionDeniedError("You are not allowed to edit the schema in the default branch")
+        if not has_permission:
+            has_permission = await permission_backend.has_permission(
+                db=db,
+                account_session=account_session,
+                permission=GlobalPermission(
+                    action=GlobalPermissions.MANAGE_SCHEMA.value,
+                    decision=(
+                        PermissionDecision.ALLOW_DEFAULT
+                        if branch.name in (GLOBAL_BRANCH_NAME, registry.default_branch)
+                        else PermissionDecision.ALLOW_OTHER
+                    ).value,
+                ),
+                branch=branch,
+            )
+
+        if not has_branch_permission:
+            has_branch_permission = await permission_backend.has_permission(
+                db=db,
+                account_session=account_session,
+                permission=GlobalPermission(
+                    action=GlobalPermissions.EDIT_DEFAULT_BRANCH.value,
+                    decision=PermissionDecision.ALLOW_DEFAULT.value,
+                ),
+                branch=branch,
+            )
+
+        if has_permission and has_branch_permission:
+            break
+
+    if not has_permission:
+        raise PermissionDeniedError("You are not allowed to manage the schema")
+    if not has_branch_permission:
+        raise PermissionDeniedError("You are not allowed to edit the schema in the default branch")
 
     service: InfrahubServices = request.app.state.service
     log.info("schema_load_request", branch=branch.name)

backend/infrahub/graphql/auth/query_permission_checker/default_branch_checker.py

@@ -37,12 +37,11 @@ async def check(
         query_parameters: GraphqlParams,
         branch: Branch,
     ) -> CheckerResolution:
-        can_edit_default_branch = False
+        has_permission = False
         for permission_backend in registry.permission_backends:
-            can_edit_default_branch = await permission_backend.has_permission(
+            if has_permission := await permission_backend.has_permission(
                 db=db, account_session=account_session, permission=self.permission_required, branch=branch
-            )
-            if can_edit_default_branch:
+            ):
                 break
 
         operates_on_default_branch = analyzed_query.branch is None or analyzed_query.branch.name in (
@@ -54,7 +53,7 @@ async def check(
         )
 
         if (
-            not can_edit_default_branch
+            not has_permission
             and operates_on_default_branch
             and analyzed_query.contains_mutation
             and not is_exempt_operation

backend/infrahub/graphql/auth/query_permission_checker/merge_operation_checker.py

@@ -31,15 +31,14 @@ async def check(
         branch: Branch,
     ) -> CheckerResolution:
         if "BranchMerge" in [operation.name for operation in analyzed_query.operations]:
-            can_merge_branch = False
+            has_permission = False
             for permission_backend in registry.permission_backends:
-                can_merge_branch = await permission_backend.has_permission(
+                if has_permission := await permission_backend.has_permission(
                     db=db, account_session=account_session, permission=self.permission_required, branch=branch
-                )
-                if can_merge_branch:
+                ):
                     break
 
-            if not can_merge_branch:
+            if not has_permission:
                 raise PermissionDeniedError("You are not allowed to merge a branch")
 
             return CheckerResolution.TERMINATE

backend/infrahub/graphql/auth/query_permission_checker/object_permission_checker.py

@@ -72,9 +72,10 @@ async def check(
         for permission in permissions:
             has_permission = False
             for permission_backend in registry.permission_backends:
-                has_permission = await permission_backend.has_permission(
+                if has_permission := await permission_backend.has_permission(
                     db=db, account_session=account_session, permission=permission, branch=branch
-                )
+                ):
+                    break
             if not has_permission:
                 raise PermissionDeniedError(f"You do not have the following permission: {permission}")
 
@@ -168,11 +169,15 @@ async def check(
         if not is_permission_operation:
             return CheckerResolution.NEXT_CHECKER
 
+        has_permission = False
         for permission_backend in registry.permission_backends:
-            if not await permission_backend.has_permission(
+            if has_permission := await permission_backend.has_permission(
                 db=db, account_session=account_session, permission=self.permission_required, branch=branch
             ):
-                raise PermissionDeniedError("You do not have the permission to manage permissions")
+                break
+
+        if not has_permission:
+            raise PermissionDeniedError("You do not have the permission to manage permissions")
 
         return CheckerResolution.NEXT_CHECKER
 
@@ -213,10 +218,14 @@ async def check(
         if not is_repository_operation or not analyzed_query.contains_mutation:
             return CheckerResolution.NEXT_CHECKER
 
+        has_permission = False
         for permission_backend in registry.permission_backends:
-            if not await permission_backend.has_permission(
+            if has_permission := await permission_backend.has_permission(
                 db=db, account_session=account_session, permission=self.permission_required, branch=branch
             ):
-                raise PermissionDeniedError("You do not have the permission to manage repositories")
+                break
+
+        if not has_permission:
+            raise PermissionDeniedError("You do not have the permission to manage repositories")
 
         return CheckerResolution.NEXT_CHECKER

changelog/+perm-check.fixed.md

@@ -0,0 +1 @@
+Fix permission check when using multiple backends, if one grants a permission the next ones must not be queried.