FastAPI中鉴权的实现

由于项目的需要，前端的 Unity 封装 WebGL 实现一些数字孪生的开发，所以，Unity 中需要请求后端的服务提供数据，所以，在请求数据时需要对请求进行鉴权，对于非法请求进行拒绝服务的操作。今天，我们分享一下，鉴权的实现。

1.实现思路

分配 AppKey 和 AppSecret，然后在前端通过将请求时间、盐值、噪声等数据连接字符串，并进行 MD5 或者 sha256 运算，形成一个结果性的字符串。在服务端使用一样的数据进行运算来校验是否合格，一般情况下，前端不提交 AppKey 和 AppSecret，或者不提交 AppSecret，只提交 AppKey 及其他计算数据及计算结果。这些数据的提交方式可以是 url 参数，也可以写到 web 请求头中。

2.后端

后端采用 Python 进行开发，但在实际调试中，遇到一些问题，记录在此，分享给大家：

Unity 的 UnityWebRequest 进行请求时，如果将传输的验证参数封装为 url 参数，没有任何问题，但这种方式使用起来是明文，所以大家都想到，应该去使用请求头来传输这些数据。但在 Python 开发中，如果使用服务器的中间件来实现的话，获取参数是没有问题的，当进行校验时，这些参数的获取就不是很稳定，分析不出来到底是 Unity 的缘由还是 Python 的缘由，所以换一种方法，不使用中间件来实现鉴权，而是采用装饰器来实现。

3.示例

服务端代码：

# coding: utf-8
import settings
import hashlib
from fastapi import Request
from functools import wraps
from fastapi.responses import JSONResponse
from logger import logger
from datetime import datetime

def common_before_after_decorator(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        # 在请求处理之前执行的代码
        print("Before request processing")

        request: Request = kwargs.get('request')
        logger.info(request.headers)

        tm = request.headers.get('tm')  # timestamp
        nc = request.headers.get('nc')  # noise
        cnc = request.headers.get('cnc')  # noise
        rs = request.headers.get('rs')  # result
        logger.info(f"
tm:{tm}
nc:{nc}
cnc:{cnc}
rs:{rs}")

        if tm is None or nc is None or cnc is None or rs is None:
            logger.info('存在空值')
            return JSONResponse(status_code=401, content={"detail": "Not authenticated"})

        # 检查请求时间是否合规
        logger.info(f'{tm}')

        request_time = datetime.strptime(tm, '%Y-%m-%d %H:%M:%S')
        current = datetime.now()
        # 计算时间差
        time_diff = current - request_time
        # 将时间差转换为秒数
        seconds_diff = time_diff.total_seconds()
        if seconds_diff > 60:
            logger.info('请求超出时限')
            return JSONResponse(status_code=401, content={"detail": "Not authenticated"})

        # 鉴权计算
        auth_str = ''.join([nc, settings.APP_KEY, tm, settings.APP_SECRET, cnc])
        sha256 = hashlib.sha256()  # 创建sha256对象
        sha256.update(auth_str.encode())
        res = sha256.hexdigest().upper()
        logger.info(f'计算的鉴权结果：{res}')
        logger.info(f'请求的鉴权结果：{rs}')

        if res != rs:
            logger.info('鉴权失败')
            return JSONResponse(status_code=401, content={"detail": "Not authenticated"})

        response = await func(*args, **kwargs)

        # 在请求处理之后执行的代码
        print("After request processing")
        return response
    return wrapper

使用方法：

@router.get(path='/heart/{hotelcode}')
@common_before_after_decorator
async def heart(request:Request, hotelcode):
    logger.info("Hotel Code : {0}".format(hotelcode))
    return {'res': 'OK'}

这里，请注意，为了在装饰器中能够获取到请求参数，在每个路由函数中，均需要增加 request:Reque 参数，大部分情况下，在路由函数中不使用该参数，而是未来提供给装饰器使用的。