IFC-670 Add global permission to protect schema against changes (#4560)

Author: gmazoyer

backend/infrahub/api/schema.py

@@ -17,6 +17,7 @@
 from infrahub.api.exceptions import SchemaNotValidError
 from infrahub.core import registry
 from infrahub.core.branch import Branch  # noqa: TCH001
+from infrahub.core.constants import GlobalPermissions, PermissionDecision
 from infrahub.core.migrations.schema.models import SchemaApplyMigrationData
 from infrahub.core.models import (  # noqa: TCH001
     SchemaBranchHash,
@@ -27,7 +28,7 @@
 from infrahub.core.schema.constants import SchemaNamespace  # noqa: TCH001
 from infrahub.core.validators.models.validate_migration import SchemaValidateMigrationData
 from infrahub.database import InfrahubDatabase  # noqa: TCH001
-from infrahub.exceptions import MigrationError
+from infrahub.exceptions import MigrationError, PermissionDeniedError
 from infrahub.log import get_logger
 from infrahub.message_bus import Meta, messages
 from infrahub.services import services
@@ -38,6 +39,7 @@
 if TYPE_CHECKING:
     from typing_extensions import Self
 
+    from infrahub.auth import AccountSession
     from infrahub.core.schema.schema_branch import SchemaBranch
     from infrahub.services import InfrahubServices
 
@@ -240,8 +242,17 @@ async def load_schema(
     background_tasks: BackgroundTasks,
     db: InfrahubDatabase = Depends(get_db),
     branch: Branch = Depends(get_branch_dep),
-    _: Any = Depends(get_current_user),
+    account_session: AccountSession = Depends(get_current_user),
 ) -> SchemaUpdate:
+    for permission_backend in registry.permission_backends:
+        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}",
+            branch=branch,
+        ):
+            raise PermissionDeniedError("You are not allowed to manage the schema")
+
     service: InfrahubServices = request.app.state.service
     log.info("schema_load_request", branch=branch.name)
 

backend/infrahub/core/constants/__init__.py

@@ -55,6 +55,7 @@ class GlobalPermissions(InfrahubStringEnum):
     SUPER_ADMIN = "super_admin"
     MERGE_BRANCH = "merge_branch"
     MERGE_PROPOSED_CHANGE = "merge_proposed_change"
+    MANAGE_SCHEMA = "manage_schema"
     MANAGE_ACCOUNTS = "manage_accounts"
     MANAGE_PERMISSIONS = "manage_permissions"
     MANAGE_REPOSITORIES = "manage_repositories"

backend/tests/unit/api/test_40_schema_api.py

@@ -4,6 +4,7 @@
 from infrahub.core.branch import Branch
 from infrahub.core.constants import InfrahubKind, SchemaPathType
 from infrahub.core.initialization import create_branch
+from infrahub.core.node import Node
 from infrahub.core.path import SchemaPath
 from infrahub.core.schema import SchemaRoot, core_models
 from infrahub.core.utils import count_relationships
@@ -221,6 +222,36 @@ async def test_schema_load_endpoint_valid_simple(
     assert relationships["tags"] == 7000
 
 
+async def test_schema_load_permission_failure(
+    db: InfrahubDatabase,
+    client: TestClient,
+    first_account,
+    default_branch: Branch,
+    prefect_test_fixture,
+    workflow_local,
+    authentication_base,
+    helper,
+):
+    token = await Node.init(db=db, schema=InfrahubKind.ACCOUNTTOKEN)
+    await token.new(db=db, token="unprivileged", account=first_account)
+    await token.save(db=db)
+
+    # Load the schema in the database
+    schema = registry.schema.get_schema_branch(name=default_branch.name)
+    await registry.schema.load_schema_to_db(schema=schema, branch=default_branch, db=db)
+
+    # Must execute in a with block to execute the startup/shutdown event
+    with client:
+        response = client.post(
+            "/api/schema/load",
+            headers={"X-INFRAHUB-KEY": "unprivileged"},
+            json={"schemas": [helper.schema_file("infra_simple_01.json")]},
+        )
+
+    assert response.status_code == 403
+    assert response.json()["errors"][0]["message"] == "You are not allowed to manage the schema"
+
+
 async def test_schema_load_restricted_namespace(
     db: InfrahubDatabase,
     client: TestClient,