mirror of
https://github.com/CloakHQ/CloakBrowser.git
synced 2026-06-23 11:41:46 +02:00
feat: auto-detect timezone/locale from proxy IP via geoip
Adds geoip=True parameter to launch(), launch_async(), and launch_context(). Resolves proxy exit IP → MaxMind GeoLite2-City lookup → timezone + locale. Downloads ~70MB DB on first use from P3TERX mirror, caches in ~/.cloakbrowser/geoip/. Optional deps: pip install cloakbrowser[geoip] / npm install mmdb-lib Explicit timezone/locale always override auto-detected values.
This commit is contained in:
CloakHQ
@@ -0,0 +1,238 @@
|
||||
"""GeoIP-based timezone and locale detection from proxy IP.
|
||||
|
||||
Optional feature — requires ``geoip2`` package::
|
||||
|
||||
pip install cloakbrowser[geoip]
|
||||
|
||||
Downloads GeoLite2-City.mmdb (~70 MB) on first use, caches in
|
||||
``~/.cloakbrowser/geoip/``. Background re-download after 30 days.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
import logging
|
||||
import socket
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
logger = logging.getLogger("cloakbrowser")
|
||||
|
||||
# P3TERX mirror of MaxMind GeoLite2-City — no license key needed
|
||||
GEOIP_DB_URL = (
|
||||
"https://github.com/P3TERX/GeoLite.mmdb/raw/download/GeoLite2-City.mmdb"
|
||||
)
|
||||
GEOIP_DB_FILENAME = "GeoLite2-City.mmdb"
|
||||
GEOIP_UPDATE_INTERVAL = 30 * 86_400 # 30 days
|
||||
|
||||
# Country ISO code → BCP 47 locale (covers ~90 % of proxy traffic)
|
||||
COUNTRY_LOCALE_MAP: dict[str, str] = {
|
||||
"US": "en-US", "GB": "en-GB", "AU": "en-AU", "CA": "en-CA", "NZ": "en-NZ",
|
||||
"IE": "en-IE", "ZA": "en-ZA", "SG": "en-SG",
|
||||
"DE": "de-DE", "AT": "de-AT", "CH": "de-CH",
|
||||
"FR": "fr-FR", "BE": "fr-BE",
|
||||
"ES": "es-ES", "MX": "es-MX", "AR": "es-AR", "CO": "es-CO", "CL": "es-CL",
|
||||
"BR": "pt-BR", "PT": "pt-PT",
|
||||
"IT": "it-IT", "NL": "nl-NL",
|
||||
"JP": "ja-JP", "KR": "ko-KR", "CN": "zh-CN", "TW": "zh-TW", "HK": "zh-HK",
|
||||
"RU": "ru-RU", "UA": "uk-UA", "PL": "pl-PL", "CZ": "cs-CZ", "RO": "ro-RO",
|
||||
"IL": "he-IL", "TR": "tr-TR", "SA": "ar-SA", "AE": "ar-AE", "EG": "ar-EG",
|
||||
"IN": "hi-IN", "ID": "id-ID", "PH": "en-PH",
|
||||
"TH": "th-TH", "VN": "vi-VN", "MY": "ms-MY",
|
||||
"SE": "sv-SE", "NO": "nb-NO", "DK": "da-DK", "FI": "fi-FI",
|
||||
"GR": "el-GR", "HU": "hu-HU", "BG": "bg-BG",
|
||||
}
|
||||
|
||||
|
||||
def resolve_proxy_geo(proxy_url: str) -> tuple[str | None, str | None]:
|
||||
"""Resolve timezone and locale from a proxy's IP address.
|
||||
|
||||
Returns ``(timezone, locale)`` — either or both may be ``None`` on
|
||||
failure (missing dep, DB download error, lookup miss). Never raises.
|
||||
"""
|
||||
try:
|
||||
import geoip2.database # noqa: F811
|
||||
except ImportError:
|
||||
raise ImportError(
|
||||
"geoip2 is required for geoip=True. Install it with:\n"
|
||||
" pip install cloakbrowser[geoip]"
|
||||
) from None
|
||||
|
||||
db_path = _ensure_geoip_db()
|
||||
if db_path is None:
|
||||
return None, None
|
||||
|
||||
# Exit IP (through proxy) is most accurate — gateway DNS may differ from exit
|
||||
ip = _resolve_exit_ip(proxy_url)
|
||||
if ip is None:
|
||||
ip = _resolve_proxy_ip(proxy_url)
|
||||
if ip is None:
|
||||
return None, None
|
||||
|
||||
try:
|
||||
with geoip2.database.Reader(str(db_path)) as reader:
|
||||
resp = reader.city(ip)
|
||||
timezone = resp.location.time_zone
|
||||
country = resp.country.iso_code
|
||||
locale = COUNTRY_LOCALE_MAP.get(country) if country else None
|
||||
logger.debug(
|
||||
"GeoIP: %s → tz=%s, country=%s, locale=%s",
|
||||
ip, timezone, country, locale,
|
||||
)
|
||||
return timezone, locale
|
||||
except Exception as exc:
|
||||
logger.debug("GeoIP lookup failed for %s: %s", ip, exc)
|
||||
return None, None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Proxy IP resolution
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _resolve_proxy_ip(proxy_url: str) -> str | None:
|
||||
"""Extract proxy hostname from URL and resolve to an IP address."""
|
||||
try:
|
||||
hostname = urlparse(proxy_url).hostname
|
||||
if not hostname:
|
||||
return None
|
||||
|
||||
# Already a literal IP?
|
||||
try:
|
||||
socket.inet_pton(socket.AF_INET, hostname)
|
||||
return hostname
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
socket.inet_pton(socket.AF_INET6, hostname)
|
||||
return hostname
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
# DNS resolve (returns first result, handles both v4/v6)
|
||||
results = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
|
||||
if results:
|
||||
ip = results[0][4][0]
|
||||
logger.debug("Resolved proxy %s → %s", hostname, ip)
|
||||
return ip
|
||||
return None
|
||||
except Exception as exc:
|
||||
logger.debug("Failed to resolve proxy hostname: %s", exc)
|
||||
return None
|
||||
|
||||
|
||||
def _is_private_ip(ip: str) -> bool:
|
||||
"""Check if an IP address is private/internal (not routable on the internet)."""
|
||||
try:
|
||||
return ipaddress.ip_address(ip).is_private
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
# IP echo services — fast, no auth, return just the IP
|
||||
_IP_ECHO_URLS = [
|
||||
"https://api.ipify.org",
|
||||
"https://checkip.amazonaws.com",
|
||||
"https://ifconfig.me/ip",
|
||||
]
|
||||
|
||||
|
||||
def _resolve_exit_ip(proxy_url: str) -> str | None:
|
||||
"""Discover the proxy's actual exit IP by connecting through it."""
|
||||
import httpx
|
||||
|
||||
for url in _IP_ECHO_URLS:
|
||||
try:
|
||||
resp = httpx.get(url, proxy=proxy_url, timeout=10.0)
|
||||
resp.raise_for_status()
|
||||
ip = resp.text.strip()
|
||||
# Validate it looks like an IP
|
||||
ipaddress.ip_address(ip)
|
||||
logger.debug("Exit IP via %s: %s", url, ip)
|
||||
return ip
|
||||
except Exception:
|
||||
continue
|
||||
logger.debug("Failed to discover exit IP through proxy")
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GeoIP database management
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _get_geoip_dir() -> Path:
|
||||
from .config import get_cache_dir
|
||||
|
||||
return get_cache_dir() / "geoip"
|
||||
|
||||
|
||||
def _ensure_geoip_db() -> Path | None:
|
||||
"""Return path to GeoLite2-City.mmdb, downloading on first use."""
|
||||
db_path = _get_geoip_dir() / GEOIP_DB_FILENAME
|
||||
|
||||
if db_path.exists():
|
||||
_maybe_trigger_update(db_path)
|
||||
return db_path
|
||||
|
||||
try:
|
||||
_download_geoip_db(db_path)
|
||||
return db_path
|
||||
except Exception as exc:
|
||||
logger.warning("Failed to download GeoIP database: %s", exc)
|
||||
return None
|
||||
|
||||
|
||||
def _download_geoip_db(dest: Path) -> None:
|
||||
"""Atomic download of GeoLite2-City.mmdb via httpx."""
|
||||
import httpx
|
||||
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
logger.info("Downloading GeoIP database (~70 MB) …")
|
||||
|
||||
tmp_fd, tmp_name = tempfile.mkstemp(dir=dest.parent, suffix=".tmp")
|
||||
tmp_path = Path(tmp_name)
|
||||
try:
|
||||
with httpx.stream(
|
||||
"GET", GEOIP_DB_URL, follow_redirects=True, timeout=300.0
|
||||
) as resp:
|
||||
resp.raise_for_status()
|
||||
total = int(resp.headers.get("content-length", 0))
|
||||
downloaded = 0
|
||||
last_pct = -1
|
||||
with open(tmp_fd, "wb") as f:
|
||||
for chunk in resp.iter_bytes(chunk_size=65_536):
|
||||
f.write(chunk)
|
||||
downloaded += len(chunk)
|
||||
if total:
|
||||
pct = downloaded * 100 // total
|
||||
if pct >= last_pct + 10:
|
||||
last_pct = pct
|
||||
logger.info("GeoIP download: %d %%", pct)
|
||||
|
||||
tmp_path.rename(dest)
|
||||
logger.info("GeoIP database ready: %s", dest)
|
||||
except Exception:
|
||||
tmp_path.unlink(missing_ok=True)
|
||||
raise
|
||||
|
||||
|
||||
def _maybe_trigger_update(db_path: Path) -> None:
|
||||
"""Re-download in background if DB is older than 30 days."""
|
||||
try:
|
||||
age = time.time() - db_path.stat().st_mtime
|
||||
if age < GEOIP_UPDATE_INTERVAL:
|
||||
return
|
||||
except OSError:
|
||||
return
|
||||
|
||||
def _bg() -> None:
|
||||
try:
|
||||
_download_geoip_db(db_path)
|
||||
except Exception:
|
||||
logger.debug("Background GeoIP update failed", exc_info=True)
|
||||
|
||||
threading.Thread(target=_bg, daemon=True).start()
|
||||
Reference in New Issue
Block a user