125 lines
4.4 KiB
Python
Executable File
125 lines
4.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""IP geolocation, cached with a TTL.
|
|
|
|
Discovers the public IP by tracerouting to 1.1.1.1 and taking the first
|
|
globally-routable hop (the ISP egress nearest the user), then resolves that IP to a
|
|
location through a public geolocation API. Falls back to locating this host's own
|
|
public IP when traceroute is unavailable or yields no public hop.
|
|
|
|
Prints JSON: {lat, lon, city, country, ip, source}. Diagnostics go to stderr,
|
|
non-zero exit on total failure. Used by services/location.py (and runnable
|
|
standalone for testing).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
CACHE = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) / "astro-menu" / "location.json"
|
|
TTL = 1800 # seconds
|
|
|
|
TARGET = "1.1.1.1"
|
|
_IPV4 = re.compile(r"\b(\d{1,3}(?:\.\d{1,3}){3})\b")
|
|
|
|
# Geolocation providers. Each builds a URL for a given IP; an empty IP asks the
|
|
# provider to resolve the caller's own public address (the self-IP fallback).
|
|
PROVIDERS = [
|
|
("ip-api",
|
|
lambda ip: f"http://ip-api.com/json/{ip}?fields=lat,lon,city,country,query",
|
|
lambda d: {"lat": d["lat"], "lon": d["lon"], "city": d.get("city"),
|
|
"country": d.get("country"), "ip": d.get("query")}),
|
|
("ipapi.co",
|
|
lambda ip: f"https://ipapi.co/{ip}/json/" if ip else "https://ipapi.co/json/",
|
|
lambda d: {"lat": d["latitude"], "lon": d["longitude"], "city": d.get("city"),
|
|
"country": d.get("country_name"), "ip": d.get("ip")}),
|
|
("ipinfo",
|
|
lambda ip: f"https://ipinfo.io/{ip}/json" if ip else "https://ipinfo.io/json",
|
|
lambda d: {"lat": float(d["loc"].split(",")[0]), "lon": float(d["loc"].split(",")[1]),
|
|
"city": d.get("city"), "country": d.get("country"), "ip": d.get("ip")}),
|
|
]
|
|
|
|
|
|
def _cached() -> dict | None:
|
|
try:
|
|
blob = json.loads(CACHE.read_text())
|
|
if time.time() - blob.get("_ts", 0) < TTL:
|
|
return blob["data"]
|
|
except (FileNotFoundError, json.JSONDecodeError, KeyError):
|
|
pass
|
|
return None
|
|
|
|
|
|
def _store(data: dict) -> None:
|
|
CACHE.parent.mkdir(parents=True, exist_ok=True)
|
|
CACHE.write_text(json.dumps({"_ts": time.time(), "data": data}))
|
|
|
|
|
|
def _public_hop_ip() -> str | None:
|
|
"""First globally-routable hop on the path to 1.1.1.1 — i.e. the ISP egress
|
|
closest to the user. The private/CGNAT hops before it and the anycast target
|
|
itself (Cloudflare, useless for locating the user) are skipped."""
|
|
try:
|
|
proc = subprocess.run(
|
|
["traceroute", "-n", "-q", "1", "-w", "2", "-m", "12", TARGET],
|
|
capture_output=True, text=True, timeout=40,
|
|
)
|
|
except (FileNotFoundError, subprocess.SubprocessError) as exc:
|
|
print(f"traceroute: {exc}", file=sys.stderr)
|
|
return None
|
|
for line in proc.stdout.splitlines():
|
|
if line.lower().startswith("traceroute to"):
|
|
continue # header line names the target; not a hop
|
|
for ip in _IPV4.findall(line):
|
|
if ip == TARGET:
|
|
continue
|
|
try:
|
|
if ipaddress.ip_address(ip).is_global:
|
|
return ip
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _fetch(url: str) -> dict:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "astro-menu/1.0"})
|
|
with urllib.request.urlopen(req, timeout=8) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
|
|
def main() -> int:
|
|
if "--no-cache" not in sys.argv:
|
|
hit = _cached()
|
|
if hit:
|
|
print(json.dumps(hit))
|
|
return 0
|
|
|
|
hop = _public_hop_ip()
|
|
# Try the traced public hop first, then fall back to our own public IP.
|
|
candidates = ([hop] if hop else []) + [""]
|
|
for ip in candidates:
|
|
for name, url_of, parse in PROVIDERS:
|
|
try:
|
|
data = parse(_fetch(url_of(ip)))
|
|
if data.get("lat") is None or data.get("lon") is None:
|
|
raise ValueError("no coordinates")
|
|
data["source"] = f"{name} via {ip}" if ip else name
|
|
_store(data)
|
|
print(json.dumps(data))
|
|
return 0
|
|
except Exception as exc: # noqa: BLE001 — try the next provider/candidate
|
|
print(f"{name}({ip or 'self'}): {exc}", file=sys.stderr)
|
|
print("all geolocation attempts failed", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|