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

from __future__ import annotations

from collections.abc import Iterable
from functools import wraps

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"
RESPONSIBILITIES_READ = "governance:responsibilities:read"
RESPONSIBILITIES_MANAGE = "governance:responsibilities:manage"
KNOWLEDGE_MANAGE = "knowledge:manage"
RULES_READ = "rules:read"
RULES_EDIT = "rules:edit"
RULES_PUBLISH = "rules:publish"
DATAFLOW_RELEASE = "dataflows:release"
DATAFLOW_DEPLOY = "dataflows:deploy"
DATAFLOW_CANARY = "dataflows:canary"
DATAFLOW_ACTIVATE = "dataflows:activate"
DATAFLOW_ROLLBACK = "dataflows:rollback"
RULES_EXECUTE = "rules:execute"
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"
DEVICE_ASSETS_EDIT = "device-assets:edit"
DEVICE_SEMANTICS_EDIT = "device-semantics:edit"
DEVICE_SEMANTICS_REVIEW = "device-semantics:review"
DEVICE_ENTITIES_EDIT = "device-entities:edit"
DEVICE_ENTITIES_REVIEW = "device-entities:review"
DEVICE_QUALITY_EDIT = "device-quality:edit"
DEVICE_QUALITY_EXECUTE = "device-quality:execute"
DEVICE_QUALITY_PUBLISH = "device-quality:publish"
QUALITY_ISSUES_EDIT = "quality-issues:edit"
QUALITY_ISSUES_REVIEW = "quality-issues:review"

ROLE_PERMISSIONS = {
    "viewer": frozenset(
        {READ_GOVERNANCE, RULES_READ, RESPONSIBILITIES_READ}
    ),
    "editor": frozenset(
        {
            READ_GOVERNANCE,
            EDIT_GOVERNANCE,
            APPROVE_REVIEW,
            OPERATE_ORDERS,
            RULES_READ,
            RULES_EDIT,
            RULES_EXECUTE,
            RESPONSIBILITIES_READ,
            INGESTION_RUN,
            DATA_ELEMENTS_EDIT,
            ONTOLOGIES_EDIT,
            DEVICE_ASSETS_EDIT,
            DEVICE_SEMANTICS_EDIT,
            DEVICE_ENTITIES_EDIT,
            DEVICE_QUALITY_EDIT,
            DEVICE_QUALITY_EXECUTE,
            QUALITY_ISSUES_EDIT,
        }
    ),
    "admin": frozenset(
        {
            READ_GOVERNANCE,
            EDIT_GOVERNANCE,
            APPROVE_REVIEW,
            MANAGE_USERS,
            ACTIVATE_WORKFLOW,
            OPERATE_ORDERS,
            DATASOURCE_POOL_MANAGE,
            RESPONSIBILITIES_READ,
            RESPONSIBILITIES_MANAGE,
            KNOWLEDGE_MANAGE,
            RULES_READ,
            RULES_EDIT,
            RULES_PUBLISH,
            DATAFLOW_RELEASE,
            DATAFLOW_DEPLOY,
            DATAFLOW_CANARY,
            DATAFLOW_ACTIVATE,
            DATAFLOW_ROLLBACK,
            RULES_EXECUTE,
            INGESTION_RUN,
            INGESTION_ADMIN,
            EVIDENCE_DOWNLOAD,
            DATA_ELEMENTS_EDIT,
            DATA_ELEMENTS_PUBLISH,
            ONTOLOGIES_EDIT,
            ONTOLOGIES_PUBLISH,
            DEVICE_ASSETS_EDIT,
            DEVICE_SEMANTICS_EDIT,
            DEVICE_SEMANTICS_REVIEW,
            DEVICE_ENTITIES_EDIT,
            DEVICE_ENTITIES_REVIEW,
            DEVICE_QUALITY_EDIT,
            DEVICE_QUALITY_EXECUTE,
            DEVICE_QUALITY_PUBLISH,
            QUALITY_ISSUES_EDIT,
            QUALITY_ISSUES_REVIEW,
        }
    ),
}

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/responsibilities/"):
        if method == "GET":
            return (RESPONSIBILITIES_READ,)
        return (RESPONSIBILITIES_MANAGE,)
    if path in {"/api/knowledge/search", "/api/knowledge/ask"}:
        return (READ_GOVERNANCE,)
    if path.startswith("/api/rules"):
        if path.startswith("/api/rules/deployments"):
            if method == "GET":
                return (RULES_READ,)
            if path.endswith("/deploy-disabled"):
                return (DATAFLOW_DEPLOY,)
            if path.endswith("/canary"):
                return (DATAFLOW_CANARY, RULES_EXECUTE)
            if path.endswith("/execute"):
                return (RULES_EXECUTE,)
            if path.endswith("/activate"):
                return (DATAFLOW_ACTIVATE,)
            if path.endswith("/rollback"):
                return (DATAFLOW_ROLLBACK,)
            if path.endswith("/reconcile-deploy"):
                return (DATAFLOW_DEPLOY,)
            if path.endswith("/reconcile-activate"):
                return (DATAFLOW_ACTIVATE,)
            if path.endswith("/reconcile-rollback"):
                return (DATAFLOW_ROLLBACK,)
            if path.endswith("/reconcile-execute"):
                return (RULES_EXECUTE,)
            if method == "POST":
                return (DATAFLOW_DEPLOY,)
            return (DATAFLOW_DEPLOY,)
        if method == "GET":
            return (RULES_READ,)
        if path.endswith("/publish"):
            return (RULES_PUBLISH,)
        if path.startswith("/api/rules/production-lines/") and path.endswith(
            "/release"
        ):
            return (DATAFLOW_RELEASE,)
        if path.endswith("/test") or path.endswith("/execute"):
            return (RULES_EXECUTE,)
        if method in {"POST", "PUT", "PATCH", "DELETE"}:
            return (RULES_EDIT,)
        return (RULES_PUBLISH,)
    if path.startswith("/api/knowledge/admin"):
        return (KNOWLEDGE_MANAGE,)
    if path.startswith("/api/system/users"):
        return (MANAGE_USERS,)
    if path.startswith("/api/development/v1/device-assets"):
        if method == "GET":
            return (READ_GOVERNANCE,)
        return (DEVICE_ASSETS_EDIT,)
    if path.startswith("/api/development/v1/device-semantics"):
        if method == "GET":
            return (READ_GOVERNANCE,)
        if path.endswith("/review"):
            return (DEVICE_SEMANTICS_REVIEW,)
        return (DEVICE_SEMANTICS_EDIT,)
    if path.startswith("/api/development/v1/device-entities"):
        if method == "GET":
            return (READ_GOVERNANCE,)
        if path.endswith("/review") or path.endswith("/rollback"):
            return (DEVICE_ENTITIES_REVIEW,)
        return (DEVICE_ENTITIES_EDIT,)
    if path.startswith("/api/development/v1/device-quality"):
        if method == "GET":
            return (READ_GOVERNANCE,)
        if path.startswith(
            "/api/development/v1/device-quality/issues"
        ):
            if path.endswith("/review"):
                return (QUALITY_ISSUES_REVIEW,)
            return (QUALITY_ISSUES_EDIT,)
        if path.endswith("/publish"):
            return (DEVICE_QUALITY_PUBLISH,)
        if path == "/api/development/v1/device-quality/runs":
            return (DEVICE_QUALITY_EXECUTE,)
        return (DEVICE_QUALITY_EDIT,)
    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/development/v1/sources/files"):
        return (INGESTION_RUN,)
    if path.startswith("/api/development/v1/evidence"):
        return (READ_GOVERNANCE,)
    if path.startswith("/api/development/v1/ontologies"):
        if method == "GET":
            return (READ_GOVERNANCE,)
        if path.endswith("/publish") or path.endswith("/rollback"):
            return (ONTOLOGIES_PUBLISH,)
        return (ONTOLOGIES_EDIT,)
    if path.startswith("/api/bd/"):
        if path in {
            "/api/bd/list",
            "/api/bd/detail",
            "/api/bd/graphall",
            "/api/bd/search",
        }:
            return (READ_GOVERNANCE,)
        return (EDIT_GOVERNANCE,)
    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