基于角色方法的中间件基本用法

import zdppy_api as api
import zdppy_apimidauth


async def index(request):
    return api.resp.success()


async def login(request):
    token = zdppy_apimidauth.get_role_token(role="admin")
    return api.resp.success(token)


auth_dict = {
    "admin": {
        "GET:/": True,
        "GET:/1": False,
    }
}


async def has_auth_func(role, method, path):
    """校验角色对method和path是否有访问权限"""
    print(role, method, path)
    if not str(path).startswith("/"):
        path = f"/{path}"

    # GET:/1
    auth = str(method).upper() + ":" + path

    # 判断是否有权限
    role_auth_dict = auth_dict.get(role)
    if not isinstance(role_auth_dict, dict):
        return False
    if not role_auth_dict.get(auth):
        return False

    return True


app1 = api.Api(
    routes=[
        api.resp.get("/", index),
        api.resp.get("/1", index),
        api.resp.post("/login", login),
    ],
    middleware=[
        # 默认是：zhangdapeng zhangdapng520
        # 可以传入账号和密码进行覆盖
        zdppy_apimidauth.roleapi(has_auth_func)
    ]
)

if __name__ == '__main__':
    app1.run()

解析Token的方法

import zdppy_jwt as jwt
import zdppy_env as env

env.load("../.env")

key = env.get("JWT_KEY")
token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjE4MzE2NTM2NjMzODEyOTkyMDAiLCJ1c2VybmFtZSI6ImFkbWluIiwicm9sZV9pZCI6bnVsbCwicm9sZSI6bnVsbCwicm9sZV9kaWN0Ijp7fSwiZXhwaXJlZCI6MTcyNTU2MDA0OC44MjI1Mjh9.23iRSnug4xHdWEIRLTaZsTi4vCzOjFS5UIFB0AT_Jmw"
userinfo = jwt.parse_token(token, key=key)
print(userinfo)

解析得到下面的东西：

{'id': '1831653663381299200', 'username': 'admin', 'role_id': None, 'role': None, 'role_dict': {}, 'expired': 1725560048.822528}

整合权限校验中间件

import zdppy_api as api
import zdppy_env as env
import zdppy_mcrud as mcrud
import routes
import zdppy_apimidauth
from zdppy_log import logger

env.load(".env")

db = mcrud.new_env()


async def has_auth_func(role, method, path):
    """校验角色对method和path是否有访问权限"""
    logger.debug("正在校验接口权限", role=role, method=method, path=path)
    return True


app = api.Api(
    routes=[
        *routes.get_routes(db),
    ],
    middleware=[
        api.middleware.cors(),
        zdppy_apimidauth.roleapi(has_auth_func, env.get("JWT_KEY"))
    ],
)

if __name__ == "__main__":
    app.run(port=18888)

如果没有传Token

带上Token

错误的Token

我传的是无效的Token，但是返回的是服务器内部错误。