IFC-761 Use flags to handle branch permissions (#4601)

backend/infrahub/api/schema.py

@@ -16,8 +16,9 @@
 from infrahub.api.dependencies import get_branch_dep, get_current_user, get_db
 from infrahub.api.exceptions import SchemaNotValidError
 from infrahub.core import registry
+from infrahub.core.account import GlobalPermission
 from infrahub.core.branch import Branch  # noqa: TCH001
-from infrahub.core.constants import GlobalPermissions, PermissionDecision
+from infrahub.core.constants import GLOBAL_BRANCH_NAME, GlobalPermissions, PermissionDecision
 from infrahub.core.migrations.schema.models import SchemaApplyMigrationData
 from infrahub.core.models import (  # noqa: TCH001
     SchemaBranchHash,
@@ -247,7 +248,16 @@ async def load_schema(
         if not await permission_backend.has_permission(
             db=db,
             account_id=account_session.account_id,
-            permission=f"global:{GlobalPermissions.MANAGE_SCHEMA.value}:{PermissionDecision.ALLOW.value}",
+            permission=GlobalPermission(
+                id="",
+                name="",
+                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")

backend/infrahub/core/account.py

@@ -3,7 +3,7 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Optional, Union
 
-from infrahub.core.constants import InfrahubKind
+from infrahub.core.constants import InfrahubKind, PermissionDecision
 from infrahub.core.query import Query
 from infrahub.core.registry import registry
 
@@ -21,22 +21,23 @@ class Permission:
     id: str
     name: str
     action: str
-    decision: str
+    decision: int
 
 
 @dataclass
 class GlobalPermission(Permission):
     def __str__(self) -> str:
-        return f"global:{self.action}:{self.decision}"
+        decision = PermissionDecision(self.decision)
+        return f"global:{self.action}:{decision.name.lower()}"
 
 
 @dataclass
 class ObjectPermission(Permission):
-    branch: str
     namespace: str
 
     def __str__(self) -> str:
-        return f"object:{self.branch}:{self.namespace}:{self.name}:{self.action}:{self.decision}"
+        decision = PermissionDecision(self.decision)
+        return f"object:{self.namespace}:{self.name}:{self.action}:{decision.name.lower()}"
 
 
 class AccountGlobalPermissionQuery(Query):
@@ -234,17 +235,6 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:
             RETURN object_permission
         }
         WITH object_permission
-        CALL {
-            WITH object_permission
-            MATCH (object_permission)-[r1:HAS_ATTRIBUTE]->(:Attribute {name: "branch"})-[r2:HAS_VALUE]->(object_permission_branch:AttributeValue)
-            WHERE all(r IN [r1, r2] WHERE (%(branch_filter)s))
-            WITH object_permission_branch, r1, r2, (r1.status = "active" AND r2.status = "active") AS is_active
-            ORDER BY object_permission_branch.uuid, r2.branch_level DESC, r2.from DESC, r1.branch_level DESC, r1.from DESC
-            WITH object_permission_branch, head(collect(is_active)) as latest_is_active
-            WHERE latest_is_active = TRUE
-            RETURN object_permission_branch
-        }
-        WITH object_permission, object_permission_branch
 
         CALL {
             WITH object_permission
@@ -254,7 +244,7 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:
             ORDER BY r2.branch_level DESC, r2.from DESC, r1.branch_level DESC, r1.from DESC
             LIMIT 1
         }
-        WITH object_permission, object_permission_branch, object_permission_namespace, is_active AS opn_is_active
+        WITH object_permission, object_permission_namespace, is_active AS opn_is_active
         WHERE opn_is_active = TRUE
         CALL {
             WITH object_permission
@@ -264,7 +254,7 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:
             ORDER BY r2.branch_level DESC, r2.from DESC, r1.branch_level DESC, r1.from DESC
             LIMIT 1
         }
-        WITH object_permission, object_permission_branch, object_permission_namespace, object_permission_name, is_active AS opn_is_active
+        WITH object_permission, object_permission_namespace, object_permission_name, is_active AS opn_is_active
         WHERE opn_is_active = TRUE
         CALL {
             WITH object_permission
@@ -274,7 +264,7 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:
             ORDER BY r2.branch_level DESC, r2.from DESC, r1.branch_level DESC, r1.from DESC
             LIMIT 1
         }
-        WITH object_permission, object_permission_branch, object_permission_namespace, object_permission_name, object_permission_action, is_active AS opa_is_active
+        WITH object_permission, object_permission_namespace, object_permission_name, object_permission_action, is_active AS opa_is_active
         WHERE opa_is_active = TRUE
         CALL {
             WITH object_permission
@@ -284,7 +274,7 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:
             ORDER BY r2.branch_level DESC, r2.from DESC, r1.branch_level DESC, r1.from DESC
             LIMIT 1
         }
-        WITH object_permission, object_permission_branch, object_permission_namespace, object_permission_name, object_permission_action, object_permission_decision, is_active AS opd_is_active
+        WITH object_permission, object_permission_namespace, object_permission_name, object_permission_action, object_permission_decision, is_active AS opd_is_active
         WHERE opd_is_active = TRUE
         """ % {
             "branch_filter": branch_filter,
@@ -298,7 +288,6 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:
 
         self.return_labels = [
             "object_permission",
-            "object_permission_branch",
             "object_permission_namespace",
             "object_permission_name",
             "object_permission_action",
@@ -311,7 +300,6 @@ def get_permissions(self) -> list[ObjectPermission]:
             permissions.append(
                 ObjectPermission(
                     id=result.get("object_permission").get("uuid"),
-                    branch=result.get("object_permission_branch").get("value"),
                     namespace=result.get("object_permission_namespace").get("value"),
                     name=result.get("object_permission_name").get("value"),
                     action=result.get("object_permission_action").get("value"),

backend/infrahub/core/attribute.py

@@ -465,9 +465,7 @@ async def to_graphql(
         """Generate GraphQL Payload for this attribute."""
         # pylint: disable=too-many-branches
 
-        response: dict[str, Any] = {
-            "id": self.id,
-        }
+        response: dict[str, Any] = {"id": self.id}
 
         if fields and isinstance(fields, dict):
             field_names = fields.keys()
@@ -508,6 +506,9 @@ async def to_graphql(
                     )
                 continue
 
+            if field_name == "permissions":
+                response["permissions"] = {"view": "ALLOW", "update": "ALLOW"}
+
             if field_name.startswith("_"):
                 field = getattr(self, field_name[1:])
             else:

backend/infrahub/core/constants/__init__.py

@@ -4,7 +4,7 @@
 
 from infrahub.core.constants import infrahubkind as InfrahubKind
 from infrahub.exceptions import ValidationError
-from infrahub.utils import InfrahubStringEnum
+from infrahub.utils import InfrahubNumberEnum, InfrahubStringEnum
 
 from .schema import FlagProperty, NodeProperty, SchemaElementPathType, UpdateSupport, UpdateValidationErrorType
 
@@ -69,9 +69,11 @@ class PermissionAction(InfrahubStringEnum):
     VIEW = "view"
 
 
-class PermissionDecision(InfrahubStringEnum):
-    ALLOW = "allow"
-    DENY = "deny"
+class PermissionDecision(InfrahubNumberEnum):
+    DENY = 1
+    ALLOW_DEFAULT = 2
+    ALLOW_OTHER = 4
+    ALLOW_ALL = 6
 
 
 class AccountRole(InfrahubStringEnum):

backend/infrahub/core/enums.py

@@ -9,7 +9,7 @@ def generate_python_enum(name: str, options: list[Any]) -> type[enum.Enum]:
     main_attrs = {}
     for option in options:
         if isinstance(option, int):
-            enum_name = str(option)
+            enum_name = f"Value_{option!s}"
         else:
             enum_name = "_".join(re.findall(ENUM_NAME_REGEX, option)).upper()
 

backend/infrahub/core/initialization.py

@@ -300,7 +300,7 @@ async def create_initial_permission(db: InfrahubDatabase) -> Node:
         db=db,
         name=format_label(GlobalPermissions.SUPER_ADMIN.value),
         action=GlobalPermissions.SUPER_ADMIN.value,
-        decision=PermissionDecision.ALLOW.value,
+        decision=PermissionDecision.ALLOW_ALL.value,
     )
     await permission.save(db=db)
     log.info(f"Created global permission: {GlobalPermissions.SUPER_ADMIN}")
@@ -329,7 +329,7 @@ async def create_super_administrator_role(db: InfrahubDatabase) -> Node:
         db=db,
         name=format_label(GlobalPermissions.SUPER_ADMIN.value),
         action=GlobalPermissions.SUPER_ADMIN.value,
-        decision=PermissionDecision.ALLOW.value,
+        decision=PermissionDecision.ALLOW_ALL.value,
     )
     await permission.save(db=db)
     log.info(f"Created global permission: {GlobalPermissions.SUPER_ADMIN}")

backend/infrahub/core/node/permissions.py

@@ -2,6 +2,8 @@
 
 from typing import TYPE_CHECKING, Optional
 
+from infrahub.permissions.constants import PermissionDecisionFlag
+
 from . import Node
 
 if TYPE_CHECKING:
@@ -27,7 +29,8 @@ async def to_graphql(
 
         if fields:
             if "identifier" in fields:
-                response["identifier"] = {"value": f"global:{self.action.value}:{self.decision.value.value}"}  # type: ignore[attr-defined]
+                decision = PermissionDecisionFlag(value=self.decision.value.value)  # type: ignore[attr-defined]
+                response["identifier"] = {"value": f"global:{self.action.value}:{decision.name.lower()}"}  # type: ignore[attr-defined,union-attr]
 
         return response
 
@@ -51,11 +54,9 @@ async def to_graphql(
 
         if fields:
             if "identifier" in fields:
+                decision = PermissionDecisionFlag(value=self.decision.value.value)  # type: ignore[attr-defined]
                 response["identifier"] = {
-                    "value": (
-                        f"object:{self.branch.value}:{self.namespace.value}:{self.name.value}:{self.action.value.value}:"  # type: ignore[attr-defined]
-                        f"{self.decision.value.value}"  # type: ignore[attr-defined]
-                    )
+                    "value": f"object:{self.namespace.value}:{self.name.value}:{self.action.value.value}:{decision.name.lower()}"  # type: ignore[attr-defined,union-attr]
                 }
 
         return response

backend/infrahub/core/protocols.py

@@ -391,7 +391,6 @@ class CoreNumberPool(CoreResourcePool, LineageSource):
 
 
 class CoreObjectPermission(CoreBasePermission):
-    branch: String
     namespace: String
     name: String
     action: Enum

backend/infrahub/core/schema/definitions/core.py

@@ -933,9 +933,9 @@
             "attributes": [
                 {
                     "name": "decision",
-                    "kind": "Text",
+                    "kind": "Number",
                     "enum": PermissionDecision.available_types(),
-                    "default_value": PermissionDecision.ALLOW.value,
+                    "default_value": PermissionDecision.ALLOW_ALL.value,
                     "order_weight": 5000,
                 },
                 {
@@ -2186,15 +2186,12 @@
             "description": "A permission that grants rights to perform actions on objects",
             "label": "Object permission",
             "include_in_menu": False,
-            "order_by": ["branch__value", "namespace__value", "name__value", "action__value", "decision__value"],
-            "display_labels": ["branch__value", "namespace__value", "name__value", "action__value", "decision__value"],
-            "uniqueness_constraints": [
-                ["branch__value", "namespace__value", "name__value", "action__value", "decision__value"]
-            ],
+            "order_by": ["namespace__value", "name__value", "action__value", "decision__value"],
+            "display_labels": ["namespace__value", "name__value", "action__value", "decision__value"],
+            "uniqueness_constraints": [["namespace__value", "name__value", "action__value", "decision__value"]],
             "generate_profile": False,
             "inherit_from": [InfrahubKind.BASEPERMISSION],
             "attributes": [
-                {"name": "branch", "kind": "Text", "order_weight": 1000},
                 {"name": "namespace", "kind": "Text", "order_weight": 2000},
                 {"name": "name", "kind": "Text", "order_weight": 3000},
                 {

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

@@ -1,5 +1,6 @@
 from infrahub.auth import AccountSession
 from infrahub.core import registry
+from infrahub.core.account import GlobalPermission
 from infrahub.core.branch import Branch
 from infrahub.core.constants import GLOBAL_BRANCH_NAME, GlobalPermissions, PermissionDecision
 from infrahub.database import InfrahubDatabase
@@ -13,7 +14,9 @@
 class DefaultBranchPermissionChecker(GraphQLQueryPermissionCheckerInterface):
     """Checker that makes sure a user account can edit data in the default branch."""
 
-    permission_required = f"global:{GlobalPermissions.EDIT_DEFAULT_BRANCH.value}:{PermissionDecision.ALLOW.value}"
+    permission_required = GlobalPermission(
+        id="", name="", action=GlobalPermissions.EDIT_DEFAULT_BRANCH.value, decision=PermissionDecision.ALLOW_ALL.value
+    )
     exempt_operations = ["BranchCreate"]
 
     async def supports(self, db: InfrahubDatabase, account_session: AccountSession, branch: Branch) -> bool:

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

@@ -1,7 +1,8 @@
 from infrahub.auth import AccountSession
 from infrahub.core import registry
+from infrahub.core.account import GlobalPermission
 from infrahub.core.branch import Branch
-from infrahub.core.constants import GlobalPermissions
+from infrahub.core.constants import GlobalPermissions, PermissionDecision
 from infrahub.database import InfrahubDatabase
 from infrahub.exceptions import PermissionDeniedError
 from infrahub.graphql.analyzer import InfrahubGraphQLQueryAnalyzer
@@ -13,7 +14,9 @@
 class MergeBranchPermissionChecker(GraphQLQueryPermissionCheckerInterface):
     """Checker that makes sure a user account can merge a branch without going through a proposed change."""
 
-    permission_required = f"global:{GlobalPermissions.MERGE_BRANCH.value}:allow"
+    permission_required = GlobalPermission(
+        id="", name="", action=GlobalPermissions.MERGE_BRANCH.value, decision=PermissionDecision.ALLOW_ALL.value
+    )
 
     async def supports(self, db: InfrahubDatabase, account_session: AccountSession, branch: Branch) -> bool:
         return account_session.authenticated