fix: AP-1 percent-encode url() creds, AP-2 reject next(session=) collision

AP-1: url()/aiohttp() percent-encode user/password so reserved chars (/ # ? @) produce a
valid url, mirroring the parse-side unquote. AP-2: next(session=...) raises a clear
ValueError instead of an opaque str.format TypeError. net.current_ip uses
json(content_type=None) + dict guard.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
disqualifier 2026-06-29 21:34:36 -04:00
parent 932fb71c95
commit 72c5342a6b
3 changed files with 17 additions and 5 deletions

View File

@ -125,6 +125,10 @@ class AioProxies:
if self._static is not None: if self._static is not None:
return self._static return self._static
if self.template is not None: if self.template is not None:
if "session" in fields:
# `session` is auto-filled with a fresh id; a caller-supplied one would
# collide in str.format with an opaque TypeError — reject it clearly
raise ValueError("'session' is filled automatically; do not pass it to next()")
try: try:
filled = self.template.format(session=self.session_id(), **fields) filled = self.template.format(session=self.session_id(), **fields)
except (KeyError, IndexError) as exc: except (KeyError, IndexError) as exc:

View File

@ -37,8 +37,10 @@ async def current_ip(
try: try:
async with aiohttp.ClientSession(timeout=t) as session: async with aiohttp.ClientSession(timeout=t) as session:
async with session.get(test_url, proxy=p.url()) as resp: async with session.get(test_url, proxy=p.url()) as resp:
data = await resp.json() # content_type=None: an echo endpoint may return text/plain json; the
return data.get("ip") # default would raise ContentTypeError and silently return None
data = await resp.json(content_type=None)
return data.get("ip") if isinstance(data, dict) else None
except Exception as exc: except Exception as exc:
log.warning("ip check failed: %s", exc) log.warning("ip check failed: %s", exc)
return None return None

View File

@ -13,7 +13,7 @@ input shape. `canonical_key()` extends that to dict/url forms.
""" """
from dataclasses import dataclass from dataclasses import dataclass
from typing import Dict, Optional, Union from typing import Dict, Optional, Union
from urllib.parse import unquote, urlsplit from urllib.parse import quote, unquote, urlsplit
SCHEME_HTTP = "http" SCHEME_HTTP = "http"
SCHEME_SOCKS5 = "socks5" SCHEME_SOCKS5 = "socks5"
@ -48,9 +48,15 @@ class Proxy:
return f"{host}:{self.port}" return f"{host}:{self.port}"
def url(self, scheme: str = SCHEME_HTTP) -> str: def url(self, scheme: str = SCHEME_HTTP) -> str:
"""render as a url, embedding auth when present""" """render as a url, embedding auth when present
credentials are percent-encoded so reserved chars (/ # ? @ :) in a user or
password produce a valid url; this mirrors the unquote() on the parse side.
"""
if self.has_auth: if self.has_auth:
return f"{scheme}://{self.user}:{self.password}@{self.host}:{self.port}" user = quote(str(self.user), safe="")
password = quote(str(self.password), safe="")
return f"{scheme}://{user}:{password}@{self.host}:{self.port}"
return f"{scheme}://{self.host}:{self.port}" return f"{scheme}://{self.host}:{self.port}"
def aiohttp(self) -> Dict[str, str]: def aiohttp(self) -> Dict[str, str]: