user-wsc/Utils/AccessControl/AccessUtil.py

import os
from typing import Optional

import casbin
from fastapi import HTTPException, Header

from Utils.Authentication.TokenUtil import decode_token


def ac_admin(token: Optional[str] = Header(...)):

    user_info = decode_token(token).get("user_info")

    if not user_info:
        raise HTTPException(status_code=400, detail="Invalid Token")

    role = user_info.get("role")

    e = casbin.Enforcer(
        os.getcwd() + r"\Utils\AccessControl\model.conf",
        os.getcwd() + r"\Utils\AccessControl\policy_index_store.csv"
    )

    sub = role
    obj = ""
    act = ""

    if not e.enforce(sub, obj, act):
        raise HTTPException(status_code=400, detail="No Access")

    # if role != "admin":
    #     raise HTTPException(status_code=400, detail="No Access")


# def ac_index_store(request: Request, token: Optional[str] = Header(...)):
#     e = casbin.Enforcer(
#         os.getcwd() + r"\Utils\AccessControl\model.conf",
#         os.getcwd() + r"\Utils\AccessControl\policy_index_store.csv"
#     )
#
#     sub = decode_token(token).get("role")
#     obj = request.url.__str__().split(request.base_url.__str__()[:-1])[-1].split("?")[0]
#     act = request.method
#
#     if not e.enforce(sub, obj, act):
#         raise HTTPException(status_code=400, detail="No Access")