IFC-672 Add global permission to protect repositories against changes (#4562)

Author: gmazoyer

backend/infrahub/core/constants/__init__.py

@@ -57,6 +57,7 @@ class GlobalPermissions(InfrahubStringEnum):
     MERGE_PROPOSED_CHANGE = "merge_proposed_change"
     MANAGE_ACCOUNTS = "manage_accounts"
     MANAGE_PERMISSIONS = "manage_permissions"
+    MANAGE_REPOSITORIES = "manage_repositories"
 
 
 class PermissionAction(InfrahubStringEnum):

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

@@ -150,9 +150,11 @@ async def check(
 
         for kind in kinds:
             schema = get_schema(db=db, branch=branch, node_schema=kind)
-            if is_permission_operation := kind in (InfrahubKind.GLOBALPERMISSION, InfrahubKind.OBJECTPERMISSION) or (
-                isinstance(schema, NodeSchema) and InfrahubKind.BASEPERMISSION in schema.inherit_from
-            ):
+            if is_permission_operation := kind in (
+                InfrahubKind.BASEPERMISSION,
+                InfrahubKind.GLOBALPERMISSION,
+                InfrahubKind.OBJECTPERMISSION,
+            ) or (isinstance(schema, NodeSchema) and InfrahubKind.BASEPERMISSION in schema.inherit_from):
                 break
 
         if not is_permission_operation:
@@ -165,3 +167,46 @@ async def check(
                 raise PermissionDeniedError("You do not have the permission to manage permissions")
 
         return CheckerResolution.NEXT_CHECKER
+
+
+class RepositoryManagerPermissionChecker(GraphQLQueryPermissionCheckerInterface):
+    """Checker that makes sure a user account can add/edit/delete repository objects.
+
+    This is similar to object permission checker except that we only care about mutations on repositories.
+    """
+
+    permission_required = f"global:{GlobalPermissions.MANAGE_REPOSITORIES.value}:{PermissionDecision.ALLOW.value}"
+
+    async def supports(self, db: InfrahubDatabase, account_session: AccountSession, branch: Branch) -> bool:
+        return account_session.authenticated
+
+    async def check(
+        self,
+        db: InfrahubDatabase,
+        account_session: AccountSession,
+        analyzed_query: InfrahubGraphQLQueryAnalyzer,
+        query_parameters: GraphqlParams,
+        branch: Branch,
+    ) -> CheckerResolution:
+        is_repository_operation = False
+        kinds = await analyzed_query.get_models_in_use(types=query_parameters.context.types)
+
+        for kind in kinds:
+            schema = get_schema(db=db, branch=branch, node_schema=kind)
+            if is_repository_operation := kind in (
+                InfrahubKind.GENERICREPOSITORY,
+                InfrahubKind.REPOSITORY,
+                InfrahubKind.READONLYREPOSITORY,
+            ) or (isinstance(schema, NodeSchema) and InfrahubKind.GENERICREPOSITORY in schema.inherit_from):
+                break
+
+        if not is_repository_operation or not analyzed_query.contains_mutation:
+            return CheckerResolution.NEXT_CHECKER
+
+        for permission_backend in registry.permission_backends:
+            if not await permission_backend.has_permission(
+                db=db, account_id=account_session.account_id, permission=self.permission_required, branch=branch
+            ):
+                raise PermissionDeniedError("You do not have the permission to manage repositories")
+
+        return CheckerResolution.NEXT_CHECKER

backend/tests/unit/graphql/auth/query_permission_checker/test_object_permission_checker.py

@@ -23,6 +23,7 @@
     AccountManagerPermissionChecker,
     ObjectPermissionChecker,
     PermissionManagerPermissionChecker,
+    RepositoryManagerPermissionChecker,
 )
 from infrahub.graphql.initialization import prepare_graphql_params
 from infrahub.permissions.local_backend import LocalPermissionBackend
@@ -182,6 +183,39 @@
 }
 """
 
+MUTATION_REPOSITORY = """
+mutation {
+  CoreRepositoryCreate(data: {
+    name: {value: "Test"}
+    location: {value: "/var/random"}
+  }) {
+    ok
+  }
+}
+"""
+
+MUTATION_READONLY_REPOSITORY = """
+mutation {
+  CoreReadOnlyRepositoryCreate(data: {
+    name: {value: "Test"}
+    location: {value: "/var/random"}
+  }) {
+    ok
+  }
+}
+"""
+
+MUTATION_GENERIC_REPOSITORY = """
+mutation {
+  CoreGenericRepositoryUpdate(data: {
+    name: {value: "Test"}
+    location: {value: "/var/random"}
+  }) {
+    ok
+  }
+}
+"""
+
 
 class TestObjectPermissions:
     async def test_setup(
@@ -560,3 +594,117 @@ async def test_account_without_permission(
                     query_parameters=gql_params,
                     branch=permissions_helper.default_branch,
                 )
+
+
+class TestRepositoryManagerPermissions:
+    async def test_setup(
+        self,
+        db: InfrahubDatabase,
+        register_core_models_schema: None,
+        default_branch: Branch,
+        permissions_helper: PermissionsHelper,
+        first_account: CoreAccount,
+        second_account: CoreAccount,
+    ):
+        registry.permission_backends = [LocalPermissionBackend()]
+        permissions_helper._default_branch = default_branch
+
+        permission = await Node.init(db=db, schema=InfrahubKind.GLOBALPERMISSION)
+        await permission.new(
+            db=db, name=GlobalPermissions.MANAGE_REPOSITORIES.value, action=GlobalPermissions.MANAGE_REPOSITORIES.value
+        )
+        await permission.save(db=db)
+
+        role = await Node.init(db=db, schema=InfrahubKind.ACCOUNTROLE)
+        await role.new(db=db, name="admin", permissions=[permission])
+        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)
+
+        permissions_helper._first = first_account
+        permissions_helper._second = second_account
+
+    @pytest.mark.parametrize(
+        "user",
+        [
+            AccountSession(account_id="abc", auth_type=AuthType.JWT, role=AccountRole.ADMIN),
+            AccountSession(authenticated=False, account_id="anonymous", auth_type=AuthType.NONE),
+        ],
+    )
+    async def test_supports_manage_repositories_permission_accounts(
+        self, user: AccountSession, db: InfrahubDatabase, permissions_helper: PermissionsHelper
+    ):
+        checker = AccountManagerPermissionChecker()
+        is_supported = await checker.supports(db=db, account_session=user, branch=permissions_helper.default_branch)
+        assert is_supported == user.authenticated
+
+    @pytest.mark.parametrize(
+        "operation", [MUTATION_REPOSITORY, MUTATION_READONLY_REPOSITORY, MUTATION_GENERIC_REPOSITORY]
+    )
+    async def test_account_with_permission(
+        self, db: InfrahubDatabase, permissions_helper: PermissionsHelper, operation: str
+    ):
+        checker = RepositoryManagerPermissionChecker()
+        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)
+        analyzed_query = InfrahubGraphQLQueryAnalyzer(
+            query=operation, schema=gql_params.schema, branch=permissions_helper.default_branch
+        )
+
+        resolution = await checker.check(
+            db=db,
+            account_session=session,
+            analyzed_query=analyzed_query,
+            query_parameters=gql_params,
+            branch=permissions_helper.default_branch,
+        )
+        assert resolution == CheckerResolution.NEXT_CHECKER
+
+    @pytest.mark.parametrize(
+        "operation,must_raise",
+        [
+            (MUTATION_REPOSITORY, True),
+            (MUTATION_READONLY_REPOSITORY, True),
+            (MUTATION_GENERIC_REPOSITORY, True),
+            (QUERY_TAGS, False),
+        ],
+    )
+    async def test_account_without_permission(
+        self, db: InfrahubDatabase, permissions_helper: PermissionsHelper, operation: str, must_raise: bool
+    ):
+        checker = RepositoryManagerPermissionChecker()
+        session = AccountSession(
+            authenticated=True, account_id=permissions_helper.second.id, session_id=str(uuid4()), auth_type=AuthType.JWT
+        )
+
+        gql_params = prepare_graphql_params(db=db, include_mutation=True, branch=permissions_helper.default_branch)
+        analyzed_query = InfrahubGraphQLQueryAnalyzer(
+            query=operation, schema=gql_params.schema, branch=permissions_helper.default_branch
+        )
+
+        if not must_raise:
+            resolution = await checker.check(
+                db=db,
+                account_session=session,
+                analyzed_query=analyzed_query,
+                query_parameters=gql_params,
+                branch=permissions_helper.default_branch,
+            )
+            assert resolution == CheckerResolution.NEXT_CHECKER
+        else:
+            with pytest.raises(PermissionDeniedError, match=r"You do not have the permission to manage repositories"):
+                await checker.check(
+                    db=db,
+                    account_session=session,
+                    analyzed_query=analyzed_query,
+                    query_parameters=gql_params,
+                    branch=permissions_helper.default_branch,
+                )