Currently, I have a working API that uses Connexion and receives an OpenAPI specification:
connexion_app.add_api(
"openapi.yaml",
options={"swagger_ui": False},
validate_responses=True,
strict_validation=True, # Changing this also didn't help
)
A response gets validated in the following order:
The verification of the API-Key is done via the OpenAPI spec:
securitySchemes:
apiKeyAuth:
type: apiKey
in: header
name: API-Key
x-apikeyInfoFunc: server.security.check_api_key
security:
- apiKeyAuth: []
The validation is also done via the OpenAPI spec.
The signature gets verified in the endpoint:
if not verify_signature(kwargs):
abort(401, "Signature could not be verified")
Where verify_signature
is basically this:
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)
For security purposes I would like to swap 2. and 3.:
The problem is that Connexion validates the body before I get to my endpoint in which I execute my Python code such as verify_signature
.
I tried adding the following to my OpenAPI.yaml:
signatureAuth:
type: http
scheme: basic
x-basicInfoFunc: server.security.verify_signature
security:
- apiKeyAuth: []
signatureAuth: []
But I think this is the wrong approach since I think this is only used as a simple verification method and I get the following error message:
No authorization token provided
.
Now to my question:
Is there a way to execute a function which receives the whole request that gets executed before Connexion validates the body?
Yes you can use the Connexion before_request
annotation so it runs a function on a new request before validating the body. Here's an example that logs the headers and content:
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())