mirror of
https://github.com/daijro/camoufox.git
synced 2026-09-08 16:01:00 +00:00
luandhgt
1934b3532d
public_ip() called requests.get with verify=False and wrapped it in a
context manager that silenced urllib3's InsecureRequestWarning, so the
disabled verification produced no output either.
These requests are routed through the user's proxy, which is the exact
position an attacker occupies. A forged response controls the value
public_ip() returns, and that value is used to spoof the WebRTC IP --
so the leak the function exists to prevent becomes attacker-selectable.
validate_ip() bounds this to a well-formed address, but the address is
still theirs to choose.
Set verify=True and drop the warning suppression. requests raises
SSLError, a subclass of RequestException, which the existing loop
already catches -- a host with a bad certificate is now skipped in
favour of the next one in URLS instead of being trusted.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
(cherry picked from commit 9de97513b0)
110 lines
2.9 KiB
Python
110 lines
2.9 KiB
Python
import re
|
|
from dataclasses import dataclass
|
|
from functools import lru_cache
|
|
from typing import Dict, Optional, Tuple
|
|
|
|
import requests
|
|
|
|
from .exceptions import InvalidIP, InvalidProxy
|
|
|
|
"""
|
|
Helpers to find the user's public IP address for geolocation.
|
|
"""
|
|
|
|
|
|
@dataclass
|
|
class Proxy:
|
|
"""
|
|
Stores proxy information.
|
|
"""
|
|
|
|
server: str
|
|
username: Optional[str] = None
|
|
password: Optional[str] = None
|
|
bypass: Optional[str] = None
|
|
|
|
@staticmethod
|
|
def parse_server(server: str) -> Tuple[str, str, Optional[str]]:
|
|
"""
|
|
Parses the proxy server string.
|
|
"""
|
|
proxy_match = re.match(r'^(?:(?P<schema>\w+)://)?(?P<url>.*?)(?:\:(?P<port>\d+))?$', server)
|
|
if not proxy_match:
|
|
raise InvalidProxy(f"Invalid proxy server: {server}")
|
|
return proxy_match['schema'], proxy_match['url'], proxy_match['port']
|
|
|
|
def as_string(self) -> str:
|
|
schema, url, port = self.parse_server(self.server)
|
|
if not schema:
|
|
schema = 'http'
|
|
result = f"{schema}://"
|
|
if self.username:
|
|
result += f"{self.username}"
|
|
if self.password:
|
|
result += f":{self.password}"
|
|
result += "@"
|
|
|
|
result += url
|
|
if port:
|
|
result += f":{port}"
|
|
return result
|
|
|
|
@staticmethod
|
|
def as_requests_proxy(proxy_string: str) -> Dict[str, str]:
|
|
"""
|
|
Converts the proxy to a requests proxy dictionary.
|
|
"""
|
|
return {
|
|
'http': proxy_string,
|
|
'https': proxy_string,
|
|
}
|
|
|
|
|
|
@lru_cache(128, typed=True)
|
|
def valid_ipv4(ip: str) -> bool:
|
|
return bool(re.match(r'^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$', ip))
|
|
|
|
|
|
@lru_cache(128, typed=True)
|
|
def valid_ipv6(ip: str) -> bool:
|
|
return bool(re.match(r'^(([0-9a-fA-F]{0,4}:){1,7}[0-9a-fA-F]{0,4})$', ip))
|
|
|
|
|
|
def validate_ip(ip: str) -> None:
|
|
if not valid_ipv4(ip) and not valid_ipv6(ip):
|
|
raise InvalidIP(f"Invalid IP address: {ip}")
|
|
|
|
|
|
@lru_cache(maxsize=None)
|
|
def public_ip(proxy: Optional[str] = None) -> str:
|
|
"""
|
|
Sends a request to a public IP api
|
|
"""
|
|
URLS = [
|
|
# Prefers IPv4
|
|
"https://api.ipify.org",
|
|
"https://checkip.amazonaws.com",
|
|
"https://ipinfo.io/ip",
|
|
# IPv4 & IPv6
|
|
"https://icanhazip.com",
|
|
"https://ifconfig.co/ip",
|
|
"https://ipecho.net/plain",
|
|
]
|
|
|
|
end_exception = None
|
|
for url in URLS:
|
|
try:
|
|
resp = requests.get(
|
|
url,
|
|
proxies=Proxy.as_requests_proxy(proxy) if proxy else None,
|
|
timeout=5,
|
|
verify=True,
|
|
)
|
|
resp.raise_for_status()
|
|
ip = resp.text.strip()
|
|
validate_ip(ip)
|
|
return ip
|
|
except (requests.exceptions.ProxyError, requests.RequestException, InvalidIP) as exception:
|
|
end_exception = exception
|
|
raise InvalidIP(f"Failed to get IP address: {end_exception}")
|