"""Exit IP intelligence: geo, ASN, datacenter heuristics. Designed to be called through the local chain proxy so the lookup itself flows over the active path (no out-of-band leak). Uses ip-api.com (free, no key) with a fallback to ipwho.is. Both return geo + ASN. """ from __future__ import annotations import logging from dataclasses import dataclass from typing import Any import httpx log = logging.getLogger(__name__) # Common datacenter / hosting ASN keywords. Used as a quick heuristic to # warn the operator before signup (residential ASNs are far less likely to # trip Google/Proton anti-fraud than DC ranges). _DC_KEYWORDS = ( "amazon", "aws", "google", "microsoft", "azure", "digitalocean", "linode", "ovh", "hetzner", "vultr", "choopa", "scaleway", "leaseweb", "contabo", "hostinger", "godaddy", "namecheap", "cloudflare", "fastly", "akamai", "datacamp", "m247", "psychz", "quadranet", "colocrossing", "rackspace", "alibaba", "tencent", "huawei", "online s.a.s", "wholesale", "datacenter", "data center", "hosting", "server", "cloud", "vps", ) @dataclass class ExitIntel: ok: bool ip: str = "" country: str = "" country_code: str = "" city: str = "" region: str = "" timezone: str = "" lat: float | None = None lon: float | None = None asn: str = "" org: str = "" isp: str = "" is_datacenter: bool = False is_mobile: bool = False is_proxy_flagged: bool = False source: str = "" detail: str = "" def summary(self) -> str: if not self.ok: return f"unknown — {self.detail}" loc_bits = [b for b in (self.city, self.region, self.country) if b] loc = ", ".join(loc_bits) if loc_bits else "—" tags: list[str] = [] if self.is_datacenter: tags.append("DATACENTER") if self.is_mobile: tags.append("MOBILE") if self.is_proxy_flagged: tags.append("PROXY-FLAGGED") tag_str = f" [{' · '.join(tags)}]" if tags else "" asn_str = f" AS{self.asn}" if self.asn else "" return f"{self.ip} {loc}{asn_str}{tag_str}" def _looks_dc(org: str, isp: str) -> bool: haystack = f"{org} {isp}".lower() return any(k in haystack for k in _DC_KEYWORDS) def _parse_ipapi(payload: dict[str, Any]) -> ExitIntel: if (payload.get("status") or "").lower() != "success": return ExitIntel(ok=False, detail=str(payload.get("message") or "ip-api error")) asn_raw = str(payload.get("as") or "") asn = asn_raw.split()[0].lstrip("AS").strip() if asn_raw else "" org = str(payload.get("org") or payload.get("isp") or "") isp = str(payload.get("isp") or "") lat_raw, lon_raw = payload.get("lat"), payload.get("lon") return ExitIntel( ok=True, ip=str(payload.get("query") or ""), country=str(payload.get("country") or ""), country_code=str(payload.get("countryCode") or ""), city=str(payload.get("city") or ""), region=str(payload.get("regionName") or ""), timezone=str(payload.get("timezone") or ""), lat=float(lat_raw) if lat_raw is not None else None, lon=float(lon_raw) if lon_raw is not None else None, asn=asn, org=org, isp=isp, is_datacenter=bool(payload.get("hosting")) or _looks_dc(org, isp), is_mobile=bool(payload.get("mobile")), is_proxy_flagged=bool(payload.get("proxy")), source="ip-api.com", detail="ok", ) def _parse_ipwhois(payload: dict[str, Any]) -> ExitIntel: if not payload.get("success", True): return ExitIntel(ok=False, detail=str(payload.get("message") or "ipwho.is error")) conn = payload.get("connection") or {} asn_val = conn.get("asn") asn = str(asn_val) if asn_val is not None else "" org = str(conn.get("org") or "") isp = str(conn.get("isp") or "") lat_raw, lon_raw = payload.get("latitude"), payload.get("longitude") return ExitIntel( ok=True, ip=str(payload.get("ip") or ""), country=str(payload.get("country") or ""), country_code=str(payload.get("country_code") or ""), city=str(payload.get("city") or ""), region=str(payload.get("region") or ""), timezone=str((payload.get("timezone") or {}).get("id") or ""), lat=float(lat_raw) if lat_raw is not None else None, lon=float(lon_raw) if lon_raw is not None else None, asn=asn, org=org, isp=isp, is_datacenter=_looks_dc(org, isp), is_mobile=False, is_proxy_flagged=False, source="ipwho.is", detail="ok", ) _IPAPI_FIELDS = ( "status,message,country,countryCode,regionName,city,lat,lon,timezone," "isp,org,as,mobile,proxy,hosting,query" ) def _fetch_intel( *, proxy_url: str | None, timeout_seconds: float, ) -> ExitIntel: """Shared ip-api / ipwho.is lookup. ``proxy_url=None`` uses a direct connection.""" timeout = httpx.Timeout(timeout_seconds, connect=min(6.0, timeout_seconds)) sources = [ (f"http://ip-api.com/json/?fields={_IPAPI_FIELDS}", _parse_ipapi), ("https://ipwho.is/", _parse_ipwhois), ] last_err = "" try: with httpx.Client( proxy=proxy_url, timeout=timeout, verify=False, follow_redirects=True, headers={"User-Agent": "Mozilla/5.0"}, ) as c: for url, parser in sources: try: r = c.get(url) if r.status_code != 200 or not r.content: last_err = f"{url} -> HTTP {r.status_code}" continue data = r.json() out = parser(data) if out.ok: return out last_err = out.detail except Exception as e: last_err = f"{type(e).__name__}: {e}" continue except Exception as e: return ExitIntel(ok=False, detail=f"{type(e).__name__}: {e}") return ExitIntel(ok=False, detail=last_err or "no source responded") def fetch_exit_intel(proxy_url: str, timeout_seconds: float = 12.0) -> ExitIntel: """Look up exit IP geo/ASN/datacenter through the given proxy. Returns an ExitIntel with ok=False and detail set on failure. Never raises. """ return _fetch_intel(proxy_url=proxy_url, timeout_seconds=timeout_seconds) def fetch_ip_geo(ip: str, timeout_seconds: float = 8.0) -> ExitIntel: """Direct geo lookup for a specific IP (used for hop / origin mapping). Never raises. """ target = (ip or "").strip() if not target: return ExitIntel(ok=False, detail="empty ip") timeout = httpx.Timeout(timeout_seconds, connect=min(5.0, timeout_seconds)) sources = [ ( f"http://ip-api.com/json/{target}?fields={_IPAPI_FIELDS}", _parse_ipapi, ), (f"https://ipwho.is/{target}", _parse_ipwhois), ] last_err = "" try: with httpx.Client( timeout=timeout, verify=False, follow_redirects=True, headers={"User-Agent": "Mozilla/5.0"}, ) as c: for url, parser in sources: try: r = c.get(url) if r.status_code != 200 or not r.content: last_err = f"{url} -> HTTP {r.status_code}" continue data = r.json() out = parser(data) if out.ok: return out last_err = out.detail except Exception as e: last_err = f"{type(e).__name__}: {e}" continue except Exception as e: return ExitIntel(ok=False, detail=f"{type(e).__name__}: {e}") return ExitIntel(ok=False, detail=last_err or "no source responded")