AI/ontology_platform/ont_platform/auth/rbac.py

"""역할 기반 액세스 제어 (RBAC).

Phase 8: 권한 관리
"""

from enum import Enum
from typing import List, Set, Dict


class Role(str, Enum):
    """사용자 역할."""

    ADMIN = "admin"
    EDITOR = "editor"
    VIEWER = "viewer"
    API = "api"


class Permission(str, Enum):
    """권한."""

    # 읽기
    READ_ENTITY = "read:entity"
    READ_RELATION = "read:relation"
    READ_GRAPH = "read:graph"

    # 쓰기
    CREATE_ENTITY = "create:entity"
    UPDATE_ENTITY = "update:entity"
    DELETE_ENTITY = "delete:entity"

    CREATE_RELATION = "create:relation"
    DELETE_RELATION = "delete:relation"

    # 분석
    RUN_ANALYSIS = "run:analysis"
    VIEW_ANALYTICS = "view:analytics"

    # LLM
    RUN_LLM_QUERY = "run:llm"

    # 관리
    MANAGE_USERS = "manage:users"
    MANAGE_API_KEYS = "manage:api_keys"
    VIEW_AUDIT_LOG = "view:audit"
    VIEW_BILLING = "view:billing"
    MANAGE_ORGANIZATION = "manage:org"


class RBAC:
    """역할 기반 액세스 제어."""

    # 역할별 권한 매핑
    ROLE_PERMISSIONS: Dict[Role, Set[Permission]] = {
        Role.ADMIN: {
            # 모든 권한
            Permission.READ_ENTITY,
            Permission.READ_RELATION,
            Permission.READ_GRAPH,
            Permission.CREATE_ENTITY,
            Permission.UPDATE_ENTITY,
            Permission.DELETE_ENTITY,
            Permission.CREATE_RELATION,
            Permission.DELETE_RELATION,
            Permission.RUN_ANALYSIS,
            Permission.VIEW_ANALYTICS,
            Permission.RUN_LLM_QUERY,
            Permission.MANAGE_USERS,
            Permission.MANAGE_API_KEYS,
            Permission.VIEW_AUDIT_LOG,
            Permission.VIEW_BILLING,
            Permission.MANAGE_ORGANIZATION,
        },
        Role.EDITOR: {
            # 읽기, 쓰기, 분석
            Permission.READ_ENTITY,
            Permission.READ_RELATION,
            Permission.READ_GRAPH,
            Permission.CREATE_ENTITY,
            Permission.UPDATE_ENTITY,
            Permission.DELETE_ENTITY,
            Permission.CREATE_RELATION,
            Permission.DELETE_RELATION,
            Permission.RUN_ANALYSIS,
            Permission.VIEW_ANALYTICS,
            Permission.RUN_LLM_QUERY,
            Permission.VIEW_BILLING,
        },
        Role.VIEWER: {
            # 읽기, 분석, LLM만
            Permission.READ_ENTITY,
            Permission.READ_RELATION,
            Permission.READ_GRAPH,
            Permission.VIEW_ANALYTICS,
            Permission.RUN_LLM_QUERY,
            Permission.VIEW_BILLING,
        },
        Role.API: {
            # API 호출 시 필요한 최소 권한
            Permission.READ_ENTITY,
            Permission.READ_RELATION,
            Permission.READ_GRAPH,
            Permission.RUN_LLM_QUERY,
        },
    }

    def get_permissions(self, role: str) -> Set[Permission]:
        """역할의 권한 반환."""
        try:
            role_enum = Role(role)
            return self.ROLE_PERMISSIONS.get(role_enum, set())
        except ValueError:
            return set()

    def has_permission(self, role: str, permission: str) -> bool:
        """사용자가 특정 권한을 가지고 있는지 확인."""
        try:
            perm_enum = Permission(permission)
            permissions = self.get_permissions(role)
            return perm_enum in permissions
        except ValueError:
            return False

    def check_permission(self, role: str, permission: str) -> None:
        """권한 확인 (없으면 예외 발생)."""
        if not self.has_permission(role, permission):
            from fastapi import HTTPException

            raise HTTPException(
                status_code=403,
                detail=f"Permission denied: {permission}",
            )

    def get_all_permissions(self) -> Dict[str, List[str]]:
        """모든 역할의 권한을 딕셔너리로 반환."""
        return {
            role.value: sorted([perm.value for perm in permissions])
            for role, permissions in self.ROLE_PERMISSIONS.items()
        }

    def can_manage_users(self, role: str) -> bool:
        """사용자 관리 권한 확인."""
        return self.has_permission(role, Permission.MANAGE_USERS.value)

    def can_view_audit(self, role: str) -> bool:
        """감시 로그 조회 권한 확인."""
        return self.has_permission(role, Permission.VIEW_AUDIT_LOG.value)

    def can_manage_organization(self, role: str) -> bool:
        """조직 관리 권한 확인."""
        return self.has_permission(role, Permission.MANAGE_ORGANIZATION.value)


# 간편 헬퍼 함수


def check_admin_role(role: str) -> bool:
    """관리자 역할 확인."""
    return role == Role.ADMIN.value


def check_editor_or_admin(role: str) -> bool:
    """편집자 이상 역할 확인."""
    return role in (Role.ADMIN.value, Role.EDITOR.value)


def require_permission(permission: str):
    """FastAPI 의존성: 권한 확인."""
    from fastapi import HTTPException, Depends
    from ont_platform.auth.auth import get_current_user

    async def permission_checker(current_user=Depends(get_current_user)):
        rbac = RBAC()
        if not rbac.has_permission(current_user.role, permission):
            raise HTTPException(
                status_code=403,
                detail=f"Permission denied: {permission}",
            )
        return current_user

    return permission_checker