server-core/app/api/routers/rbac_admin.py

# -*- coding: utf-8 -*-
"""运营方管理端点（证明 RBAC 角色/权限/审计）。"""
from __future__ import annotations

from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel

from ..dependencies import get_db
from ...infrastructure.models import SetUserRoleRequest, SetUserStatusRequest
from ...rbac import require_permission, require_roles, write_audit
from ...infrastructure.repositories import Database

router = APIRouter(prefix="/admin", tags=["admin"])


class UserCreateRequest(BaseModel):
    username: str          # 手机号
    password: str
    nickname: str = ""
    role: str = "opc_member"
    sub_role: str | None = None
    org_id: str | None = None
    region_id: str | None = None


class RolePermissionRequest(BaseModel):
    permissions: list[str] = []


async def _validate_role_assignment(db: Database, actor: dict, role: str, sub_role: str | None) -> None:
    """白名单 + 提权防护：拒绝非法角色组合，及无权者创建/提升特权角色。"""
    key = role if not sub_role else f"{role}|{sub_role}"
    if not await db.roles.role_exists(role):
        raise HTTPException(status_code=400, detail=f"无效角色: {role}")
    if sub_role and not await db.roles.role_exists(f"{role}|{sub_role}"):
        raise HTTPException(status_code=400, detail=f"无效角色组合: {key}")
    # 目标角色若授予「配置角色权限」，仅持该权限者（超管）可创建/分配
    if "action:role.grant_perm" in await db.roles.permissions_for(role, sub_role) and \
            "action:role.grant_perm" not in actor.get("permissions", []):
        raise HTTPException(
            status_code=403,
            detail="无权创建/分配特权角色（仅超级管理员）",
        )


@router.post("/users", summary="新增账号（按类型/角色）")
async def create_user(
    req: UserCreateRequest,
    request: Request,
    db: Database = Depends(get_db),
    actor: dict = Depends(require_permission("action:user.assign_role")),
):
    if not req.username.strip() or not req.password.strip():
        raise HTTPException(status_code=400, detail="账号与密码必填")
    if await db.users.get_by_username(req.username) is not None:
        raise HTTPException(status_code=400, detail="账号已存在")
    await _validate_role_assignment(db, actor, req.role, req.sub_role)
    user = await db.users.create(
        req.username, req.password, nickname=req.nickname,
        role=req.role, sub_role=req.sub_role, org_id=req.org_id, region_id=req.region_id,
    )
    # 自动创建该账号的默认端口身份
    await db.identities.create(user["id"], port=_port_for_role(req.role), role=req.role,
                         sub_role=req.sub_role, org_id=req.org_id, region_id=req.region_id,
                         name=req.nickname or req.username)
    await write_audit(db, action="user.create", resource="user", resource_id=user["id"],
                detail=f"role={req.role}", user=actor, request=request)
    return await db.users.to_profile(user)


def _port_for_role(role: str) -> str:
    return {
        "opc_member": "opc", "carrier": "carrier", "enterprise": "enterprise",
        "provider": "provider", "government": "government", "operator": "operator",
        "investor": "investor", "developer": "developer",
    }.get(role, "opc")


@router.get("/users", summary="用户列表（运营方）")
async def list_users(
    request: Request,
    db: Database = Depends(get_db),
    _role: dict = Depends(require_roles("operator")),
    _perm: dict = Depends(require_permission("menu:admin_user_mgmt")),
):
    return await db.users.list()


@router.post("/users/{user_id}/role", summary="分配角色")
async def set_user_role(
    user_id: str,
    req: SetUserRoleRequest,
    request: Request,
    db: Database = Depends(get_db),
    actor: dict = Depends(require_permission("action:user.assign_role")),
):
    if await db.users.get_by_id(user_id) is None:
        raise HTTPException(status_code=404, detail="User not found")
    await _validate_role_assignment(db, actor, req.role, req.sub_role)
    updated = await db.users.set_role(
        user_id, req.role, req.sub_role, req.org_id, req.region_id,
    )
    await write_audit(
        db, action="role.assign", resource="user", resource_id=user_id,
        detail=f"{actor['username']} -> role={req.role} sub={req.sub_role}",
        user=actor, request=request,
    )
    return {"ok": True, "user": await db.users.to_profile(updated)}


@router.post("/users/{user_id}/status", summary="禁用/启用用户")
async def set_user_status(
    user_id: str,
    req: SetUserStatusRequest,
    request: Request,
    db: Database = Depends(get_db),
    actor: dict = Depends(require_permission("action:user.disable")),
):
    if await db.users.get_by_id(user_id) is None:
        raise HTTPException(status_code=404, detail="User not found")
    updated = await db.users.set_status(user_id, req.status)
    await write_audit(
        db, action="user.disable", resource="user", resource_id=user_id,
        detail=f"{actor['username']} -> status={req.status}",
        user=actor, request=request,
    )
    return {"ok": True, "user": await db.users.to_profile(updated)}


@router.get("/roles", summary="角色列表")
async def list_roles(
    db: Database = Depends(get_db),
    _user: dict = Depends(require_permission("menu:admin_role_mgmt")),
):
    return await db.roles.list_roles()


@router.get("/permissions", summary="权限列表")
async def list_permissions(
    db: Database = Depends(get_db),
    _user: dict = Depends(require_permission("menu:admin_role_mgmt")),
):
    return await db.roles.list_permissions()


@router.post("/roles/{role_id}/permissions", summary="配置角色权限")
async def set_role_permissions(
    role_id: str,
    req: RolePermissionRequest,
    request: Request,
    db: Database = Depends(get_db),
    actor: dict = Depends(require_permission("action:role.grant_perm")),
):
    if not await db.roles.role_exists(role_id):
        raise HTTPException(status_code=404, detail="Role not found")
    perms = await db.roles.set_role_permissions(role_id, req.permissions)
    await write_audit(db, action="role.grant_perm", resource="role", resource_id=role_id,
                detail=f"permissions={len(req.permissions)}", user=actor, request=request)
    return {"ok": True, "role_id": role_id, "permissions": perms}


@router.get("/audit-logs", summary="审计日志")
async def list_audit_logs(
    limit: int = 100,
    offset: int = 0,
    db: Database = Depends(get_db),
    _user: dict = Depends(require_permission("action:audit.view")),
):
    return await db.audit.list(limit=min(limit, 500), offset=offset)