fix: fail loud on unreachable domain overwrites and jar-dropped IP cookies

overwrite_domain() now rejects a port-bearing replacement immediately at
registration time with a clear ValueError, instead of only surfacing a
deep yarl error from inside with_host() the first time a matching request
is made, far from the misconfiguration site.

set_cookie() now raises ValueError when domain resolves to a bare IP host
and the jar is the default CookieJar(unsafe=False), which silently drops
IP-bound cookies with no store, no send, and no log. Non-IP domains and a
caller-supplied unsafe=True jar are unaffected.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
2026-07-06 00:09:37 -04:00
parent 78dfba0962
commit 59eadabe16
+42 -2
View File
@@ -9,6 +9,7 @@ headers, domain rewriting, previews, retry/backoff. See README for usage and con
""" """
import asyncio import asyncio
import ipaddress
import logging import logging
import warnings import warnings
from http.cookies import SimpleCookie from http.cookies import SimpleCookie
@@ -28,6 +29,17 @@ def _route_body(data):
return data, None return data, None
def _is_ip_host(host) -> bool:
"""whether host is a literal IPv4/IPv6 address rather than a hostname"""
if not host:
return False
try:
ipaddress.ip_address(host)
except ValueError:
return False
return True
class _RetryStatus(Exception): class _RetryStatus(Exception):
"""internal signal: a retryable HTTP status; carries the real Response through aretry""" """internal signal: a retryable HTTP status; carries the real Response through aretry"""
@@ -174,7 +186,19 @@ class ExtendedSession:
# domain overwrites # domain overwrites
def overwrite_domain(self, target, replacement): def overwrite_domain(self, target, replacement):
"""register a host-substring rewrite (target -> replacement)""" """register a host-substring rewrite (target -> replacement)
raises:
ValueError: if replacement contains a port (e.g. 'host:1234') - yarl's
with_host() rejects any ':' in a host at request time, deep inside
yarl with a stack trace far from this call; caught here instead
"""
if ":" in replacement:
raise ValueError(
f"overwrite_domain replacement {replacement!r} contains a port - "
"yarl's URL.with_host() cannot take a host:port string; rewrite the "
"port separately or via a full URL override, not domain_overwrites"
)
self.domain_overwrites[target] = replacement self.domain_overwrites[target] = replacement
def _apply_domain_overwrites(self, url: str) -> str: def _apply_domain_overwrites(self, url: str) -> str:
@@ -229,6 +253,12 @@ class ExtendedSession:
(scheme optional, defaulted to http://) scopes it to that host - a bare hostname (scheme optional, defaulted to http://) scopes it to that host - a bare hostname
is normalized into a URL so the jar binds by host instead of silently storing is normalized into a URL so the jar binds by host instead of silently storing
another domain-less shared cookie. `path` is honored via the morsel itself. another domain-less shared cookie. `path` is honored via the morsel itself.
raises:
ValueError: if domain is a bare IP host (e.g. '127.0.0.1') and the jar is
the default aiohttp.CookieJar(unsafe=False), which drops cookies bound
to IP hosts with no store, no send, and no log - construct the session
with cookie_jar=aiohttp.CookieJar(unsafe=True) to allow IP-host cookies
""" """
cookie = SimpleCookie() cookie = SimpleCookie()
cookie[name] = value cookie[name] = value
@@ -238,7 +268,17 @@ class ExtendedSession:
return return
if "://" not in domain: if "://" not in domain:
domain = "http://" + domain domain = "http://" + domain
self.session.cookie_jar.update_cookies(cookie, response_url=URL(domain)) url = URL(domain)
jar = self.session.cookie_jar
if _is_ip_host(url.raw_host) and not getattr(jar, "unsafe", False):
raise ValueError(
f"set_cookie domain {domain!r} resolves to a bare IP host, but "
"aiohttp's cookie jar rejects IP-address domains under unsafe=False "
"(the default) - the cookie would be silently dropped; construct "
"ExtendedSession with cookie_jar=aiohttp.CookieJar(unsafe=True) to "
"allow IP-host cookies"
)
jar.update_cookies(cookie, response_url=url)
def clear_cookies(self): def clear_cookies(self):
"""clear the session cookie jar""" """clear the session cookie jar"""