feat(server): add project RBAC and guarded workflows

app/auth/permissions.py

@@ -0,0 +1,162 @@
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from fastapi import Depends, HTTPException, Request, status
+
+from app.auth.project_dependencies import ProjectContext, get_project_context
+
+WEBGIS_VIEW = "webgis.view"
+WEBGIS_EDIT = "webgis.edit"
+SCADA_VIEW = "scada.view"
+SCADA_CLEAN = "scada.clean"
+SIMULATION_VIEW = "simulation.view"
+SIMULATION_RUN = "simulation.run"
+BURST_VIEW = "burst.view"
+BURST_RUN = "burst.run"
+RISK_VIEW = "risk.view"
+RISK_RUN = "risk.run"
+OPTIMIZATION_VIEW = "optimization.view"
+OPTIMIZATION_RUN = "optimization.run"
+MODEL_IMPORT = "model.import"
+AUDIT_VIEW = "audit.view"
+ENVIRONMENT_MANAGE = "environment.manage"
+MEMBERSHIP_MANAGE = "membership.manage"
+
+PROJECT_MEMBER_PERMISSIONS = frozenset(
+    {
+        WEBGIS_VIEW,
+        WEBGIS_EDIT,
+        SCADA_VIEW,
+        SCADA_CLEAN,
+        SIMULATION_VIEW,
+        SIMULATION_RUN,
+        BURST_VIEW,
+        BURST_RUN,
+        RISK_VIEW,
+        RISK_RUN,
+        OPTIMIZATION_VIEW,
+        OPTIMIZATION_RUN,
+    }
+)
+
+PROJECT_VIEWER_PERMISSIONS = frozenset(
+    {
+        WEBGIS_VIEW,
+        SCADA_VIEW,
+        SIMULATION_VIEW,
+    }
+)
+
+SYSTEM_ADMIN_PERMISSIONS = frozenset(
+    {
+        MODEL_IMPORT,
+        AUDIT_VIEW,
+        ENVIRONMENT_MANAGE,
+        MEMBERSHIP_MANAGE,
+    }
+)
+
+PROJECT_ROLE_PERMISSIONS: dict[str, frozenset[str]] = {
+    "member": PROJECT_MEMBER_PERMISSIONS,
+    "viewer": PROJECT_VIEWER_PERMISSIONS,
+}
+
+
+def resolve_permissions(
+    *,
+    project_role: str | None,
+    system_role: str,
+    is_superuser: bool,
+) -> frozenset[str]:
+    permissions = set(PROJECT_ROLE_PERMISSIONS.get(project_role or "", frozenset()))
+    if is_superuser or system_role == "admin":
+        permissions.update(SYSTEM_ADMIN_PERMISSIONS)
+    return frozenset(permissions)
+
+
+def permissions_for_context(ctx: ProjectContext) -> frozenset[str]:
+    return resolve_permissions(
+        project_role=ctx.project_role,
+        system_role=ctx.system_role,
+        is_superuser=ctx.is_superuser,
+    )
+
+
+def _permission_denied(permission: str) -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail={
+            "code": "permission_denied",
+            "permission": permission,
+        },
+    )
+
+
+async def _enforce_project_scope(request: Request, ctx: ProjectContext) -> None:
+    requested_network = (
+        request.path_params.get("network")
+        or request.query_params.get("network")
+    )
+    if not requested_network:
+        content_type = request.headers.get("content-type", "")
+        if content_type.startswith("application/json"):
+            try:
+                payload = await request.json()
+            except (ValueError, RuntimeError):
+                payload = None
+            if isinstance(payload, dict):
+                requested_network = payload.get("network")
+
+    if requested_network and str(requested_network) != ctx.project_code:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail={
+                "code": "project_scope_denied",
+                "project_id": str(ctx.project_id),
+            },
+        )
+
+
+def require_permission(
+    permission: str,
+) -> Callable[..., Awaitable[ProjectContext]]:
+    async def dependency(
+        request: Request,
+        ctx: ProjectContext = Depends(get_project_context),
+    ) -> ProjectContext:
+        if permission not in permissions_for_context(ctx):
+            raise _permission_denied(permission)
+        await _enforce_project_scope(request, ctx)
+        return ctx
+
+    return dependency
+
+
+def require_method_permission(
+    *,
+    read_permission: str,
+    write_permission: str,
+) -> Callable[..., Awaitable[ProjectContext]]:
+    async def dependency(
+        request: Request,
+        ctx: ProjectContext = Depends(get_project_context),
+    ) -> ProjectContext:
+        permission = (
+            read_permission
+            if request.method.upper() in {"GET", "HEAD", "OPTIONS"}
+            else write_permission
+        )
+        if permission not in permissions_for_context(ctx):
+            raise _permission_denied(permission)
+        await _enforce_project_scope(request, ctx)
+        return ctx
+
+    return dependency
+
+
+def has_permission(user: Any, project_role: str | None, permission: str) -> bool:
+    return permission in resolve_permissions(
+        project_role=project_role,
+        system_role=str(getattr(user, "role", "user")),
+        is_superuser=bool(getattr(user, "is_superuser", False)),
+    )