83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import typing
|
|
|
|
import quart
|
|
import werkzeug
|
|
from werkzeug.http import parse_list_header
|
|
|
|
from .app import app
|
|
|
|
|
|
def _get_first_in_list(value_list: str | None, allowed: typing.Sequence[str] = ()) -> str | None:
|
|
if not value_list:
|
|
return None
|
|
values = parse_list_header(value_list)
|
|
if values and values[0]:
|
|
if not allowed or values[0] in allowed:
|
|
return values[0]
|
|
return None
|
|
|
|
|
|
def local_proxy_fix(request: quart.Request):
|
|
if not request.remote_addr:
|
|
return
|
|
try:
|
|
addr = ipaddress.ip_address(request.remote_addr)
|
|
except ValueError:
|
|
# TODO: accept unix sockets somehow too?
|
|
return
|
|
if not addr.is_loopback:
|
|
return
|
|
client = _get_first_in_list(request.headers.get("X-Forwarded-For"))
|
|
if not client:
|
|
# assume this is always set behind reverse proxies supporting any of the headers
|
|
return
|
|
request.remote_addr = client
|
|
scheme = _get_first_in_list(request.headers.get("X-Forwarded-Proto"), ("http", "https"))
|
|
port: int | None = None
|
|
if scheme:
|
|
port = 443 if scheme == "https" else 80
|
|
request.scheme = scheme
|
|
host = _get_first_in_list(request.headers.get("X-Forwarded-Host"))
|
|
port_s: str | None
|
|
if host:
|
|
request.host = host
|
|
if ":" in host and not host.endswith("]"):
|
|
try:
|
|
_, port_s = host.rsplit(":", maxsplit=1)
|
|
port = int(port_s)
|
|
except ValueError:
|
|
# ignore invalid port in host header
|
|
pass
|
|
port_s = _get_first_in_list(request.headers.get("X-Forwarded-Port"))
|
|
if port_s:
|
|
try:
|
|
port = int(port_s)
|
|
except ValueError:
|
|
# ignore invalid port in header
|
|
pass
|
|
if port:
|
|
if request.server and len(request.server) == 2:
|
|
request.server = (request.server[0], port)
|
|
root_path = _get_first_in_list(request.headers.get("X-Forwarded-Prefix"))
|
|
if root_path:
|
|
request.root_path = root_path
|
|
|
|
|
|
class LocalProxyFixRequestHandler:
|
|
def __init__(
|
|
self,
|
|
orig_handle_request: typing.Callable[[quart.Request], typing.Awaitable[quart.Response | werkzeug.Response]],
|
|
):
|
|
self._orig_handle_request = orig_handle_request
|
|
|
|
async def __call__(self, request: quart.Request) -> quart.Response | werkzeug.Response:
|
|
# need to patch request before url_adapter is built
|
|
local_proxy_fix(request)
|
|
return await self._orig_handle_request(request)
|
|
|
|
|
|
app.handle_request = LocalProxyFixRequestHandler(app.handle_request) # type: ignore
|