在 Connexion 中验证请求正文之前验证签名消息

Question

目前，我有一个使用 Connexion 并接收 OpenAPI 规范的工作 API：

connexion_app.add_api(
    "openapi.yaml",
    options={"swagger_ui": False},
    validate_responses=True,
    strict_validation=True,  # Changing this also didn't help
)

按以下顺序验证响应：

1. 检查 API 密钥是否有效
2. 验证请求正文是否包含所有必要的参数
3. 验证消息签名
4. 处理请求并发送响应

API-Key 的验证是通过 OpenAPI 规范完成的：

  securitySchemes:
    apiKeyAuth:
      type: apiKey
      in: header
      name: API-Key
      x-apikeyInfoFunc: server.security.check_api_key
security:
  - apiKeyAuth: []

验证也是通过 OpenAPI 规范完成的。

签名在端点中进行验证：

if not verify_signature(kwargs):
    abort(401, "Signature could not be verified")

其中 verify_signature 基本上是这样的：

def verify_signature(request) -> bool:
    """Calculate the signature using the header and data."""
    signature = re.findall(r'"([A-Za-z0-9+/=]+)"', connexion.request.headers.get("Message-Signature", ""))
    created = re.findall(r"created=(\d+)", connexion.request.headers.get("Message-Signature", ""))
    if len(signature) == 0:
        abort(401, "No valid Signature found.")
    if len(created) == 0:
        abort(401, "No valid created timestamp found.")

    signature = signature[0]
    created = int(created[0])
    method, path, host, api_key, content_type = _get_attributes_from_request()
    message = create_signature_message(request["body"], created, method, path, host, api_key, content_type)
    recreated_signature = _encode_message(message)
    return recreated_signature == str(signature)

---

出于安全目的，我想交换 2. 和 3.：

1. 检查 API 密钥是否有效
2. 验证消息签名
3. 验证请求是否 有效正文包含所有必需的参数
4. 处理请求并发送响应

问题是，Connexion 在我到达执行 Python 代码（例如 verify_signature）的端点之前验证正文。

我尝试将以下内容添加到我的 OpenAPI.yaml 中：

    signatureAuth:
      type: http
      scheme: basic
      x-basicInfoFunc: server.security.verify_signature
security:
  - apiKeyAuth: []
    signatureAuth: []

但我认为这是错误的方法，因为我认为这仅用作简单的验证方法，并且我收到以下错误消息： 未提供授权令牌。

现在我的问题是：

有没有办法执行一个函数，该函数接收在 Connexion 验证正文之前执行的整个请求？

Answer 1

是的，您可以使用 Connexion before_request 注释，以便它在验证正文之前对新请求运行一个函数。这是记录标题和内容的示例：

import connexion
import logging
from flask import request

logger = logging.getLogger(__name__)
conn_app = connexion.FlaskApp(__name__)

@conn_app.app.before_request
def before_request():
    for h in request.headers:
        logger.debug('header %s', h)
   logger.debug('data %s', request.get_data())