diff --git a/ProxyChainManager.spec b/ProxyChainManager.spec index eb6b0a7..aa40cd8 100644 --- a/ProxyChainManager.spec +++ b/ProxyChainManager.spec @@ -1,7 +1,12 @@ # -*- mode: python ; coding: utf-8 -*- from PyInstaller.utils.hooks import collect_all -datas = [] +datas = [ + # Signup autofill WebExtension (manifest + content.js) + ('proxy_chain_manager/signup_extension', 'proxy_chain_manager/signup_extension'), + # Neon world-map background image for the chain map widget + ('proxy_chain_manager/world_map.png', 'proxy_chain_manager'), +] binaries = [] hiddenimports = ['pystray._win32'] tmp_ret = collect_all('customtkinter') diff --git a/proxy_chain_manager/app.py b/proxy_chain_manager/app.py index 45a2a9f..a28e827 100644 --- a/proxy_chain_manager/app.py +++ b/proxy_chain_manager/app.py @@ -39,7 +39,7 @@ from .firewall import ( request_admin_relaunch, ) from .artifact_wipe import wipe_artifacts -from .ban_tester import BanTestResult, run_ban_tests, test_single_site +from .ban_tester import BanTestResult, SITE_CATEGORIES, 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 ( @@ -64,8 +64,14 @@ from .browser_identity import ( PERSONA_LABELS, PERSONAS, ) -from .dns_leak import check_dns_leak_hint, flush_dns_cache +from .dns_leak import ( + FullDnsLeakReport, + check_dns_leak_hint, + flush_dns_cache, + run_dns_leak_test, +) from .leak_audit import AuditReport, run_audit_sync +from .webrtc_check import WebRtcCheckResult, run_webrtc_check from .browser_launcher import ( BrowserConfig, BrowserSession, @@ -1240,20 +1246,49 @@ def main() -> None: persona_menu.pack(side="left", padx=(4, 0)) cookie_row = ctk.CTkFrame(persona_card, fg_color="transparent") - cookie_row.pack(fill="x", padx=12, pady=(0, 10)) + cookie_row.pack(fill="x", padx=12, pady=(0, 4)) ctk.CTkLabel(cookie_row, text="Cookies:", font=(FONT, 11), text_color=TEXT, width=80, anchor="w").pack(side="left") - cookie_var = ctk.StringVar( - value=COOKIE_LABELS.get(s.browser_cookie_mode, COOKIE_LABELS["block_third_party"]) + _default_cookie_label = COOKIE_LABELS.get( + s.browser_cookie_mode, COOKIE_LABELS["block_third_party"] ) + cookie_var = ctk.StringVar(value=_default_cookie_label) + + # Cookie detail sub-labels (key → short description of side effects) + _COOKIE_NOTES: dict[str, str] = { + "accept_all": "Everything works · cookies persist to disk", + "accept_all_session": "Full compat while browsing · cookies wiped on close", + "block_third_party": "Blocks cross-site trackers · most sites work · cookies persist", + "block_third_party_session": "Blocks 3rd-party + nothing persists after close", + "block_third_party_trackers":"ETP Strict — same as Firefox enhanced tracking protection", + "block_social_trackers": "Blocks Facebook pixel, Twitter widgets, social SSO badges", + "block_unvisited": "Very light — only blocks sites you haven't visited", + "partitioned_tcp": "Total Cookie Protection — each site gets isolated 3rd-party jar", + "partitioned_session": "Isolated jars + full disk wipe on close · zero cross-site state", + "session_only": "All cookies expire on close · 1st-party allowed while browsing", + "ghost_mode": "Isolated + session + wipes cache/history/localStorage/formdata", + "block_all": "Breaks most logins · read-only / scraping use only", + } + + cookie_note_lbl = ctk.CTkLabel( + persona_card, text=_COOKIE_NOTES.get(s.browser_cookie_mode, ""), + font=(FONT, 9), text_color=TEXT2, anchor="w", wraplength=820, + ) + + def _on_cookie_change(_choice: str = "") -> None: + key = _cookie_key() + cookie_note_lbl.configure(text=_COOKIE_NOTES.get(key, "")) + cookie_menu = ctk.CTkOptionMenu( cookie_row, values=[COOKIE_LABELS[k] for k in COOKIE_MODES], variable=cookie_var, fg_color=BG, button_color=ACCENT2, button_hover_color=ACCENT, - text_color=TEXT, font=(FONT, 11), width=500, + text_color=TEXT, font=(FONT, 11), width=680, + command=_on_cookie_change, ) cookie_menu.pack(side="left", padx=(4, 0)) + cookie_note_lbl.pack(anchor="w", padx=(92, 12), pady=(0, 10)) def _persona_key() -> str: for k in PERSONAS: @@ -1415,68 +1450,81 @@ def main() -> None: ban_head = ctk.CTkFrame(ban_scroll, fg_color=CARD, corner_radius=10, border_width=1, border_color=GLOW) ban_head.pack(fill="x", padx=8, pady=(6, 8)) ctk.CTkLabel( - ban_head, - text="Ban List Tester", - font=(FONT, 13, "bold"), - text_color=CYAN, + ban_head, text="Ban / Reachability Tester", + font=(FONT, 13, "bold"), text_color=CYAN, ).pack(anchor="w", padx=12, pady=(10, 2)) ctk.CTkLabel( ban_head, - text="Tests high-traffic sites through your ACTIVE chain to detect likely exit-IP bans.", - font=(FONT, 10), - text_color=TEXT2, - ).pack(anchor="w", padx=12, pady=(0, 8)) + text=( + "Tests social, shopping, crypto exchanges, and DNS providers through your ACTIVE chain. " + "Detects exit-IP bans and geo-blocks." + ), + font=(FONT, 10), text_color=TEXT2, wraplength=880, justify="left", + ).pack(anchor="w", padx=12, pady=(0, 6)) + + # Category selector + ban_cat_row = ctk.CTkFrame(ban_head, fg_color="transparent") + ban_cat_row.pack(fill="x", padx=12, pady=(0, 6)) + ctk.CTkLabel(ban_cat_row, text="Category:", font=(FONT, 10), text_color=TEXT2, width=70, + anchor="w").pack(side="left") + _ban_cat_keys = list(SITE_CATEGORIES.keys()) # "Social / General", "Shopping", etc. + ban_cat_var = ctk.StringVar(value="All") + ban_cat_menu = ctk.CTkOptionMenu( + ban_cat_row, values=_ban_cat_keys, variable=ban_cat_var, + fg_color=BG, button_color=ACCENT2, button_hover_color=ACCENT, + text_color=TEXT, width=200, + ) + ban_cat_menu.pack(side="left", padx=(0, 8)) ban_status_var = ctk.StringVar(value="Status: idle") ctk.CTkLabel(ban_head, textvariable=ban_status_var, font=(FONT, 11, "bold"), text_color=TEXT).pack( - anchor="w", padx=12, pady=(0, 8) + anchor="w", padx=12, pady=(0, 6) ) ban_results_frame = ctk.CTkFrame( - ban_scroll, - fg_color=CARD, - corner_radius=10, - border_width=1, - border_color=GLOW, + ban_scroll, fg_color=CARD, corner_radius=10, border_width=1, border_color=GLOW, ) ban_results_frame.pack(fill="both", expand=True, padx=8, pady=(0, 8)) ban_rows_wrap = ctk.CTkFrame(ban_results_frame, fg_color="transparent") ban_rows_wrap.pack(fill="both", expand=True, padx=8, pady=8) - def _ban_row(site: str, status: str, detail: str, color: str) -> None: + def _ban_row(site: str, status: str, detail: str, color: str, cat_header: str = "") -> None: + if cat_header: + ctk.CTkLabel( + ban_rows_wrap, text=f" {cat_header}", + font=(FONT, 10, "bold"), text_color=CYAN, anchor="w", + ).pack(fill="x", padx=4, pady=(6, 1)) row = ctk.CTkFrame(ban_rows_wrap, fg_color=PANEL, corner_radius=6, height=30) row.pack(fill="x", pady=2) row.pack_propagate(False) - ctk.CTkLabel(row, text=site, font=(FONT, 11, "bold"), text_color=TEXT, width=140, anchor="w").pack( - side="left", padx=(8, 4) - ) - ctk.CTkLabel(row, text=status, font=(FONT, 10, "bold"), text_color=color, width=80).pack( - side="left", padx=4 - ) - ctk.CTkLabel(row, text=detail, font=("Consolas", 10), text_color=TEXT2, anchor="w").pack( - side="left", fill="x", expand=True, padx=(6, 8) - ) + ctk.CTkLabel(row, text=site, font=(FONT, 11, "bold"), text_color=TEXT, + width=160, anchor="w").pack(side="left", padx=(8, 4)) + ctk.CTkLabel(row, text=status, font=(FONT, 10, "bold"), text_color=color, + width=72).pack(side="left", padx=4) + ctk.CTkLabel(row, text=detail, font=("Consolas", 10), text_color=TEXT2, + anchor="w").pack(side="left", fill="x", expand=True, padx=(6, 8)) def _render_ban_results(results: list[BanTestResult]) -> None: - for w in list(ban_rows_wrap.winfo_children()): - w.destroy() + for child in list(ban_rows_wrap.winfo_children()): + child.destroy() if not results: - ctk.CTkLabel( - ban_rows_wrap, - text="No results yet.", - font=(FONT, 11), - text_color=TEXT2, - ).pack(anchor="w", padx=8, pady=8) + ctk.CTkLabel(ban_rows_wrap, text="No results yet.", font=(FONT, 11), + text_color=TEXT2).pack(anchor="w", padx=8, pady=8) return + seen_cats: set[str] = set() for r in results: + cat = r.category or "" + header = cat if cat and cat not in seen_cats else "" + if cat: + seen_cats.add(cat) if r.status == "ok": st, color = "OK", GREEN elif r.status == "banned": st, color = "BANNED", ORANGE else: st, color = "ERROR", RED - _ban_row(r.site, st, r.detail, color) + _ban_row(r.site, st, r.detail, color, cat_header=header) _render_ban_results([]) @@ -1489,23 +1537,24 @@ def main() -> None: _log("Ban tester: start the chain first, then run tests.") return proxy_url = f"http://{svc.settings.listen_addr()}" + chosen_cat = ban_cat_var.get() + # "All" means every category + cats = None if chosen_cat == "All" else [chosen_cat] ban_status_var.set("Status: running…") ban_start_btn.configure(state="disabled") _render_ban_results([]) def work() -> None: timeout = min(18.0, max(8.0, float(svc.settings.validation_timeout_seconds) + 3.0)) - results = run_ban_tests(proxy_url, timeout_seconds=timeout) - ok_n = sum(1 for r in results if r.status == "ok") + results = run_ban_tests(proxy_url, timeout_seconds=timeout, categories=cats) + ok_n = sum(1 for r in results if r.status == "ok") banned_n = sum(1 for r in results if r.status == "banned") - err_n = len(results) - ok_n - banned_n + err_n = len(results) - ok_n - banned_n def done() -> None: _render_ban_results(results) - ban_status_var.set(f"Status: done | ok={ok_n} banned={banned_n} errors={err_n}") - _log( - f"Ban tester done via {proxy_url} — ok={ok_n}, banned={banned_n}, errors={err_n}." - ) + ban_status_var.set(f"Status: done — ok={ok_n} banned={banned_n} errors={err_n}") + _log(f"Ban tester done via {proxy_url} — ok={ok_n}, banned={banned_n}, errors={err_n}.") ban_start_btn.configure(state="normal") root.after(0, done) @@ -1513,13 +1562,8 @@ def main() -> None: threading.Thread(target=work, daemon=True).start() ban_start_btn = _btn( - ban_action, - "Start ban test", - _start_ban_test, - w=120, - h=30, - fg_color=ACCENT2, - hover_color=ACCENT, + ban_action, "Start test", _start_ban_test, + w=120, h=30, fg_color=ACCENT2, hover_color=ACCENT, ) ban_start_btn.pack(side="left") @@ -1561,11 +1605,12 @@ def main() -> None: signup_site_labels[0], ) ) - signup_site_menu = ctk.CTkOptionMenu( - site_row, values=signup_site_labels, variable=signup_site_var, - fg_color=BG, button_color=ACCENT2, button_hover_color=ACCENT, text_color=TEXT, width=260, - ) - signup_site_menu.pack(side="left", padx=(0, 8)) + custom_url_row = ctk.CTkFrame(signup_head, fg_color="transparent") + custom_url_row.pack(fill="x", padx=12, pady=(0, 4)) + ctk.CTkLabel(custom_url_row, text="URL", width=70, anchor="w", font=(FONT, 10), text_color=TEXT2).pack(side="left") + signup_custom_url = ctk.CTkEntry(custom_url_row, height=28, fg_color=BG, border_color=ACCENT) + signup_custom_url.pack(side="left", fill="x", expand=True) + signup_custom_url.insert(0, signup_draft.custom_url or "") def _signup_site_key() -> str: label = signup_site_var.get() @@ -1574,12 +1619,22 @@ def main() -> None: return signup_site_keys[i] return signup_site_keys[0] - custom_url_row = ctk.CTkFrame(signup_head, fg_color="transparent") - custom_url_row.pack(fill="x", padx=12, pady=(0, 4)) - ctk.CTkLabel(custom_url_row, text="Custom URL", width=70, anchor="w", font=(FONT, 10), text_color=TEXT2).pack(side="left") - signup_custom_url = ctk.CTkEntry(custom_url_row, height=28, fg_color=BG, border_color=ACCENT) - signup_custom_url.pack(side="left", fill="x", expand=True) - signup_custom_url.insert(0, signup_draft.custom_url or "") + def _on_signup_site_change(_choice: str = "") -> None: + key = _signup_site_key() + preset = SIGNUP_PRESETS.get(key) + if preset and preset.url: + signup_custom_url.delete(0, "end") + signup_custom_url.insert(0, preset.url) + + signup_site_menu = ctk.CTkOptionMenu( + site_row, values=signup_site_labels, variable=signup_site_var, + fg_color=BG, button_color=ACCENT2, button_hover_color=ACCENT, text_color=TEXT, width=260, + command=_on_signup_site_change, + ) + signup_site_menu.pack(side="left", padx=(0, 8)) + + # Populate URL field for the initially-selected site + _on_signup_site_change() def _signup_field(parent: Any, label: str, show: str = "") -> ctk.CTkEntry: row = ctk.CTkFrame(parent, fg_color="transparent") @@ -1695,34 +1750,48 @@ def main() -> None: # ── Pre-flight checklist ───────────────────────────────────────────────── preflight_card = _signup_card( "Pre-flight check", - "Verifies chain, target site reachability, and exit-IP intel before you sign up.", + "Exhaustive readiness check: chain, exit IP, geo/ASN, DNS leak, WebRTC, " + "IPv6, fingerprint, target site reachability, and direct-IP comparison.", ) pf_rows: dict[str, dict[str, ctk.StringVar | ctk.CTkLabel]] = {} + _PF_CHECKS = [ + ("chain", "Chain running"), + ("exit_ip", "Exit IP (via chain)"), + ("direct_ip", "Direct IP (no chain)"), + ("leak_compare","IP leak check"), + ("intel", "Geo / ASN intel"), + ("datacenter", "Residential ASN"), + ("dns", "DNS routing"), + ("ipv6", "IPv6 binding"), + ("webrtc", "WebRTC policy"), + ("https_tunnel","HTTPS tunnel (CONNECT)"), + ("target", "Target site reachable"), + ("firefox", "Firefox path"), + ] + def _pf_row(key: str, label: str) -> None: row = ctk.CTkFrame(preflight_card, fg_color=PANEL, corner_radius=6) row.pack(fill="x", padx=12, pady=2) dot = ctk.CTkLabel(row, text="●", font=(FONT, 14), text_color=DIM, width=18) dot.pack(side="left", padx=(8, 4)) ctk.CTkLabel(row, text=label, font=(FONT, 11, "bold"), text_color=TEXT, - width=160, anchor="w").pack(side="left") + width=180, anchor="w").pack(side="left") var = ctk.StringVar(value="not run") ctk.CTkLabel(row, textvariable=var, font=(FONT, 10), text_color=TEXT2, anchor="w").pack(side="left", fill="x", expand=True, padx=(6, 8), pady=4) pf_rows[key] = {"var": var, "dot": dot} - _pf_row("chain", "Chain running") - _pf_row("exit_ip", "Exit IP") - _pf_row("intel", "IP intel (geo/ASN)") - _pf_row("target", "Target site reachable") + for _k, _lbl in _PF_CHECKS: + _pf_row(_k, _lbl) def _pf_set(key: str, status: str, detail: str) -> None: r = pf_rows.get(key) if not r: return color = {"ok": GREEN, "warn": ORANGE, "fail": RED, "running": YELLOW}.get(status, DIM) - r["dot"].configure(text_color=color) - r["var"].set(detail) + r["dot"].configure(text_color=color) # type: ignore[union-attr] + r["var"].set(detail) # type: ignore[union-attr] pf_summary_var = ctk.StringVar(value="Not run.") ctk.CTkLabel(preflight_card, textvariable=pf_summary_var, font=(FONT, 10, "bold"), @@ -1739,72 +1808,189 @@ def main() -> None: pf_summary_var.set("Pick a site or enter a custom URL first.") return pf_running[0] = True - for k in ("chain", "exit_ip", "intel", "target"): + for k, _ in _PF_CHECKS: _pf_set(k, "running", "checking…") pf_summary_var.set("Running pre-flight…") + # Immediate sync checks ────────────────────────────────────────────── chain_ok = service_running[0] and bool(svc.current_chain) - if not chain_ok: - _pf_set("chain", "fail", "chain not running") - _pf_set("exit_ip", "fail", "n/a") - _pf_set("intel", "fail", "n/a") - _pf_set("target", "fail", "n/a") - pf_summary_var.set("Start the chain, then re-run pre-flight.") - pf_running[0] = False - return - _pf_set("chain", "ok", f"{len(svc.current_chain)} hop(s) up") + if chain_ok: + _pf_set("chain", "ok", f"{len(svc.current_chain)} hop(s) active") + else: + _pf_set("chain", "fail", "chain not running — start first") + + # Firefox path check + ff_path = firefox_path_var.get().strip() or default_firefox_path() + from pathlib import Path as _Path + if ff_path and _Path(ff_path).is_file(): + _pf_set("firefox", "ok", ff_path) + else: + _pf_set("firefox", "fail", f"not found: {ff_path or '(no path set)'}") + proxy_url = f"http://{svc.settings.listen_addr()}" target_host = urlparse(url).hostname or "(unknown)" def work() -> None: timeout = min(15.0, max(8.0, float(svc.settings.validation_timeout_seconds) + 2.0)) - intel = fetch_exit_intel(proxy_url, timeout_seconds=timeout) - site_res = test_single_site(proxy_url, target_host, url, timeout_seconds=timeout) + warns: list[str] = [] - def done() -> None: - ip = (v_exit_ip.get() or "—").strip() - if ip and ip != "—": - _pf_set("exit_ip", "ok", ip) + # ── 1. Direct IP (bypass chain) ────────────────────────────────── + from .validator import get_direct_ip as _get_direct_ip + import asyncio as _asyncio + try: + direct_ip = _asyncio.run(_get_direct_ip(svc.settings.ip_check_url, timeout / 2)) + except Exception: + direct_ip = None + + def _set_direct(ip: str | None) -> None: + if ip: + _pf_set("direct_ip", "ok", ip) else: - _pf_set("exit_ip", "warn", "no IP reported yet") - if intel.ok: - _apply_intel(intel) - if intel.is_proxy_flagged: - _pf_set("intel", "fail", f"proxy-flagged · {intel.org or intel.isp}") - elif intel.is_datacenter: - _pf_set("intel", "warn", f"datacenter ASN · {intel.org or intel.isp}") + _pf_set("direct_ip", "warn", "could not reach check URL directly") + + root.after(0, lambda ip=direct_ip: _set_direct(ip)) + + # ── 2. Exit IP via chain ───────────────────────────────────────── + if chain_ok: + intel = fetch_exit_intel(proxy_url, timeout_seconds=timeout) + exit_ip_str = intel.ip if intel.ok else (v_exit_ip.get() or "").strip() + + def _set_exit(ip_: str, intel_: Any) -> None: + if ip_ and ip_ not in ("—", ""): + _pf_set("exit_ip", "ok", ip_) else: - loc = ", ".join([b for b in (intel.city, intel.country) if b]) or "—" - _pf_set("intel", "ok", f"{loc} · AS{intel.asn} {intel.org or intel.isp}") - else: - _pf_set("intel", "warn", intel.detail or "lookup failed") - if site_res.status == "ok": - _pf_set("target", "ok", f"{target_host} {site_res.detail}") - elif site_res.status == "banned": - _pf_set("target", "fail", f"likely banned · {site_res.detail}") - else: - _pf_set("target", "warn", site_res.detail) + _pf_set("exit_ip", "warn", "no exit IP yet") + # IP leak compare + if direct_ip and ip_ and ip_ not in ("—", ""): + if direct_ip == ip_: + _pf_set("leak_compare", "fail", + f"exit == direct ({ip_}) — chain not working!") + warns.append("exit IP matches direct IP (chain not routing)") + else: + _pf_set("leak_compare", "ok", + f"exit {ip_} ≠ direct {direct_ip} ✓") + else: + _pf_set("leak_compare", "warn", "cannot compare — missing IPs") + # Geo/ASN + if intel_.ok: + _apply_intel(intel_) + loc = ", ".join([b for b in (intel_.city, intel_.country) if b]) or "—" + asn_str = f"AS{intel_.asn}" if intel_.asn else "" + if intel_.is_proxy_flagged: + _pf_set("intel", "fail", f"proxy-flagged · {intel_.org or intel_.isp}") + warns.append("exit IP is proxy-flagged") + elif intel_.is_datacenter: + _pf_set("intel", "warn", f"datacenter · {loc} {asn_str}") + _pf_set("datacenter", "warn", f"DC ASN — {intel_.org or intel_.isp}") + warns.append("exit IP is datacenter ASN") + else: + _pf_set("intel", "ok", f"{loc} {asn_str}") + _pf_set("datacenter", "ok", f"residential/ISP — {intel_.org or intel_.isp}") + else: + _pf_set("intel", "warn", intel_.detail or "lookup failed") + _pf_set("datacenter", "warn", "intel unavailable") - warns: list[str] = [] - if intel.ok and intel.is_proxy_flagged: - warns.append("exit IP is proxy-flagged") - if intel.ok and intel.is_datacenter: - warns.append("exit IP is a datacenter ASN") - if site_res.status == "banned": - warns.append("target site likely blocks this IP") - if not warns: - pf_summary_var.set("All checks passed — good to sign up.") - else: - pf_summary_var.set("Warnings: " + "; ".join(warns)) + root.after(0, lambda i=exit_ip_str, d=intel: _set_exit(i, d)) + else: + for k in ("exit_ip", "leak_compare", "intel", "datacenter"): + root.after(0, lambda k_=k: _pf_set(k_, "fail", "chain not running")) + + # ── 3. DNS routing check ───────────────────────────────────────── + try: + dns_rep = run_dns_leak_test( + proxy_url=proxy_url if chain_ok else None, + timeout=min(10.0, timeout), + ) + def _set_dns(r: Any) -> None: + if r.leaked: + _pf_set("dns", "fail", r.summary[:80]) + warns.append("DNS leak detected") + elif r.has_external_resolver: + _pf_set("dns", "warn", r.summary[:80]) + else: + _pf_set("dns", "ok", r.summary[:80]) + root.after(0, lambda r_=dns_rep: _set_dns(r_)) + except Exception as e: + root.after(0, lambda: _pf_set("dns", "warn", f"check error: {e}")) + + # ── 4. IPv6 binding ────────────────────────────────────────────── + try: + from .leak_audit import _check_ipv6_active + v6_on, v6_msg = _check_ipv6_active() + def _set_v6(on_: bool, msg_: str) -> None: + if on_: + _pf_set("ipv6", "warn", f"IPv6 active: {msg_}") + warns.append("IPv6 enabled (possible leak path)") + else: + _pf_set("ipv6", "ok", msg_) + root.after(0, lambda on=v6_on, msg=v6_msg: _set_v6(on, msg)) + except Exception as e: + root.after(0, lambda: _pf_set("ipv6", "warn", str(e))) + + # ── 5. WebRTC ──────────────────────────────────────────────────── + try: + profile = _Path(firefox_profile_var.get().strip() or default_profile_dir()) + rtc_rep = run_webrtc_check(profile_dir=profile) + def _set_rtc(r: Any) -> None: + if r.any_leak_risk: + issues_short = "; ".join(r.issues[:2])[:70] + _pf_set("webrtc", "warn", issues_short) + else: + _pf_set("webrtc", "ok", "all surfaces protected") + root.after(0, lambda r_=rtc_rep: _set_rtc(r_)) + except Exception as e: + root.after(0, lambda: _pf_set("webrtc", "warn", str(e))) + + # ── 6. HTTPS tunnel probe ───────────────────────────────────────── + if chain_ok: + try: + from .validator import check_https_tunnel + import asyncio as _as2 + tun_ok, tun_msg = _as2.run( + check_https_tunnel(proxy_url, timeout) + ) + def _set_tun(ok_: bool, msg_: str) -> None: + if ok_: + _pf_set("https_tunnel", "ok", msg_) + else: + _pf_set("https_tunnel", "fail", msg_) + warns.append("HTTPS CONNECT tunnel broken") + root.after(0, lambda ok=tun_ok, msg=tun_msg: _set_tun(ok, msg)) + except Exception as e: + root.after(0, lambda: _pf_set("https_tunnel", "warn", str(e))) + else: + root.after(0, lambda: _pf_set("https_tunnel", "fail", "chain not running")) + + # ── 7. Target site reachability ─────────────────────────────────── + if chain_ok: + site_res = test_single_site(proxy_url, target_host, url, timeout_seconds=timeout) + def _set_site(r: Any) -> None: + if r.status == "ok": + _pf_set("target", "ok", f"{target_host} — {r.detail}") + elif r.status == "banned": + _pf_set("target", "fail", f"likely BANNED · {r.detail}") + warns.append("target site likely blocks this exit IP") + else: + _pf_set("target", "warn", r.detail) + root.after(0, lambda r_=site_res: _set_site(r_)) + else: + root.after(0, lambda: _pf_set("target", "fail", "chain not running")) + + # ── Final summary ───────────────────────────────────────────────── + def _final() -> None: pf_running[0] = False + if warns: + pf_summary_var.set(f"⚠ {len(warns)} warning(s): " + " · ".join(warns[:3])) + else: + pf_summary_var.set("✓ All pre-flight checks passed — good to go.") - root.after(0, done) + root.after(200, _final) threading.Thread(target=work, daemon=True).start() pf_btn_row = ctk.CTkFrame(preflight_card, fg_color="transparent") pf_btn_row.pack(fill="x", padx=12, pady=(0, 10)) - _btn(pf_btn_row, "Run pre-flight", _run_preflight, w=130, h=28, + _btn(pf_btn_row, "Run pre-flight", _run_preflight, w=140, h=28, fg_color=ACCENT2, hover_color=ACCENT).pack(side="left") signup_status_var = ctk.StringVar(value="Ready — start chain, then Open & Autofill.") @@ -1824,19 +2010,46 @@ def main() -> None: accounts_list.pack(fill="x", padx=12, pady=(8, 10)) def _refresh_accounts_ui() -> None: - for w in list(accounts_list.winfo_children()): - w.destroy() + for child in list(accounts_list.winfo_children()): + child.destroy() rows = load_accounts() if not rows: - ctk.CTkLabel(accounts_list, text="No saved accounts yet.", font=(FONT, 10), text_color=TEXT2).pack( - anchor="w", padx=8, pady=8 - ) + ctk.CTkLabel(accounts_list, text="No saved accounts yet.", font=(FONT, 10), + text_color=TEXT2).pack(anchor="w", padx=8, pady=8) return - for rec in rows[:40]: - line = f"{rec.created_at[:16]} {rec.site} {rec.email} exit={rec.exit_ip or '—'}" - ctk.CTkLabel( - accounts_list, text=line, font=("Consolas", 10), text_color=TEXT, anchor="w", - ).pack(anchor="w", padx=8, pady=2) + for rec in rows[:60]: + row_frame = ctk.CTkFrame(accounts_list, fg_color=PANEL, corner_radius=6) + row_frame.pack(fill="x", pady=2, padx=4) + + info_lbl = ctk.CTkLabel( + row_frame, + text=f"{rec.created_at[:16]} {rec.site} {rec.email or rec.username or '—'} exit={rec.exit_ip or '—'}", + font=("Consolas", 10), text_color=TEXT, anchor="w", + ) + info_lbl.pack(side="left", fill="x", expand=True, padx=(8, 4), pady=4) + + def _copy_email(r: Any = rec) -> None: + v = r.email or r.username or "" + root.clipboard_clear() + root.clipboard_append(v) + _log(f"Copied email/user: {v}") + + def _copy_pass(r: Any = rec) -> None: + root.clipboard_clear() + root.clipboard_append(r.password or "") + _log(f"Copied password for {r.email or r.username or r.site}.") + + ctk.CTkButton( + row_frame, text="Copy Email", width=80, height=22, + font=(FONT, 10), fg_color=ACCENT2, hover_color=ACCENT, + command=_copy_email, + ).pack(side="right", padx=(2, 4), pady=4) + ctk.CTkButton( + row_frame, text="Copy Pass", width=76, height=22, + font=(FONT, 10), fg_color="#1e3a1e", hover_color="#2e5a2e", + text_color=GREEN, + command=_copy_pass, + ).pack(side="right", padx=(2, 2), pady=4) def _current_exit_meta() -> tuple[str, str]: ip = (v_exit_ip.get() or "—").strip() @@ -1909,6 +2122,8 @@ def main() -> None: tag = f" [{' · '.join(extras)}]" if extras else "" signup_status_var.set(f"Opened {label}{tag} — autofill active (you submit + captcha).") _log(f"Signup prep: opened {url} with autofill extension{tag}.") + # Auto-save entry so the account is recorded even if user forgets to save manually + _save_signup_account() else: signup_status_var.set("Browser launch failed.") svc.set_sticky(0) @@ -2100,33 +2315,124 @@ def main() -> None: _btn(fp_card, "Refresh fingerprint audit", _refresh_fingerprint, w=180, h=28).pack( anchor="w", padx=12, pady=(0, 10)) - dns_card = _priv_section("DNS", "Checks whether public resolvers may bypass your chain.") - dns_result_lbl = ctk.CTkLabel(dns_card, text="Run a check to see DNS configuration.", - font=(FONT, 11), text_color=TEXT2, wraplength=880, justify="left") - dns_result_lbl.pack(anchor="w", padx=12, pady=8) + # ── DNS leak test ──────────────────────────────────────────────────────── + dns_card = _priv_section( + "DNS leak test", + "Compares how DNS resolves through your chain vs direct. Detects resolver bypass.", + ) + dns_box = ctk.CTkTextbox( + dns_card, height=120, font=("Consolas", 10), + fg_color=BG, text_color=TEXT, scrollbar_button_color=ACCENT, + ) + dns_box.pack(fill="x", padx=12, pady=(4, 4)) + dns_box.insert("end", "Click 'Full DNS leak test' or 'Quick check' to probe.") + dns_status_lbl = ctk.CTkLabel(dns_card, text="", font=(FONT, 10, "bold"), text_color=TEXT2) + dns_status_lbl.pack(anchor="w", padx=12, pady=(0, 2)) - def _run_dns_check() -> None: - listen = f"http://{svc.settings.listen_addr()}" if svc.current_chain else None - r = check_dns_leak_hint(listen) - color = GREEN if r.ok else RED - dns_result_lbl.configure( - text=r.message + (f"\nResolvers: {', '.join(r.system_resolvers)}" if r.system_resolvers else ""), + def _render_dns_report(rep: FullDnsLeakReport) -> None: + dns_box.delete("1.0", "end") + mark = "LEAK" if rep.leaked else "OK " + dns_box.insert("end", f"[{mark}] {rep.summary}\n\n") + if rep.system_resolvers: + dns_box.insert("end", f"Configured resolvers: {', '.join(rep.system_resolvers)}\n") + for host, addrs in rep.direct_resolution.items(): + via_proxy = rep.proxy_resolution.get(host, []) + proxy_str = ", ".join(via_proxy) if via_proxy else "(no proxy answer)" + dns_box.insert("end", f" {host:<22} direct={', '.join(addrs) or '—':<20} via-proxy={proxy_str}\n") + if rep.errors: + dns_box.insert("end", "\nErrors: " + "; ".join(rep.errors[:3])) + color = RED if rep.leaked else GREEN + dns_status_lbl.configure( + text="DNS LEAK DETECTED" if rep.leaked else "DNS looks clean", text_color=color, ) - dns_btn_row = ctk.CTkFrame(dns_card, fg_color="transparent") - dns_btn_row.pack(fill="x", padx=12, pady=(0, 10)) - _btn(dns_btn_row, "DNS leak check", _run_dns_check, w=120, h=28).pack(side="left", padx=(0, 8)) + def _run_full_dns_test() -> None: + proxy = f"http://{svc.settings.listen_addr()}" if svc.current_chain else None + dns_status_lbl.configure(text="Running…", text_color=YELLOW) + dns_box.delete("1.0", "end") + dns_box.insert("end", "Probing DNS via proxy vs direct…\n") + + def work() -> None: + try: + rep = run_dns_leak_test(proxy_url=proxy, + timeout=min(12.0, svc.settings.validation_timeout_seconds)) + except Exception as e: + root.after(0, lambda: dns_status_lbl.configure( + text=f"Error: {e}", text_color=RED)) + return + root.after(0, lambda: _render_dns_report(rep)) + + threading.Thread(target=work, daemon=True).start() + + def _run_quick_dns_check() -> None: + listen = f"http://{svc.settings.listen_addr()}" if svc.current_chain else None + r = check_dns_leak_hint(listen) + color = GREEN if r.ok else RED + dns_box.delete("1.0", "end") + dns_box.insert("end", r.message) + if r.system_resolvers: + dns_box.insert("end", f"\nResolvers: {', '.join(r.system_resolvers)}") + dns_status_lbl.configure(text="clean" if r.ok else "issue detected", text_color=color) def _manual_dns_flush() -> None: ok, msg = flush_dns_cache() - dns_result_lbl.configure( - text=msg, text_color=GREEN if ok else RED, - ) + dns_status_lbl.configure(text=msg, text_color=GREEN if ok else RED) + dns_btn_row = ctk.CTkFrame(dns_card, fg_color="transparent") + dns_btn_row.pack(fill="x", padx=12, pady=(0, 10)) + _btn(dns_btn_row, "Full DNS leak test", _run_full_dns_test, w=140, h=28, + fg_color=ACCENT2, hover_color=ACCENT).pack(side="left", padx=(0, 6)) + _btn(dns_btn_row, "Quick check", _run_quick_dns_check, w=100, h=28).pack(side="left", padx=(0, 6)) _btn(dns_btn_row, "Flush DNS now", _manual_dns_flush, w=110, h=28, fg_color=DIM).pack(side="left") + # ── WebRTC leak test ───────────────────────────────────────────────────── + rtc_card = _priv_section( + "WebRTC leak test", + "Scans Chrome/Edge policy, Firefox profile prefs, and live STUN server reachability.", + ) + rtc_box = ctk.CTkTextbox( + rtc_card, height=110, font=("Consolas", 10), + fg_color=BG, text_color=TEXT, scrollbar_button_color=ACCENT, + ) + rtc_box.pack(fill="x", padx=12, pady=(4, 4)) + rtc_box.insert("end", "Click 'WebRTC leak test' to probe.") + rtc_status_lbl = ctk.CTkLabel(rtc_card, text="", font=(FONT, 10, "bold"), text_color=TEXT2) + rtc_status_lbl.pack(anchor="w", padx=12, pady=(0, 2)) + + def _render_rtc_report(rep: WebRtcCheckResult) -> None: + rtc_box.delete("1.0", "end") + for name, ok, detail in rep.checks: + mark = "OK " if ok else "RISK" + rtc_box.insert("end", f"[{mark}] {name:<40} {detail}\n") + color = RED if rep.any_leak_risk else GREEN + rtc_status_lbl.configure( + text="WebRTC LEAK RISK — see items above" if rep.any_leak_risk else "WebRTC looks protected", + text_color=color, + ) + + def _run_rtc_test() -> None: + rtc_status_lbl.configure(text="Running…", text_color=YELLOW) + rtc_box.delete("1.0", "end") + rtc_box.insert("end", "Scanning WebRTC surfaces…\n") + profile = Path(firefox_profile_var.get().strip() or default_profile_dir()) + + def work() -> None: + try: + rep = run_webrtc_check(profile_dir=profile) + except Exception as e: + root.after(0, lambda: rtc_status_lbl.configure(text=f"Error: {e}", text_color=RED)) + return + root.after(0, lambda: _render_rtc_report(rep)) + + threading.Thread(target=work, daemon=True).start() + + rtc_btn_row = ctk.CTkFrame(rtc_card, fg_color="transparent") + rtc_btn_row.pack(fill="x", padx=12, pady=(0, 10)) + _btn(rtc_btn_row, "WebRTC leak test", _run_rtc_test, w=140, h=28, + fg_color=ACCENT2, hover_color=ACCENT).pack(side="left") + wipe_card = _priv_section( "Forensic artifact wipe", "One-button purge of common Windows breadcrumb trails. Irreversible.", diff --git a/proxy_chain_manager/ban_tester.py b/proxy_chain_manager/ban_tester.py index dbdfac1..206aa33 100644 --- a/proxy_chain_manager/ban_tester.py +++ b/proxy_chain_manager/ban_tester.py @@ -1,29 +1,73 @@ from __future__ import annotations +import concurrent.futures +import socket from dataclasses import dataclass from typing import Iterable import httpx -# High-traffic destinations that often reveal proxy bans quickly. -POPULAR_SITES: tuple[tuple[str, str], ...] = ( - ("Google", "https://www.google.com/generate_204"), - ("YouTube", "https://www.youtube.com/"), - ("Facebook", "https://www.facebook.com/"), - ("Instagram", "https://www.instagram.com/"), - ("X", "https://x.com/"), - ("Reddit", "https://www.reddit.com/"), - ("Wikipedia", "https://www.wikipedia.org/"), - ("Amazon", "https://www.amazon.com/"), - ("Netflix", "https://www.netflix.com/"), - ("GitHub", "https://github.com/"), - ("Cloudflare", "https://www.cloudflare.com/"), - ("Microsoft", "https://www.microsoft.com/"), - ("TikTok", "https://www.tiktok.com/"), - ("BBC", "https://www.bbc.com/"), - ("DuckDuckGo", "https://duckduckgo.com/"), +# ── Site lists ──────────────────────────────────────────────────────────────── + +SITES_SOCIAL: tuple[tuple[str, str], ...] = ( + ("Google", "https://www.google.com/generate_204"), + ("YouTube", "https://www.youtube.com/"), + ("Facebook", "https://www.facebook.com/"), + ("Instagram", "https://www.instagram.com/"), + ("X / Twitter", "https://x.com/"), + ("Reddit", "https://www.reddit.com/"), + ("TikTok", "https://www.tiktok.com/"), + ("Wikipedia", "https://www.wikipedia.org/"), + ("DuckDuckGo", "https://duckduckgo.com/"), + ("BBC", "https://www.bbc.com/"), + ("GitHub", "https://github.com/"), ) +SITES_SHOPPING: tuple[tuple[str, str], ...] = ( + ("Amazon", "https://www.amazon.com/"), + ("eBay", "https://www.ebay.com/"), + ("Walmart", "https://www.walmart.com/"), + ("Etsy", "https://www.etsy.com/"), + ("AliExpress", "https://www.aliexpress.com/"), + ("Best Buy", "https://www.bestbuy.com/"), + ("Target", "https://www.target.com/"), + ("Newegg", "https://www.newegg.com/"), +) + +SITES_CRYPTO: tuple[tuple[str, str], ...] = ( + ("Coinbase", "https://www.coinbase.com/"), + ("Binance", "https://www.binance.com/"), + ("Kraken", "https://www.kraken.com/"), + ("Bybit", "https://www.bybit.com/"), + ("OKX", "https://www.okx.com/"), + ("KuCoin", "https://www.kucoin.com/"), + ("Bitfinex", "https://www.bitfinex.com/"), + ("Gemini", "https://www.gemini.com/"), +) + +SITES_DNS: tuple[tuple[str, str], ...] = ( + ("Cloudflare DNS (1.1.1.1)", "https://1.1.1.1/"), + ("Google DNS (8.8.8.8)", "https://dns.google/"), + ("Quad9 DNS (9.9.9.9)", "https://quad9.net/"), + ("AdGuard DNS", "https://adguard-dns.io/"), + ("NextDNS", "https://nextdns.io/"), + ("Mullvad DNS", "https://mullvad.net/en/help/dns-over-https-and-dns-over-tls/"), + ("OpenDNS", "https://www.opendns.com/"), +) + +# Combined default: all categories +POPULAR_SITES: tuple[tuple[str, str], ...] = ( + SITES_SOCIAL + SITES_SHOPPING + SITES_CRYPTO + SITES_DNS +) + +SITE_CATEGORIES: dict[str, tuple[tuple[str, str], ...]] = { + "Social / General": SITES_SOCIAL, + "Shopping": SITES_SHOPPING, + "Crypto Exchanges": SITES_CRYPTO, + "DNS Providers": SITES_DNS, + "All": POPULAR_SITES, +} + _BANNED_STATUS_CODES = {401, 403, 407, 418, 429, 451} _BANNED_TEXT_HINTS = ( "access denied", @@ -33,6 +77,15 @@ _BANNED_TEXT_HINTS = ( "unusual traffic", "captcha", "challenge required", + "bot detected", + "suspicious activity", + "geo-blocked", + "not available in your region", + "your ip", + "ip address", + "cloudflare ray", + "ddos protection", + "attention required", ) @@ -40,9 +93,10 @@ _BANNED_TEXT_HINTS = ( class BanTestResult: site: str url: str - status: str # ok | banned | error + status: str # ok | banned | error code: int | None detail: str + category: str = "" def _looks_banned_body(body: str) -> bool: @@ -50,7 +104,13 @@ def _looks_banned_body(body: str) -> bool: return any(h in t for h in _BANNED_TEXT_HINTS) -def _test_one(proxy_url: str, site: str, url: str, timeout_seconds: float) -> BanTestResult: +def _test_one( + proxy_url: str, + site: str, + url: str, + timeout_seconds: float, + category: str = "", +) -> BanTestResult: timeout = httpx.Timeout(timeout_seconds, connect=min(8.0, timeout_seconds)) try: with httpx.Client( @@ -58,27 +118,145 @@ def _test_one(proxy_url: str, site: str, url: str, timeout_seconds: float) -> Ba timeout=timeout, verify=False, follow_redirects=True, - headers={"User-Agent": "Mozilla/5.0"}, + headers={ + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" + ), + "Accept": "text/html,application/xhtml+xml,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9", + }, ) as c: r = c.get(url) - body = r.text[:1800] if r.text else "" + body = r.text[:2400] if r.text else "" if r.status_code in _BANNED_STATUS_CODES or _looks_banned_body(body): - return BanTestResult(site, url, "banned", r.status_code, f"HTTP {r.status_code}") + return BanTestResult( + site, url, "banned", r.status_code, + f"HTTP {r.status_code}", category, + ) if 200 <= r.status_code < 400: - return BanTestResult(site, url, "ok", r.status_code, f"HTTP {r.status_code}") - return BanTestResult(site, url, "error", r.status_code, f"HTTP {r.status_code}") + return BanTestResult(site, url, "ok", r.status_code, f"HTTP {r.status_code}", category) + return BanTestResult(site, url, "error", r.status_code, f"HTTP {r.status_code}", category) except Exception as e: - return BanTestResult(site, url, "error", None, f"{type(e).__name__}: {e}") + short = str(e)[:80] + return BanTestResult(site, url, "error", None, f"{type(e).__name__}: {short}", category) + + +def _dns_resolve_via_system(hostname: str, timeout: float = 5.0) -> str | None: + """Resolve a hostname using system DNS (does not go through proxy — exposes leak).""" + try: + socket.setdefaulttimeout(timeout) + info = socket.getaddrinfo(hostname, None) + for entry in info: + addr = entry[4][0] + if addr: + return addr + except Exception: + pass + return None + + +@dataclass(frozen=True) +class DnsLeakResult: + resolver: str + ip_via_proxy: str | None + ip_direct: str | None + leaked: bool + detail: str + + +def check_dns_leak( + proxy_url: str, + timeout_seconds: float = 8.0, +) -> list[DnsLeakResult]: + """ + Detect DNS leaks: compare hostname resolution seen through proxy vs direct. + A mismatch means DNS is escaping the tunnel. + """ + test_hosts = [ + ("Cloudflare (1.1.1.1)", "one.one.one.one"), + ("Google (8.8.8.8)", "dns.google"), + ("OpenDNS", "resolver1.opendns.com"), + ] + results: list[DnsLeakResult] = [] + timeout = httpx.Timeout(timeout_seconds, connect=min(6.0, timeout_seconds)) + + for label, host in test_hosts: + # Get IP via direct system DNS + direct_ip = _dns_resolve_via_system(host) + + # Get IP as seen from the proxy path (via http://dns-endpoint) + proxy_ip: str | None = None + try: + with httpx.Client(proxy=proxy_url, timeout=timeout, verify=False, follow_redirects=True) as c: + r = c.get(f"https://{host}/") + proxy_ip = str(r.headers.get("x-real-ip") or "") + if not proxy_ip: + # fall back: grab connected IP from response + proxy_ip = None + except Exception: + pass + + # Simple leak heuristic: if direct resolution works but proxy connection fails, possible leak path + leaked = bool(direct_ip and not proxy_ip) + if leaked: + detail = f"DNS resolved directly to {direct_ip} but proxy could not reach it — possible bypass" + elif not direct_ip: + detail = "Could not resolve directly" + leaked = False + else: + detail = f"Direct: {direct_ip}" + + results.append(DnsLeakResult( + resolver=label, + ip_via_proxy=proxy_ip, + ip_direct=direct_ip, + leaked=leaked, + detail=detail, + )) + + return results def run_ban_tests( proxy_url: str, timeout_seconds: float = 12.0, - sites: Iterable[tuple[str, str]] = POPULAR_SITES, + sites: Iterable[tuple[str, str]] | None = None, + categories: Iterable[str] | None = None, + max_workers: int = 8, ) -> list[BanTestResult]: + """ + Run ban tests in parallel across requested site categories. + + ``categories`` accepts names from SITE_CATEGORIES (e.g. "Shopping", "Crypto Exchanges"). + Defaults to all categories when neither ``sites`` nor ``categories`` is given. + """ + if sites is not None: + work = [(s, u, "") for s, u in sites] + elif categories is not None: + work = [] + for cat in categories: + for s, u in SITE_CATEGORIES.get(cat, ()): + work.append((s, u, cat)) + else: + work = [(s, u, cat) for cat, pairs in SITE_CATEGORIES.items() if cat != "All" for s, u in pairs] + out: list[BanTestResult] = [] - for site, url in sites: - out.append(_test_one(proxy_url, site, url, timeout_seconds)) + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool: + futures = { + pool.submit(_test_one, proxy_url, site, url, timeout_seconds, cat): (site, url, cat) + for site, url, cat in work + } + for fut in concurrent.futures.as_completed(futures): + try: + out.append(fut.result()) + except Exception: + site, url, cat = futures[fut] + out.append(BanTestResult(site, url, "error", None, "internal error", cat)) + + # Sort: category then site name + out.sort(key=lambda r: (r.category, r.site)) return out diff --git a/proxy_chain_manager/browser_identity.py b/proxy_chain_manager/browser_identity.py index 816d32b..0d3518e 100644 --- a/proxy_chain_manager/browser_identity.py +++ b/proxy_chain_manager/browser_identity.py @@ -15,84 +15,78 @@ Two strategies are exposed: • **Hardened** — full RFP / blocked WebRTC / strict tracking / FPI. Stand-out but maximum block. Same behavior as the original profile. -Cookie modes are independent of persona and let the user pick per-session -behavior from a dropdown. +Cookie modes are independent of persona. Each mode expresses a complete +policy: which cookies to accept, lifetime, whether to wipe on close, and +whether to partition 3rd-party storage. """ from __future__ import annotations from dataclasses import dataclass -# Order matters: dropdown order in the UI. + +# ── Personas ────────────────────────────────────────────────────────────────── + PERSONAS: list[str] = [ "blend_windows_chrome", + "blend_windows_chrome_laptop", "blend_windows_firefox", + "blend_windows_edge", "blend_mac_safari", + "blend_mac_chrome", + "blend_linux_firefox", "hardened", "custom", ] PERSONA_LABELS: dict[str, str] = { - "blend_windows_chrome": "Blend in — Windows 10 + Chrome (US-en, most common)", - "blend_windows_firefox": "Blend in — Windows 10 + Firefox (US-en)", - "blend_mac_safari": "Blend in — macOS + Safari (US-en)", - "hardened": "Hardened — block + RFP (stands out; max anti-fingerprint)", - "custom": "Custom — use individual hardening toggles", -} - -COOKIE_MODES: list[str] = [ - "accept_all", - "block_third_party", - "block_third_party_trackers", - "session_only", - "block_all", -] - -COOKIE_LABELS: dict[str, str] = { - "accept_all": "Accept all cookies (most sites work)", - "block_third_party": "Block 3rd-party cookies (recommended)", - "block_third_party_trackers": "Block 3rd-party trackers only (Firefox default-strict)", - "session_only": "Session only — clear on close", - "block_all": "Block ALL cookies (breaks logins)", -} - -# Firefox network.cookie.cookieBehavior values -_COOKIE_BEHAVIOR = { - "accept_all": 0, - "block_third_party": 1, - "block_third_party_trackers": 4, - "session_only": 0, - "block_all": 2, + "blend_windows_chrome": "Blend — Win10 + Chrome 124 (most common, US-en)", + "blend_windows_chrome_laptop": "Blend — Win11 + Chrome 124 laptop 1366×768", + "blend_windows_firefox": "Blend — Win10 + Firefox 128 (US-en)", + "blend_windows_edge": "Blend — Win11 + Edge 124 (corporate look)", + "blend_mac_safari": "Blend — macOS 14 + Safari 17 (US-en)", + "blend_mac_chrome": "Blend — macOS 14 + Chrome 124 (US-en)", + "blend_linux_firefox": "Blend — Ubuntu + Firefox 128 (en-US)", + "hardened": "Hardened — RFP + FPI + block all (stands out)", + "custom": "Custom — use individual hardening toggles", } @dataclass(frozen=True) class Persona: """A user-agent / locale / timezone / screen tuple that mimics a real, - common configuration. All fields are deliberately popular values so the - persona blends in. - """ + common configuration.""" key: str user_agent: str accept_language: str - timezone: str # IANA tz; used only when persona drives env TZ + timezone: str screen_w: int screen_h: int - platform: str # navigator.platform - locale: str # general.useragent.locale equivalent + platform: str + locale: str PERSONA_DATA: dict[str, Persona] = { "blend_windows_chrome": Persona( key="blend_windows_chrome", - # Stable Chrome on Win10 x64 — by far the most common UA on the web. user_agent=( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36" ), accept_language="en-US,en;q=0.9", timezone="America/New_York", - screen_w=1920, - screen_h=1080, + screen_w=1920, screen_h=1080, + platform="Win32", + locale="en-US", + ), + "blend_windows_chrome_laptop": Persona( + key="blend_windows_chrome_laptop", + user_agent=( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36" + ), + accept_language="en-US,en;q=0.9", + timezone="America/Chicago", + screen_w=1366, screen_h=768, platform="Win32", locale="en-US", ), @@ -104,8 +98,19 @@ PERSONA_DATA: dict[str, Persona] = { ), accept_language="en-US,en;q=0.5", timezone="America/New_York", - screen_w=1920, - screen_h=1080, + screen_w=1920, screen_h=1080, + platform="Win32", + locale="en-US", + ), + "blend_windows_edge": Persona( + key="blend_windows_edge", + user_agent=( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0" + ), + accept_language="en-US,en;q=0.9", + timezone="America/Chicago", + screen_w=1920, screen_h=1080, platform="Win32", locale="en-US", ), @@ -117,20 +122,201 @@ PERSONA_DATA: dict[str, Persona] = { ), accept_language="en-US,en;q=0.9", timezone="America/Los_Angeles", - screen_w=1680, - screen_h=1050, + screen_w=1680, screen_h=1050, platform="MacIntel", locale="en-US", ), + "blend_mac_chrome": Persona( + key="blend_mac_chrome", + user_agent=( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36" + ), + accept_language="en-US,en;q=0.9", + timezone="America/Los_Angeles", + screen_w=1440, screen_h=900, + platform="MacIntel", + locale="en-US", + ), + "blend_linux_firefox": Persona( + key="blend_linux_firefox", + user_agent=( + "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:128.0) " + "Gecko/20100101 Firefox/128.0" + ), + accept_language="en-US,en;q=0.5", + timezone="America/New_York", + screen_w=1920, screen_h=1080, + platform="Linux x86_64", + locale="en-US", + ), } +# ── Cookie policies ─────────────────────────────────────────────────────────── + +@dataclass(frozen=True) +class CookiePolicy: + """ + Complete Firefox cookie configuration expressed as a single composable unit. + + behavior : network.cookie.cookieBehavior + 0 = accept all + 1 = block 3rd-party + 2 = block ALL cookies + 3 = block 3rd-party from unvisited sites + 4 = block cross-site tracking (ETP Strict) + 5 = block cross-site + social media trackers + lifetime : network.cookie.lifetimePolicy + 0 = normal (respect Set-Cookie expires) + 2 = session-only (all cookies expire on close) + partition : Total Cookie Protection — partition 3rd-party storage + by the top-level site (network.cookie.cookieBehavior.optInPartitioning) + clear_cookies : privacy.clearOnShutdown.cookies + clear_cache : privacy.clearOnShutdown.cache + clear_history : privacy.clearOnShutdown.history + clear_ls : privacy.clearOnShutdown.localStorage + indexedDB + clear_formdata : privacy.clearOnShutdown.formdata + sanitize : privacy.sanitize.sanitizeOnShutdown (master switch) + """ + behavior: int = 1 + lifetime: int = 0 + partition: bool = False + sanitize: bool = False + clear_cookies: bool = False + clear_cache: bool = False + clear_history: bool = False + clear_ls: bool = False + clear_formdata: bool = False + + +# Every cookie mode is a complete CookiePolicy — no guesswork in profile.py. +COOKIE_POLICIES: dict[str, CookiePolicy] = { + + # ── Permissive ────────────────────────────────────────────────────────── + + "accept_all": CookiePolicy( + behavior=0, lifetime=0, + # No clearing, no blocking — maximum site compatibility. + ), + + "accept_all_session": CookiePolicy( + behavior=0, lifetime=2, + sanitize=True, clear_cookies=True, clear_ls=True, + # Accept all cookies while browsing but wipe them when the + # browser closes. Good for signup / form-filling sessions. + ), + + # ── Standard 3rd-party blocking ──────────────────────────────────────── + + "block_third_party": CookiePolicy( + behavior=1, lifetime=0, + # Block 3rd-party, accept 1st-party with normal lifetime. + # Recommended baseline — most sites work fine. + ), + + "block_third_party_session": CookiePolicy( + behavior=1, lifetime=2, + sanitize=True, clear_cookies=True, clear_cache=True, clear_ls=True, + # Block 3rd-party AND treat all accepted cookies as session-only. + # Nothing persists. Use for anonymous sessions. + ), + + # ── Tracker-focused blocking ──────────────────────────────────────────── + + "block_third_party_trackers": CookiePolicy( + behavior=4, lifetime=0, + # Firefox ETP Strict mode: only blocks known trackers, not ALL + # 3rd-party cookies. Best blend-in option — same as Firefox default. + ), + + "block_social_trackers": CookiePolicy( + behavior=5, lifetime=0, + # Block cross-site cookies AND social-media tracking cookies + # (Facebook, Twitter pixels, etc.). Stricter than ETP Strict but + # still allows most logins. + ), + + "block_unvisited": CookiePolicy( + behavior=3, lifetime=0, + # Block 3rd-party from sites the user has never visited. Very light + # — almost invisible to sites; good when you need max compatibility. + ), + + # ── Partitioned storage (Total Cookie Protection) ─────────────────────── + + "partitioned_tcp": CookiePolicy( + behavior=1, lifetime=0, partition=True, + # Block 3rd-party AND enable Total Cookie Protection: each site + # gets its own isolated cookie jar so cross-site tracking via + # cookie sync is impossible. Recommended for blend-in + privacy. + ), + + "partitioned_session": CookiePolicy( + behavior=1, lifetime=2, partition=True, + sanitize=True, clear_cookies=True, clear_cache=True, + clear_ls=True, clear_formdata=True, + # Partitioned + session-only + full wipe on close. Leaves no + # persistent state on disk after the browser exits. + ), + + # ── Maximum wipe ─────────────────────────────────────────────────────── + + "session_only": CookiePolicy( + behavior=0, lifetime=2, + sanitize=True, clear_cookies=True, + # All cookies accepted but expire when the browser closes. + # Note: localStorage is NOT cleared here — only cookies. + ), + + "ghost_mode": CookiePolicy( + behavior=1, lifetime=2, + sanitize=True, clear_cookies=True, clear_cache=True, + clear_history=True, clear_ls=True, clear_formdata=True, + partition=True, + # Maximum ephemeral session: block 3rd-party, TCP partitioning, + # session-only lifetime, full wipe of cookies + cache + history + + # localStorage/IndexedDB + formdata on every close. Zero disk trace. + ), + + "block_all": CookiePolicy( + behavior=2, lifetime=0, + # Block ALL cookies. Most login-dependent sites will break. + # Only useful for read-only scraping sessions. + ), +} + +COOKIE_MODES: list[str] = list(COOKIE_POLICIES) + +COOKIE_LABELS: dict[str, str] = { + "accept_all": "Accept all — maximum compatibility (cookies persist)", + "accept_all_session": "Accept all, session only — wipe cookies on close", + "block_third_party": "Block 3rd-party — recommended baseline (persists 1st-party)", + "block_third_party_session": "Block 3rd-party + session — nothing persists", + "block_third_party_trackers":"Block trackers only — ETP Strict (Firefox default-strict)", + "block_social_trackers": "Block social + cross-site trackers (Facebook pixel, etc.)", + "block_unvisited": "Block 3rd-party from unvisited sites — lightest option", + "partitioned_tcp": "Partitioned (TCP) — each site gets isolated 3rd-party jar", + "partitioned_session": "Partitioned + session — isolated jars, full wipe on close", + "session_only": "Session only — all cookies expire on close (no partition)", + "ghost_mode": "Ghost mode — partitioned + session + full disk wipe on close", + "block_all": "Block ALL cookies — breaks most logins", +} + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + +def get_cookie_policy(mode: str) -> CookiePolicy: + return COOKIE_POLICIES.get(mode, COOKIE_POLICIES["block_third_party"]) + + +# Legacy shims kept for code that still calls these directly. def cookie_behavior_value(mode: str) -> int: - return _COOKIE_BEHAVIOR.get(mode, 1) + return get_cookie_policy(mode).behavior def cookie_session_only(mode: str) -> bool: - return mode == "session_only" + return get_cookie_policy(mode).lifetime == 2 def get_persona(key: str) -> Persona | None: diff --git a/proxy_chain_manager/browser_launcher.py b/proxy_chain_manager/browser_launcher.py index cc7bc20..ef65322 100644 --- a/proxy_chain_manager/browser_launcher.py +++ b/proxy_chain_manager/browser_launcher.py @@ -5,6 +5,7 @@ import os import shutil import subprocess import time +import winreg from dataclasses import dataclass from pathlib import Path @@ -14,6 +15,14 @@ from .signup_prep import SignupDraft, install_signup_extension log = logging.getLogger(__name__) +# CREATE_NO_WINDOW suppresses the console; DETACHED_PROCESS fully separates +# the child from Python's job object so Firefox lives past Python's exit. +_CREATE_FLAGS = ( + getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000) + | getattr(subprocess, "DETACHED_PROCESS", 0x00000008) + | getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0x00000200) +) + @dataclass class BrowserConfig: @@ -35,11 +44,74 @@ class BrowserConfig: cookie_mode: str = "block_third_party" +def _find_firefox_via_registry() -> str: + """Look up Firefox install path in the Windows registry.""" + hives = [ + (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Mozilla\Mozilla Firefox"), + (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\WOW6432Node\Mozilla\Mozilla Firefox"), + (winreg.HKEY_CURRENT_USER, r"SOFTWARE\Mozilla\Mozilla Firefox"), + ] + for hive, key_path in hives: + try: + with winreg.OpenKey(hive, key_path) as root: + version = winreg.QueryValue(root, None) + for sub in (f"{version}\\Main", "bin"): + try: + with winreg.OpenKey(root, sub) as k: + path = winreg.QueryValueEx(k, "PathToExe")[0] + if path and Path(path).is_file(): + return str(path) + except OSError: + continue + except OSError: + continue + + # Also try the "Uninstall" key + for hive in (winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER): + for root_path in ( + r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall", + r"SOFTWARE\WOW6432Node\Microsoft\Windows\CurrentVersion\Uninstall", + ): + try: + with winreg.OpenKey(hive, root_path) as base: + i = 0 + while True: + try: + sub = winreg.EnumKey(base, i) + i += 1 + if "firefox" not in sub.lower(): + continue + with winreg.OpenKey(base, sub) as k: + loc = winreg.QueryValueEx(k, "InstallLocation")[0] + candidate = Path(loc) / "firefox.exe" + if candidate.is_file(): + return str(candidate) + except OSError: + break + except OSError: + continue + return "" + + def default_firefox_path() -> str: - candidates = [ + """Find Firefox.exe — registry first, then well-known paths.""" + # 1) Registry + reg = _find_firefox_via_registry() + if reg: + return reg + + # 2) Standard install locations + candidates: list[Path] = [ Path(r"C:\Program Files\Mozilla Firefox\firefox.exe"), Path(r"C:\Program Files (x86)\Mozilla Firefox\firefox.exe"), + Path(os.path.expandvars(r"%LOCALAPPDATA%\Mozilla Firefox\firefox.exe")), + Path(os.path.expandvars(r"%APPDATA%\Mozilla Firefox\firefox.exe")), ] + # 3) Scoop / winget / portable in PATH + which = shutil.which("firefox") or shutil.which("firefox.exe") + if which: + candidates.insert(0, Path(which)) + for c in candidates: if c.is_file(): return str(c) @@ -50,13 +122,39 @@ def default_profile_dir() -> str: return str(app_data_dir() / "browser_profiles" / "firefox_hardened") +def _firefox_is_running_on_system() -> bool: + """Check if any firefox.exe process is alive (tasklist, no admin needed).""" + try: + r = subprocess.run( + ["tasklist", "/FI", "IMAGENAME eq firefox.exe", "/NH", "/FO", "CSV"], + capture_output=True, text=True, timeout=6, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000), + ) + return "firefox.exe" in (r.stdout or "").lower() + except Exception: + return False + + class BrowserSession: def __init__(self) -> None: self._proc: subprocess.Popen[str] | None = None self._profile_path: Path | None = None + self._launched_at: float = 0.0 def is_running(self) -> bool: - return self._proc is not None and self._proc.poll() is None + if self._proc is None: + return False + # Firefox's initial launcher process exits quickly (code 0) while + # browser child processes carry on. Poll the parent process but also + # fall back to checking the system process list so the GUI doesn't + # falsely report "stopped". + if self._proc.poll() is None: + return True + # Parent has exited — check if any firefox.exe is still alive AND we + # launched within the last 5 minutes (to avoid false positives from + # unrelated browser sessions). + age = time.monotonic() - self._launched_at + return age < 300 and _firefox_is_running_on_system() def pid(self) -> int | None: if self._proc is None: @@ -72,11 +170,22 @@ class BrowserSession: signup_draft: SignupDraft | None = None, ) -> tuple[bool, str]: raw_exe = (cfg.firefox_path or default_firefox_path()).strip().strip('"').strip("'") + if not raw_exe: + return ( + False, + "Firefox not found. Install Firefox then set the path in the Browser tab " + "(or browse to it with the Browse button).", + ) exe = Path(raw_exe) if not exe.is_file(): - return False, f"Firefox executable not found: {raw_exe or '(empty path)'}" + return False, ( + f"Firefox executable not found: {raw_exe} " + "Use the Browse button on the Browser tab to locate firefox.exe." + ) + raw_profile = (cfg.profile_dir or default_profile_dir()).strip().strip('"').strip("'") profile_dir = Path(raw_profile) + hard = FirefoxHardening( force_proxy=cfg.force_proxy, disable_webrtc=cfg.disable_webrtc, @@ -89,46 +198,88 @@ class BrowserSession: persona_key=cfg.persona, cookie_mode=cfg.cookie_mode, ) - if cfg.lock_managed_profile: - ensure_firefox_profile(profile_dir, proxy_host, int(proxy_port), hard) - if signup_draft is not None: - install_signup_extension(profile_dir, signup_draft) + + try: + if cfg.lock_managed_profile: + ensure_firefox_profile(profile_dir, proxy_host, int(proxy_port), hard) + if signup_draft is not None: + install_signup_extension(profile_dir, signup_draft) + except Exception as e: + return False, f"Profile setup failed: {e!s}" + if self.is_running(): self.stop() + if cfg.lock_managed_profile: args = [str(exe), "-no-remote", "-profile", str(profile_dir)] else: args = [str(exe)] + url = (start_url or "").strip() if url: args.append(url) + env = os.environ.copy() env.update(persona_env(hard)) + try: self._proc = subprocess.Popen( args, - creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + # DETACHED_PROCESS ensures Firefox outlives Python and is not + # killed when Python's job object exits. CREATE_NO_WINDOW + # suppresses any console window. CREATE_NEW_PROCESS_GROUP + # isolates Ctrl-C handling. + creationflags=_CREATE_FLAGS, + stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, text=True, env=env, ) - # Smoke-check: catch immediate startup crashes and report clearly. - time.sleep(0.4) - rc = self._proc.poll() - if rc is not None: - self._proc = None - return False, f"Firefox exited immediately (code {rc}). Check path/profile." - self._profile_path = profile_dir if cfg.lock_managed_profile else None - log.info("Launched hardened Firefox pid=%s profile=%s", self._proc.pid, profile_dir) - if cfg.lock_managed_profile: - return True, f"Hardened Firefox started (pid {self._proc.pid})." - return True, f"Firefox started (unlocked mode, pid {self._proc.pid})." except Exception as e: self._proc = None return False, f"Launch failed: {e!s}" + self._launched_at = time.monotonic() + + # Firefox's multi-process launcher exits (code 0) within ~0.5 s while + # the actual browser window loads in child processes. We therefore do + # NOT treat a quick exit of the parent as an error — we just verify + # that *some* firefox.exe appears in the process list within 3 s. + deadline = time.monotonic() + 3.0 + found = False + while time.monotonic() < deadline: + time.sleep(0.3) + rc = self._proc.poll() + # If it already exited with an error code, report it. + if rc is not None and rc not in (0, None): + self._proc = None + return False, f"Firefox exited with error code {rc}." + if _firefox_is_running_on_system(): + found = True + break + + if not found: + # One last check — maybe the profile write was the only action needed + # and Firefox opened fast; accept success if parent is still alive. + if self._proc.poll() is None: + found = True + + if not found: + self._proc = None + return False, ( + "Firefox did not appear in the process list after 3 s. " + "Check that the path is correct and Firefox is not already blocked by another -profile lock." + ) + + self._profile_path = profile_dir if cfg.lock_managed_profile else None + pid = self._proc.pid + log.info("Launched hardened Firefox pid=%s profile=%s", pid, profile_dir) + mode = "hardened profile" if cfg.lock_managed_profile else "unlocked mode" + return True, f"Firefox started ({mode}, pid {pid})." + def stop(self, dispose: bool = False) -> tuple[bool, str]: + # Terminate the Popen handle if still alive if self._proc and self._proc.poll() is None: try: self._proc.terminate() @@ -140,9 +291,24 @@ class BrowserSession: pass finally: self._proc = None + + # Also kill any remaining firefox.exe processes (handles the + # detached-child case where the parent already exited naturally) + try: + subprocess.run( + ["taskkill", "/F", "/IM", "firefox.exe"], + capture_output=True, + timeout=10, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000), + ) + except Exception: + pass + + self._proc = None if dispose and self._profile_path and self._profile_path.exists(): try: shutil.rmtree(self._profile_path, ignore_errors=True) except Exception: pass + self._launched_at = 0.0 return True, "Browser stopped." diff --git a/proxy_chain_manager/browser_profile.py b/proxy_chain_manager/browser_profile.py index 8ea96fb..1d83e02 100644 --- a/proxy_chain_manager/browser_profile.py +++ b/proxy_chain_manager/browser_profile.py @@ -4,9 +4,11 @@ from dataclasses import dataclass, field from pathlib import Path from .browser_identity import ( + CookiePolicy, Persona, cookie_behavior_value, cookie_session_only, + get_cookie_policy, get_persona, is_blend_persona, ) @@ -180,28 +182,62 @@ def build_user_js( f'user_pref("general.useragent.override", "{hard.user_agent_override}");' ) - # Cookie policy (persona-independent) - behavior = cookie_behavior_value(hard.cookie_mode) - lines.extend( - [ - f'user_pref("network.cookie.cookieBehavior", {behavior});', - # 0 = expire normally, 2 = current session only - f'user_pref("network.cookie.lifetimePolicy", {2 if cookie_session_only(hard.cookie_mode) else 0});', - ] - ) + # ── Cookie policy ──────────────────────────────────────────────────────── + cp: CookiePolicy = get_cookie_policy(hard.cookie_mode) - if hard.clear_on_shutdown: - lines.extend( - [ - 'user_pref("privacy.sanitize.sanitizeOnShutdown", true);', - 'user_pref("privacy.clearOnShutdown.history", true);', - 'user_pref("privacy.clearOnShutdown.cookies", true);', - 'user_pref("privacy.clearOnShutdown.cache", true);', - 'user_pref("privacy.clearOnShutdown.downloads", true);', - 'user_pref("privacy.clearOnShutdown.formdata", true);', - 'user_pref("privacy.clearOnShutdown.sessions", true);', - ] - ) + lines.extend([ + f'user_pref("network.cookie.cookieBehavior", {cp.behavior});', + # pbmode = private-browsing mode uses same behavior + f'user_pref("network.cookie.cookieBehavior.pbmode", {cp.behavior});', + # lifetime: 0 = normal, 2 = session-only + f'user_pref("network.cookie.lifetimePolicy", {cp.lifetime});', + ]) + + # Total Cookie Protection (TCP / Firefox partitioning) + if cp.partition: + lines.extend([ + 'user_pref("network.cookie.cookieBehavior.optInPartitioning", true);', + # Also isolate cache, localStorage, IndexedDB per top-level site + 'user_pref("privacy.partition.serviceWorkers", true);', + 'user_pref("privacy.partition.network_state", true);', + 'user_pref("privacy.partition.bloburl_by_default", true);', + ]) + else: + lines.extend([ + 'user_pref("network.cookie.cookieBehavior.optInPartitioning", false);', + ]) + + # SameSite=None cookies must be Secure (good hygiene regardless of mode) + lines.append('user_pref("network.cookie.sameSite.noneRequiresSecure", true);') + + # ── Sanitize-on-shutdown ───────────────────────────────────────────────── + # The CookiePolicy sanitize flag takes priority; the hard.clear_on_shutdown + # toggle acts as an additional override for cookies+cache even when the + # cookie mode doesn't request it. + do_sanitize = cp.sanitize or hard.clear_on_shutdown + if do_sanitize: + lines.append('user_pref("privacy.sanitize.sanitizeOnShutdown", true);') + # Cookies + lines.append(f'user_pref("privacy.clearOnShutdown.cookies", {_bool(cp.clear_cookies or hard.clear_on_shutdown)});') + # Cache + lines.append(f'user_pref("privacy.clearOnShutdown.cache", {_bool(cp.clear_cache or hard.clear_on_shutdown)});') + # History + lines.append(f'user_pref("privacy.clearOnShutdown.history", {_bool(cp.clear_history or hard.clear_on_shutdown)});') + # Downloads list + lines.append('user_pref("privacy.clearOnShutdown.downloads", true);') + # Form / search bar autofill data + lines.append(f'user_pref("privacy.clearOnShutdown.formdata", {_bool(cp.clear_formdata or hard.clear_on_shutdown)});') + # Active sessions (HTTP auth, logged-in sites) + lines.append('user_pref("privacy.clearOnShutdown.sessions", true);') + # localStorage + IndexedDB + lines.append(f'user_pref("privacy.clearOnShutdown.localStorage", {_bool(cp.clear_ls or hard.clear_on_shutdown)});') + lines.append(f'user_pref("privacy.clearOnShutdown.indexedDB", {_bool(cp.clear_ls or hard.clear_on_shutdown)});') + # Service worker registrations + lines.append('user_pref("privacy.clearOnShutdown.serviceWorkers", true);') + # Offline web app cache + lines.append('user_pref("privacy.clearOnShutdown.offlineApps", true);') + else: + lines.append('user_pref("privacy.sanitize.sanitizeOnShutdown", false);') if hard.timezone_utc: lines.append('user_pref("privacy.resistFingerprinting.reduceTimerPrecision", true);') diff --git a/proxy_chain_manager/chain_map.py b/proxy_chain_manager/chain_map.py index cac1688..ccadbf8 100644 --- a/proxy_chain_manager/chain_map.py +++ b/proxy_chain_manager/chain_map.py @@ -1,4 +1,8 @@ -"""World-map visualization for proxy chains (Pillow + equirectangular projection).""" +"""World-map visualization for proxy chains. + +Renders hop arcs over the bundled neon world map image. Falls back to a +drawn landmass map if the image file is absent. +""" from __future__ import annotations import ipaddress @@ -6,10 +10,11 @@ import logging import math import socket from dataclasses import dataclass +from pathlib import Path from typing import Literal from urllib.parse import urlparse -from PIL import Image, ImageDraw, ImageFont +from PIL import Image, ImageDraw, ImageFilter, ImageFont try: _RESAMPLE = Image.Resampling.LANCZOS @@ -23,21 +28,42 @@ 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" +# Theme palette +_BG = "#070713" +_GRID = "#1a1f3a" +_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" +_LINE_DEAD = "#ff2a6d" +_LINE_IDLE = "#3052ff" +_YOU = "#00ff9c" +_HOP = "#00e5ff" +_EXIT = "#ffd000" +_LABEL = "#eaf2ff" +_LABEL_DIM = "#7a8aab" +_UNKNOWN = "#2a2f4a" + +def _map_img_path() -> Path: + """Locate world_map.png whether running from source or a frozen PyInstaller bundle.""" + import sys + if getattr(sys, "frozen", False) and hasattr(sys, "_MEIPASS"): + return Path(sys._MEIPASS) / "proxy_chain_manager" / "world_map.png" + return Path(__file__).resolve().parent / "world_map.png" + +_cached_map: Image.Image | None = None + + +def _load_world_map(width: int, height: int) -> Image.Image | None: + """Return the neon world map resized to (width × height), or None if unavailable.""" + global _cached_map + try: + if _cached_map is None: + _cached_map = Image.open(_map_img_path()).convert("RGB") + return _cached_map.resize((width, height), _RESAMPLE) + except Exception as exc: + log.debug("world_map load failed: %s", exc) + return None @dataclass(frozen=True) @@ -134,26 +160,23 @@ def build_chain_map_points( 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 != "—": + exit_str = (exit_ip or "").strip() + if exit_str and exit_str != "—": if ( exit_intel and exit_intel.ok - and exit_intel.ip == exit + and exit_intel.ip == exit_str 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)) + pt = _intel_to_point("exit", "EXIT", fetch_ip_geo(exit_str, timeout_seconds)) if pt: points.append(pt) @@ -173,7 +196,7 @@ def _great_circle_points( lon1: float, lat2: float, lon2: float, - steps: int = 24, + steps: int = 32, ) -> list[tuple[float, float]]: """Interpolate along a great circle for curved hop arcs.""" phi1, lam1 = math.radians(lat1), math.radians(lon1) @@ -194,21 +217,50 @@ def _great_circle_points( 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)))) + 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 _line_color(status: ChainStatus) -> tuple[int, int, int]: + table = { + "healthy": (0, 229, 255), + "connecting": (255, 208, 0), + "dead": (255, 42, 109), + "idle": (48, 82, 255), + } + return table.get(status, (48, 82, 255)) -def _node_color(role: str) -> str: - return {"you": _YOU, "hop": _HOP, "exit": _EXIT}.get(role, _HOP) +def _node_color(role: str) -> tuple[int, int, int]: + table = {"you": (0, 255, 156), "hop": (0, 229, 255), "exit": (255, 208, 0)} + return table.get(role, (0, 229, 255)) + + +def _hex_to_rgb(h: str) -> tuple[int, int, int]: + h = h.lstrip("#") + return int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16) + + +def _draw_glow_line( + draw: ImageDraw.ImageDraw, + pts: list[tuple[float, float]], + color: tuple[int, int, int], +) -> None: + """Draw neon glow arc: fat dim outer halo → bright core line.""" + if len(pts) < 2: + return + r, g, b = color + # outer halo — wider, dim + halo = (r // 4, g // 4, b // 4) + draw.line(pts, fill=halo, width=7) + # mid glow + mid = (r // 2, g // 2, b // 2) + draw.line(pts, fill=mid, width=4) + # bright core + draw.line(pts, fill=color, width=2) def _load_font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont: @@ -221,18 +273,13 @@ def _load_font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont: def _draw_landmasses(draw: ImageDraw.ImageDraw, width: int, height: int, margin: int) -> None: - """Minimal continent silhouettes — stylized, not survey-accurate.""" + """Fallback minimal continent silhouettes.""" 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: @@ -246,58 +293,75 @@ def render_chain_map( width: int = 960, height: int = 220, status: ChainStatus = "idle", - margin: int = 18, + margin: int = 12, ) -> 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) + """Render chain hops over the neon world map (or a drawn fallback).""" + w = max(320, width) + h = max(120, height) - # 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) + # ── background ────────────────────────────────────────────────────────── + world = _load_world_map(w, h) + if world is not None: + # Slightly darken so overlaid lines pop + bg = Image.new("RGB", (w, h), (0, 0, 0)) + img = Image.blend(world, bg, alpha=0.22) + else: + img = Image.new("RGB", (w, h), _BG) + draw_bg = ImageDraw.Draw(img) + _draw_landmasses(draw_bg, w, h, margin) + # grid + for lat in range(-60, 91, 30): + y = _project(lat, 0, w, h, margin)[1] + draw_bg.line([(margin, y), (w - margin, y)], fill=_GRID, width=1) + for lon in range(-150, 181, 30): + x = _project(0, lon, w, h, margin)[0] + draw_bg.line([(x, margin), (x, h - margin)], fill=_GRID, width=1) + + draw = ImageDraw.Draw(img) if len(points) < 2: - font = _load_font(13) + font = _load_font(12) 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) + # drop shadow + draw.text((margin + 1, h // 2 - 7), msg, fill=(0, 0, 0), font=font) + draw.text((margin, h // 2 - 8), msg, fill=_hex_to_rgb(_LABEL_DIM), font=font) return img - line_c = _line_color(status) - projected = [_project(p.lat, p.lon, width, height, margin) for p in points] + line_rgb = _line_color(status) + projected = [_project(p.lat, p.lon, w, h, margin) for p in points] - # Glow underlay + # ── arc lines between consecutive hops ───────────────────────────────── 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) + arc = _great_circle_points( + points[i].lat, points[i].lon, + points[i + 1].lat, points[i + 1].lon, + ) + arc_xy = [_project(lat, lon, w, h, margin) for lat, lon in arc] + _draw_glow_line(draw, arc_xy, line_rgb) - label_font = _load_font(11) + # ── nodes ─────────────────────────────────────────────────────────────── + label_font = _load_font(11) detail_font = _load_font(9) for pt, (x, y) in zip(points, projected): - color = _node_color(pt.role) + nc = _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) + # outer glow ring + glow = (nc[0] // 3, nc[1] // 3, nc[2] // 3) + draw.ellipse((x - r - 4, y - r - 4, x + r + 4, y + r + 4), fill=glow) + # filled node with dark center + draw.ellipse((x - r, y - r, x + r, y + r), fill=nc) + draw.ellipse((x - r + 2, y - r + 2, x + r - 2, y + r - 2), fill=(5, 5, 20)) + + tx, ty = int(x + r + 5), int(y - 8) + # text drop-shadow + draw.text((tx + 1, ty + 1), pt.label, fill=(0, 0, 0), font=label_font) + draw.text((tx, ty), pt.label, fill=_hex_to_rgb(_LABEL), font=label_font) if pt.detail: - draw.text((tx, ty + 13), pt.detail[:28], fill=_LABEL_DIM, font=detail_font) + draw.text((tx, ty + 12), pt.detail[:30], fill=_hex_to_rgb(_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) + return pil_img.resize((max(320, int(width)), max(120, int(height))), _RESAMPLE) diff --git a/proxy_chain_manager/dns_leak.py b/proxy_chain_manager/dns_leak.py index 7782459..f8aa2d0 100644 --- a/proxy_chain_manager/dns_leak.py +++ b/proxy_chain_manager/dns_leak.py @@ -1,41 +1,118 @@ -"""DNS leak checks and cache flush.""" +"""DNS leak detection and cache flushing. + +Two levels of testing: + 1. check_dns_leak_hint() — fast, no network: inspect configured DNS resolvers + 2. run_dns_leak_test() — real network test: compare DNS answers seen through + the proxy vs direct, and probe resolver identity via + a dedicated leak-test API. +""" from __future__ import annotations +import concurrent.futures +import ipaddress import logging +import socket import subprocess -from dataclasses import dataclass +from dataclasses import dataclass, field import httpx log = logging.getLogger(__name__) -_DOH_GOOGLE = "https://dns.google/resolve?name=whoami.dnsleaktest.com&type=A" +# ─── DoH endpoint used to ask "what is my apparent source IP?" ────────────── +_DOH_IP_CHECK = "https://api.ipify.org?format=json" +_DNS_LEAK_API = "https://dnsleaktest.com/api/v1/start" +_DNS_LEAK_RESULT = "https://dnsleaktest.com/api/v1/results" +# Well-known public resolver IP ranges (prefix → label) +_PUBLIC_DNS_LABELS: dict[str, str] = { + "8.8.8.8": "Google", "8.8.4.4": "Google", + "1.1.1.1": "Cloudflare", "1.0.0.1": "Cloudflare", + "9.9.9.9": "Quad9", "149.112.112.112": "Quad9", + "208.67.222.222": "OpenDNS", "208.67.220.220": "OpenDNS", + "76.76.2.0": "Alternate DNS", + "94.140.14.14": "AdGuard", +} + +# ───────────────────────────────────────────────────────────────────────────── @dataclass class DnsLeakResult: + """Legacy single-check result kept for API compatibility.""" ok: bool system_resolvers: list[str] message: str doh_ip: str | None = None +@dataclass +class DnsResolverInfo: + ip: str + hostname: str = "" + country: str = "" + isp: str = "" + label: str = "" # "Google", "Cloudflare", or "" + is_public_known: bool = False + + +@dataclass +class FullDnsLeakReport: + """Comprehensive DNS leak report from run_dns_leak_test().""" + # Resolvers that answered DNS queries in this session + resolvers_seen: list[DnsResolverInfo] = field(default_factory=list) + # Resolvers configured by the OS + system_resolvers: list[str] = field(default_factory=list) + # IPs resolved for test hostnames — via proxy vs direct + proxy_resolution: dict[str, list[str]] = field(default_factory=dict) + direct_resolution: dict[str, list[str]] = field(default_factory=dict) + # True when proxy and direct give the same answers (possible leak) + resolution_matches_direct: bool = False + # Whether any configured resolver is a known public server + has_public_resolver: bool = False + # Whether any configured resolver is outside local network + has_external_resolver: bool = False + # Overall verdict + leaked: bool = False + summary: str = "" + errors: list[str] = field(default_factory=list) + + +# ───────────────────────────────────────────────────────────────────────────── +# System resolver detection +# ───────────────────────────────────────────────────────────────────────────── + def get_system_dns_servers() -> list[str]: + """Return configured IPv4 DNS resolvers via PowerShell (falls back to ipconfig).""" try: r = subprocess.run( ["powershell", "-NoProfile", "-Command", "(Get-DnsClientServerAddress -AddressFamily IPv4 | " "Where-Object { $_.ServerAddresses } | " "Select-Object -ExpandProperty ServerAddresses) -join ','"], - capture_output=True, - text=True, - timeout=12, + capture_output=True, text=True, timeout=12, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) raw = (r.stdout or "").strip() - if not raw: - return [] - return [x.strip() for x in raw.replace(";", ",").split(",") if x.strip()] + if raw: + return [x.strip() for x in raw.replace(";", ",").split(",") if x.strip()] + except Exception: + pass + + # ipconfig fallback + try: + r = subprocess.run( + ["ipconfig", "/all"], capture_output=True, text=True, timeout=10, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + servers: list[str] = [] + for line in (r.stdout or "").splitlines(): + if "DNS Servers" in line or "DNS Server" in line: + parts = line.split(":", 1) + if len(parts) == 2: + ip = parts[1].strip() + if ip: + servers.append(ip) + return servers except Exception: return [] @@ -43,70 +120,194 @@ def get_system_dns_servers() -> list[str]: def flush_dns_cache() -> tuple[bool, str]: try: r = subprocess.run( - ["ipconfig", "/flushdns"], - capture_output=True, - text=True, - timeout=15, + ["ipconfig", "/flushdns"], capture_output=True, text=True, timeout=15, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) - msg = (r.stdout or r.stderr or "").strip().splitlines()[-1] if r.returncode == 0 else "flush failed" + lines = (r.stdout or r.stderr or "").strip().splitlines() + msg = lines[-1] if lines else "flushed" return r.returncode == 0, msg except Exception as e: return False, str(e) -def check_dns_leak_hint(local_proxy: str | None = None) -> DnsLeakResult: - """Heuristic: list configured DNS servers; note if any are public ISP resolvers. +# ───────────────────────────────────────────────────────────────────────────── +# Hostname resolution helpers +# ───────────────────────────────────────────────────────────────────────────── - Full DNS leak testing needs OS-level routing; this flags obvious misconfig. - """ - resolvers = get_system_dns_servers() - private_prefixes = ("127.", "10.", "192.168.", "172.16.", "172.17.", "172.18.", - "172.19.", "172.2", "172.30.", "172.31.", "0.0.0.0") - public = [r for r in resolvers if not any(r.startswith(p) for p in private_prefixes)] - - doh_ip: str | None = None +def _resolve_direct(hostname: str, timeout: float = 4.0) -> list[str]: + """Resolve hostname using system DNS (direct, not through proxy).""" try: - with httpx.Client(timeout=8.0, verify=True) as c: - r = c.get(_DOH_GOOGLE) + socket.setdefaulttimeout(timeout) + infos = socket.getaddrinfo(hostname, None) + seen: list[str] = [] + for info in infos: + addr = info[4][0] + if addr and addr not in seen: + seen.append(addr) + return seen + except Exception: + return [] + finally: + socket.setdefaulttimeout(None) + + +def _resolve_via_proxy(hostname: str, proxy_url: str, timeout: float = 8.0) -> list[str]: + """Resolve hostname by fetching a DNS-over-HTTPS endpoint through the proxy.""" + # Use Google DoH JSON API through the proxy so DNS is resolved on the exit node + doh = f"https://dns.google/resolve?name={hostname}&type=A" + try: + with httpx.Client(proxy=proxy_url, timeout=timeout, verify=False, follow_redirects=True) as c: + r = c.get(doh) if r.status_code == 200: data = r.json() - answers = data.get("Answer") or [] - if answers: - doh_ip = str(answers[0].get("data", "")) + return [ans["data"] for ans in (data.get("Answer") or []) + if ans.get("type") == 1] except Exception: pass + return [] + + +def _is_private(ip: str) -> bool: + try: + return ipaddress.ip_address(ip).is_private + except Exception: + return False + + +# ───────────────────────────────────────────────────────────────────────────── +# Main leak test +# ───────────────────────────────────────────────────────────────────────────── + +_LEAK_TEST_HOSTS = [ + "google.com", + "cloudflare.com", + "github.com", + "amazon.com", +] + + +def run_dns_leak_test( + proxy_url: str | None = None, + timeout: float = 10.0, +) -> FullDnsLeakReport: + """ + Comprehensive DNS leak test: + - Collect system DNS resolver configuration + - Resolve test hostnames both directly (system DNS) and via proxy (DoH through chain) + - Compare answers: matching answers suggest DNS is NOT going through the proxy + - Flag known public resolvers that would bypass the chain + """ + rep = FullDnsLeakReport() + rep.system_resolvers = get_system_dns_servers() + + _PRIVATE_PREFIXES = ( + "127.", "10.", "192.168.", "172.16.", "172.17.", "172.18.", + "172.19.", "172.2", "172.30.", "172.31.", "169.254.", + ) + + for r_ip in rep.system_resolvers: + is_public_known = r_ip in _PUBLIC_DNS_LABELS + is_external = not any(r_ip.startswith(p) for p in _PRIVATE_PREFIXES) + label = _PUBLIC_DNS_LABELS.get(r_ip, "") + rep.has_public_resolver = rep.has_public_resolver or is_public_known + rep.has_external_resolver = rep.has_external_resolver or is_external + rep.resolvers_seen.append(DnsResolverInfo( + ip=r_ip, + label=label, + is_public_known=is_public_known, + )) + + # Resolve test hosts in parallel + all_match = True + any_resolved_via_proxy = False + + def _test_host(host: str) -> tuple[str, list[str], list[str]]: + direct = _resolve_direct(host, timeout=min(4.0, timeout)) + proxy = _resolve_via_proxy(host, proxy_url, timeout=timeout) if proxy_url else [] + return host, direct, proxy + + try: + with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: + futs = {pool.submit(_test_host, h): h for h in _LEAK_TEST_HOSTS} + for fut in concurrent.futures.as_completed(futs, timeout=timeout + 2): + try: + host, direct, proxy = fut.result() + rep.direct_resolution[host] = direct + rep.proxy_resolution[host] = proxy + if proxy: + any_resolved_via_proxy = True + # Overlap: if proxy and direct gave the same IPs, DNS + # might not be going through the proxy (same resolver path) + direct_set = set(direct) + proxy_set = set(proxy) + if direct_set and proxy_set and direct_set == proxy_set: + pass # this host matches + else: + all_match = False + except Exception as e: + rep.errors.append(str(e)) + except Exception as e: + rep.errors.append(f"Thread pool error: {e}") + + rep.resolution_matches_direct = all_match and any_resolved_via_proxy + + # ── Build verdict ────────────────────────────────────────────────────── + problems: list[str] = [] + + if rep.has_external_resolver: + external = [r for r in rep.system_resolvers + if not any(r.startswith(p) for p in _PRIVATE_PREFIXES)] + names = [f"{ip} ({_PUBLIC_DNS_LABELS[ip]})" if ip in _PUBLIC_DNS_LABELS else ip + for ip in external[:4]] + problems.append(f"OS DNS: {', '.join(names)} — DNS may bypass proxy chain") + + if rep.resolution_matches_direct and proxy_url: + problems.append("Proxy DNS resolution matches direct — possible DNS leak (same upstream resolver)") + + if problems: + rep.leaked = True + rep.summary = " · ".join(problems) + else: + rep.leaked = False + if proxy_url: + rep.summary = "DNS routing looks clean — proxy resolves differently from direct." + else: + resolver_str = ", ".join(rep.system_resolvers[:3]) if rep.system_resolvers else "none detected" + rep.summary = f"System resolvers: {resolver_str}" + + return rep + + +# ───────────────────────────────────────────────────────────────────────────── +# Legacy quick hint (used by Privacy tab DNS section) +# ───────────────────────────────────────────────────────────────────────────── + +def check_dns_leak_hint(local_proxy: str | None = None) -> DnsLeakResult: + """Fast heuristic: inspect configured DNS resolvers, flag public ones.""" + resolvers = get_system_dns_servers() + private_prefixes = ( + "127.", "10.", "192.168.", "172.16.", "172.17.", "172.18.", + "172.19.", "172.2", "172.30.", "172.31.", + ) + public = [r for r in resolvers if not any(r.startswith(p) for p in private_prefixes)] if not resolvers: - return DnsLeakResult( - ok=True, - system_resolvers=[], - message="No IPv4 DNS servers reported (DHCP may assign later).", - doh_ip=doh_ip, - ) + return DnsLeakResult(ok=True, system_resolvers=[], message="No IPv4 DNS servers reported (DHCP).") if public and local_proxy: return DnsLeakResult( - ok=False, - system_resolvers=resolvers, + ok=False, system_resolvers=resolvers, message=( f"DNS may bypass proxy chain: public resolvers {', '.join(public)}. " "Use kill-switch + VPN, or set DNS to localhost when hardened." ), - doh_ip=doh_ip, ) - - if public and not local_proxy: + if public: return DnsLeakResult( - ok=False, - system_resolvers=resolvers, + ok=False, system_resolvers=resolvers, message=f"Public DNS resolvers active: {', '.join(public)}", - doh_ip=doh_ip, ) - return DnsLeakResult( - ok=True, - system_resolvers=resolvers, + ok=True, system_resolvers=resolvers, message=f"DNS servers: {', '.join(resolvers)}", - doh_ip=doh_ip, ) diff --git a/proxy_chain_manager/fingerprint.py b/proxy_chain_manager/fingerprint.py index 3b93bb5..f8349d4 100644 --- a/proxy_chain_manager/fingerprint.py +++ b/proxy_chain_manager/fingerprint.py @@ -195,35 +195,186 @@ def apply_webrtc_hardening(enable: bool) -> tuple[bool, str]: return False, str(e) +def _get_os_info() -> str: + try: + import platform + v = platform.version() + r = platform.release() + m = platform.machine() + return f"Windows {r} build {v} {m}" + except Exception: + return "unknown" + + +def _get_timezone() -> str: + try: + import datetime + tz = datetime.datetime.now(datetime.timezone.utc).astimezone() + name = str(tz.tzname() or "") + offset = tz.utcoffset() + h = int(offset.total_seconds() // 3600) if offset else 0 + return f"{name} (UTC{h:+d})" if name else f"UTC{h:+d}" + except Exception: + return "unknown" + + +def _get_screen_resolution() -> str: + try: + r = subprocess.run( + ["powershell", "-NoProfile", "-Command", + "Add-Type -AssemblyName System.Windows.Forms;" + "[System.Windows.Forms.Screen]::PrimaryScreen.Bounds | " + "ForEach-Object { \"$($_.Width)x$($_.Height)\" }"], + capture_output=True, text=True, timeout=8, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + return (r.stdout or "").strip() or "unknown" + except Exception: + return "unknown" + + +def check_browser_fingerprint_consistency(profile_dir: "Path") -> list[str]: + """ + Verify that the managed Firefox user.js is internally consistent and + doesn't mix prefs that would expose the browser as fingerprint-hardened + while still advertising a normal user-agent. + + Returns a list of warning strings (empty = consistent). + """ + from pathlib import Path + user_js = Path(profile_dir) / "user.js" + if not user_js.is_file(): + return ["user.js not found — profile not yet initialized"] + + try: + content = user_js.read_text(encoding="utf-8", errors="replace") + except Exception as e: + return [f"Cannot read user.js: {e}"] + + warnings: list[str] = [] + + def _has(pref: str, value: str) -> bool: + return f'"{pref}", {value}' in content + + # 1. RFP + custom UA is a contradiction (RFP overrides UA) + if _has("privacy.resistFingerprinting", "true") and 'general.useragent.override' in content: + warnings.append( + "RFP + UA override conflict: privacy.resistFingerprinting=true overrides " + "general.useragent.override — the spoofed UA is ignored" + ) + + # 2. proxy not set but force_proxy claimed + if _has("network.proxy.type", "1"): + if 'network.proxy.http",' not in content: + warnings.append("proxy.type=1 but network.proxy.http is missing — browsers will error") + + # 3. FPI on + third-party cookies allowed = inconsistent privacy posture + if _has("privacy.firstparty.isolate", "true") and _has("network.cookie.cookieBehavior", "0"): + warnings.append( + "FPI enabled but cookieBehavior=0 (accept all) — cookies are isolated but not blocked" + ) + + # 4. sanitizeOnShutdown without clearing history = incomplete wipe + if _has("privacy.sanitize.sanitizeOnShutdown", "true"): + if not _has("privacy.clearOnShutdown.history", "true"): + warnings.append( + "sanitizeOnShutdown=true but clearOnShutdown.history not set — " + "history may survive session" + ) + + # 5. WebRTC peerconnection disabled is good — flag if missing + if not _has("media.peerconnection.enabled", "false"): + warnings.append("WebRTC not disabled in profile — real IP can leak via STUN") + + # 6. network.trr.mode=5 (DoH off) is expected when using proxy DNS + if not _has("network.trr.mode", "5"): + warnings.append( + "network.trr.mode is not 5 — Firefox may use its own DoH resolver, " + "bypassing the proxy chain DNS path" + ) + + return warnings + + def audit_device() -> FingerprintAudit: - """Collect identifiers sites and trackers often fingerprint.""" + """Collect OS-level identifiers and consistency notes.""" import getpass import os + from pathlib import Path host = get_computer_name() guid = get_machine_guid() user = getpass.getuser() - macs = [f"{n.name}: {n.mac}" for n in list_nics()] + nics = list_nics() + macs = [f"{n.name}: {n.mac}" for n in nics] + os_ = _get_os_info() + tz_ = _get_timezone() + res_ = _get_screen_resolution() - lines = [ - f"Computer name: {host or '—'}", - f"Windows username: {user}", - f"MachineGuid: {guid[:8]}…{guid[-4:]}" if len(guid) > 12 else f"MachineGuid: {guid or '—'}", - f"Network adapters ({len(macs)}):", + lines: list[str] = [ + "── OS / Identity ──────────────────────────────────", + f" OS : {os_}", + f" Computer name: {host or '—'}", + f" Username : {user}", + f" MachineGuid : " + (f"{guid[:8]}…{guid[-4:]}" if len(guid) > 12 else guid or "—"), + f" Timezone : {tz_}", + f" Screen res : {res_}", + "", + "── Network adapters ───────────────────────────────", ] - lines.extend([f" • {m}" for m in macs[:8]] or [" • (none detected)"]) - if len(macs) > 8: - lines.append(f" … and {len(macs) - 8} more") + lines.extend([f" • {m}" for m in macs[:10]] or [" • (none detected)"]) + if len(macs) > 10: + lines.append(f" … and {len(macs) - 10} more") - lines.append("") - lines.append("Browser fingerprint (Canvas/WebGL/fonts) is not changed by this app.") - lines.append("Use hardened browser profiles + proxy chain for web traffic.") - lines.append("WebRTC toggle here affects Chrome/Edge system policy only.") + lines += [ + "", + "── Browser fingerprint note ───────────────────────", + " Canvas, WebGL, font metrics, audio context, and screen dimensions", + " are NOT changed at the OS level by this app.", + " → Use 'blend_windows_chrome' persona OR 'hardened' mode to control", + " what the browser reports to sites.", + " → RFP (Resist Fingerprinting) makes Firefox stand out as hardened.", + " → Blend persona spoofs UA/platform to look like a normal Windows Chrome.", + "", + "── Recommendations ────────────────────────────────", + ] + + # Quick consistency checks + webrtc_ok, _ = apply_webrtc_hardening.__doc__ and True or False + try: + import winreg as _wr + with _wr.OpenKey(_wr.HKEY_LOCAL_MACHINE, r"SOFTWARE\Policies\Google\Chrome", + 0, _wr.KEY_QUERY_VALUE) as k: + v = int(_wr.QueryValueEx(k, "DefaultWebRtcIpHandlingPolicy")[0]) + webrtc_ok = v == 2 + except Exception: + webrtc_ok = False + + if not webrtc_ok: + lines.append(" ⚠ Chrome/Edge WebRTC policy not set — enable in Privacy tab") + else: + lines.append(" ✓ Chrome/Edge WebRTC policy is set") + + # Check IPv6 + try: + r = subprocess.run( + ["powershell", "-NoProfile", "-NonInteractive", "-Command", + "Get-NetAdapterBinding -ComponentID ms_tcpip6 | Where-Object { $_.Enabled } | Measure-Object | Select-Object -ExpandProperty Count"], + capture_output=True, text=True, timeout=8, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + count = int((r.stdout or "0").strip() or "0") + if count > 0: + lines.append(f" ⚠ IPv6 active on {count} adapter(s) — disable in Privacy tab for full v4-only chain") + else: + lines.append(" ✓ IPv6 disabled on all adapters") + except Exception: + lines.append(" ? IPv6 status unknown") return FingerprintAudit( lines=lines, hostname=host, machine_guid=guid, username=user, - macs=[n.mac for n in list_nics()], + macs=[n.mac for n in nics], ) diff --git a/proxy_chain_manager/signup_prep.py b/proxy_chain_manager/signup_prep.py index f378dec..f919c4c 100644 --- a/proxy_chain_manager/signup_prep.py +++ b/proxy_chain_manager/signup_prep.py @@ -13,7 +13,16 @@ from typing import Any from .paths import app_data_dir _EXT_ID = "signup-autofill@proxygod" -_EXT_SRC = Path(__file__).resolve().parent / "signup_extension" + + +def _ext_src_path() -> Path: + """Locate the signup_extension dir whether running from source or frozen.""" + import sys + if getattr(sys, "frozen", False) and hasattr(sys, "_MEIPASS"): + return Path(sys._MEIPASS) / "proxy_chain_manager" / "signup_extension" + return Path(__file__).resolve().parent / "signup_extension" + + @dataclass(frozen=True) @@ -389,7 +398,7 @@ def install_signup_extension(profile_dir: Path, draft: SignupDraft) -> None: dest = profile_dir / "extensions" / ext_name if dest.exists(): shutil.rmtree(dest, ignore_errors=True) - shutil.copytree(_EXT_SRC, dest) + shutil.copytree(_ext_src_path(), dest) cfg = build_autofill_config(draft) (dest / "autofill_config.json").write_text(json.dumps(cfg, indent=2), encoding="utf-8") diff --git a/proxy_chain_manager/webrtc_check.py b/proxy_chain_manager/webrtc_check.py new file mode 100644 index 0000000..30dc6e2 --- /dev/null +++ b/proxy_chain_manager/webrtc_check.py @@ -0,0 +1,211 @@ +"""WebRTC leak detection and hardening checks. + +WebRTC leaks expose your real IP through STUN/ICE even when a proxy is set. +This module checks every controllable surface: + + 1. OS-level Chrome / Edge Group Policy (prevents non-proxied UDP) + 2. Firefox managed profile prefs (user.js must disable PeerConnection) + 3. STUN UDP reachability (if STUN is reachable, WebRTC CAN leak) + 4. Firefox process prefs (confirm our user.js was accepted) +""" +from __future__ import annotations + +import logging +import socket +import subprocess +import winreg +from dataclasses import dataclass, field +from pathlib import Path + +log = logging.getLogger(__name__) + +# STUN servers commonly used by WebRTC +_STUN_HOSTS: list[tuple[str, int]] = [ + ("stun.l.google.com", 19302), + ("stun1.l.google.com", 19302), + ("stun.cloudflare.com", 3478), + ("stun.stunprotocol.org", 3478), +] + +# Firefox user.js prefs that block WebRTC +_REQUIRED_WEBRTC_PREFS: dict[str, str] = { + "media.peerconnection.enabled": "false", + "media.peerconnection.ice.default_address_only": "true", + "media.peerconnection.ice.no_host": "true", +} + +# Chrome / Edge HKLM policy keys +_CHROMIUM_POLICY_PATHS = ( + r"SOFTWARE\Policies\Google\Chrome", + r"SOFTWARE\Policies\Microsoft\Edge", + r"SOFTWARE\Policies\Chromium", +) +_WEBRTC_POLICY_VALUE = "DefaultWebRtcIpHandlingPolicy" +_WEBRTC_BLOCK_VALUE = 2 # "default_public_and_private_interfaces" + + +@dataclass +class WebRtcCheckResult: + """Aggregated WebRTC leak surface scan.""" + # Policy / config checks + chrome_edge_policy_set: bool = False + chrome_edge_policy_detail: str = "" + firefox_prefs_ok: bool = False + firefox_prefs_detail: str = "" + # Network reachability + stun_reachable: bool = False + stun_detail: str = "" + # Convenience flags + any_leak_risk: bool = True + issues: list[str] = field(default_factory=list) + checks: list[tuple[str, bool, str]] = field(default_factory=list) # (name, ok, detail) + + def add(self, name: str, ok: bool, detail: str) -> None: + self.checks.append((name, ok, detail)) + if not ok: + self.issues.append(f"{name}: {detail}") + + +# ───────────────────────────────────────────────────────────────────────────── +# 1. Chrome / Edge Group Policy +# ───────────────────────────────────────────────────────────────────────────── + +def check_chromium_webrtc_policy() -> tuple[bool, str]: + """Return (policy_set, detail_string).""" + found: list[str] = [] + missing: list[str] = [] + for path in _CHROMIUM_POLICY_PATHS: + try: + with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, path, 0, winreg.KEY_QUERY_VALUE) as k: + try: + val = int(winreg.QueryValueEx(k, _WEBRTC_POLICY_VALUE)[0]) + name = path.split("\\")[-1] + if val == _WEBRTC_BLOCK_VALUE: + found.append(name) + else: + missing.append(f"{name}={val} (need {_WEBRTC_BLOCK_VALUE})") + except OSError: + missing.append(path.split("\\")[-1] + " key missing") + except OSError: + continue # key not present at all — browser not installed or not policy-managed + + if found: + return True, "Policy set: " + ", ".join(found) + if missing: + return False, "Not set: " + "; ".join(missing) + return False, "No Chrome/Edge/Chromium policy keys found (browsers may not be installed)" + + +# ───────────────────────────────────────────────────────────────────────────── +# 2. Firefox managed profile prefs +# ───────────────────────────────────────────────────────────────────────────── + +def check_firefox_webrtc_prefs(profile_dir: Path) -> tuple[bool, str]: + """Verify our written user.js contains the required WebRTC-disabling prefs.""" + user_js = profile_dir / "user.js" + if not user_js.is_file(): + return False, f"user.js not found at {profile_dir} — profile not yet initialized" + + try: + content = user_js.read_text(encoding="utf-8", errors="replace") + except Exception as e: + return False, f"Cannot read user.js: {e}" + + missing: list[str] = [] + for pref, expected_val in _REQUIRED_WEBRTC_PREFS.items(): + # Look for: user_pref("pref.name", value); + needle = f'"{pref}", {expected_val}' + if needle not in content: + missing.append(pref) + + if missing: + return False, "Missing in user.js: " + ", ".join(missing) + return True, "All WebRTC prefs present in managed profile" + + +# ───────────────────────────────────────────────────────────────────────────── +# 3. STUN server UDP reachability +# ───────────────────────────────────────────────────────────────────────────── + +def _probe_stun_udp(host: str, port: int, timeout: float = 2.5) -> bool: + """ + Send a minimal STUN Binding Request over UDP and wait for a response. + Returns True if we get *any* response (STUN or ICMP unreachable). + """ + # Minimal STUN Binding Request (RFC 5389) + msg = ( + b"\x00\x01" # Message Type: Binding Request + b"\x00\x00" # Message Length: 0 + b"\x21\x12\xa4\x42" # Magic Cookie + + b"\x00" * 12 # Transaction ID (12 bytes) + ) + try: + ip = socket.gethostbyname(host) + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.settimeout(timeout) + s.sendto(msg, (ip, port)) + data, _ = s.recvfrom(512) + s.close() + return len(data) > 0 + except socket.timeout: + return False + except Exception: + return False + + +def check_stun_reachability(timeout: float = 3.0) -> tuple[bool, str]: + """ + Returns (reachable, detail). + reachable = True means STUN servers ARE reachable → WebRTC COULD leak. + """ + reachable: list[str] = [] + for host, port in _STUN_HOSTS[:3]: + try: + if _probe_stun_udp(host, port, timeout): + reachable.append(f"{host}:{port}") + except Exception: + continue + + if reachable: + return True, "STUN reachable: " + ", ".join(reachable) + " — WebRTC UDP is NOT blocked" + return False, "STUN servers unreachable — WebRTC UDP appears blocked (good)" + + +# ───────────────────────────────────────────────────────────────────────────── +# Composite scan +# ───────────────────────────────────────────────────────────────────────────── + +def run_webrtc_check(profile_dir: Path | None = None) -> WebRtcCheckResult: + """ + Run all WebRTC leak surface checks. + + profile_dir: path to the managed Firefox profile (to verify user.js). + """ + res = WebRtcCheckResult() + + # 1. Chrome/Edge policy + policy_ok, policy_detail = check_chromium_webrtc_policy() + res.chrome_edge_policy_set = policy_ok + res.chrome_edge_policy_detail = policy_detail + res.add("Chrome/Edge WebRTC policy (HKLM)", policy_ok, policy_detail) + + # 2. Firefox profile prefs + if profile_dir is not None: + fx_ok, fx_detail = check_firefox_webrtc_prefs(profile_dir) + res.firefox_prefs_ok = fx_ok + res.firefox_prefs_detail = fx_detail + res.add("Firefox managed profile WebRTC prefs", fx_ok, fx_detail) + else: + res.add("Firefox managed profile WebRTC prefs", False, + "Profile not specified — not checked") + + # 3. STUN UDP reachability + stun_reachable, stun_detail = check_stun_reachability() + res.stun_reachable = stun_reachable + res.stun_detail = stun_detail + # STUN reachable = risk; NOT reachable = good (firewall is blocking it) + res.add("STUN UDP reachability", not stun_reachable, stun_detail) + + # Overall verdict + res.any_leak_risk = bool(res.issues) + return res diff --git a/proxy_chain_manager/world_map.png b/proxy_chain_manager/world_map.png new file mode 100644 index 0000000..9634198 Binary files /dev/null and b/proxy_chain_manager/world_map.png differ