0

当我使用 Flask 时，每个 API 调用在处理之前都经过身份验证：

app = connexion.App(__name__, specification_dir='./swagger/', swagger_json=True, swagger_ui=True, server='tornado')
app.app.json_encoder = encoder.JSONEncoder
app.add_api('swagger.yaml', arguments={'title': 'ABCD API'})
# add CORS support
CORS(app.app)

@app.app.before_request
def before_request_func():
    app_id = request.headers.get("X-AppId")
    token = request.headers.get("X-Token")
    user, success = security.Security().authorize(token)
    if not success:
        status_code = 401
        response = {
            'code': status_code,
            'message': 'Unauthorized user'
        }
        return jsonify(response), status_code
    g.user = user

当我将其更改为 AioHttp 时，我的身份验证未正确设置：

options = {'swagger_path': 'swagger/', "swagger_ui": True}
app = connexion.AioHttpApp(__name__, specification_dir='swagger/', options=options)
app.add_api('swagger.yaml', arguments={'title': ' ABCD API'})
app = web.Application(middlewares=[auth_through_token])

async def auth_through_token(app: web.Application, handler: Any) -> Callable:
    @web.middleware
    async def middleware_handler(request: web.Request) -> web.Response:
        headers = request.headers
        x_auth_token = headers.get("X-Token")
        app_id = headers.get("X-AppId")
        user, success = security.Security().authorize(x_auth_token)
        if not success:
            return web.json_response(status=401, data={
                "error": {
                    "message": ("Not authorized. Reason: {}"
                                )
                }
            })
        response = await handler(request)
        return response

    return middleware_handler

我的请求没有被重定向到 API 方法。

谁能帮我设置每个 API 的 before_request 身份验证？谢谢。

4

1 回答 1

0

首先，您必须将middleware_handler从auth_through_token 移出。

然后，引用您的代码：

options = {'swagger_path': 'swagger/', "swagger_ui": True}
app = connexion.AioHttpApp(__name__, specification_dir='swagger/', options=options)
app.add_api('swagger.yaml', arguments={'title': ' ABCD API'})
app = web.Application(middlewares=[auth_through_token])

您必须删除最后一行并将第一行更改为：

options = {'swagger_path': 'swagger/', "swagger_ui": True, 'middlewares': [middleware_handler]}

所以最后代码应该是这样的：

options = {'swagger_path': 'swagger/', "swagger_ui": True, 'middlewares': [middleware_handler]}
app = connexion.AioHttpApp(__name__, specification_dir='swagger/', options=options)
app.add_api('swagger.yaml', arguments={'title': ' ABCD API'})

@web.middleware
async def middleware_handler(request: web.Request, handler: Any) -> web.Response:
    headers = request.headers
    x_auth_token = headers.get("X-Token")
    app_id = headers.get("X-AppId")
    user, success = security.Security().authorize(x_auth_token)
    if not success:
        return web.json_response(status=401, data={
            "error": {
                "message": ("Not authorized. Reason: {}"
                            )
            }
        })
    response = await handler(request)
    return response
于 2020-03-19T08:30:12.053 回答