Source code for aiohttp_middlewares.https
"""
================
HTTPS Middleware
================
Change scheme for current request when aiohttp application deployed behind
reverse proxy with HTTPS enabled.
Usage
=====
.. code-block:: python
from aiohttp import web
from aiohttp_middlewares import https_middleware
# Basic usage
app = web.Application(middlewares=[https_middleware()])
# Specify custom headers to match, not `X-Forwarded-Proto: https`
app = web.Application(
middlewares=https_middleware({"Forwarded": "https"})
)
"""
import logging
from typing import Union
from aiohttp import web
from aiohttp_middlewares.annotations import DictStrStr, Handler, Middleware
DEFAULT_MATCH_HEADERS = {"X-Forwarded-Proto": "https"}
logger = logging.getLogger(__name__)
[docs]def https_middleware(
match_headers: Union[DictStrStr, None] = None
) -> Middleware:
"""
Change scheme for current request when aiohttp application deployed behind
reverse proxy with HTTPS enabled.
This middleware is required to use, when your aiohttp app deployed behind
nginx with HTTPS enabled, after aiohttp discounted
``secure_proxy_ssl_header`` keyword argument in
https://github.com/aio-libs/aiohttp/pull/2299.
:param match_headers:
Dict of header(s) from reverse proxy to specify that aiohttp run behind
HTTPS. By default:
.. code-block:: python
{"X-Forwarded-Proto": "https"}
"""
@web.middleware
async def middleware(
request: web.Request, handler: Handler
) -> web.StreamResponse:
"""Change scheme of current request when HTTPS headers matched."""
headers = DEFAULT_MATCH_HEADERS
if match_headers is not None:
headers = match_headers
matched = any(
request.headers.get(key) == value for key, value in headers.items()
)
if matched:
logger.debug(
"Substitute request URL scheme to https",
extra={
"headers": headers,
"request_headers": dict(request.headers),
},
)
request = request.clone(scheme="https")
return await handler(request)
return middleware