"""GeoIP lookup helpers for Squid Web Manager. Two backends are supported: 1. **Online** — public free `ip-api.com` JSON endpoint (no API key, 45 requests/min from the same source IP). This is the default because it requires no extra files and no extra dependencies. 2. **Offline** — a local MaxMind GeoLite2-City ``.mmdb`` file. We only use the optional ``maxminddb`` package if it's installed; otherwise we report an error rather than failing the page render. Both backends write to a small in-process dict cache so the same IP is never looked up twice within ``CACHE_TTL`` seconds (default 24 h). Private IPs (RFC 1918 / loopback / link-local) are short-circuited and return a GeoIPResult with ``error="private IP"`` so they never hit the network or DB. """ from __future__ import annotations import ipaddress import json import socket import time import urllib.error import urllib.request from typing import Any # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- #: In-process cache. Maps ``ip`` -> ``(GeoIPResult, ts)``. CACHE: dict[str, tuple["GeoIPResult", float]] = {} #: Cache entry lifetime, in seconds. Geo data is stable for ~weeks. CACHE_TTL = 86400 # 24h #: ip-api.com endpoint. Field list pulled to a minimum to stay polite. _API_FIELDS = ( "status,message,country,countryCode,region,regionName,city," "lat,lon,isp,org,as,query" ) _API_URL = "http://ip-api.com/json/{ip}?fields={fields}" #: Per-request timeout for the online endpoint. DEFAULT_TIMEOUT = 5 #: User agent — ip-api.com prefers something identifying. _USER_AGENT = "squid-web-manager/1.0 (geoip-lookup)" # --------------------------------------------------------------------------- # Result container # --------------------------------------------------------------------------- class GeoIPResult: """Lightweight GeoIP result. Plain attributes; safe to JSON-serialize.""" __slots__ = ( "ip", "country", "country_name", "city", "region", "latitude", "longitude", "asn", "isp", "error", ) def __init__( self, ip: str = "", country: str = "", country_name: str = "", city: str = "", region: str = "", latitude: float = 0.0, longitude: float = 0.0, asn: str = "", isp: str = "", error: str = "", ) -> None: self.ip = ip self.country = country self.country_name = country_name self.city = city self.region = region self.latitude = latitude self.longitude = longitude self.asn = asn self.isp = isp self.error = error # -- helpers ----------------------------------------------------------- def ok(self) -> bool: return not self.error and self.latitude != 0.0 and self.longitude != 0.0 def to_dict(self) -> dict[str, Any]: return { "ip": self.ip, "country": self.country, "country_name": self.country_name, "city": self.city, "region": self.region, "latitude": self.latitude, "longitude": self.longitude, "asn": self.asn, "isp": self.isp, "error": self.error, } def __repr__(self) -> str: if self.error: return f"GeoIPResult({self.ip!r}, error={self.error!r})" return ( f"GeoIPResult({self.ip!r}, {self.country_name!r}, " f"{self.city!r}, ({self.latitude}, {self.longitude}))" ) # --------------------------------------------------------------------------- # Private-IP classification (RFC 1918 + loopback + link-local) # --------------------------------------------------------------------------- def _is_private_ip(ip: str) -> bool: """Return True for RFC 1918 / loopback / link-local addresses.""" if not ip: return True # cheap pre-filter so we don't pay ipaddress.parse for garbage head = ip.split(".", 1)[0] if "." in ip else "" if head in ("127", "10"): return True try: addr = ipaddress.ip_address(ip) except ValueError: # not an IP at all (could be a hostname) — treat as "not private" return False return bool( addr.is_private or addr.is_loopback or addr.is_link_local or addr.is_multicast or addr.is_reserved or addr.is_unspecified ) # --------------------------------------------------------------------------- # Online backend — ip-api.com # --------------------------------------------------------------------------- def lookup_online(ip: str, timeout: int = DEFAULT_TIMEOUT) -> GeoIPResult: """Look up ``ip`` against the public ip-api.com endpoint. Returns a :class:`GeoIPResult` whose ``error`` attribute is set on any failure (network error, bad JSON, private IP, rate-limit…). Never raises. """ result = GeoIPResult(ip=ip) if not ip: result.error = "empty ip" return result if _is_private_ip(ip): result.error = "private IP" return result url = _API_URL.format(ip=ip, fields=_API_FIELDS) req = urllib.request.Request( url, headers={ "User-Agent": _USER_AGENT, "Accept": "application/json", }, ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read() payload = json.loads(raw.decode("utf-8", errors="replace")) except urllib.error.HTTPError as e: # 429 = rate limited result.error = f"http {e.code} {e.reason}" return result except urllib.error.URLError as e: result.error = f"url error: {e.reason}" return result except (socket.timeout, TimeoutError): result.error = "timeout" return result except json.JSONDecodeError as e: result.error = f"json decode: {e.msg}" return result except Exception as e: # noqa: BLE001 result.error = f"unexpected: {type(e).__name__}: {e}" return result if not isinstance(payload, dict): result.error = "unexpected payload shape" return result if payload.get("status") == "fail": result.error = payload.get("message") or "lookup failed" return result result.country = (payload.get("countryCode") or "").strip() result.country_name = (payload.get("country") or "").strip() result.city = (payload.get("city") or "").strip() result.region = (payload.get("regionName") or "").strip() try: result.latitude = float(payload.get("lat") or 0.0) except (TypeError, ValueError): result.latitude = 0.0 try: result.longitude = float(payload.get("lon") or 0.0) except (TypeError, ValueError): result.longitude = 0.0 result.asn = (payload.get("as") or "").split()[0] if payload.get("as") else "" result.isp = (payload.get("isp") or payload.get("org") or "").strip() return result # --------------------------------------------------------------------------- # Offline backend — MaxMind GeoLite2-City .mmdb # --------------------------------------------------------------------------- def lookup_offline(ip: str, db_path: str = "") -> GeoIPResult: """Look up ``ip`` against a local MaxMind GeoLite2-City ``.mmdb``. Requires the optional ``maxminddb`` package; if it's not installed, the function returns a result with ``error="maxminddb not installed"`` rather than raising. """ result = GeoIPResult(ip=ip) if not ip: result.error = "empty ip" return result if _is_private_ip(ip): result.error = "private IP" return result if not db_path: result.error = "no db_path provided" return result try: import maxminddb # type: ignore except ImportError: result.error = "maxminddb not installed" return result try: reader = maxminddb.open_database(db_path) except Exception as e: # noqa: BLE001 result.error = f"cannot open db: {e}" return result try: record = reader.get(ip) except Exception as e: # noqa: BLE001 reader.close() result.error = f"db error: {e}" return result reader.close() if not record: result.error = "no record" return result country = record.get("country") or {} iso = country.get("iso_code") or "" name = (country.get("names") or {}).get("en") or "" city = ((record.get("city") or {}).get("names") or {}).get("en") or "" sub = ((record.get("subdivisions") or [{}])[0]) if record.get("subdivisions") else {} region = (sub.get("names") or {}).get("en") or "" loc = record.get("location") or {} try: lat = float(loc.get("latitude") or 0.0) lon = float(loc.get("longitude") or 0.0) except (TypeError, ValueError): lat = lon = 0.0 result.country = iso result.country_name = name result.city = city result.region = region result.latitude = lat result.longitude = lon return result # --------------------------------------------------------------------------- # Cached unified entry point # --------------------------------------------------------------------------- def _cache_get(ip: str) -> GeoIPResult | None: entry = CACHE.get(ip) if not entry: return None result, ts = entry if (time.time() - ts) > CACHE_TTL: CACHE.pop(ip, None) return None return result def _cache_put(result: GeoIPResult) -> None: if not result.ip: return CACHE[result.ip] = (result, time.time()) def lookup( ip: str, *, db_path: str = "", use_online: bool = True, ) -> GeoIPResult: """Unified GeoIP entry point. Returns a cached :class:`GeoIPResult` if present; otherwise performs an online or offline lookup depending on ``use_online``. - Private IPs short-circuit and are cached too (cheap; also lets us avoid repeatedly network-attempting them). - Failures are cached as well — but only briefly. Negative caching for 60 s is plenty. """ cached = _cache_get(ip) if cached is not None: return cached if _is_private_ip(ip): result = GeoIPResult(ip=ip, error="private IP") _cache_put(result) return result if use_online: result = lookup_online(ip) elif db_path: result = lookup_offline(ip, db_path) else: result = GeoIPResult(ip=ip, error="no backend") _cache_put(result) return result def cache_clear() -> int: """Clear the in-process cache. Returns the number of entries purged.""" n = len(CACHE) CACHE.clear() return n def cache_stats() -> dict[str, int]: """Return a small snapshot of the cache for diagnostics pages.""" now = time.time() total = len(CACHE) fresh = 0 stale = 0 for _ip, (_, ts) in CACHE.items(): if (now - ts) <= CACHE_TTL: fresh += 1 else: stale += 1 return {"total": total, "fresh": fresh, "stale": stale} __all__ = [ "GeoIPResult", "CACHE", "CACHE_TTL", "lookup", "lookup_online", "lookup_offline", "cache_clear", "cache_stats", ]