starred/camoufox
1
0
Fork 0
mirror of https://github.com/daijro/camoufox.git synced 2026-10-04 16:00:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(python): NewContext looks up a proxy's exit IP through the right URL, or fails

Browse Source 
NewContext derives the context's WebRTC IP and timezone from the proxy's exit
IP. That lookup had two defects, and both left the context showing the
host's values while its traffic went through the proxy:

- It built its own proxy URL with urlparse, which reads a scheme-less server
  such as "1.2.3.4:8080" (a form Playwright accepts) as scheme "1.2.3.4" with
  no host. urllib could not use a SOCKS proxy at all.
- Any failure was swallowed, and the context opened without the values.

The URL is now built with Proxy.as_string(), which the geoip launch path
already uses (scheme-less means http). The lookup goes through requests,
which handles SOCKS, and a failed lookup raises InvalidIP, naming the two
options that skip it. The tests cover scheme-less, http and socks5 servers
with credentials, both failure modes, and the case where no lookup is needed,
for NewContext and AsyncNewContext.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Jake Writer
2026-09-25 15:22:28 -06:00
co-authored by Claude Opus 5.5
parent 470d256484
commit 6a22248c0b
4 changed files with 142 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pythonlib/camoufox/async_api.py
+10 -34
View File
@@ -1,9 +1,6 @@
import asyncio
import json as _json
import urllib.request
from functools import partial
from typing import Any, Dict, Optional, Union, overload
from urllib.parse import urlparse
from typing import Any, Dict, Optional, Tuple, Union, overload
from playwright.async_api import (
Browser,
@@ -16,6 +13,7 @@ from typing_extensions import Literal
from camoufox.virtdisplay import VirtualDisplay
from .fingerprints import generate_context_fingerprint
from .ip import Proxy, proxy_exit_geo
from .utils import (
async_attach_vd,
attach_no_viewport_default,
@@ -178,31 +176,9 @@ async def _launch(
return await async_attach_vd(browser, virtual_display)
def _proxy_url_with_creds(proxy: Dict[str, str]) -> str:
"""Builds a proxy URL string with embedded credentials."""
parsed = urlparse(proxy.get("server", ""))
user = proxy.get("username", "")
pwd = proxy.get("password", "")
if user and pwd:
return f"{parsed.scheme}://{user}:{pwd}@{parsed.netloc}"
return proxy.get("server", "")
async def _resolve_proxy_geo(proxy: Dict[str, str]) -> Dict[str, Optional[str]]:
"""Queries ip-api.com through the proxy for the exit IP and timezone."""
proxy_url = _proxy_url_with_creds(proxy)
def _fetch() -> Dict[str, Optional[str]]:
handler = urllib.request.ProxyHandler({"http": proxy_url, "https": proxy_url})
opener = urllib.request.build_opener(handler)
try:
with opener.open("http://ip-api.com/json?fields=query,timezone", timeout=10) as resp:
data = _json.loads(resp.read())
return {"ip": data.get("query") or None, "timezone": data.get("timezone") or None}
except Exception:
return {"ip": None, "timezone": None}
return await asyncio.get_event_loop().run_in_executor(None, _fetch)
async def _resolve_proxy_geo(proxy: Dict[str, str]) -> Tuple[str, str]:
"""The proxy's exit IP and timezone, looked up off the event loop."""
return await asyncio.to_thread(proxy_exit_geo, Proxy(**proxy).as_string())
async def AsyncNewContext(
@@ -230,16 +206,16 @@ async def AsyncNewContext(
ff_version: Firefox version string for UA patching.
webrtc_ip: IPv4 address to spoof for WebRTC ICE candidates.
proxy: Per-context proxy (Playwright format: {"server": "...", "username": "...", "password": "..."}).
Unless webrtc_ip and timezone_id are both given, they are looked up from the
proxy's exit IP; InvalidIP is raised if that lookup fails.
geolocation: Per-context geolocation ({"latitude": float, "longitude": float}).
**context_kwargs: Additional Playwright new_context() options.
"""
# Auto-derive WebRTC IP and timezone from proxy's exit IP when not explicitly provided
if proxy and (not webrtc_ip or "timezone_id" not in context_kwargs):
geo = await _resolve_proxy_geo(proxy)
if not webrtc_ip:
webrtc_ip = geo["ip"]
if "timezone_id" not in context_kwargs and geo["timezone"]:
context_kwargs["timezone_id"] = geo["timezone"]
exit_ip, timezone = await _resolve_proxy_geo(proxy)
webrtc_ip = webrtc_ip or exit_ip
context_kwargs.setdefault("timezone_id", timezone)
fp = await asyncio.get_event_loop().run_in_executor(
None,
pythonlib/camoufox/ip.py
+28
View File
@@ -75,6 +75,34 @@ def validate_ip(ip: str) -> None:
raise InvalidIP(f"Invalid IP address: {ip}")
def proxy_exit_geo(proxy: str) -> Tuple[str, str]:
"""
The exit IP of `proxy` and that IP's timezone, looked up through the proxy.
Raises InvalidIP when the lookup fails: a context that silently kept the
host's WebRTC IP and timezone behind a proxy would be a leak.
"""
try:
resp = requests.get(
"http://ip-api.com/json?fields=status,message,query,timezone",
proxies=Proxy.as_requests_proxy(proxy),
timeout=10,
)
resp.raise_for_status()
data = resp.json()
except requests.RequestException as exception:
raise InvalidIP(f"{PROXY_LOOKUP_FAILED}: {exception}") from exception
if data.get("status") != "success" or not data.get("timezone"):
raise InvalidIP(f"{PROXY_LOOKUP_FAILED}: {data.get('message') or data}")
validate_ip(data["query"])
return data["query"], data["timezone"]
PROXY_LOOKUP_FAILED = (
"Could not look up the proxy's exit IP and timezone. Pass webrtc_ip and "
"timezone_id explicitly to skip the lookup"
)
@lru_cache(maxsize=None)
def public_ip(proxy: Optional[str] = None) -> str:
"""
pythonlib/camoufox/sync_api.py
+10 -30
View File
@@ -1,7 +1,4 @@
import json as _json
import urllib.request
from typing import Any, Dict, Optional, Union, overload
from urllib.parse import urlparse
from typing import Any, Dict, Optional, Tuple, Union, overload
from playwright.sync_api import (
Browser,
@@ -14,6 +11,7 @@ from typing_extensions import Literal
from camoufox.virtdisplay import VirtualDisplay
from .fingerprints import generate_context_fingerprint
from .ip import Proxy, proxy_exit_geo
from .utils import (
attach_no_viewport_default,
attach_stock_media_defaults,
@@ -156,27 +154,9 @@ def NewBrowser(
cpu_affinity.restore(pid, previous)
def _proxy_url_with_creds(proxy: Dict[str, str]) -> str:
"""Builds a proxy URL string with embedded credentials."""
parsed = urlparse(proxy.get("server", ""))
user = proxy.get("username", "")
pwd = proxy.get("password", "")
if user and pwd:
return f"{parsed.scheme}://{user}:{pwd}@{parsed.netloc}"
return proxy.get("server", "")
def _resolve_proxy_geo(proxy: Dict[str, str]) -> Dict[str, Optional[str]]:
"""Queries ip-api.com through the proxy for the exit IP and timezone."""
proxy_url = _proxy_url_with_creds(proxy)
handler = urllib.request.ProxyHandler({"http": proxy_url, "https": proxy_url})
opener = urllib.request.build_opener(handler)
try:
with opener.open("http://ip-api.com/json?fields=query,timezone", timeout=10) as resp:
data = _json.loads(resp.read())
return {"ip": data.get("query") or None, "timezone": data.get("timezone") or None}
except Exception:
return {"ip": None, "timezone": None}
def _resolve_proxy_geo(proxy: Dict[str, str]) -> Tuple[str, str]:
"""The proxy's exit IP and timezone."""
return proxy_exit_geo(Proxy(**proxy).as_string())
def NewContext(
@@ -204,16 +184,16 @@ def NewContext(
ff_version: Firefox version string for UA patching.
webrtc_ip: IPv4 address to spoof for WebRTC ICE candidates.
proxy: Per-context proxy (Playwright format: {"server": "...", "username": "...", "password": "..."}).
Unless webrtc_ip and timezone_id are both given, they are looked up from the
proxy's exit IP; InvalidIP is raised if that lookup fails.
geolocation: Per-context geolocation ({"latitude": float, "longitude": float}).
**context_kwargs: Additional Playwright new_context() options.
"""
# Auto-derive WebRTC IP and timezone from proxy's exit IP when not explicitly provided
if proxy and (not webrtc_ip or "timezone_id" not in context_kwargs):
geo = _resolve_proxy_geo(proxy)
if not webrtc_ip:
webrtc_ip = geo["ip"]
if "timezone_id" not in context_kwargs and geo["timezone"]:
context_kwargs["timezone_id"] = geo["timezone"]
exit_ip, timezone = _resolve_proxy_geo(proxy)
webrtc_ip = webrtc_ip or exit_ip
context_kwargs.setdefault("timezone_id", timezone)
fp = generate_context_fingerprint(preset=preset, os=os, ff_version=ff_version, webrtc_ip=webrtc_ip)
pythonlib/tests/test_proxy_geo.py
+94
View File
@@ -0,0 +1,94 @@
"""NewContext derives the WebRTC IP and timezone from the proxy's exit IP.
Two defects, both of which left a context with the host's WebRTC IP and
timezone while its traffic went through the proxy:
- The lookup built its own proxy URL with urlparse, which reads a scheme-less
server such as "1.2.3.4:8080" (a form Playwright accepts) as scheme "1.2.3.4"
and drops the host.
- A failed lookup was swallowed, and the context launched without the values.
"""
import asyncio
from unittest import mock
import pytest
from camoufox import async_api, ip, sync_api
from camoufox.exceptions import InvalidIP
class _Response:
def __init__(self, payload):
self._payload = payload
def raise_for_status(self):
pass
def json(self):
return self._payload
EXIT = {"status": "success", "query": "203.0.113.7", "timezone": "Europe/Paris"}
def _sync_browser():
browser = mock.MagicMock()
return browser
def _async_browser():
context = mock.MagicMock()
context.add_init_script = mock.AsyncMock()
browser = mock.MagicMock()
browser.new_context = mock.AsyncMock(return_value=context)
return browser
def _new_context(api, proxy, **kwargs):
if api == "sync":
browser = _sync_browser()
sync_api.NewContext(browser, os="linux", proxy=proxy, **kwargs)
return browser.new_context.call_args.kwargs, browser.new_context.return_value
browser = _async_browser()
asyncio.run(async_api.AsyncNewContext(browser, os="linux", proxy=proxy, **kwargs))
return browser.new_context.call_args.kwargs, browser.new_context.return_value
@pytest.mark.parametrize("api", ["sync", "async"])
@pytest.mark.parametrize(
"server, expected",
[
("1.2.3.4:8080", "http://u:p@1.2.3.4:8080"),
("proxy.example.com:8080", "http://u:p@proxy.example.com:8080"),
("http://proxy.example.com:8080", "http://u:p@proxy.example.com:8080"),
("socks5://proxy.example.com:1080", "socks5://u:p@proxy.example.com:1080"),
],
)
def test_lookup_goes_through_the_proxy_with_its_credentials(api, server, expected):
with mock.patch.object(ip.requests, "get", return_value=_Response(EXIT)) as get:
options, context = _new_context(api, {"server": server, "username": "u", "password": "p"})
assert get.call_args.kwargs["proxies"] == {"http": expected, "https": expected}
assert options["timezone_id"] == "Europe/Paris"
assert "203.0.113.7" in context.add_init_script.call_args.args[0]
@pytest.mark.parametrize("api", ["sync", "async"])
@pytest.mark.parametrize(
"failure",
[
ip.requests.ConnectionError("proxy refused"),
_Response({"status": "fail", "message": "private range"}),
],
)
def test_a_failed_lookup_raises_instead_of_launching_without_the_values(api, failure):
get = mock.Mock(side_effect=failure) if isinstance(failure, Exception) else mock.Mock(return_value=failure)
with mock.patch.object(ip.requests, "get", get), pytest.raises(InvalidIP, match="webrtc_ip"):
_new_context(api, {"server": "1.2.3.4:8080"})
@pytest.mark.parametrize("api", ["sync", "async"])
def test_no_lookup_when_both_values_are_given(api):
with mock.patch.object(ip.requests, "get") as get:
_new_context(api, {"server": "1.2.3.4:8080"}, webrtc_ip="198.51.100.1", timezone_id="UTC")
get.assert_not_called()