CloakBrowser/cloakbrowser/geoip.py

"""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 math
import os
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
DEFAULT_GEOIP_TIMEOUT_SECONDS = 5.0
GEOIP_TIMEOUT_ENV = "CLOAKBROWSER_GEOIP_TIMEOUT_SECONDS"

# 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.
    """
    tz, locale, _ip = resolve_proxy_geo_with_ip(proxy_url)
    return tz, locale


def resolve_proxy_geo_with_ip(
    proxy_url: str,
) -> tuple[str | None, str | None, str | None]:
    """Resolve timezone, locale, and exit IP from a proxy.

    Returns ``(timezone, locale, exit_ip)``.  The exit IP is a free bonus
    from the lookup — reused for WebRTC spoofing without an extra HTTP call.
    """
    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, None

    timeout = _get_geoip_timeout_seconds()
    deadline = _deadline_from_timeout(timeout)

    # Exit IP (through proxy) is most accurate — gateway DNS may differ from exit
    ip = _resolve_exit_ip(proxy_url, timeout=_remaining_seconds(deadline))
    if ip is None and not _deadline_expired(deadline):
        ip = _resolve_proxy_ip(proxy_url)
    if ip is None or _deadline_expired(deadline):
        if deadline is not None and _deadline_expired(deadline):
            logger.warning("GeoIP resolution timed out after %.1fs; continuing without GeoIP", timeout)
        return None, 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, ip
    except Exception as exc:
        logger.warning("GeoIP lookup failed for %s: %s", ip, exc)
        return None, None, ip


# ---------------------------------------------------------------------------
# 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.warning("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 _get_geoip_timeout_seconds() -> float:
    raw = os.getenv(GEOIP_TIMEOUT_ENV)
    if not raw:
        return DEFAULT_GEOIP_TIMEOUT_SECONDS
    try:
        timeout = float(raw)
    except ValueError:
        timeout = float("nan")
    if not math.isfinite(timeout):
        logger.warning(
            "Invalid %s=%r; using %.1fs",
            GEOIP_TIMEOUT_ENV,
            raw,
            DEFAULT_GEOIP_TIMEOUT_SECONDS,
        )
        return DEFAULT_GEOIP_TIMEOUT_SECONDS
    return max(timeout, 0.0)


def _deadline_from_timeout(timeout: float) -> float | None:
    if timeout <= 0:
        return None
    return time.monotonic() + timeout


def _remaining_seconds(deadline: float | None) -> float | None:
    if deadline is None:
        return None
    return max(deadline - time.monotonic(), 0.0)


def _deadline_expired(deadline: float | None) -> bool:
    return deadline is not None and time.monotonic() >= deadline


def resolve_proxy_exit_ip(proxy_url: str) -> str | None:
    """Resolve only the proxy exit IP, bounded by the GeoIP timeout."""
    timeout = _get_geoip_timeout_seconds()
    deadline = _deadline_from_timeout(timeout)
    ip = _resolve_exit_ip(proxy_url, timeout=timeout)
    if ip is None and _deadline_expired(deadline):
        logger.warning("GeoIP resolution timed out after %.1fs; continuing without GeoIP", timeout)
    return ip


def _resolve_exit_ip(proxy_url: str, timeout: float | None = None) -> str | None:
    """Discover the proxy's actual exit IP by connecting through it."""
    import httpx

    deadline = _deadline_from_timeout(timeout or 0)

    for url in _IP_ECHO_URLS:
        try:
            remaining = _remaining_seconds(deadline)
            if remaining is not None and remaining <= 0:
                return None
            request_timeout = min(10.0, remaining) if remaining is not None else 10.0
            resp = httpx.get(url, proxy=proxy_url, timeout=request_timeout)
            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 httpx.UnsupportedProtocol:
            logger.warning(
                "SOCKS5 proxy requires socksio: pip install cloakbrowser[geoip]"
            )
            return None
        except Exception:
            continue
    logger.warning("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()