Proof of concept of permission enforcement

backend/infrahub/graphql/analyzer.py

@@ -12,8 +12,19 @@
 
 
 class InfrahubGraphQLQueryAnalyzer(GraphQLQueryAnalyzer):
-    def __init__(self, query: str, schema: Optional[GraphQLSchema] = None, branch: Optional[Branch] = None):
+    def __init__(
+        self,
+        query: str,
+        variables: Optional[dict[str, Any]] = None,
+        schema: Optional[GraphQLSchema] = None,
+        operation_name: Optional[str] = None,
+        branch: Optional[Branch] = None,
+    ):
         self.branch: Optional[Branch] = branch
+        self.operation_name: Optional[str] = operation_name
+        # FIXME: base class as variables property
+        self.vars: dict[str, Any] = variables or {}
+
         super().__init__(query=query, schema=schema)
 
     async def get_models_in_use(self, types: dict[str, Any]) -> set[str]:

backend/infrahub/graphql/api/dependencies.py

@@ -5,7 +5,10 @@
 from ..app import InfrahubGraphQLApp
 from ..auth.query_permission_checker.anonymous_checker import AnonymousGraphQLPermissionChecker
 from ..auth.query_permission_checker.checker import GraphQLQueryPermissionChecker
-from ..auth.query_permission_checker.default_checker import DefaultGraphQLPermissionChecker
+from ..auth.query_permission_checker.default_checker import (
+    DefaultBranchPermissionChecker,
+    DefaultGraphQLPermissionChecker,
+)
 from ..auth.query_permission_checker.read_only_checker import ReadOnlyGraphQLPermissionChecker
 from ..auth.query_permission_checker.read_write_checker import ReadWriteGraphQLPermissionChecker
 
@@ -17,6 +20,7 @@ def get_anonymous_access_setting() -> bool:
 def build_graphql_query_permission_checker() -> GraphQLQueryPermissionChecker:
     return GraphQLQueryPermissionChecker(
         [
+            DefaultBranchPermissionChecker(),
             ReadWriteGraphQLPermissionChecker(),
             ReadOnlyGraphQLPermissionChecker(),
             AnonymousGraphQLPermissionChecker(get_anonymous_access_setting),

backend/infrahub/graphql/app.py

@@ -197,16 +197,21 @@ async def _handle_http_request(
 
         operation = operations
         query = operation["query"]
+        variable_values = operation.get("variables")
+        operation_name = operation.get("operationName")
 
         at = request.query_params.get("at", None)
         graphql_params = prepare_graphql_params(
             db=db, branch=branch, at=at, account_session=account_session, request=request
         )
-        analyzed_query = InfrahubGraphQLQueryAnalyzer(query=query, schema=graphql_params.schema, branch=branch)
-        await self.permission_checker.check(account_session=account_session, analyzed_query=analyzed_query)
-
-        variable_values = operation.get("variables")
-        operation_name = operation.get("operationName")
+        analyzed_query = InfrahubGraphQLQueryAnalyzer(
+            query=query,
+            schema=graphql_params.schema,
+            branch=branch,
+            operation_name=operation_name,
+            variables=variable_values,
+        )
+        await self._evaluate_permissions(request=request, query=analyzed_query, account_session=account_session)
 
         # if the query contains some mutation, it's not currently supported to set AT manually
         if analyzed_query.contains_mutation:
@@ -221,13 +226,7 @@ async def _handle_http_request(
                 "Processing IntrospectionQuery .. ", branch=branch.name, nbr_object_in_schema=nbr_object_in_schema
             )
 
-        labels = {
-            "type": "mutation" if analyzed_query.contains_mutation else "query",
-            "branch": branch.name,
-            "operation": operation_name if operation_name is not None else "",
-            "name": analyzed_query.operations[0].name,
-            "query_id": "",
-        }
+        labels = self._set_labels(request=request, branch=branch, query=analyzed_query)
 
         with trace.get_tracer(__name__).start_as_current_span("execute_graphql") as span:
             span.set_attributes(labels)
@@ -272,6 +271,20 @@ async def _handle_http_request(
 
         return json_response
 
+    def _set_labels(self, request: Request, branch: Branch, query: InfrahubGraphQLQueryAnalyzer) -> dict[str, Any]:
+        return {
+            "type": "mutation" if query.contains_mutation else "query",
+            "branch": branch.name,
+            "operation": query.operation_name if query.operation_name is not None else "",
+            "name": query.operations[0].name,
+            "query_id": "",
+        }
+
+    async def _evaluate_permissions(
+        self, request: Request, query: InfrahubGraphQLQueryAnalyzer, account_session: AccountSession
+    ) -> None:
+        await self.permission_checker.check(account_session=account_session, analyzed_query=query)
+
     def _log_error(self, error: Exception) -> None:
         if isinstance(error, Error):
             if 500 <= error.HTTP_CODE <= 500:

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

@@ -1,5 +1,8 @@
+from graphql import OperationType
+
+from infrahub import config
 from infrahub.auth import AccountSession
-from infrahub.exceptions import AuthorizationError
+from infrahub.exceptions import AuthorizationError, PermissionDeniedError
 from infrahub.graphql.analyzer import InfrahubGraphQLQueryAnalyzer
 
 from .interface import GraphQLQueryPermissionCheckerInterface
@@ -11,3 +14,24 @@ async def supports(self, account_session: AccountSession) -> bool:
 
     async def check(self, analyzed_query: InfrahubGraphQLQueryAnalyzer) -> None:
         raise AuthorizationError("Authentication is required to perform this operation")
+
+
+class DefaultBranchPermissionChecker(GraphQLQueryPermissionCheckerInterface):
+    def __init__(self) -> None:
+        self.can_edit_default_branch: bool = False
+
+    async def supports(self, account_session: AccountSession) -> bool:
+        if account_session.permissions:
+            self.can_edit_default_branch = "edit_default_branch" in account_session.permissions
+        return account_session.authenticated
+
+    async def check(self, analyzed_query: InfrahubGraphQLQueryAnalyzer) -> None:
+        for operation in analyzed_query.operations:
+            if (
+                not self.can_edit_default_branch
+                and analyzed_query.branch.name == config.SETTINGS.initial.default_branch
+                and operation.operation_type == OperationType.MUTATION
+            ):
+                raise PermissionDeniedError(
+                    f"You are not allowed to change data in the default branch '{config.SETTINGS.initial.default_branch}'"
+                )