xbackend/api/api_v1/authz/authz.py

from typing import Any

from fastapi import APIRouter, Depends, Request
from motor.motor_asyncio import AsyncIOMotorDatabase

import crud
import schemas
from api import deps
from db import get_database
from db.ckdb import CKDrive, get_ck_db
from db.redisdb import RedisDrive, get_redis_pool
from models.behavior_analysis import BehaviorAnalysis
from utils import casbin_enforcer

router = APIRouter()


@router.post("/add_role_domain")
async def add_role_domain(
        request: Request,
        data_in: schemas.AddRoleForUserInDomain,
        db: AsyncIOMotorDatabase = Depends(get_database),
        current_user: schemas.UserDB = Depends(deps.get_current_user)):
    """
    在域内为用户添加角色
    """

    # username role dom
    is_exists_role = await crud.role.check(db, _id=data_in.role_id, game=data_in.game)
    if not is_exists_role:
        return schemas.Msg(code='-1', msg='角色不存在')
    casbin_enforcer.add_role_for_user_in_domain(user=data_in.username,
                                                role=data_in.role_id,
                                                domain=data_in.game)
    return schemas.Msg(code='-1', msg='添加成功')


@router.post("/del_role_user_domain")
async def del_role_domain(
        request: Request,
        data_in: schemas.DeleteRolesForUserInDomain,
        db: AsyncIOMotorDatabase = Depends(get_database),
        current_user: schemas.UserDB = Depends(deps.get_current_user)):
    """
    删除用户角色域
    """

    # username role dom

    res = casbin_enforcer.delete_roles_for_user_in_domain(user=data_in.username,
                                                          role=data_in.role_id,
                                                          domain=data_in.game)
    return schemas.Msg(code=0, msg='ok', data=res)


@router.post("/add_policy")
async def add_policy(
        request: Request,
        data_id: schemas.AddPolicy,
        db: AsyncIOMotorDatabase = Depends(get_database),
        current_user: schemas.UserDB = Depends(deps.get_current_user)):
    """
    向当前策略添加授权规则
    """

    res = casbin_enforcer.add_policy(data_id.role_id, data_id.game, data_id.path, data_id.act)
    return schemas.Msg(code=0, msg='ok', data=res)


@router.post("/del_policy")
async def remove_policy(
        request: Request,
        data_id: schemas.DelPolicy,
        current_user: schemas.UserDB = Depends(deps.get_current_user)):
    """
    向当前策略添加授权规则
    """

    res = casbin_enforcer.remove_policy(data_id.role_id, data_id.game, data_id.path, data_id.act)
    return schemas.Msg(code=0, msg='ok', data=res)


@router.get("/api_list")
async def api_list(
        request: Request,
        db: AsyncIOMotorDatabase = Depends(get_database),
        current_user: schemas.UserDB = Depends(deps.get_current_user)):
    """

    GetPermissionsForUserInDomain
    已经添加的api
    标记 已添加的权限
    """
    res = await crud.api_list.all_api(db)
    return schemas.Msg(code=0, msg='ok', data=res)


@router.post("/add_api")
async def add_api(
        request: Request,
        data_in: schemas.AddApi,
        db: AsyncIOMotorDatabase = Depends(get_database),
        current_user: schemas.UserDB = Depends(deps.get_current_user)
) -> schemas.Msg:
    """
    添加api
    """
    try:
        res = await crud.api_list.add_api(db, data_in)
    except Exception as e:
        return schemas.Msg(code=-1, msg='已经存在')
    return schemas.Msg(code=0, msg='ok', data=res.matched_count)


@router.post("/del_api")
async def del_api(
        request: Request,
        data_in: schemas.DelApi,
        db: AsyncIOMotorDatabase = Depends(get_database),
        current_user: schemas.UserDB = Depends(deps.get_current_user)) -> schemas.Msg:
    """
    删除api
    """
    # 删除规则
    paths = await crud.api_list.find_ids(db, data_in.ids, {'path': 1})
    for item in paths:
        casbin_enforcer.remove_filtered_policy(2, item['path'])

    # 删除保存的记录
    res = await crud.api_list.del_api(db, data_in)

    return schemas.Msg(code=0, msg='ok', data=res.deleted_count)


@router.post("/edit_api")
async def edit_api(
        request: Request,
        data_in: schemas.EditApi,
        db: AsyncIOMotorDatabase = Depends(get_database),
        current_user: schemas.UserDB = Depends(deps.get_current_user)) -> schemas.Msg:
    """
    编辑api
    """
    res = await crud.api_list.edit_api(db, data_in)
    return schemas.Msg(code=0, msg='ok', data=res.matched_count)


@router.get("/domain")
async def domain_list(
        request: Request,
        db: AsyncIOMotorDatabase = Depends(get_database),
        current_user: schemas.UserDB = Depends(deps.get_current_user)
) -> schemas.Msg:
    """
    可选择域  游戏代号
    """
    # roel  dom path *
    res = await crud.project.all_game(db)
    return schemas.Msg(code=0, msg='ok', data=res)


@router.post("/add_roles")
async def add_roles(
        request: Request,
        data_in: schemas.AddRole,
        db: AsyncIOMotorDatabase = Depends(get_database),
        current_user: schemas.UserDB = Depends(deps.get_current_user)
) -> schemas.Msg:
    """
    新建角色
    """
    try:
        res = await crud.role.add_role(db, data_in)
        return schemas.Msg(code=0, msg='ok', data=res.matched_count)
    except Exception as e:
        return schemas.Msg(code=-1, msg='添加失败', data=str(e))


@router.get("/roles")
async def roles(
        request: Request,
        game: str,
        db: AsyncIOMotorDatabase = Depends(get_database),
        current_user: schemas.UserDB = Depends(deps.get_current_user)
) -> schemas.Msg:
    """
    域内所有角色
    """
    res = await crud.role.dom_roles(db, game)
    return schemas.Msg(code=0, msg='ok', data=res)