From 23e4b0b0196830b099ba541b058063db73fff3a4 Mon Sep 17 00:00:00 2001 From: Indiana Holmes Date: Tue, 19 May 2026 15:43:18 -0700 Subject: [PATCH] Add Live tab world chain map with hop geo arcs. Plots YOU through each hop to EXIT on an equirectangular map with great-circle connectors, extends exit intel with lat/lon, and auto-refreshes on chain rotation. Co-authored-by: Cursor --- proxy_chain_manager/app.py | 99 +++++++++- proxy_chain_manager/chain_map.py | 303 ++++++++++++++++++++++++++++++ proxy_chain_manager/exit_intel.py | 80 +++++++- tests/test_proxy_god.py | 76 ++++++++ 4 files changed, 547 insertions(+), 11 deletions(-) create mode 100644 proxy_chain_manager/chain_map.py diff --git a/proxy_chain_manager/app.py b/proxy_chain_manager/app.py index 95e5fb9..45a2a9f 100644 --- a/proxy_chain_manager/app.py +++ b/proxy_chain_manager/app.py @@ -1,6 +1,7 @@ """ Proxy God — GUI Tabs: Live | Chain Builder | Browser | Ban Tester | Signup Prep | Privacy | Settings +(Live tab includes exit intel + world chain map) """ from __future__ import annotations @@ -40,6 +41,7 @@ from .firewall import ( from .artifact_wipe import wipe_artifacts from .ban_tester import BanTestResult, run_ban_tests, test_single_site from .exit_intel import ExitIntel, fetch_exit_intel +from .chain_map import build_chain_map_points, render_chain_map, resize_map_display from .signup_prep import ( SIGNUP_PRESET_KEYS, SIGNUP_PRESETS, @@ -139,8 +141,8 @@ def main() -> None: root = ctk.CTk() root.title("Proxy God v2") - root.geometry("1080x760") - root.minsize(900, 620) + root.geometry("1080x860") + root.minsize(900, 720) root.configure(fg_color=BG) # ── 3D-ish neon header banner (drawn on a Canvas for a true gradient) ──── @@ -539,6 +541,7 @@ def main() -> None: color = GREEN v_intel_tags.set("Flags: " + " · ".join(tags)) intel_tags_lbl.configure(text_color=color) + root.after(200, lambda: _refresh_chain_map(silent=True)) def _refresh_exit_intel(silent: bool = False) -> None: if intel_running[0]: @@ -577,6 +580,92 @@ def main() -> None: _btn(intel_btn_row, "Refresh intel", lambda: _refresh_exit_intel(False), w=110, h=24, fg_color=ACCENT2, hover_color=ACCENT).pack(side="left") + # ── Chain world map ─────────────────────────────────────────────────────── + map_card = ctk.CTkFrame(tab_live, fg_color=CARD, corner_radius=8) + map_card.pack(fill="x", pady=(0, 6)) + + map_head = ctk.CTkFrame(map_card, fg_color="transparent") + map_head.pack(fill="x", padx=10, pady=(8, 2)) + ctk.CTkLabel( + map_head, text="Chain map", font=(FONT, 11, "bold"), text_color=CYAN, + ).pack(side="left") + v_map_status = ctk.StringVar(value="Plotting: waiting for chain…") + ctk.CTkLabel( + map_head, textvariable=v_map_status, font=(FONT, 10), text_color=TEXT2, + ).pack(side="right") + + map_frame = ctk.CTkFrame(map_card, fg_color=BG, corner_radius=6) + map_frame.pack(fill="x", padx=10, pady=(0, 8)) + map_img_label = ctk.CTkLabel(map_frame, text="") + map_img_label.pack(padx=2, pady=2) + + map_ctk_image: dict[str, ctk.CTkImage | None] = {"img": None} + last_chain_state: dict[str, Any] = { + "hops": [], + "status": "idle", + "exit_ip": "", + "direct_ip": "", + } + map_running = [False] + + def _show_chain_map(pil_img: Any) -> None: + try: + map_frame.update_idletasks() + w = max(640, map_frame.winfo_width() - 8) + h = max(180, min(240, int(w * 0.23))) + resized = resize_map_display(pil_img, w, h) + map_ctk_image["img"] = ctk.CTkImage( + light_image=resized, dark_image=resized, size=(w, h), + ) + map_img_label.configure(image=map_ctk_image["img"], text="") + except Exception as e: + logging.getLogger(__name__).debug("chain map display: %s", e) + + def _refresh_chain_map(silent: bool = True) -> None: + if map_running[0]: + return + hops = list(last_chain_state["hops"]) + status = str(last_chain_state["status"] or "idle") + if not service_running[0] or not hops: + v_map_status.set("Plotting: chain not running") + _show_chain_map(render_chain_map([], status="idle")) + return + map_running[0] = True + v_map_status.set("Plotting: resolving hop locations…") + direct_ip = str(last_chain_state.get("direct_ip") or "") + exit_ip = str(last_chain_state.get("exit_ip") or "") + exit_intel = last_intel.get("intel") + timeout = min(12.0, svc.settings.validation_timeout_seconds + 2.0) + + def work() -> None: + points = build_chain_map_points( + hops, + direct_ip=direct_ip, + exit_ip=exit_ip, + exit_intel=exit_intel if isinstance(exit_intel, ExitIntel) else None, + timeout_seconds=timeout, + ) + img = render_chain_map(points, width=960, height=220, status=status) # type: ignore[arg-type] + + def done() -> None: + _show_chain_map(img) + if points: + v_map_status.set(f"Plotting: {len(points)} nodes · {status}") + if not silent: + _log(f"Chain map updated — {len(points)} geo nodes ({status}).") + else: + v_map_status.set("Plotting: no geo data yet (hop DNS/geo pending)") + map_running[0] = False + + root.after(0, done) + + threading.Thread(target=work, daemon=True).start() + + _btn(map_head, "Refresh map", lambda: _refresh_chain_map(False), w=100, h=22, + fg_color=ACCENT2, hover_color=ACCENT).pack(side="right", padx=(8, 0)) + + _show_chain_map(render_chain_map([], status="idle")) + # Log box log_frame = ctk.CTkFrame(tab_live, fg_color=CARD, corner_radius=8) log_frame.pack(fill="both", expand=True) @@ -2385,6 +2474,8 @@ def main() -> None: ip = str(m.get("ip", "—")) your_ip_lbl.configure(text=f"Direct IP: {ip}") v_real_ip.set(ip) + last_chain_state["direct_ip"] = ip + root.after(300, lambda: _refresh_chain_map(silent=True)) elif t == "vpn": nonlocal outer_hop_label @@ -2447,6 +2538,10 @@ def main() -> None: tray.set_state(tray_color, exit_ip=exit_s) prev_exit = v_exit_ip.get() v_exit_ip.set(exit_s if exit_s else "—") + last_chain_state["hops"] = hops + last_chain_state["status"] = status + last_chain_state["exit_ip"] = exit_s or "" + root.after(250, lambda: _refresh_chain_map(silent=True)) if status == "healthy" and exit_s and exit_s != prev_exit: root.after(400, lambda: _refresh_exit_intel(silent=True)) elif status != "healthy": diff --git a/proxy_chain_manager/chain_map.py b/proxy_chain_manager/chain_map.py new file mode 100644 index 0000000..cac1688 --- /dev/null +++ b/proxy_chain_manager/chain_map.py @@ -0,0 +1,303 @@ +"""World-map visualization for proxy chains (Pillow + equirectangular projection).""" +from __future__ import annotations + +import ipaddress +import logging +import math +import socket +from dataclasses import dataclass +from typing import Literal +from urllib.parse import urlparse + +from PIL import Image, ImageDraw, ImageFont + +try: + _RESAMPLE = Image.Resampling.LANCZOS +except AttributeError: + _RESAMPLE = Image.LANCZOS # type: ignore[attr-defined] + +from .config import normalize_proxy_url, redact_proxy_url +from .exit_intel import ExitIntel, fetch_ip_geo + +log = logging.getLogger(__name__) + +ChainStatus = Literal["healthy", "connecting", "dead", "idle"] + +# Theme colors (match app.py palette) +_BG = "#070713" +_GRID = "#141432" +_LAND = "#12182e" +_LAND_EDGE = "#1e2a4a" +_LINE_HEALTHY = "#00e5ff" +_LINE_CONNECTING = "#ffd000" +_LINE_DEAD = "#ff2a6d" +_LINE_IDLE = "#3052ff" +_YOU = "#00ff9c" +_HOP = "#00e5ff" +_EXIT = "#ffd000" +_LABEL = "#eaf2ff" +_LABEL_DIM = "#7a8aab" +_UNKNOWN = "#2a2f4a" + + +@dataclass(frozen=True) +class MapPoint: + role: Literal["you", "hop", "exit"] + label: str + lat: float + lon: float + detail: str = "" + hop_index: int = -1 + + +def extract_hop_host(proxy_url: str) -> str | None: + """Return hostname or IP from a proxy URL.""" + p = urlparse(normalize_proxy_url(proxy_url)) + host = (p.hostname or "").strip() + return host or None + + +def resolve_host_ip(host: str, timeout_seconds: float = 4.0) -> str | None: + """Resolve host to IPv4/IPv6 string. Never raises.""" + h = (host or "").strip() + if not h: + return None + try: + ipaddress.ip_address(h) + return h + except ValueError: + pass + try: + prev = socket.getdefaulttimeout() + socket.setdefaulttimeout(max(1.0, float(timeout_seconds))) + try: + infos = socket.getaddrinfo(h, None, type=socket.SOCK_STREAM) + finally: + socket.setdefaulttimeout(prev) + for info in infos: + addr = info[4][0] + try: + ipaddress.ip_address(addr) + return addr + except ValueError: + continue + except Exception as e: + log.debug("resolve_host_ip(%s): %s", h, e) + return None + + +def _loc_label(intel: ExitIntel) -> str: + bits = [b for b in (intel.city, intel.region, intel.country) if b] + return ", ".join(bits) if bits else intel.country_code or intel.ip or "—" + + +def _intel_to_point( + role: Literal["you", "hop", "exit"], + label: str, + intel: ExitIntel, + *, + hop_index: int = -1, +) -> MapPoint | None: + if not intel.ok or intel.lat is None or intel.lon is None: + return None + return MapPoint( + role=role, + label=label, + lat=float(intel.lat), + lon=float(intel.lon), + detail=_loc_label(intel), + hop_index=hop_index, + ) + + +def build_chain_map_points( + hops: list[str], + *, + direct_ip: str = "", + exit_ip: str = "", + exit_intel: ExitIntel | None = None, + timeout_seconds: float = 8.0, +) -> list[MapPoint]: + """Resolve geo for YOU → hop₁ → … → EXIT. Never raises.""" + points: list[MapPoint] = [] + + origin_ip = (direct_ip or "").strip() + if origin_ip and origin_ip != "—": + you = _intel_to_point("you", "YOU", fetch_ip_geo(origin_ip, timeout_seconds)) + if you: + points.append(you) + + for i, hop in enumerate(hops): + host = extract_hop_host(hop) + if not host: + continue + ip = resolve_host_ip(host, timeout_seconds=min(4.0, timeout_seconds)) + if not ip: + continue + short = redact_proxy_url(hop) + if len(short) > 22: + short = short[:20] + "…" + intel = fetch_ip_geo(ip, timeout_seconds) + pt = _intel_to_point("hop", f"H{i + 1}", intel, hop_index=i) + if pt: + points.append(pt) + + exit = (exit_ip or "").strip() + if exit and exit != "—": + if ( + exit_intel + and exit_intel.ok + and exit_intel.ip == exit + and exit_intel.lat is not None + and exit_intel.lon is not None + ): + pt = _intel_to_point("exit", "EXIT", exit_intel) + else: + pt = _intel_to_point("exit", "EXIT", fetch_ip_geo(exit, timeout_seconds)) + if pt: + points.append(pt) + + return points + + +def _project(lat: float, lon: float, width: int, height: int, margin: int) -> tuple[float, float]: + inner_w = max(1, width - margin * 2) + inner_h = max(1, height - margin * 2) + x = margin + (float(lon) + 180.0) / 360.0 * inner_w + y = margin + (90.0 - float(lat)) / 180.0 * inner_h + return x, y + + +def _great_circle_points( + lat1: float, + lon1: float, + lat2: float, + lon2: float, + steps: int = 24, +) -> list[tuple[float, float]]: + """Interpolate along a great circle for curved hop arcs.""" + phi1, lam1 = math.radians(lat1), math.radians(lon1) + phi2, lam2 = math.radians(lat2), math.radians(lon2) + d = 2 * math.asin( + math.sqrt( + math.sin((phi2 - phi1) / 2) ** 2 + + math.cos(phi1) * math.cos(phi2) * math.sin((lam2 - lam1) / 2) ** 2 + ) + ) + if d < 1e-9: + return [(lat1, lon1), (lat2, lon2)] + out: list[tuple[float, float]] = [] + for i in range(steps + 1): + f = i / steps + a = math.sin((1 - f) * d) / math.sin(d) + b = math.sin(f * d) / math.sin(d) + x = a * math.cos(phi1) * math.cos(lam1) + b * math.cos(phi2) * math.cos(lam2) + y = a * math.cos(phi1) * math.sin(lam1) + b * math.cos(phi2) * math.sin(lam2) + z = a * math.sin(phi1) + b * math.sin(phi2) + out.append((math.degrees(math.atan2(z, math.sqrt(x * x + y * y))), math.degrees(math.atan2(y, x)))) + return out + + +def _line_color(status: ChainStatus) -> str: + return { + "healthy": _LINE_HEALTHY, + "connecting": _LINE_CONNECTING, + "dead": _LINE_DEAD, + "idle": _LINE_IDLE, + }.get(status, _LINE_IDLE) + + +def _node_color(role: str) -> str: + return {"you": _YOU, "hop": _HOP, "exit": _EXIT}.get(role, _HOP) + + +def _load_font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont: + for name in ("segoeui.ttf", "arial.ttf", "DejaVuSans.ttf"): + try: + return ImageFont.truetype(name, size) + except OSError: + continue + return ImageFont.load_default() + + +def _draw_landmasses(draw: ImageDraw.ImageDraw, width: int, height: int, margin: int) -> None: + """Minimal continent silhouettes — stylized, not survey-accurate.""" + blobs: list[list[tuple[float, float]]] = [ + # North America + [(-168, 72), (-140, 75), (-100, 72), (-80, 50), (-75, 25), (-95, 15), (-110, 20), (-125, 48), (-168, 72)], + # South America + [(-82, 12), (-72, -5), (-55, -35), (-65, -55), (-75, -18), (-82, 12)], + # Europe / Africa + [(-10, 72), (30, 70), (40, 55), (35, 30), (20, 5), (-5, 5), (-18, 28), (-10, 72)], + [(15, 35), (35, 32), (50, 12), (42, -5), (18, -35), (15, 35)], + # Asia + [(40, 72), (100, 75), (140, 55), (130, 35), (110, 10), (80, 8), (60, 25), (40, 45), (40, 72)], + # Australia + [(115, -12), (135, -12), (150, -25), (145, -38), (115, -38), (115, -12)], + ] + for poly in blobs: + pts = [_project(lat, lon, width, height, margin) for lat, lon in poly] + draw.polygon(pts, fill=_LAND, outline=_LAND_EDGE) + + +def render_chain_map( + points: list[MapPoint], + *, + width: int = 960, + height: int = 220, + status: ChainStatus = "idle", + margin: int = 18, +) -> Image.Image: + """Render chain hops on a dark equirectangular world map.""" + img = Image.new("RGB", (max(320, width), max(120, height)), _BG) + draw = ImageDraw.Draw(img) + _draw_landmasses(draw, width, height, margin) + + # Lat/lon grid + for lat in range(-60, 91, 30): + y = _project(lat, 0, width, height, margin)[1] + draw.line([(margin, y), (width - margin, y)], fill=_GRID, width=1) + for lon in range(-150, 181, 30): + x = _project(0, lon, width, height, margin)[0] + draw.line([(x, margin), (x, height - margin)], fill=_GRID, width=1) + + if len(points) < 2: + font = _load_font(13) + msg = "Start the chain to plot hops on the map." if not points else "Need 2+ geo points to draw path." + draw.text((margin, height // 2 - 8), msg, fill=_LABEL_DIM, font=font) + return img + + line_c = _line_color(status) + projected = [_project(p.lat, p.lon, width, height, margin) for p in points] + + # Glow underlay + for i in range(len(projected) - 1): + arc = _great_circle_points(points[i].lat, points[i].lon, points[i + 1].lat, points[i + 1].lon) + arc_xy = [_project(lat, lon, width, height, margin) for lat, lon in arc] + draw.line(arc_xy, fill=line_c, width=5) + for i in range(len(projected) - 1): + arc = _great_circle_points(points[i].lat, points[i].lon, points[i + 1].lat, points[i + 1].lon) + arc_xy = [_project(lat, lon, width, height, margin) for lat, lon in arc] + draw.line(arc_xy, fill=line_c, width=2) + + label_font = _load_font(11) + detail_font = _load_font(9) + for pt, (x, y) in zip(points, projected): + color = _node_color(pt.role) + r = 7 if pt.role == "exit" else (6 if pt.role == "you" else 5) + draw.ellipse((x - r - 2, y - r - 2, x + r + 2, y + r + 2), fill=color) + draw.ellipse((x - r, y - r, x + r, y + r), fill=_BG, outline=color, width=2) + tx, ty = x + 10, y - 14 + draw.text((tx + 1, ty + 1), pt.label, fill="#000000", font=label_font) + draw.text((tx, ty), pt.label, fill=_LABEL, font=label_font) + if pt.detail: + draw.text((tx, ty + 13), pt.detail[:28], fill=_LABEL_DIM, font=detail_font) + + return img + + +def resize_map_display(pil_img: Image.Image, width: int, height: int) -> Image.Image: + """Resize rendered map for the GUI widget.""" + w = max(320, int(width)) + h = max(120, int(height)) + return pil_img.resize((w, h), _RESAMPLE) diff --git a/proxy_chain_manager/exit_intel.py b/proxy_chain_manager/exit_intel.py index c149e3a..857d9d3 100644 --- a/proxy_chain_manager/exit_intel.py +++ b/proxy_chain_manager/exit_intel.py @@ -38,6 +38,8 @@ class ExitIntel: city: str = "" region: str = "" timezone: str = "" + lat: float | None = None + lon: float | None = None asn: str = "" org: str = "" isp: str = "" @@ -76,6 +78,7 @@ def _parse_ipapi(payload: dict[str, Any]) -> ExitIntel: 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 ""), @@ -84,6 +87,8 @@ def _parse_ipapi(payload: dict[str, Any]) -> ExitIntel: 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, @@ -103,6 +108,7 @@ def _parse_ipwhois(payload: dict[str, Any]) -> ExitIntel: 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 ""), @@ -111,6 +117,8 @@ def _parse_ipwhois(payload: dict[str, Any]) -> ExitIntel: 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, @@ -122,18 +130,21 @@ def _parse_ipwhois(payload: dict[str, Any]) -> ExitIntel: ) -def fetch_exit_intel(proxy_url: str, timeout_seconds: float = 12.0) -> ExitIntel: - """Look up exit IP geo/ASN/datacenter through the given proxy. +_IPAPI_FIELDS = ( + "status,message,country,countryCode,regionName,city,lat,lon,timezone," + "isp,org,as,mobile,proxy,hosting,query" +) - Returns an ExitIntel with ok=False and detail set on failure. Never raises. - """ + +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 = [ - ( - "http://ip-api.com/json/?fields=status,message,country,countryCode," - "regionName,city,timezone,isp,org,as,mobile,proxy,hosting,query", - _parse_ipapi, - ), + (f"http://ip-api.com/json/?fields={_IPAPI_FIELDS}", _parse_ipapi), ("https://ipwho.is/", _parse_ipwhois), ] last_err = "" @@ -162,3 +173,54 @@ def fetch_exit_intel(proxy_url: str, timeout_seconds: float = 12.0) -> ExitIntel 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") diff --git a/tests/test_proxy_god.py b/tests/test_proxy_god.py index fcd0251..9cdc65a 100644 --- a/tests/test_proxy_god.py +++ b/tests/test_proxy_god.py @@ -344,5 +344,81 @@ class TestLeakAuditPure(unittest.TestCase): self.assertEqual(len(rep.findings), 2) +class TestExitIntelGeo(unittest.TestCase): + def test_parse_ipapi_lat_lon(self) -> None: + from proxy_chain_manager.exit_intel import _parse_ipapi + + out = _parse_ipapi( + { + "status": "success", + "query": "8.8.8.8", + "country": "United States", + "countryCode": "US", + "regionName": "Virginia", + "city": "Ashburn", + "lat": 39.03, + "lon": -77.5, + "timezone": "America/New_York", + "isp": "Google LLC", + "org": "Google Public DNS", + "as": "AS15169 Google LLC", + "mobile": False, + "proxy": False, + "hosting": True, + } + ) + self.assertTrue(out.ok) + self.assertAlmostEqual(out.lat or 0, 39.03) + self.assertAlmostEqual(out.lon or 0, -77.5) + + def test_parse_ipwhois_lat_lon(self) -> None: + from proxy_chain_manager.exit_intel import _parse_ipwhois + + out = _parse_ipwhois( + { + "success": True, + "ip": "1.1.1.1", + "country": "Australia", + "country_code": "AU", + "city": "Sydney", + "region": "New South Wales", + "latitude": -33.86, + "longitude": 151.2, + "timezone": {"id": "Australia/Sydney"}, + "connection": {"asn": 13335, "org": "Cloudflare", "isp": "Cloudflare"}, + } + ) + self.assertTrue(out.ok) + self.assertAlmostEqual(out.lat or 0, -33.86) + self.assertAlmostEqual(out.lon or 0, 151.2) + + +class TestChainMap(unittest.TestCase): + def test_extract_hop_host(self) -> None: + from proxy_chain_manager.chain_map import extract_hop_host + + self.assertEqual(extract_hop_host("http://203.0.113.10:8080"), "203.0.113.10") + self.assertEqual(extract_hop_host("socks5://proxy.example:1080"), "proxy.example") + self.assertIsNone(extract_hop_host("")) + + def test_project_corners(self) -> None: + from proxy_chain_manager.chain_map import _project + + x, y = _project(0, 0, 360, 180, 0) + self.assertAlmostEqual(x, 180.0) + self.assertAlmostEqual(y, 90.0) + + def test_render_chain_map_with_points(self) -> None: + from proxy_chain_manager.chain_map import MapPoint, render_chain_map + + pts = [ + MapPoint("you", "YOU", 40.0, -74.0, detail="NYC"), + MapPoint("hop", "H1", 51.5, -0.1, detail="London"), + MapPoint("exit", "EXIT", 35.6, 139.7, detail="Tokyo"), + ] + img = render_chain_map(pts, width=640, height=200, status="healthy") + self.assertEqual(img.size, (640, 200)) + + if __name__ == "__main__": unittest.main()