refactor(proxy): derive the bind policy once and pin the refusal status

The bind address never changes for an app's lifetime, so create_app now
canonicalises it and computes the wildcard flag and allowed Host set once
instead of on every request. _WILDCARD_HOSTS duplicated what
ip_address().is_unspecified already answers (only "" needs a special case),
the Sec-Fetch-Site sets become named module constants, and fetch_site is a
required argument so no caller can silently skip that check.

The 403 message now says what is actually refused (browser requests and
foreign Host names) rather than implying every non-local client is.

The parametrized proxy test asserted 200-vs-"some 4xx" via a tuple
comparison; it now pins status == 200/403 and the exact upstream request
count, so a regression to a different status or a leaked forward fails.
This commit is contained in:
kshitijk4poor
2026-10-01 18:09:51 +05:30
committed by kshitij
parent 47d4e3cba6
commit 98f7ec3cf4
2 changed files with 27 additions and 12 deletions

View File

@@ -56,7 +56,10 @@ def _json_error(status: int, message: str, code: str = "proxy_error") -> "web.Re
_LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"})
_WILDCARD_HOSTS = frozenset({"", "0.0.0.0", "::"})
# Sec-Fetch-Site values a browser sends for a request no foreign page made (typed URL / own page).
_UNFORGED_FETCH_SITES = frozenset({"none", "same-origin"})
# On a wildcard bind a DNS-rebound page is same-origin with its Host, so only "none" is trusted.
_WILDCARD_FETCH_SITES = frozenset({"none"})
def _canonical_host(host: str) -> str:
@@ -83,13 +86,18 @@ def _parse_authority(authority: str) -> Optional[tuple[str, int]]:
def _is_wildcard(bound: str) -> bool:
try:
return bound in _WILDCARD_HOSTS or ipaddress.ip_address(bound).is_unspecified
return bound == "" or ipaddress.ip_address(bound).is_unspecified
except ValueError:
return False
def _local_request_error(
host_header: str, origin: Optional[str], bound_host: str, fetch_site: Optional[str] = None
host_header: str,
origin: Optional[str],
fetch_site: Optional[str],
*,
wildcard: bool,
allowed_hosts: frozenset,
) -> Optional[str]:
"""Why a request must be refused, or None. The proxy attaches the operator's subscription
credential to whatever reaches it, so a web page in the operator's browser must not be able
@@ -101,15 +109,14 @@ def _local_request_error(
its own to match, and a DNS-rebound page is same-origin with its Host, so every page-made
browser request is refused there."""
target = _parse_authority(host_header)
bound = _canonical_host(bound_host.strip("[]"))
wildcard = _is_wildcard(bound)
if target is None or (not wildcard and target[0] not in _LOOPBACK_HOSTS | {bound}):
if target is None or (not wildcard and target[0] not in allowed_hosts):
return "host_not_allowed"
if origin is not None:
scheme, sep, rest = origin.strip().partition("://")
if wildcard or not sep or scheme.lower() != "http" or _parse_authority(rest) != target:
return "origin_not_allowed"
if fetch_site is not None and fetch_site.strip().lower() not in ({"none"} if wildcard else {"none", "same-origin"}):
trusted_sites = _WILDCARD_FETCH_SITES if wildcard else _UNFORGED_FETCH_SITES
if fetch_site is not None and fetch_site.strip().lower() not in trusted_sites:
return "origin_not_allowed"
return None
@@ -193,17 +200,26 @@ def create_app(adapter: UpstreamAdapter, bound_host: str = DEFAULT_HOST) -> "web
the single loop and every other in-flight streaming completion.
"""
_require_aiohttp()
# The bind address is fixed for the app's lifetime, so derive the Host policy once.
bound = _canonical_host(bound_host.strip("[]"))
wildcard = _is_wildcard(bound)
allowed_hosts = _LOOPBACK_HOSTS | {bound}
@web.middleware
async def local_only(request: "web.Request", handler):
refusal = _local_request_error(
request.headers.get("Host", ""),
request.headers.get("Origin"),
bound_host,
request.headers.get("Sec-Fetch-Site"),
wildcard=wildcard,
allowed_hosts=allowed_hosts,
)
if refusal:
return _json_error(403, "Request refused: only local, non-browser clients may use this proxy.", code=refusal)
return _json_error(
403,
"Request refused: browser requests and foreign Host names are not accepted by this proxy.",
code=refusal,
)
return await handler(request)
app = web.Application(client_max_size=MAX_REQUEST_BYTES, middlewares=[local_only])

View File

@@ -407,9 +407,8 @@ def test_loopback_proxy_serves_only_local_non_browser_requests(bound, headers, a
finally:
await proxy_runner.cleanup()
await upstream_runner.cleanup()
assert (status == 200, len(captured["requests"]) == 1) == (allowed, allowed), status
if not allowed:
assert 400 <= status < 500
assert status == (200 if allowed else 403)
assert len(captured["requests"]) == int(allowed)
asyncio.run(run())