DataOps-platform/app/core/system/permissions.py

from __future__ import annotations

from functools import wraps
from typing import Iterable

from flask import current_app, g, jsonify, request

from app.models.result import failed


READ_GOVERNANCE = "governance:read"
EDIT_GOVERNANCE = "governance:edit"
APPROVE_REVIEW = "review:approve"
MANAGE_USERS = "users:manage"
ACTIVATE_WORKFLOW = "workflow:activate"
OPERATE_ORDERS = "orders:operate"
DATASOURCE_POOL_MANAGE = "datasources:pools:manage"
INGESTION_RUN = "ingestion:run"
INGESTION_ADMIN = "ingestion:admin"
EVIDENCE_DOWNLOAD = "evidence:download"
DATA_ELEMENTS_EDIT = "data-elements:edit"
DATA_ELEMENTS_PUBLISH = "data-elements:publish"
ONTOLOGIES_EDIT = "ontologies:edit"
ONTOLOGIES_PUBLISH = "ontologies:publish"

ROLE_PERMISSIONS = {
    "viewer": frozenset({READ_GOVERNANCE}),
    "editor": frozenset(
        {
            READ_GOVERNANCE,
            EDIT_GOVERNANCE,
            APPROVE_REVIEW,
            OPERATE_ORDERS,
            INGESTION_RUN,
            DATA_ELEMENTS_EDIT,
            ONTOLOGIES_EDIT,
        }
    ),
    "admin": frozenset(
        {
            READ_GOVERNANCE,
            EDIT_GOVERNANCE,
            APPROVE_REVIEW,
            MANAGE_USERS,
            ACTIVATE_WORKFLOW,
            OPERATE_ORDERS,
            DATASOURCE_POOL_MANAGE,
            INGESTION_RUN,
            INGESTION_ADMIN,
            EVIDENCE_DOWNLOAD,
            DATA_ELEMENTS_EDIT,
            DATA_ELEMENTS_PUBLISH,
            ONTOLOGIES_EDIT,
            ONTOLOGIES_PUBLISH,
        }
    ),
}

PUBLIC = "public"


def permission_for_request(path: str, method: str) -> tuple[str, ...]:
    """Classify every API request in one auditable, deny-by-default policy."""
    method = method.upper()
    if path in {"/api/system/health", "/api/system/auth/login"}:
        return (PUBLIC,)
    if path.startswith("/api/system/users"):
        return (MANAGE_USERS,)
    if path.startswith("/api/development/v1/ingestion-jobs"):
        if method == "GET":
            return (READ_GOVERNANCE,)
        if path.endswith("/retry"):
            return (INGESTION_ADMIN,)
        if path.endswith("/cancel"):
            return (INGESTION_RUN,)
        if method == "POST":
            return (INGESTION_RUN,)
    if path.startswith("/api/development/v1/data-elements"):
        if method == "GET":
            return (READ_GOVERNANCE,)
        return (DATA_ELEMENTS_EDIT,)
    if path.startswith("/api/development/v1/candidate-decisions"):
        return (DATA_ELEMENTS_EDIT,)
    if path.startswith("/api/system/workbench"):
        return (READ_GOVERNANCE,)
    if (
        path == "/api/datasource/pools"
        or (
            path.startswith("/api/datasource/")
            and (
                path.endswith("/pool")
                or path.endswith("/pool/invalidate")
            )
        )
    ):
        return (DATASOURCE_POOL_MANAGE,)
    if path == "/api/datasource/list":
        return (READ_GOVERNANCE,)
    if path.startswith("/api/system/") and not path.startswith("/api/system/auth/me"):
        return (MANAGE_USERS,)
    if path.startswith("/api/dataflow/") and any(
        marker in path.lower() for marker in ("activate", "publish", "execute")
    ):
        return (ACTIVATE_WORKFLOW,)
    if "order" in path.lower() and method != "GET":
        return (OPERATE_ORDERS,)
    if method == "GET":
        return (READ_GOVERNANCE,)
    if method in {"POST", "PUT", "PATCH", "DELETE"}:
        return (EDIT_GOVERNANCE,)
    return (MANAGE_USERS,)


def permissions_for_roles(roles: Iterable[str]) -> frozenset[str]:
    result: set[str] = set()
    for role in roles:
        result.update(ROLE_PERMISSIONS.get(role, ()))
    return frozenset(result)


def authenticate_request() -> dict | None:
    from app.core.system.auth import load_identity_from_token

    existing = getattr(g, "current_user", None)
    if existing:
        return existing
    header = request.headers.get("Authorization", "")
    if not header.startswith("Bearer "):
        return None
    token = header[7:].strip()
    if not token:
        return None
    return load_identity_from_token(token, secret=current_app.config["SECRET_KEY"])


def configure_api_authorization(app) -> None:
    @app.before_request
    def enforce_api_policy():
        if not request.path.startswith("/api/") or request.method == "OPTIONS":
            return None
        required = permission_for_request(request.path, request.method)
        if required == (PUBLIC,):
            return None
        identity = authenticate_request()
        if identity is None:
            return jsonify(failed("未登录或登录已过期", code=401)), 401
        permissions = permissions_for_roles(identity["roles"])
        if not set(required).issubset(permissions):
            return jsonify(failed("权限不足", code=403)), 403
        identity["permissions"] = sorted(permissions)
        g.current_user = identity
        return None


def require_permissions(*required: str):
    def decorator(view):
        @wraps(view)
        def wrapped(*args, **kwargs):
            identity = authenticate_request()
            if identity is None:
                return jsonify(failed("未登录或登录已过期", code=401)), 401
            permissions = permissions_for_roles(identity["roles"])
            if not set(required).issubset(permissions):
                return jsonify(failed("权限不足", code=403)), 403
            identity["permissions"] = sorted(permissions)
            g.current_user = identity
            return view(*args, **kwargs)

        wrapped.required_permissions = tuple(required)
        return wrapped

    return decorator