From d4296763eef0078bc0d0dbd18f2f4076e87c4ba5 Mon Sep 17 00:00:00 2001 From: Indiana Holmes Date: Sat, 16 May 2026 13:50:05 -0700 Subject: [PATCH] Build comprehensive privacy suite and hardened browser controls. Adds VPN-aware leak handling, chain testing UX improvements, hardened Firefox launch/profile management, privacy/device hardening modules, and tray/status upgrades so the app is production-ready as the new baseline. Co-authored-by: Cursor --- ProxyChainManager.spec | 2 +- proxy_chain_manager/app.py | 773 ++++++++++++++++++++---- proxy_chain_manager/browser_launcher.py | 132 ++++ proxy_chain_manager/browser_profile.py | 135 +++++ proxy_chain_manager/config.py | 25 +- proxy_chain_manager/dns_leak.py | 112 ++++ proxy_chain_manager/fingerprint.py | 221 +++++++ proxy_chain_manager/firewall.py | 29 +- proxy_chain_manager/gui_validate.py | 41 ++ proxy_chain_manager/leak_detect.py | 55 ++ proxy_chain_manager/mac_spoof.py | 113 ++++ proxy_chain_manager/pool_ops.py | 106 ++++ proxy_chain_manager/proxy_picker.py | 201 ++++++ proxy_chain_manager/service.py | 151 +++-- proxy_chain_manager/tray.py | 116 +++- proxy_chain_manager/vpn_detect.py | 156 +++++ tests/test_proxy_god.py | 28 +- 17 files changed, 2158 insertions(+), 238 deletions(-) create mode 100644 proxy_chain_manager/browser_launcher.py create mode 100644 proxy_chain_manager/browser_profile.py create mode 100644 proxy_chain_manager/dns_leak.py create mode 100644 proxy_chain_manager/fingerprint.py create mode 100644 proxy_chain_manager/gui_validate.py create mode 100644 proxy_chain_manager/leak_detect.py create mode 100644 proxy_chain_manager/mac_spoof.py create mode 100644 proxy_chain_manager/pool_ops.py create mode 100644 proxy_chain_manager/proxy_picker.py create mode 100644 proxy_chain_manager/vpn_detect.py diff --git a/ProxyChainManager.spec b/ProxyChainManager.spec index 7f7c3bd..eb6b0a7 100644 --- a/ProxyChainManager.spec +++ b/ProxyChainManager.spec @@ -9,7 +9,7 @@ datas += tmp_ret[0]; binaries += tmp_ret[1]; hiddenimports += tmp_ret[2] a = Analysis( - ['run.py'], + ['C:\\Users\\india\\Desktop\\proxy god\\proxy-god\\run.py'], pathex=[], binaries=binaries, datas=datas, diff --git a/proxy_chain_manager/app.py b/proxy_chain_manager/app.py index 072f341..d38ec57 100644 --- a/proxy_chain_manager/app.py +++ b/proxy_chain_manager/app.py @@ -1,14 +1,18 @@ """ Proxy God — GUI -Tabs: Live | Chain Builder | Settings +Tabs: Live | Chain Builder | Browser | Privacy | Settings """ from __future__ import annotations import logging import queue import sys +import threading +import time +from dataclasses import asdict from datetime import datetime -from typing import Any +from typing import Any, Literal +from tkinter import filedialog import customtkinter as ctk @@ -30,10 +34,21 @@ from .firewall import ( is_engaged as fw_is_engaged, request_admin_relaunch, ) +from .dns_leak import check_dns_leak_hint, flush_dns_cache +from .browser_launcher import ( + BrowserConfig, + BrowserSession, + default_firefox_path, + default_profile_dir, +) +from .fingerprint import audit_device +from .gui_validate import run_validate_each_sync from .paths import app_data_dir +from .proxy_picker import show_proxy_picker from .service import ChainService from .sysproxy import clear_system_proxy, is_system_proxy_set from .tray import TrayIcon +from .vpn_detect import detect_vpn from .windows_task import install_logon_task, task_exists, uninstall_logon_task LOG_PATH = app_data_dir() / "proxy_chain_manager.log" @@ -61,6 +76,7 @@ class UILogHandler(logging.Handler): except Exception: self.handleError(record) + # ── palette ────────────────────────────────────────────────────────────────── BG = "#0d0d1a" CARD = "#12122a" @@ -97,6 +113,9 @@ def main() -> None: ui_q: queue.Queue[dict[str, Any]] = queue.Queue() svc = ChainService(notify=lambda m: ui_q.put(m)) s = load_settings() + browser = BrowserSession() + browser_should_run = [False] + browser_last_launch_ts = [0.0] # ── tray ───────────────────────────────────────────────────────────────── tray = TrayIcon( @@ -106,12 +125,8 @@ def main() -> None: ) tray.start() - # ───────────────────────────────────────────────────────────────────────── - # SHARED STATE - # ───────────────────────────────────────────────────────────────────────── - _hop_count = [s.chain_length] # mutable ref so topbar buttons can mutate it - pulse_job: dict[str, str | None] = {"id": None} + HopState = Literal["untested", "testing", "ok", "fail"] def _cancel_pulse() -> None: if pulse_job["id"]: @@ -171,62 +186,15 @@ def main() -> None: _save_settings() svc.start() - _btn(topbar, "▶ Start", _start, w=86).pack(side="left", padx=2) - _btn(topbar, "■ Stop", svc.stop, w=76).pack(side="left", padx=2) - _btn(topbar, "↻ Rotate", svc.rotate_now, w=76).pack(side="left", padx=2) + start_btn = _btn(topbar, "▶ Start", _start, w=86) + start_btn.pack(side="left", padx=2) + stop_btn = _btn(topbar, "■ Stop", svc.stop, w=76) + stop_btn.pack(side="left", padx=2) + rotate_btn = _btn(topbar, "↻ Rotate", svc.rotate_now, w=76) + rotate_btn.pack(side="left", padx=2) _btn(topbar, "Quit", lambda: root.after(0, _close), w=52, fg_color="#7f1d1d", hover_color="#991b1b").pack(side="left", padx=(10, 2)) - # Hop count +/- control - ctk.CTkLabel(topbar, text="│", text_color=DIM).pack(side="left", padx=6) - ctk.CTkLabel(topbar, text="Hops:", font=(FONT, 11), text_color=TEXT2).pack(side="left", padx=(0, 2)) - - hop_count_lbl = ctk.CTkLabel(topbar, text=str(_hop_count[0]), - font=(FONT, 15, "bold"), text_color=CYAN, width=24) - hop_count_lbl.pack(side="left") - - def _hop_adjust(delta: int) -> None: - new = max(1, min(8, _hop_count[0] + delta)) - _hop_count[0] = new - hop_count_lbl.configure(text=str(new)) - # Update hop slider in Chain Builder if it exists - try: - hop_slider.set(new) - hop_val_lbl.configure(text=str(new)) - except Exception: - pass - # Immediately apply to settings - _apply_hop_count(new) - - _btn(topbar, "−", lambda: _hop_adjust(-1), w=26, h=26, - fg_color=PANEL, hover_color=ACCENT).pack(side="left", padx=1) - _btn(topbar, "+", lambda: _hop_adjust(+1), w=26, h=26, - fg_color=PANEL, hover_color=ACCENT).pack(side="left", padx=(1, 6)) - - def _apply_hop_count(n: int) -> None: - """Save hop count change to live settings immediately.""" - cur = svc.settings - ns = Settings( - local_host=cur.local_host, - local_port=cur.local_port, - chain_length=n, - obfuscation_mode=cur.obfuscation_mode, - use_pinned_chain=cur.use_pinned_chain, - pinned_chain=cur.pinned_chain, - manual_exit_proxy=cur.manual_exit_proxy, - health_check_seconds=cur.health_check_seconds, - full_refresh_seconds=cur.full_refresh_seconds, - validation_concurrency=cur.validation_concurrency, - max_candidates=cur.max_candidates, - validation_timeout_seconds=cur.validation_timeout_seconds, - prefer_elite=cur.prefer_elite, - kill_switch_enabled=cur.kill_switch_enabled, - proxy_bypass=cur.proxy_bypass, - sources=cur.sources, - ip_check_url=cur.ip_check_url, - ) - svc.update_settings(ns) - # Boot buttons ctk.CTkLabel(topbar, text="│", text_color=DIM).pack(side="left", padx=4) boot_lbl = ctk.CTkLabel(topbar, text="Boot:?", font=(FONT, 10), text_color=TEXT2) @@ -301,6 +269,7 @@ def main() -> None: exit_ip_lbl.pack(side="right", padx=(8, 14)) hop_dot_labels: list[ctk.CTkLabel] = [] + outer_hop_label = "VPN" # updated when VPN detected; "Direct" if no tunnel def _pulse(step: int = 0) -> None: if not hop_dot_labels: @@ -333,7 +302,7 @@ def main() -> None: _node("YOU", GREEN, bold=True) _arr() - _node("NORD", PURPLE, bold=True) + _node(outer_hop_label[:12], PURPLE, bold=True) _arr() last_i = len(hops) - 1 @@ -391,10 +360,12 @@ def main() -> None: ) tabs.pack(fill="both", expand=True, padx=8, pady=(4, 8)) - tab_live = tabs.add(" Live ") - tab_chain = tabs.add(" Chain Builder ") + tab_live = tabs.add(" Live ") + tab_chain = tabs.add(" Chain Builder ") + tab_browser = tabs.add(" Browser ") + tab_privacy = tabs.add(" Privacy ") tab_settings = tabs.add(" Settings ") - for tab in (tab_live, tab_chain, tab_settings): + for tab in (tab_live, tab_chain, tab_browser, tab_privacy, tab_settings): tab.configure(fg_color=BG) # ─── TAB: LIVE ─────────────────────────────────────────────────────────── @@ -415,7 +386,7 @@ def main() -> None: v_pool = ctk.StringVar(value="0") v_rotation = ctk.StringVar(value="0") - _stat_card(stats_row, "Your IP (through Nord)", v_real_ip) + _stat_card(stats_row, "Direct / VPN IP", v_real_ip) _stat_card(stats_row, "Exit IP (last hop)", v_exit_ip) _stat_card(stats_row, "Valid pool size", v_pool) _stat_card(stats_row, "Rotations this session", v_rotation) @@ -498,35 +469,6 @@ def main() -> None: text_color=TEXT, ).pack(anchor="w", padx=16, pady=3) - ctk.CTkFrame(cb_left, fg_color="transparent", height=8).pack() - ctk.CTkLabel(cb_left, text="Hop Count (also: topbar +/−)", - font=(FONT, 12, "bold"), text_color=TEXT).pack(anchor="w", padx=12) - - hop_val_lbl = ctk.CTkLabel(cb_left, text=str(_hop_count[0]), - font=(FONT, 26, "bold"), text_color=CYAN) - hop_val_lbl.pack(pady=(2, 0)) - - def _on_hop_slider(val: float) -> None: - n = int(val) - _hop_count[0] = n - hop_val_lbl.configure(text=str(n)) - hop_count_lbl.configure(text=str(n)) - _apply_hop_count(n) - - hop_slider = ctk.CTkSlider( - cb_left, from_=1, to=8, number_of_steps=7, - command=_on_hop_slider, - fg_color=PANEL, progress_color=ACCENT2, - button_color=CYAN, button_hover_color=BLUE, - ) - hop_slider.set(s.chain_length) - hop_slider.pack(fill="x", padx=16, pady=(2, 6)) - - hop_legend = ctk.CTkFrame(cb_left, fg_color="transparent") - hop_legend.pack(fill="x", padx=16, pady=(0, 8)) - ctk.CTkLabel(hop_legend, text="1", font=(FONT, 9), text_color=DIM).pack(side="left") - ctk.CTkLabel(hop_legend, text="8", font=(FONT, 9), text_color=DIM).pack(side="right") - elite_var = ctk.BooleanVar(value=s.prefer_elite) ctk.CTkCheckBox( cb_left, text="Elite proxies only (slower but more anonymous)", @@ -536,12 +478,16 @@ def main() -> None: exit_fix_frame = ctk.CTkFrame(cb_left, fg_color=PANEL, corner_radius=8) exit_fix_frame.pack(fill="x", padx=12, pady=(0, 12)) + exit_hdr = ctk.CTkFrame(exit_fix_frame, fg_color="transparent") + exit_hdr.pack(fill="x", padx=10, pady=(8, 2)) ctk.CTkLabel( - exit_fix_frame, - text="Fixed exit — last hop only", + exit_hdr, + text="Exit node (optional final hop)", font=(FONT, 11, "bold"), text_color=ORANGE, - ).pack(anchor="w", padx=10, pady=(8, 2)) + ).pack(side="left") + exit_test_dot = ctk.CTkLabel(exit_hdr, text="●", font=(FONT, 14), text_color=RED) + exit_test_dot.pack(side="left", padx=(8, 4)) exit_proxy_entry = ctk.CTkEntry( exit_fix_frame, placeholder_text="http://host:port or socks5://host:port (optional)", @@ -578,53 +524,122 @@ def main() -> None: if _epw: exit_pass_entry.insert(0, _epw) + exit_btn_row = ctk.CTkFrame(exit_fix_frame, fg_color="transparent") + exit_btn_row.pack(fill="x", padx=10, pady=(0, 4)) + + def _exit_url() -> str: + raw = exit_proxy_entry.get().strip() + if not raw: + return "" + if "://" not in raw: + raw = "http://" + raw + return merge_proxy_credentials(raw, exit_user_entry.get(), exit_pass_entry.get()) + + def _test_exit() -> None: + url = _exit_url() + if not url: + _log("Exit node: enter a proxy URL first.") + return + exit_test_dot.configure(text_color=YELLOW) + test_exit_btn.configure(state="disabled") + + def work() -> None: + def on_result(_u: str, ok: bool) -> None: + root.after(0, lambda: exit_test_dot.configure(text_color=GREEN if ok else RED)) + root.after(0, lambda: _log( + f"Exit node: {'OK' if ok else 'FAILED'} — {_short(url)}" + )) + + run_validate_each_sync( + [url], + svc.settings.ip_check_url, + min(12.0, svc.settings.validation_timeout_seconds), + on_result, + concurrency=1, + ) + root.after(0, lambda: test_exit_btn.configure(state="normal")) + + threading.Thread(target=work, daemon=True).start() + + test_exit_btn = _btn(exit_btn_row, "Test exit", _test_exit, w=88, h=26, + fg_color=ACCENT2, hover_color=ACCENT) + test_exit_btn.pack(side="left") + ctk.CTkLabel( exit_fix_frame, - text="Earlier hops still rotate randomly. Empty = fully automatic exit.\n" - "Auth: use User/Pass (saved in settings) or paste user:pass@host in the URL field.\n" - "Ignored while “Pin & use this chain” is enabled.", + text="Appended after your chain hops when running. Leave empty to use last chain hop as exit.\n" + "Auth: User/Pass fields or user:pass@host in the URL.", font=(FONT, 9), text_color=TEXT2, justify="left", ).pack(anchor="w", padx=10, pady=(0, 8)) - # Right — manual chain editor + # Right — your chain cb_right = ctk.CTkFrame(cb_split, fg_color=CARD, corner_radius=8, width=400) cb_right.pack(side="left", fill="both", expand=True) hdr_row = ctk.CTkFrame(cb_right, fg_color="transparent") - hdr_row.pack(fill="x", padx=12, pady=(12, 4)) - ctk.CTkLabel(hdr_row, text="Manual Chain", - font=(FONT, 12, "bold"), text_color=TEXT).pack(side="left") + hdr_row.pack(fill="x", padx=12, pady=(12, 2)) + ctk.CTkLabel( + hdr_row, text="Your chain", + font=(FONT, 12, "bold"), text_color=TEXT, + ).pack(side="left") + chain_count_lbl = ctk.CTkLabel(hdr_row, text="", font=(FONT, 10), text_color=TEXT2) + chain_count_lbl.pack(side="left", padx=(8, 0)) - use_manual_var = ctk.BooleanVar(value=s.use_pinned_chain) - ctk.CTkCheckBox( - hdr_row, text="Pin & use this chain", + use_manual_var = ctk.BooleanVar(value=True if s.pinned_chain else s.use_pinned_chain) + use_chain_cb = ctk.CTkCheckBox( + hdr_row, text="Use this chain on Start", variable=use_manual_var, font=(FONT, 10), fg_color=ACCENT2, hover_color=ACCENT, text_color=TEXT2, - ).pack(side="right") + ) + use_chain_cb.pack(side="right") + + ctk.CTkLabel( + cb_right, + text="Traffic flows top → bottom. ● red = not tested ● green = live", + font=(FONT, 9), text_color=TEXT2, + ).pack(anchor="w", padx=12, pady=(0, 4)) manual_chain: list[str] = list(s.pinned_chain) + hop_status: dict[str, HopState] = {u: "untested" for u in manual_chain} + status_dots: dict[str, ctk.CTkLabel] = {} + + _HOP_COLOR = {"untested": RED, "testing": YELLOW, "ok": GREEN, "fail": RED} chain_list_frame = ctk.CTkScrollableFrame( - cb_right, fg_color=BG, height=170, corner_radius=6, + cb_right, fg_color=BG, height=200, corner_radius=6, scrollbar_button_color=ACCENT, ) - chain_list_frame.pack(fill="x", padx=12, pady=(0, 6)) + chain_list_frame.pack(fill="x", padx=12, pady=(0, 4)) + + def _update_chain_count() -> None: + n = len(manual_chain) + chain_count_lbl.configure(text=f"({n} hop{'s' if n != 1 else ''})") def _rebuild_chain_ui() -> None: + status_dots.clear() for w in list(chain_list_frame.winfo_children()): w.destroy() for i, hop in enumerate(manual_chain): + if hop not in hop_status: + hop_status[hop] = "untested" _chain_row(i, hop) + _update_chain_count() def _chain_row(idx: int, hop: str) -> None: fr = ctk.CTkFrame(chain_list_frame, fg_color=PANEL, corner_radius=6, height=32) fr.pack(fill="x", pady=2) fr.pack_propagate(False) - ctk.CTkLabel(fr, text=f" {idx+1}.", font=(FONT, 10, "bold"), - text_color=CYAN, width=26).pack(side="left") + st = hop_status.get(hop, "untested") + dot = ctk.CTkLabel(fr, text="●", font=(FONT, 12), + text_color=_HOP_COLOR[st], width=18) + dot.pack(side="left", padx=(4, 0)) + status_dots[hop] = dot + + ctk.CTkLabel(fr, text=f"{idx+1}", font=(FONT, 10, "bold"), + text_color=CYAN, width=22).pack(side="left") proto = hop.split("://")[0].upper() if "://" in hop else "HTTP" badge_c = {"HTTP": "#1e3a8a", "SOCKS5": "#064e3b", "SOCKS4": "#3b1a64"}.get(proto, DIM) @@ -646,7 +661,9 @@ def main() -> None: _rebuild_chain_ui() def _rm(i: int = idx) -> None: + removed = manual_chain[i] del manual_chain[i] + hop_status.pop(removed, None) _rebuild_chain_ui() _btn(fr, "↑", _up, w=26, h=24, fg_color=DIM, hover_color=ACCENT).pack(side="right", padx=1) @@ -655,6 +672,52 @@ def main() -> None: _rebuild_chain_ui() + test_row = ctk.CTkFrame(cb_right, fg_color="transparent") + test_row.pack(fill="x", padx=12, pady=(0, 6)) + + def _set_all_status(state: HopState) -> None: + for u in manual_chain: + hop_status[u] = state + if u in status_dots: + status_dots[u].configure(text_color=_HOP_COLOR[state]) + + def _test_chain() -> None: + if not manual_chain: + _log("Add proxies to your chain first.") + return + test_chain_btn.configure(state="disabled") + _set_all_status("testing") + check_url = svc.settings.ip_check_url + timeout = min(12.0, svc.settings.validation_timeout_seconds) + ok_count = [0] + + def on_result(url: str, ok: bool) -> None: + hop_status[url] = "ok" if ok else "fail" + if ok: + ok_count[0] += 1 + + def ui() -> None: + if url in status_dots: + status_dots[url].configure(text_color=GREEN if ok else RED) + + root.after(0, ui) + + def work() -> None: + run_validate_each_sync(manual_chain, check_url, timeout, on_result) + n_ok, n_tot = ok_count[0], len(manual_chain) + + def done() -> None: + test_chain_btn.configure(state="normal") + _log(f"Chain test: {n_ok}/{n_tot} proxies responded OK.") + + root.after(0, done) + + threading.Thread(target=work, daemon=True).start() + + test_chain_btn = _btn(test_row, "▶ Test entire chain", _test_chain, w=150, h=28, + fg_color=ACCENT2, hover_color=ACCENT) + test_chain_btn.pack(side="left", padx=(0, 8)) + # Add proxy input add_row = ctk.CTkFrame(cb_right, fg_color="transparent") add_row.pack(fill="x", padx=12, pady=(0, 4)) @@ -688,6 +751,7 @@ def main() -> None: else: v = normalize_proxy_url(v) manual_chain.append(v) + hop_status[v] = "untested" _rebuild_chain_ui() add_entry.delete(0, "end") add_user_entry.delete(0, "end") @@ -695,15 +759,27 @@ def main() -> None: _btn(add_row, "+ Add", _add_proxy, w=68, h=28).pack(side="left") - def _paste_current() -> None: - for h in svc.current_chain: - if h not in manual_chain: - manual_chain.append(h) - _rebuild_chain_ui() - _log("Active chain pasted into manual editor.") + chain_btn_row = ctk.CTkFrame(cb_right, fg_color="transparent") + chain_btn_row.pack(pady=(0, 10)) - _btn(cb_right, "↙ Paste active chain", _paste_current, w=180, h=26, - fg_color=DIM, hover_color=ACCENT).pack(pady=(0, 10)) + def _on_picker_done(urls: list[str], mode: str) -> None: + if mode == "replace": + manual_chain.clear() + hop_status.clear() + for u in urls: + if u not in manual_chain: + manual_chain.append(u) + hop_status[u] = "untested" + _rebuild_chain_ui() + _log(f"{'Replaced' if mode == 'replace' else 'Added'} {len(urls)} proxies to your chain.") + + def _open_picker() -> None: + _save_settings() + show_proxy_picker(root, svc.settings, bool(elite_var.get()), _on_picker_done) + + browse_btn = _btn(chain_btn_row, "Browse proxy lists…", _open_picker, w=150, h=26, + fg_color=ACCENT2, hover_color=ACCENT) + browse_btn.pack(side="left") # Sources src_frame = ctk.CTkFrame(tab_chain, fg_color=CARD, corner_radius=8) @@ -718,6 +794,338 @@ def main() -> None: sources_box.pack(fill="x", padx=12, pady=(0, 8)) sources_box.insert("end", "\n".join(s.sources)) + # ─── TAB: BROWSER ───────────────────────────────────────────────────────── + br_scroll = ctk.CTkScrollableFrame(tab_browser, fg_color=BG, scrollbar_button_color=ACCENT) + br_scroll.pack(fill="both", expand=True, padx=4, pady=4) + + def _br_section(title: str, subtitle: str = "") -> ctk.CTkFrame: + ctk.CTkLabel(br_scroll, text=title, font=(FONT, 13, "bold"), text_color=CYAN).pack( + anchor="w", padx=8, pady=(12, 0) + ) + if subtitle: + ctk.CTkLabel( + br_scroll, text=subtitle, font=(FONT, 10), text_color=TEXT2, wraplength=900, justify="left" + ).pack(anchor="w", padx=8, pady=(0, 4)) + f = ctk.CTkFrame(br_scroll, fg_color=CARD, corner_radius=8) + f.pack(fill="x", padx=8, pady=(4, 8)) + return f + + browser_status_card = _br_section( + "Hardened Firefox", + "Launches Firefox in an app-controlled profile that enforces proxy/hardening prefs on every launch.", + ) + browser_status_lbl = ctk.CTkLabel( + browser_status_card, text="Browser status: stopped", font=(FONT, 12, "bold"), text_color=TEXT2 + ) + browser_status_lbl.pack(anchor="w", padx=12, pady=(10, 2)) + browser_hint_lbl = ctk.CTkLabel( + browser_status_card, text="Exit IP on hover in tray while chain is healthy.", font=(FONT, 10), text_color=TEXT2 + ) + browser_hint_lbl.pack(anchor="w", padx=12, pady=(0, 10)) + + browser_path_card = _br_section("Executable and profile", "Set your Firefox path once; profile is kept isolated.") + path_row = ctk.CTkFrame(browser_path_card, fg_color="transparent") + path_row.pack(fill="x", padx=12, pady=(10, 4)) + ctk.CTkLabel(path_row, text="Firefox exe", width=90, anchor="w", font=(FONT, 10), text_color=TEXT2).pack(side="left") + firefox_path_var = ctk.StringVar(value=s.firefox_path or default_firefox_path()) + firefox_path_entry = ctk.CTkEntry(path_row, textvariable=firefox_path_var, height=28, fg_color=BG, border_color=ACCENT) + firefox_path_entry.pack(side="left", fill="x", expand=True, padx=(0, 6)) + + def _pick_firefox_path() -> None: + p = filedialog.askopenfilename( + title="Select firefox.exe", + filetypes=[("Firefox", "firefox.exe"), ("Executable", "*.exe"), ("All files", "*.*")], + ) + if p: + firefox_path_var.set(p) + + _btn(path_row, "Browse", _pick_firefox_path, w=72, h=28, fg_color=DIM, hover_color=ACCENT).pack(side="left") + + profile_row = ctk.CTkFrame(browser_path_card, fg_color="transparent") + profile_row.pack(fill="x", padx=12, pady=(0, 10)) + ctk.CTkLabel(profile_row, text="Profile dir", width=90, anchor="w", font=(FONT, 10), text_color=TEXT2).pack(side="left") + firefox_profile_var = ctk.StringVar(value=s.firefox_profile_dir or default_profile_dir()) + firefox_profile_entry = ctk.CTkEntry( + profile_row, textvariable=firefox_profile_var, height=28, fg_color=BG, border_color=ACCENT + ) + firefox_profile_entry.pack(side="left", fill="x", expand=True) + + browser_toggles_card = _br_section( + "Browser controls", + "These settings are applied each time you launch from this tab.", + ) + br_force_proxy_var = ctk.BooleanVar(value=s.browser_force_proxy) + br_disable_webrtc_var = ctk.BooleanVar(value=s.browser_disable_webrtc) + br_rfp_var = ctk.BooleanVar(value=s.browser_resist_fingerprinting) + br_telemetry_var = ctk.BooleanVar(value=s.browser_disable_telemetry) + br_fpi_var = ctk.BooleanVar(value=s.browser_first_party_isolation) + br_strict_tp_var = ctk.BooleanVar(value=s.browser_strict_tracking_protection) + br_tz_utc_var = ctk.BooleanVar(value=s.browser_timezone_utc) + br_clear_var = ctk.BooleanVar(value=s.browser_clear_on_close) + br_disposable_var = ctk.BooleanVar(value=s.browser_disposable_profile) + br_kill_drop_var = ctk.BooleanVar(value=s.browser_kill_on_chain_drop) + br_auto_relaunch_var = ctk.BooleanVar(value=s.browser_auto_relaunch) + br_lock_profile_var = ctk.BooleanVar(value=s.browser_lock_managed_profile) + + def _br_toggle(parent: Any, text: str, var: ctk.BooleanVar, tip: str = "") -> ctk.CTkCheckBox: + cb = ctk.CTkCheckBox( + parent, text=text, variable=var, font=(FONT, 11), + fg_color=ACCENT2, hover_color=ACCENT, text_color=TEXT, + ) + cb.pack(anchor="w", padx=12, pady=5) + return cb + + _br_toggle( + browser_toggles_card, "Force browser proxy to local chain", br_force_proxy_var, + "Forces Firefox traffic through Proxy God listener. Prevents direct bypass.", + ) + _br_toggle( + browser_toggles_card, "Disable WebRTC peer connections", br_disable_webrtc_var, + "Stops common WebRTC local-IP leaks in browser calls.", + ) + _br_toggle( + browser_toggles_card, "Enable Resist Fingerprinting", br_rfp_var, + "Turns on Firefox anti-fingerprinting mode (slightly changes site behavior).", + ) + _br_toggle( + browser_toggles_card, "Disable telemetry and studies", br_telemetry_var, + "Disables browser diagnostics/study pings.", + ) + _br_toggle( + browser_toggles_card, "Enable First-Party Isolation", br_fpi_var, + "Separates site storage per domain to reduce cross-site tracking.", + ) + _br_toggle( + browser_toggles_card, "Strict tracking protection", br_strict_tp_var, + "Uses stronger tracker/cookie blocking in Firefox.", + ) + _br_toggle( + browser_toggles_card, "Timezone hardening (UTC-friendly timer precision)", br_tz_utc_var, + "Reduces timing precision used in fingerprinting.", + ) + _br_toggle( + browser_toggles_card, "Clear cookies/cache/history on close", br_clear_var, + "Wipes session traces from the managed profile when browser closes.", + ) + _br_toggle( + browser_toggles_card, "Disposable profile (delete on close)", br_disposable_var, + "Deletes the entire managed profile folder on stop/close.", + ) + _br_toggle( + browser_toggles_card, "Kill browser if chain drops", br_kill_drop_var, + "Stops browser immediately if chain becomes unhealthy.", + ) + _br_toggle( + browser_toggles_card, "Auto-relaunch browser while chain is healthy", br_auto_relaunch_var, + "If browser exits unexpectedly, relaunch it automatically while chain stays up.", + ) + _br_toggle( + browser_toggles_card, "Lock to managed profile (recommended)", br_lock_profile_var, + "Launches Firefox only with Proxy God profile and re-applies settings each launch.", + ) + + browser_action_row = ctk.CTkFrame(br_scroll, fg_color="transparent") + browser_action_row.pack(fill="x", padx=8, pady=(2, 14)) + + def _browser_cfg() -> BrowserConfig: + return BrowserConfig( + firefox_path=firefox_path_var.get().strip(), + profile_dir=firefox_profile_var.get().strip(), + clear_on_close=bool(br_clear_var.get()), + disposable_profile=bool(br_disposable_var.get()), + kill_on_chain_drop=bool(br_kill_drop_var.get()), + auto_relaunch=bool(br_auto_relaunch_var.get()), + lock_managed_profile=bool(br_lock_profile_var.get()), + force_proxy=bool(br_force_proxy_var.get()), + disable_webrtc=bool(br_disable_webrtc_var.get()), + resist_fingerprinting=bool(br_rfp_var.get()), + disable_telemetry=bool(br_telemetry_var.get()), + first_party_isolation=bool(br_fpi_var.get()), + strict_tracking_protection=bool(br_strict_tp_var.get()), + timezone_utc=bool(br_tz_utc_var.get()), + ) + + def _refresh_browser_status() -> None: + if browser.is_running(): + browser_status_lbl.configure( + text=f"Browser status: running (pid {browser.pid()})", + text_color=GREEN, + ) + else: + if browser_should_run[0]: + browser_status_lbl.configure(text="Browser status: waiting/restarting", text_color=YELLOW) + else: + browser_status_lbl.configure(text="Browser status: stopped", text_color=TEXT2) + + def _launch_browser_with_cfg(cfg: BrowserConfig) -> bool: + if not svc.current_chain: + _log("Chain not ready yet — starting chain and launching browser now.") + _start() + ok, msg = browser.launch(cfg, LISTEN_HOST, svc.settings.local_port) + _log(msg) + _refresh_browser_status() + return bool(ok) + + def _launch_browser() -> None: + _save_settings() + cfg = _browser_cfg() + browser_should_run[0] = True + if _launch_browser_with_cfg(cfg): + browser_last_launch_ts[0] = time.monotonic() + + def _stop_browser() -> None: + cfg = _browser_cfg() + browser_should_run[0] = False + ok, msg = browser.stop(dispose=cfg.disposable_profile) + _log(msg) + _refresh_browser_status() + + launch_browser_btn = _btn( + browser_action_row, "Launch Hardened Firefox", _launch_browser, w=190, h=34, font=(FONT, 12) + ) + launch_browser_btn.pack(side="left", padx=(0, 8)) + stop_browser_btn = _btn( + browser_action_row, "Stop Browser", _stop_browser, w=110, h=34, + fg_color="#7f1d1d", hover_color="#991b1b", + ) + stop_browser_btn.pack(side="left") + + # ─── TAB: PRIVACY ────────────────────────────────────────────────────────── + priv_scroll = ctk.CTkScrollableFrame(tab_privacy, fg_color=BG, + scrollbar_button_color=ACCENT) + priv_scroll.pack(fill="both", expand=True, padx=4, pady=4) + + def _priv_section(title: str, subtitle: str = "") -> ctk.CTkFrame: + ctk.CTkLabel(priv_scroll, text=title, font=(FONT, 13, "bold"), + text_color=CYAN).pack(anchor="w", padx=8, pady=(12, 0)) + if subtitle: + ctk.CTkLabel(priv_scroll, text=subtitle, font=(FONT, 10), + text_color=TEXT2, wraplength=900, justify="left").pack( + anchor="w", padx=8, pady=(0, 4)) + f = ctk.CTkFrame(priv_scroll, fg_color=CARD, corner_radius=8) + f.pack(fill="x", padx=8, pady=(4, 8)) + return f + + status_card = _priv_section( + "Network identity", + "VPN detection drives leak checks: strict without VPN, subnet-aware with VPN.", + ) + status_row = ctk.CTkFrame(status_card, fg_color="transparent") + status_row.pack(fill="x", padx=12, pady=10) + vpn_status_lbl = ctk.CTkLabel(status_row, text="VPN: scanning…", + font=(FONT, 12, "bold"), text_color=TEXT) + vpn_status_lbl.pack(anchor="w") + leak_mode_lbl = ctk.CTkLabel(status_row, text="Leak check: —", + font=(FONT, 11), text_color=TEXT2) + leak_mode_lbl.pack(anchor="w", pady=(4, 0)) + + def _refresh_vpn_ui() -> None: + vs = detect_vpn() + nonlocal outer_hop_label + if vs.active: + outer_hop_label = vs.short_label()[:12] + vpn_status_lbl.configure( + text=f"VPN: {vs.label}" + (f" ({vs.adapter})" if vs.adapter else ""), + text_color=GREEN, + ) + leak_mode_lbl.configure(text="Leak check: VPN-aware (/16 subnet)") + else: + outer_hop_label = "Direct" + vpn_status_lbl.configure(text="VPN: not detected (direct connection)", + text_color=YELLOW) + leak_mode_lbl.configure(text="Leak check: strict (exit must differ from your IP)") + + _btn(status_card, "Refresh VPN status", _refresh_vpn_ui, w=140, h=28).pack( + anchor="w", padx=12, pady=(0, 10)) + + toggles_card = _priv_section( + "Protections while chain is running", + "Applied on Start, restored on Stop. Admin required for MAC, hostname, IPv6, and WebRTC.", + ) + mac_var = ctk.BooleanVar(value=s.mac_spoof_enabled) + host_var = ctk.BooleanVar(value=s.spoof_hostname_enabled) + dns_flush_var = ctk.BooleanVar(value=s.flush_dns_on_rotate) + ipv6_var = ctk.BooleanVar(value=s.disable_ipv6_while_active) + webrtc_var = ctk.BooleanVar(value=s.harden_webrtc_enabled) + + def _priv_toggle(parent: Any, text: str, var: ctk.BooleanVar, tip: str = "") -> ctk.CTkCheckBox: + cb = ctk.CTkCheckBox( + parent, text=text, variable=var, font=(FONT, 11), + fg_color=ACCENT2, hover_color=ACCENT, text_color=TEXT, + ) + cb.pack(anchor="w", padx=12, pady=6) + return cb + + _priv_toggle( + toggles_card, "Randomize MAC addresses on physical adapters", mac_var, + "Changes NIC MAC values while chain runs (admin required).", + ) + _priv_toggle( + toggles_card, "Spoof computer / NetBIOS hostname", host_var, + "Temporarily renames machine identity while active (admin required).", + ) + _priv_toggle( + toggles_card, "Flush DNS cache when starting or rotating chains", dns_flush_var, + "Clears cached resolver results to reduce stale direct routes.", + ) + _priv_toggle( + toggles_card, "Disable IPv6 on active adapters", ipv6_var, + "Turns off IPv6 bindings while active to reduce IPv6 leak paths.", + ) + _priv_toggle( + toggles_card, + "Harden WebRTC in Chrome / Edge (block non-proxied UDP)", + webrtc_var, + "Applies Windows policy to block direct WebRTC UDP bypass in Chromium browsers.", + ) + + fp_card = _priv_section( + "Device fingerprint", + "OS-level identifiers. Browsers still fingerprint Canvas, fonts, and WebGL separately.", + ) + fp_box = ctk.CTkTextbox(fp_card, height=140, font=("Consolas", 10), + fg_color=BG, text_color=TEXT, + scrollbar_button_color=ACCENT) + fp_box.pack(fill="x", padx=12, pady=(4, 8)) + + def _refresh_fingerprint() -> None: + fp_box.delete("1.0", "end") + audit = audit_device() + fp_box.insert("end", "\n".join(audit.lines)) + + _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) + + 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 ""), + 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 _manual_dns_flush() -> None: + ok, msg = flush_dns_cache() + dns_result_lbl.configure( + text=msg, text_color=GREEN if ok else RED, + ) + + _btn(dns_btn_row, "Flush DNS now", _manual_dns_flush, w=110, h=28, + fg_color=DIM).pack(side="left") + + _btn(priv_scroll, "Save privacy settings", lambda: _save_settings(verbose=True), + w=200, h=34, font=(FONT, 12)).pack(anchor="w", padx=8, pady=12) + # ─── TAB: SETTINGS ─────────────────────────────────────────────────────── sf_outer = ctk.CTkScrollableFrame(tab_settings, fg_color=BG, scrollbar_button_color=ACCENT) @@ -776,6 +1184,46 @@ def main() -> None: fg_color=ACCENT2, hover_color=ACCENT, text_color=TEXT, ).pack(side="left") + def _apply_point_and_shoot() -> None: + """Simple safe defaults: secure + low-friction launch profile.""" + use_manual_var.set(True) + mode_var.set("auto") + elite_var.set(False) + ks_var.set(True) + dns_flush_var.set(True) + mac_var.set(False) + host_var.set(False) + ipv6_var.set(False) + webrtc_var.set(True) + br_force_proxy_var.set(True) + br_disable_webrtc_var.set(True) + br_rfp_var.set(True) + br_telemetry_var.set(True) + br_fpi_var.set(True) + br_strict_tp_var.set(True) + br_tz_utc_var.set(True) + br_clear_var.set(True) + br_disposable_var.set(False) + br_kill_drop_var.set(True) + br_auto_relaunch_var.set(False) + br_lock_profile_var.set(True) + if not firefox_path_var.get().strip(): + firefox_path_var.set(default_firefox_path()) + if not firefox_profile_var.get().strip(): + firefox_profile_var.set(default_profile_dir()) + _save_settings(verbose=False) + _log("⚡ Point-and-shoot preset applied.") + + ready_btn = _btn( + topbar, + "⚡ Ready", + _apply_point_and_shoot, + w=78, + fg_color="#0f766e", + hover_color="#0d9488", + ) + ready_btn.pack(side="left", padx=(8, 2)) + ctk.CTkFrame(sf_outer, fg_color="transparent", height=4).pack() _btn(sf_outer, "Save All Settings", lambda: _save_settings(verbose=True), w=220, h=36, @@ -790,10 +1238,7 @@ def main() -> None: src_list = [ln.strip() for ln in src_raw.splitlines() if ln.strip().startswith("http")] - n_hops = _hop_count[0] - hop_slider.set(n_hops) - hop_val_lbl.configure(text=str(n_hops)) - hop_count_lbl.configure(text=str(n_hops)) + n_hops = max(1, len(manual_chain)) if manual_chain else 1 ns = Settings( local_host=LISTEN_HOST, @@ -818,6 +1263,25 @@ def main() -> None: proxy_bypass=entries["bypass"].get().strip() or Settings().proxy_bypass, sources=src_list or Settings().sources, ip_check_url=entries["check_url"].get().strip() or Settings().ip_check_url, + mac_spoof_enabled=bool(mac_var.get()), + spoof_hostname_enabled=bool(host_var.get()), + flush_dns_on_rotate=bool(dns_flush_var.get()), + disable_ipv6_while_active=bool(ipv6_var.get()), + harden_webrtc_enabled=bool(webrtc_var.get()), + firefox_path=firefox_path_var.get().strip(), + firefox_profile_dir=firefox_profile_var.get().strip(), + browser_clear_on_close=bool(br_clear_var.get()), + browser_disposable_profile=bool(br_disposable_var.get()), + browser_kill_on_chain_drop=bool(br_kill_drop_var.get()), + browser_auto_relaunch=bool(br_auto_relaunch_var.get()), + browser_lock_managed_profile=bool(br_lock_profile_var.get()), + browser_force_proxy=bool(br_force_proxy_var.get()), + browser_disable_webrtc=bool(br_disable_webrtc_var.get()), + browser_resist_fingerprinting=bool(br_rfp_var.get()), + browser_disable_telemetry=bool(br_telemetry_var.get()), + browser_first_party_isolation=bool(br_fpi_var.get()), + browser_strict_tracking_protection=bool(br_strict_tp_var.get()), + browser_timezone_utc=bool(br_tz_utc_var.get()), ) if not (1 <= ns.local_port <= 65535): raise ValueError("Port must be 1-65535") @@ -857,11 +1321,13 @@ def main() -> None: status_dot.configure(text_color=GREEN if running else RED) status_txt.configure(text="RUNNING" if running else "STOPPED") _refresh_sysproxy() + if running: + tray.set_state("yellow", exit_ip=None) if not running: prog_bar.set(0) phase_lbl.configure(text="idle") countdown_lbl.configure(text="next: —", text_color=TEXT2) - tray.set_state("gray") + tray.set_state("gray", exit_ip=None) v_exit_ip.set("—") v_rotation.set("0") @@ -907,9 +1373,32 @@ def main() -> None: elif t == "real_ip": ip = str(m.get("ip", "—")) - your_ip_lbl.configure(text=f"Your IP: {ip}") + your_ip_lbl.configure(text=f"Direct IP: {ip}") v_real_ip.set(ip) + elif t == "vpn": + nonlocal outer_hop_label + if m.get("active"): + outer_hop_label = str(m.get("label", "VPN"))[:12] + vpn_status_lbl.configure( + text=f"VPN: {m.get('label', 'Active')}" + + (f" ({m.get('adapter')})" if m.get("adapter") else ""), + text_color=GREEN, + ) + leak_mode_lbl.configure( + text=f"Leak check: {m.get('leak_mode', 'VPN-aware')}", + text_color=TEXT2, + ) + else: + outer_hop_label = "Direct" + vpn_status_lbl.configure( + text="VPN: not detected (direct connection)", text_color=YELLOW, + ) + leak_mode_lbl.configure( + text=f"Leak check: {m.get('leak_mode', 'strict')}", + text_color=TEXT2, + ) + elif t == "firewall": _refresh_fw(bool(m.get("engaged"))) @@ -933,14 +1422,20 @@ def main() -> None: hops = [str(x) for x in (m.get("hops") or [])] status = str(m.get("status", "connecting")) exit_ip = m.get("exit_ip") + exit_s = str(exit_ip) if exit_ip else None fix = normalize_proxy_url(svc.settings.manual_exit_proxy) fixed_last = bool(fix) and not svc.settings.use_pinned_chain _render_chain(hops, status, exit_ip, fixed_last=fixed_last) _refresh_sysproxy() - tray.set_state( - {"healthy": "green", "dead": "red", "connecting": "yellow"}.get(status, "yellow") + tray_color = {"healthy": "green", "dead": "red", "connecting": "yellow"}.get( + status, "yellow" ) - v_exit_ip.set(str(exit_ip) if exit_ip else "—") + tray.set_state(tray_color, exit_ip=exit_s) + v_exit_ip.set(exit_s if exit_s else "—") + if status == "dead" and browser.is_running() and bool(br_kill_drop_var.get()): + browser.stop(dispose=bool(br_disposable_var.get())) + _refresh_browser_status() + _log("Browser stopped because chain dropped (Kill-on-drop enabled).") def _pump() -> None: try: @@ -948,6 +1443,12 @@ def main() -> None: _handle(ui_q.get_nowait()) except queue.Empty: pass + if browser_should_run[0] and not browser.is_running(): + cfg = _browser_cfg() + now = time.monotonic() + if cfg.auto_relaunch and (now - browser_last_launch_ts[0]) >= 3.0: + if _launch_browser_with_cfg(cfg): + browser_last_launch_ts[0] = now root.after(80, _pump) # ───────────────────────────────────────────────────────────────────────── @@ -955,6 +1456,8 @@ def main() -> None: # ───────────────────────────────────────────────────────────────────────── def _close() -> None: _cancel_pulse() + browser_should_run[0] = False + browser.stop(dispose=bool(br_disposable_var.get())) tray.stop() svc.stop() clear_system_proxy() @@ -970,14 +1473,18 @@ def main() -> None: _refresh_boot() _refresh_sysproxy() _refresh_fw() + _refresh_vpn_ui() + _refresh_browser_status() + _refresh_fingerprint() _log("─" * 60) _log("Proxy God v2 — ready.") - _log(f"Admin: {'YES — kill-switch available' if is_admin() else 'NO — run as Admin for kill-switch'}") + _log(f"Admin: {'YES — kill-switch + privacy hardening available' if is_admin() else 'NO — run as Admin for full privacy tools'}") _log(f"Log file: {LOG_PATH}") _log("Press START to fetch, validate and chain proxies.") - _log("Use +/− in top bar to change hop count instantly.") - _log("Chain Builder → Fixed exit: set the last hop (optional); rest still rotates.") - _log("Verbose logs (Live tab): DEBUG lines from the engine + validator + fetcher (httpx kept quiet).") + _log("Chain Builder → build your chain, Test entire chain, then Start.") + _log("Browser tab → launch hardened Firefox profile that follows your chain.") + _log("Privacy tab → MAC, hostname, IPv6, WebRTC, fingerprint audit, DNS checks.") + _log("Works with or without VPN — leak detection adapts automatically.") _log("─" * 60) _pump() root.mainloop() diff --git a/proxy_chain_manager/browser_launcher.py b/proxy_chain_manager/browser_launcher.py new file mode 100644 index 0000000..844ccd8 --- /dev/null +++ b/proxy_chain_manager/browser_launcher.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +import logging +import shutil +import subprocess +import time +from dataclasses import dataclass +from pathlib import Path + +from .browser_profile import FirefoxHardening, ensure_firefox_profile +from .paths import app_data_dir + +log = logging.getLogger(__name__) + + +@dataclass +class BrowserConfig: + firefox_path: str = "" + profile_dir: str = "" + clear_on_close: bool = True + disposable_profile: bool = False + kill_on_chain_drop: bool = True + auto_relaunch: bool = False + lock_managed_profile: bool = True + force_proxy: bool = True + disable_webrtc: bool = True + resist_fingerprinting: bool = True + disable_telemetry: bool = True + first_party_isolation: bool = True + strict_tracking_protection: bool = True + timezone_utc: bool = True + + +def default_firefox_path() -> str: + candidates = [ + Path(r"C:\Program Files\Mozilla Firefox\firefox.exe"), + Path(r"C:\Program Files (x86)\Mozilla Firefox\firefox.exe"), + ] + for c in candidates: + if c.is_file(): + return str(c) + return "" + + +def default_profile_dir() -> str: + return str(app_data_dir() / "browser_profiles" / "firefox_hardened") + + +class BrowserSession: + def __init__(self) -> None: + self._proc: subprocess.Popen[str] | None = None + self._profile_path: Path | None = None + + def is_running(self) -> bool: + return self._proc is not None and self._proc.poll() is None + + def pid(self) -> int | None: + if self._proc is None: + return None + return self._proc.pid + + def launch( + self, + cfg: BrowserConfig, + proxy_host: str, + proxy_port: int, + ) -> tuple[bool, str]: + raw_exe = (cfg.firefox_path or default_firefox_path()).strip().strip('"').strip("'") + exe = Path(raw_exe) + if not exe.is_file(): + return False, f"Firefox executable not found: {raw_exe or '(empty path)'}" + 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, + resist_fingerprinting=cfg.resist_fingerprinting, + disable_telemetry=cfg.disable_telemetry, + first_party_isolation=cfg.first_party_isolation, + strict_tracking_protection=cfg.strict_tracking_protection, + clear_on_shutdown=cfg.clear_on_close, + timezone_utc=cfg.timezone_utc, + ) + if cfg.lock_managed_profile: + ensure_firefox_profile(profile_dir, proxy_host, int(proxy_port), hard) + if self.is_running(): + self.stop() + if cfg.lock_managed_profile: + args = [str(exe), "-no-remote", "-profile", str(profile_dir)] + else: + args = [str(exe)] + try: + self._proc = subprocess.Popen( + args, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + text=True, + ) + # 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}" + + def stop(self, dispose: bool = False) -> tuple[bool, str]: + if self._proc and self._proc.poll() is None: + try: + self._proc.terminate() + self._proc.wait(timeout=8) + except Exception: + try: + self._proc.kill() + except Exception: + pass + finally: + 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 + return True, "Browser stopped." diff --git a/proxy_chain_manager/browser_profile.py b/proxy_chain_manager/browser_profile.py new file mode 100644 index 0000000..fc99bc9 --- /dev/null +++ b/proxy_chain_manager/browser_profile.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + + +@dataclass +class FirefoxHardening: + force_proxy: bool = True + disable_webrtc: bool = True + resist_fingerprinting: bool = True + disable_telemetry: bool = True + first_party_isolation: bool = True + strict_tracking_protection: bool = True + clear_on_shutdown: bool = True + timezone_utc: bool = True + + +def _bool(v: bool) -> str: + return "true" if v else "false" + + +def build_user_js( + proxy_host: str, + proxy_port: int, + hard: FirefoxHardening, +) -> str: + lines: list[str] = [ + "// Managed by Proxy God. Changes are overwritten on next launch.", + 'user_pref("app.normandy.enabled", false);', + 'user_pref("app.shield.optoutstudies.enabled", false);', + 'user_pref("browser.newtabpage.activity-stream.feeds.telemetry", false);', + 'user_pref("browser.newtabpage.activity-stream.telemetry", false);', + 'user_pref("browser.ping-centre.telemetry", false);', + 'user_pref("toolkit.telemetry.archive.enabled", false);', + 'user_pref("toolkit.telemetry.bhrPing.enabled", false);', + 'user_pref("toolkit.telemetry.enabled", false);', + 'user_pref("toolkit.telemetry.firstShutdownPing.enabled", false);', + 'user_pref("toolkit.telemetry.hybridContent.enabled", false);', + 'user_pref("toolkit.telemetry.newProfilePing.enabled", false);', + 'user_pref("toolkit.telemetry.shutdownPingSender.enabled", false);', + 'user_pref("toolkit.telemetry.unified", false);', + 'user_pref("datareporting.healthreport.uploadEnabled", false);', + 'user_pref("datareporting.policy.dataSubmissionEnabled", false);', + 'user_pref("network.trr.mode", 5);', + 'user_pref("network.captive-portal-service.enabled", false);', + 'user_pref("geo.enabled", false);', + 'user_pref("media.peerconnection.enabled", false);', + 'user_pref("media.navigator.enabled", false);', + 'user_pref("dom.battery.enabled", false);', + 'user_pref("dom.gamepad.enabled", false);', + 'user_pref("dom.netinfo.enabled", false);', + 'user_pref("webgl.disabled", false);', + 'user_pref("webgl.enable-debug-renderer-info", false);', + ] + if hard.force_proxy: + lines.extend( + [ + 'user_pref("network.proxy.type", 1);', + f'user_pref("network.proxy.http", "{proxy_host}");', + f'user_pref("network.proxy.http_port", {int(proxy_port)});', + f'user_pref("network.proxy.ssl", "{proxy_host}");', + f'user_pref("network.proxy.ssl_port", {int(proxy_port)});', + f'user_pref("network.proxy.socks", "{proxy_host}");', + f'user_pref("network.proxy.socks_port", {int(proxy_port)});', + 'user_pref("network.proxy.socks_version", 5);', + 'user_pref("network.proxy.socks_remote_dns", true);', + 'user_pref("network.proxy.no_proxies_on", "");', + ] + ) + if hard.disable_webrtc: + lines.extend( + [ + 'user_pref("media.peerconnection.enabled", false);', + 'user_pref("media.peerconnection.ice.default_address_only", true);', + 'user_pref("media.peerconnection.ice.no_host", true);', + ] + ) + if hard.resist_fingerprinting: + lines.extend( + [ + 'user_pref("privacy.resistFingerprinting", true);', + 'user_pref("privacy.resistFingerprinting.letterboxing", true);', + 'user_pref("privacy.window.maxInnerWidth", 1600);', + 'user_pref("privacy.window.maxInnerHeight", 900);', + ] + ) + if hard.first_party_isolation: + lines.append('user_pref("privacy.firstparty.isolate", true);') + if hard.disable_telemetry: + lines.append('user_pref("browser.send_pings", false);') + if hard.strict_tracking_protection: + lines.extend( + [ + 'user_pref("privacy.trackingprotection.enabled", true);', + 'user_pref("privacy.trackingprotection.pbmode.enabled", true);', + ] + ) + 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);', + ] + ) + if hard.timezone_utc: + lines.append('user_pref("privacy.resistFingerprinting.reduceTimerPrecision", true);') + lines.append("") + return "\n".join(lines) + + +def build_prefs_js_marker() -> str: + return ( + "// Proxy God managed profile\n" + '// Do not edit manually; user.js is rewritten on launch.\n' + ) + + +def ensure_firefox_profile( + profile_dir: Path, + proxy_host: str, + proxy_port: int, + hard: FirefoxHardening, +) -> None: + profile_dir.mkdir(parents=True, exist_ok=True) + (profile_dir / "user.js").write_text( + build_user_js(proxy_host, proxy_port, hard), + encoding="utf-8", + ) + (profile_dir / "prefs.js").write_text(build_prefs_js_marker(), encoding="utf-8") diff --git a/proxy_chain_manager/config.py b/proxy_chain_manager/config.py index d3d6b47..8f90f98 100644 --- a/proxy_chain_manager/config.py +++ b/proxy_chain_manager/config.py @@ -115,7 +115,7 @@ class Settings: # ── chain ───────────────────────────────────────────────────────────── chain_length: int = 3 # number of hops (2-8) obfuscation_mode: str = "auto" # see OBFUSCATION_MODES - use_pinned_chain: bool = False # use manually ordered chain + use_pinned_chain: bool = True # use manually ordered chain from Chain Builder pinned_chain: list[str] = field(default_factory=list) # user-ordered hop list # Fixed last hop only (ignored when use_pinned_chain is True — full manual chain wins) manual_exit_proxy: str = "" @@ -135,6 +135,29 @@ class Settings: kill_switch_enabled: bool = True # engage Windows Firewall kill-switch when running proxy_bypass: str = "localhost;127.*;10.*;192.168.*;" + # ── privacy / device hardening (Privacy tab) ─────────────────────────── + mac_spoof_enabled: bool = False # randomize NIC MAC while chain runs (Admin) + spoof_hostname_enabled: bool = False # temporary computer name while chain runs (Admin) + flush_dns_on_rotate: bool = True # ipconfig /flushdns on each rotation + disable_ipv6_while_active: bool = False # disable IPv6 bindings while chain runs (Admin) + harden_webrtc_enabled: bool = False # Chrome/Edge WebRTC policy (Admin) + + # ── hardened browser ──────────────────────────────────────────────────── + firefox_path: str = "" + firefox_profile_dir: str = "" + browser_clear_on_close: bool = True + browser_disposable_profile: bool = False + browser_kill_on_chain_drop: bool = True + browser_auto_relaunch: bool = False + browser_lock_managed_profile: bool = True + browser_force_proxy: bool = True + browser_disable_webrtc: bool = True + browser_resist_fingerprinting: bool = True + browser_disable_telemetry: bool = True + browser_first_party_isolation: bool = True + browser_strict_tracking_protection: bool = True + browser_timezone_utc: bool = True + # ── sources ─────────────────────────────────────────────────────────── sources: list[str] = field( default_factory=lambda: [ diff --git a/proxy_chain_manager/dns_leak.py b/proxy_chain_manager/dns_leak.py new file mode 100644 index 0000000..7782459 --- /dev/null +++ b/proxy_chain_manager/dns_leak.py @@ -0,0 +1,112 @@ +"""DNS leak checks and cache flush.""" +from __future__ import annotations + +import logging +import subprocess +from dataclasses import dataclass + +import httpx + +log = logging.getLogger(__name__) + +_DOH_GOOGLE = "https://dns.google/resolve?name=whoami.dnsleaktest.com&type=A" + + +@dataclass +class DnsLeakResult: + ok: bool + system_resolvers: list[str] + message: str + doh_ip: str | None = None + + +def get_system_dns_servers() -> list[str]: + 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, + 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()] + except Exception: + return [] + + +def flush_dns_cache() -> tuple[bool, str]: + try: + r = subprocess.run( + ["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" + 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. + + 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 + try: + with httpx.Client(timeout=8.0, verify=True) as c: + r = c.get(_DOH_GOOGLE) + if r.status_code == 200: + data = r.json() + answers = data.get("Answer") or [] + if answers: + doh_ip = str(answers[0].get("data", "")) + except Exception: + pass + + if not resolvers: + return DnsLeakResult( + ok=True, + system_resolvers=[], + message="No IPv4 DNS servers reported (DHCP may assign later).", + doh_ip=doh_ip, + ) + + if public and local_proxy: + return DnsLeakResult( + 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: + return DnsLeakResult( + ok=False, + system_resolvers=resolvers, + message=f"Public DNS resolvers active: {', '.join(public)}", + doh_ip=doh_ip, + ) + + return DnsLeakResult( + 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 new file mode 100644 index 0000000..99405b2 --- /dev/null +++ b/proxy_chain_manager/fingerprint.py @@ -0,0 +1,221 @@ +"""Device fingerprint audit and OS-level hardening helpers.""" +from __future__ import annotations + +import logging +import random +import re +import string +import subprocess +import winreg +from dataclasses import dataclass, field + +from .firewall import is_admin +from .mac_spoof import list_nics + +log = logging.getLogger(__name__) + +_WEBRTC_CHROME = r"SOFTWARE\Policies\Google\Chrome" +_WEBRTC_EDGE = r"SOFTWARE\Policies\Microsoft\Edge" +_WEBRTC_VALUE = "DefaultWebRtcIpHandlingPolicy" +_WEBRTC_DISABLE = 2 # disable_non_proxied_udp + + +@dataclass +class FingerprintAudit: + lines: list[str] = field(default_factory=list) + hostname: str = "" + machine_guid: str = "" + username: str = "" + macs: list[str] = field(default_factory=list) + + +def _run_ps(script: str, timeout: float = 15.0) -> str: + try: + r = subprocess.run( + ["powershell", "-NoProfile", "-Command", script], + capture_output=True, + text=True, + timeout=timeout, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + return (r.stdout or "").strip() + except Exception as e: + log.debug("fingerprint ps: %s", e) + return "" + + +def get_computer_name() -> str: + try: + import os + return os.environ.get("COMPUTERNAME", "") or "" + except Exception: + return "" + + +def get_machine_guid() -> str: + try: + with winreg.OpenKey( + winreg.HKEY_LOCAL_MACHINE, + r"SOFTWARE\Microsoft\Cryptography", + ) as key: + val, _ = winreg.QueryValueEx(key, "MachineGuid") + return str(val) + except OSError: + return "" + + +def random_hostname(prefix: str = "PC") -> str: + suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=7)) + return f"{prefix}-{suffix}"[:15] + + +def set_computer_name(name: str) -> tuple[bool, str]: + """Set NetBIOS / computer name (Admin). Reboot may be required for all apps.""" + if not is_admin(): + return False, "Administrator required to change computer name." + name = re.sub(r"[^A-Za-z0-9\-]", "", name)[:15] + if len(name) < 1: + return False, "Invalid hostname." + code = subprocess.run( + ["powershell", "-NoProfile", "-Command", + f'Rename-Computer -NewName "{name}" -Force -ErrorAction Stop'], + capture_output=True, + text=True, + timeout=30, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ).returncode + if code != 0: + # NetBIOS name via WMI (often works without full rename) + wmi = ( + f'$n = Get-WmiObject Win32_ComputerSystem; ' + f'$r = $n.Rename("{name}"); if ($r.ReturnValue -ne 0) {{ exit $r.ReturnValue }}' + ) + code = subprocess.run( + ["powershell", "-NoProfile", "-Command", wmi], + capture_output=True, + text=True, + timeout=30, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ).returncode + if code == 0: + return True, f"Computer name set to {name} (some apps need reconnect/reboot)." + return False, "Could not change computer name." + + +def disable_ipv6_on_adapters() -> tuple[list[str], list[str]]: + """Disable IPv6 binding on up physical adapters. Returns (adapter names, log lines).""" + if not is_admin(): + return [], ["IPv6 disable skipped (not Admin)."] + script = ( + "Get-NetAdapter | Where-Object { $_.Status -eq 'Up' } | " + "ForEach-Object { $_.Name }" + ) + names = [n.strip() for n in _run_ps(script).splitlines() if n.strip()] + logs: list[str] = [] + changed: list[str] = [] + skip = ("virtual", "vmware", "hyper-v", "loopback", "bluetooth") + for name in names: + if any(h in name.lower() for h in skip): + continue + cmd = ( + f'Disable-NetAdapterBinding -Name "{name}" -ComponentID ms_tcpip6 -Confirm:$false ' + f'-ErrorAction SilentlyContinue' + ) + subprocess.run( + ["powershell", "-NoProfile", "-Command", cmd], + capture_output=True, + timeout=20, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + changed.append(name) + logs.append(f"IPv6 disabled on {name}") + return changed, logs + + +def enable_ipv6_on_adapters(adapters: list[str]) -> list[str]: + if not is_admin() or not adapters: + return [] + logs: list[str] = [] + for name in adapters: + cmd = ( + f'Enable-NetAdapterBinding -Name "{name}" -ComponentID ms_tcpip6 -Confirm:$false ' + f'-ErrorAction SilentlyContinue' + ) + subprocess.run( + ["powershell", "-NoProfile", "-Command", cmd], + capture_output=True, + timeout=20, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + logs.append(f"IPv6 re-enabled on {name}") + return logs + + +def apply_webrtc_hardening(enable: bool) -> tuple[bool, str]: + """Chrome/Edge: disable WebRTC non-proxied UDP (Admin, HKLM policies).""" + if not is_admin(): + return False, "Administrator required for browser WebRTC policy." + paths = [_WEBRTC_CHROME, _WEBRTC_EDGE] + try: + for path in paths: + if enable: + try: + key = winreg.CreateKeyEx( + winreg.HKEY_LOCAL_MACHINE, path, 0, winreg.KEY_SET_VALUE + ) + except OSError: + continue + with key: + winreg.SetValueEx(key, _WEBRTC_VALUE, 0, winreg.REG_DWORD, _WEBRTC_DISABLE) + else: + try: + with winreg.OpenKey( + winreg.HKEY_LOCAL_MACHINE, path, 0, winreg.KEY_SET_VALUE + ) as key: + try: + winreg.DeleteValue(key, _WEBRTC_VALUE) + except OSError: + pass + except OSError: + pass + return True, ( + "WebRTC hardened (Chrome/Edge: non-proxied UDP disabled)." + if enable + else "WebRTC policy removed." + ) + except OSError as e: + return False, str(e) + + +def audit_device() -> FingerprintAudit: + """Collect identifiers sites and trackers often fingerprint.""" + import getpass + import os + + host = get_computer_name() + guid = get_machine_guid() + user = getpass.getuser() + macs = [f"{n.name}: {n.mac}" for n in list_nics()] + + 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.extend([f" • {m}" for m in macs[:8]] or [" • (none detected)"]) + if len(macs) > 8: + lines.append(f" … and {len(macs) - 8} 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.") + + return FingerprintAudit( + lines=lines, + hostname=host, + machine_guid=guid, + username=user, + macs=[n.mac for n in list_nics()], + ) diff --git a/proxy_chain_manager/firewall.py b/proxy_chain_manager/firewall.py index b7d2db7..b0a13a9 100644 --- a/proxy_chain_manager/firewall.py +++ b/proxy_chain_manager/firewall.py @@ -13,24 +13,18 @@ When disengaged: from __future__ import annotations import ctypes -import glob import logging import subprocess import sys from pathlib import Path from .paths import gost_exe_path +from .vpn_detect import expand_vpn_executables log = logging.getLogger(__name__) RULE_PREFIX = "PCM_" -NORD_GLOBS = [ - r"C:\Program Files\NordVPN\*.exe", - r"C:\Program Files\NordUpdater\*.exe", - r"C:\Program Files\NordVPN\NordSec ThreatProtection\*.exe", -] - def is_admin() -> bool: try: @@ -99,13 +93,6 @@ def _resolve_self_exe() -> Path: return Path(sys.executable).resolve() -def _expand_nord_exes() -> list[str]: - out: list[str] = [] - for pattern in NORD_GLOBS: - out.extend(glob.glob(pattern)) - return out - - def engage(gost_path: Path | None = None) -> tuple[bool, str]: """Activate kill-switch firewall. Returns (success, message).""" if not is_admin(): @@ -113,7 +100,7 @@ def engage(gost_path: Path | None = None) -> tuple[bool, str]: gost = gost_path or gost_exe_path() self_exe = _resolve_self_exe() - nord_exes = _expand_nord_exes() + vpn_exes = expand_vpn_executables() _delete_rules() @@ -137,10 +124,10 @@ def engage(gost_path: Path | None = None) -> tuple[bool, str]: _add_rule(f"Python_{sibling}", dir="out", action="allow", program=f'"{p}"', protocol="any") - # Allow all NordVPN executables - for i, npath in enumerate(nord_exes): - _add_rule(f"Nord_{i}", dir="out", action="allow", - program=f'"{npath}"', protocol="any") + # Allow VPN client executables (Nord, WireGuard, OpenVPN, etc.) + for i, vpath in enumerate(vpn_exes): + _add_rule(f"VPN_{i}", dir="out", action="allow", + program=f'"{vpath}"', protocol="any") # Allow DHCP (or you lose your adapter) _add_rule("DHCP", dir="out", action="allow", @@ -155,8 +142,8 @@ def engage(gost_path: Path | None = None) -> tuple[bool, str]: # Set default outbound to BLOCK _set_outbound_policy("blockinbound,blockoutbound") - log.info("Firewall kill-switch engaged. %d Nord exes whitelisted.", len(nord_exes)) - return True, f"Kill-switch ON. {len(nord_exes)} Nord processes whitelisted." + log.info("Firewall kill-switch engaged. %d VPN exes whitelisted.", len(vpn_exes)) + return True, f"Kill-switch ON. {len(vpn_exes)} VPN client(s) whitelisted." def disengage() -> tuple[bool, str]: diff --git a/proxy_chain_manager/gui_validate.py b/proxy_chain_manager/gui_validate.py new file mode 100644 index 0000000..d273bdb --- /dev/null +++ b/proxy_chain_manager/gui_validate.py @@ -0,0 +1,41 @@ +"""Fast parallel proxy checks for the GUI (per-hop callbacks).""" +from __future__ import annotations + +import asyncio +from typing import Callable + +from .validator import _check_one + +# High concurrency for interactive chain tests +GUI_VALIDATE_CONCURRENCY = 48 + + +async def validate_each_parallel( + urls: list[str], + check_url: str, + concurrency: int, + timeout_seconds: float, + on_result: Callable[[str, bool], None], +) -> None: + if not urls: + return + sem = asyncio.Semaphore(max(1, concurrency)) + + async def one(u: str) -> None: + async with sem: + ok = await _check_one(u, check_url, timeout_seconds) + on_result(u, ok) + + await asyncio.gather(*(one(u) for u in urls)) + + +def run_validate_each_sync( + urls: list[str], + check_url: str, + timeout_seconds: float, + on_result: Callable[[str, bool], None], + concurrency: int = GUI_VALIDATE_CONCURRENCY, +) -> None: + asyncio.run( + validate_each_parallel(urls, check_url, concurrency, timeout_seconds, on_result) + ) diff --git a/proxy_chain_manager/leak_detect.py b/proxy_chain_manager/leak_detect.py new file mode 100644 index 0000000..853b26f --- /dev/null +++ b/proxy_chain_manager/leak_detect.py @@ -0,0 +1,55 @@ +"""VPN-aware chain leak detection.""" +from __future__ import annotations + + +def is_same_subnet(ip_a: str | None, ip_b: str | None, prefix_len: int = 16) -> bool: + """True if two IPv4 addresses share the same /prefix_len subnet.""" + if not ip_a or not ip_b: + return False + try: + a_parts = [int(x) for x in ip_a.split(".")] + b_parts = [int(x) for x in ip_b.split(".")] + if len(a_parts) != 4 or len(b_parts) != 4: + return False + + def to_int(parts: list[int]) -> int: + return (parts[0] << 24) | (parts[1] << 16) | (parts[2] << 8) | parts[3] + + mask = (0xFFFFFFFF << (32 - prefix_len)) & 0xFFFFFFFF + return (to_int(a_parts) & mask) == (to_int(b_parts) & mask) + except Exception: + return False + + +def is_chain_leak(exit_ip: str | None, real_ip: str | None, vpn_active: bool) -> bool: + """True when the chain is not forwarding (traffic still looks like direct/VPN exit). + + With VPN: compare /16 — VPN IPs rotate but stay in-provider ranges. + Without VPN: exact IP match only (avoid false positives on same ISP /16). + """ + if not exit_ip: + return True + if not real_ip: + return False + if exit_ip == real_ip: + return True + if vpn_active: + return is_same_subnet(exit_ip, real_ip) + return False + + +def leak_reason(exit_ip: str | None, real_ip: str | None, vpn_active: bool) -> str: + if not exit_ip: + return "exit IP unreachable" + if not real_ip: + return "unknown direct IP" + if exit_ip == real_ip: + if vpn_active: + return f"exit {exit_ip} equals VPN/direct IP (chain not forwarding)" + return f"exit {exit_ip} equals your real IP (no anonymization)" + if vpn_active and is_same_subnet(exit_ip, real_ip): + return ( + f"exit {exit_ip} shares /16 with direct {real_ip} " + "(likely exiting via VPN tunnel, not proxy chain)" + ) + return "ok" diff --git a/proxy_chain_manager/mac_spoof.py b/proxy_chain_manager/mac_spoof.py new file mode 100644 index 0000000..d2f111d --- /dev/null +++ b/proxy_chain_manager/mac_spoof.py @@ -0,0 +1,113 @@ +"""Randomize and restore NIC MAC addresses (Admin required).""" +from __future__ import annotations + +import logging +import random +import re +import subprocess +from dataclasses import dataclass + +from .firewall import is_admin + +log = logging.getLogger(__name__) + +_MAC_RE = re.compile(r"^([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})$") + + +@dataclass +class NicMac: + name: str + mac: str + description: str = "" + + +def _run(args: list[str], timeout: float = 20.0) -> tuple[int, str, str]: + try: + r = subprocess.run( + args, + capture_output=True, + text=True, + timeout=timeout, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + return r.returncode, r.stdout or "", r.stderr or "" + except Exception as e: + return 1, "", str(e) + + +def list_nics() -> list[NicMac]: + """Physical/up adapters with current MAC.""" + script = ( + "Get-NetAdapter | Where-Object { $_.Status -ne 'Disabled' } | " + "Select-Object Name, MacAddress, InterfaceDescription | " + "ConvertTo-Json -Compress" + ) + code, out, _ = _run(["powershell", "-NoProfile", "-Command", script]) + if code != 0 or not out.strip(): + return [] + import json + + try: + data = json.loads(out) + except json.JSONDecodeError: + return [] + rows = data if isinstance(data, list) else [data] + nics: list[NicMac] = [] + for row in rows: + if not isinstance(row, dict): + continue + name = str(row.get("Name", "")).strip() + mac = str(row.get("MacAddress", "")).strip().replace("-", ":") + desc = str(row.get("InterfaceDescription", "")).strip() + if name and mac and mac != "00:00:00:00:00:00": + nics.append(NicMac(name=name, mac=mac, description=desc)) + return nics + + +def random_mac() -> str: + """Locally administered unicast MAC.""" + b = [random.randint(0, 255) for _ in range(6)] + b[0] = (b[0] | 0x02) & 0xFE + return ":".join(f"{x:02X}" for x in b) + + +def set_mac(adapter: str, mac: str) -> tuple[bool, str]: + if not is_admin(): + return False, "Administrator required to change MAC." + mac = mac.replace("-", ":").upper() + if not _MAC_RE.match(mac.replace(":", "-")): + return False, f"Invalid MAC: {mac}" + ps = ( + f'$a = Get-NetAdapter -Name "{adapter}" -ErrorAction Stop; ' + f'Set-NetAdapter -Name $a.Name -MacAddress "{mac}" -Confirm:$false' + ) + code, _, err = _run(["powershell", "-NoProfile", "-Command", ps]) + if code != 0: + return False, err or "Set-NetAdapter failed" + log.info("MAC set %s → %s", adapter, mac) + return True, f"{adapter} → {mac}" + + +def spoof_all_physical(snapshot: dict[str, str] | None = None) -> tuple[dict[str, str], list[str]]: + """Randomize MAC on non-virtual adapters. Returns (original_map, log lines).""" + originals: dict[str, str] = dict(snapshot or {}) + logs: list[str] = [] + skip_hints = ("virtual", "vmware", "hyper-v", "loopback", "bluetooth", "wan miniport") + for nic in list_nics(): + low = (nic.description + nic.name).lower() + if any(h in low for h in skip_hints): + continue + if nic.name not in originals: + originals[nic.name] = nic.mac + new_mac = random_mac() + ok, msg = set_mac(nic.name, new_mac) + logs.append(msg if ok else f"{nic.name}: {msg}") + return originals, logs + + +def restore_macs(originals: dict[str, str]) -> list[str]: + logs: list[str] = [] + for name, mac in originals.items(): + ok, msg = set_mac(name, mac) + logs.append(msg if ok else f"{name}: {msg}") + return logs diff --git a/proxy_chain_manager/pool_ops.py b/proxy_chain_manager/pool_ops.py new file mode 100644 index 0000000..2989a83 --- /dev/null +++ b/proxy_chain_manager/pool_ops.py @@ -0,0 +1,106 @@ +"""Fetch proxy lists for UI picker and shared pool building.""" +from __future__ import annotations + +import asyncio +import logging +import random +from concurrent.futures import ThreadPoolExecutor +from typing import Callable + +from .config import Settings +from .fetcher import fetch_proxy_json, normalize_entries +from .validator import validate_proxies + +log = logging.getLogger(__name__) +_POOL_EXEC = ThreadPoolExecutor(max_workers=6, thread_name_prefix="pool_ops") + +PICKER_DISPLAY_CAP = 200 + + +class ProxySourceCache: + """Cache full source lists; each UI pull shows a new random batch.""" + + def __init__(self) -> None: + self._full: list[str] = [] + self.total_cached: int = 0 + + def clear(self) -> None: + self._full.clear() + self.total_cached = 0 + + @property + def loaded(self) -> bool: + return bool(self._full) + + def load(self, sources: list[str], prefer_elite: bool, + on_status: Callable[[str], None] | None = None) -> int: + self._full = fetch_raw_proxies(sources, prefer_elite, on_status=on_status) + self.total_cached = len(self._full) + return self.total_cached + + def random_batch(self, cap: int = PICKER_DISPLAY_CAP) -> list[str]: + if not self._full: + return [] + k = min(cap, len(self._full)) + return random.sample(self._full, k) + + +def fetch_raw_proxies( + sources: list[str], + prefer_elite: bool, + on_status: Callable[[str], None] | None = None, +) -> list[str]: + """Blocking fetch from all JSON sources; deduped, not validated.""" + if not sources: + sources = list(Settings().sources) + seen: set[str] = set() + out: list[str] = [] + for url in sources: + if on_status: + on_status(f"Fetching {url[:70]}…") + try: + rows = fetch_proxy_json(url, timeout=45.0) + entries = normalize_entries(rows, prefer_elite) + if on_status: + on_status(f" → {len(entries)} proxies") + for u in entries: + if u not in seen: + seen.add(u) + out.append(u) + except Exception as e: + log.warning("fetch_raw_proxies %s: %s", url[:60], e) + if on_status: + on_status(f" → error: {e!s}") + return out + + +async def validate_proxy_subset( + urls: list[str], + settings: Settings, + target: int = 0, + on_progress: Callable[[int, int], None] | None = None, +) -> list[str]: + if not urls: + return [] + sample = list(urls) + random.shuffle(sample) + if len(sample) > settings.max_candidates: + sample = sample[: settings.max_candidates] + tgt = target or settings.min_pool_size + return await validate_proxies( + sample, + settings.ip_check_url, + settings.validation_concurrency, + settings.validation_timeout_seconds, + on_progress=on_progress, + target=tgt, + ) + + +def run_validate_sync( + urls: list[str], + settings: Settings, + target: int = 0, + on_progress: Callable[[int, int], None] | None = None, +) -> list[str]: + return asyncio.run(validate_proxy_subset(urls, settings, target=target, on_progress=on_progress)) diff --git a/proxy_chain_manager/proxy_picker.py b/proxy_chain_manager/proxy_picker.py new file mode 100644 index 0000000..1c9b007 --- /dev/null +++ b/proxy_chain_manager/proxy_picker.py @@ -0,0 +1,201 @@ +"""Modal dialog: random batch from cached sources, multi-select.""" +from __future__ import annotations + +import threading +from typing import Any, Callable + +import customtkinter as ctk + +from .config import Settings, redact_proxy_url +from .pool_ops import PICKER_DISPLAY_CAP, ProxySourceCache + +BG = "#0d0d1a" +CARD = "#12122a" +PANEL = "#1a1a3e" +ACCENT = "#1e3a8a" +ACCENT2 = "#2563eb" +GREEN = "#00e676" +RED = "#ff1744" +YELLOW = "#ffab00" +DIM = "#4a5568" +TEXT = "#e2e8f0" +TEXT2 = "#94a3b8" +FONT = "Segoe UI" +_UI_CHUNK = 50 + +# One cache per app session — avoids re-downloading 800+ URLs every open +_SESSION_CACHE = ProxySourceCache() + + +def show_proxy_picker( + parent: Any, + settings: Settings, + prefer_elite: bool, + on_done: Callable[[list[str], str], None], +) -> None: + """``on_done(urls, mode)`` where mode is ``append`` or ``replace``.""" + win = ctk.CTkToplevel(parent) + win.title("Browse proxy lists") + win.geometry("700x500") + win.configure(fg_color=BG) + win.transient(parent) + win.grab_set() + + status = ctk.CTkLabel( + win, + text=f"Pull loads sources once, then shows {PICKER_DISPLAY_CAP} random proxies per batch.", + font=(FONT, 11), + text_color=TEXT2, + ) + status.pack(fill="x", padx=12, pady=(10, 4)) + + filter_var = ctk.StringVar(value="") + filt_row = ctk.CTkFrame(win, fg_color="transparent") + filt_row.pack(fill="x", padx=12, pady=4) + ctk.CTkLabel(filt_row, text="Filter:", font=(FONT, 10), text_color=TEXT2).pack(side="left") + ctk.CTkEntry( + filt_row, textvariable=filter_var, width=180, height=28, + fg_color=BG, border_color=ACCENT, + ).pack(side="left", padx=6) + + proto_var = ctk.StringVar(value="all") + for label, val in [("All", "all"), ("HTTP", "http"), ("SOCKS5", "socks5")]: + ctk.CTkRadioButton( + filt_row, text=label, variable=proto_var, value=val, + font=(FONT, 10), fg_color=ACCENT2, text_color=TEXT, + ).pack(side="left", padx=4) + + scroll = ctk.CTkScrollableFrame(win, fg_color=CARD, height=300, + scrollbar_button_color=ACCENT) + scroll.pack(fill="both", expand=True, padx=12, pady=6) + + display_proxies: list[str] = [] + check_vars: dict[str, ctk.BooleanVar] = {} + _build_gen = [0] + + def _filtered_urls() -> list[str]: + q = filter_var.get().strip().lower() + pv = proto_var.get() + out: list[str] = [] + for u in display_proxies: + if pv == "http" and not u.startswith("http://"): + continue + if pv == "socks5" and not u.startswith("socks5://"): + continue + if q and q not in u.lower(): + continue + out.append(u) + return out + + def _rebuild_list() -> None: + _build_gen[0] += 1 + gen = _build_gen[0] + for w in scroll.winfo_children(): + w.destroy() + urls = _filtered_urls() + if not urls: + status.configure( + text="No proxies match filter — pull a batch or change filter.", + text_color=TEXT2, + ) + return + + def _chunk(start: int) -> None: + if gen != _build_gen[0]: + return + end = min(start + _UI_CHUNK, len(urls)) + for u in urls[start:end]: + if u not in check_vars: + check_vars[u] = ctk.BooleanVar(value=False) + fr = ctk.CTkFrame(scroll, fg_color=PANEL, corner_radius=4) + fr.pack(fill="x", pady=1) + ctk.CTkCheckBox( + fr, text=redact_proxy_url(u), variable=check_vars[u], + font=("Consolas", 10), fg_color=ACCENT2, text_color=TEXT, + ).pack(anchor="w", padx=8, pady=2) + if end < len(urls): + win.after(1, lambda: _chunk(end)) + else: + cached = _SESSION_CACHE.total_cached + status.configure( + text=f"Showing {len(urls)} of batch ({cached} cached in memory). Select → OK.", + text_color=GREEN, + ) + + _chunk(0) + + def _select_all(on: bool) -> None: + for u in _filtered_urls(): + if u in check_vars: + check_vars[u].set(on) + + def _pull(force_reload: bool = False) -> None: + pull_btn.configure(state="disabled") + reload_btn.configure(state="disabled") + status.configure(text="Loading sources…", text_color=YELLOW) + + def work() -> None: + def stat(msg: str) -> None: + win.after(0, lambda m=msg: status.configure(text=m, text_color=TEXT2)) + + if force_reload: + _SESSION_CACHE.clear() + if not _SESSION_CACHE.loaded: + _SESSION_CACHE.load(list(settings.sources), prefer_elite, on_status=stat) + batch = _SESSION_CACHE.random_batch(PICKER_DISPLAY_CAP) + + def finish() -> None: + nonlocal display_proxies + display_proxies = batch + check_vars.clear() + for u in batch: + check_vars[u] = ctk.BooleanVar(value=False) + pull_btn.configure(state="normal") + reload_btn.configure(state="normal") + _rebuild_list() + + win.after(0, finish) + + threading.Thread(target=work, daemon=True).start() + + def _selected() -> list[str]: + return [u for u, v in check_vars.items() if v.get()] + + def _ok(mode: str) -> None: + sel = _selected() + if not sel: + status.configure(text="Select at least one proxy.", text_color=RED) + return + win.grab_release() + win.destroy() + on_done(sel, mode) + + btn_row = ctk.CTkFrame(win, fg_color="transparent") + btn_row.pack(fill="x", padx=12, pady=(4, 12)) + + pull_btn = ctk.CTkButton( + btn_row, text="↻ Random batch", command=lambda: _pull(False), + width=120, fg_color=ACCENT, hover_color=ACCENT2, + ) + pull_btn.pack(side="left", padx=2) + reload_btn = ctk.CTkButton( + btn_row, text="Reload sources", command=lambda: _pull(True), + width=110, fg_color=DIM, hover_color=ACCENT, + ) + reload_btn.pack(side="left", padx=2) + ctk.CTkButton(btn_row, text="Select all", command=lambda: _select_all(True), + width=72, fg_color=DIM).pack(side="left", padx=2) + ctk.CTkButton(btn_row, text="Clear", command=lambda: _select_all(False), + width=52, fg_color=DIM).pack(side="left", padx=2) + ctk.CTkButton(btn_row, text="Cancel", + command=lambda: (win.grab_release(), win.destroy()), + width=64, fg_color="#7f1d1d").pack(side="right", padx=2) + ctk.CTkButton(btn_row, text="Add to chain", command=lambda: _ok("append"), + width=100, fg_color=GREEN, hover_color=ACCENT2).pack(side="right", padx=2) + ctk.CTkButton(btn_row, text="Replace chain", command=lambda: _ok("replace"), + width=100, fg_color=ACCENT2).pack(side="right", padx=2) + + filter_var.trace_add("write", lambda *_: _rebuild_list()) + proto_var.trace_add("write", lambda *_: _rebuild_list()) + + _pull(False) diff --git a/proxy_chain_manager/service.py b/proxy_chain_manager/service.py index f4036a4..6356a8a 100644 --- a/proxy_chain_manager/service.py +++ b/proxy_chain_manager/service.py @@ -16,11 +16,23 @@ from .config import ( redact_proxy_url, save_settings, ) +from .dns_leak import flush_dns_cache from .fetcher import fetch_proxy_json, normalize_entries +from .fingerprint import ( + apply_webrtc_hardening, + disable_ipv6_on_adapters, + enable_ipv6_on_adapters, + get_computer_name, + random_hostname, + set_computer_name, +) from .firewall import disengage as fw_disengage, engage as fw_engage, is_admin from .gost_util import build_gost_cmd, ensure_gost, popen_no_window, read_gost_log_tail, terminate_process +from .leak_detect import is_chain_leak, leak_reason +from .mac_spoof import restore_macs, spoof_all_physical from .sysproxy import clear_system_proxy, set_system_proxy from .validator import check_chain_exit_ip, get_direct_ip, validate_proxies +from .vpn_detect import VpnStatus, detect_vpn log = logging.getLogger(__name__) @@ -30,30 +42,8 @@ Notify = Callable[[dict[str, Any]], None] _FETCH_POOL = ThreadPoolExecutor(max_workers=8, thread_name_prefix="fetcher") -def _is_same_network(ip_a: str | None, ip_b: str | None, prefix_len: int = 16) -> bool: - """True if two IPv4 addresses share the same / subnet. - - With NordVPN (or any VPN), the VPN provider rotates IPs so an exact-match - comparison misses leaks where the chain exits through the VPN tunnel directly. - A /16 check catches same-ISP/same-VPN exit while still allowing genuine - unrelated proxies that happen to share a /24 with the VPN exit. - Returns False if either address is None or non-IPv4. - """ - if not ip_a or not ip_b: - return False - try: - a_parts = [int(x) for x in ip_a.split(".")] - b_parts = [int(x) for x in ip_b.split(".")] - if len(a_parts) != 4 or len(b_parts) != 4: - return False - - def to_int(parts: list[int]) -> int: - return (parts[0] << 24) | (parts[1] << 16) | (parts[2] << 8) | parts[3] - - mask = (0xFFFFFFFF << (32 - prefix_len)) & 0xFFFFFFFF - return (to_int(a_parts) & mask) == (to_int(b_parts) & mask) - except Exception: - return False +# Back-compat for tests +from .leak_detect import is_same_subnet as _is_same_network # noqa: F401 class ChainService: @@ -74,6 +64,12 @@ class ChainService: # Per-session blacklist: proxies that crashed GOST immediately self._blacklist: set[str] = set() + self._vpn: VpnStatus = VpnStatus() + self._mac_originals: dict[str, str] = {} + self._hostname_original: str | None = None + self._ipv6_adapters: list[str] = [] + self._webrtc_was_applied: bool = False + def _manual_exit_url(self) -> str | None: u = normalize_proxy_url(self._settings.manual_exit_proxy) return u if u else None @@ -116,11 +112,61 @@ class ChainService: def _teardown_network(self) -> None: clear_system_proxy() self._notify({"type": "log", "text": "System proxy cleared."}) + self._restore_privacy() if is_admin() and self._settings.kill_switch_enabled: ok, msg = fw_disengage() self._notify({"type": "log", "text": msg}) self._notify({"type": "firewall", "engaged": False}) + def _apply_privacy(self) -> None: + s = self._settings + if s.mac_spoof_enabled and is_admin(): + self._mac_originals, logs = spoof_all_physical(self._mac_originals or None) + for ln in logs: + self._notify({"type": "log", "text": f"MAC: {ln}"}) + elif s.mac_spoof_enabled: + self._notify({"type": "log", "text": "MAC spoof enabled but not Admin — skipped."}) + + if s.spoof_hostname_enabled and is_admin(): + self._hostname_original = get_computer_name() + new_name = random_hostname() + ok, msg = set_computer_name(new_name) + self._notify({"type": "log", "text": f"Hostname: {msg}"}) + elif s.spoof_hostname_enabled: + self._notify({"type": "log", "text": "Hostname spoof enabled but not Admin — skipped."}) + + if s.disable_ipv6_while_active and is_admin(): + self._ipv6_adapters, logs = disable_ipv6_on_adapters() + for ln in logs: + self._notify({"type": "log", "text": ln}) + elif s.disable_ipv6_while_active: + self._notify({"type": "log", "text": "IPv6 disable enabled but not Admin — skipped."}) + + if s.harden_webrtc_enabled and is_admin(): + ok, msg = apply_webrtc_hardening(True) + self._webrtc_was_applied = ok + self._notify({"type": "log", "text": msg}) + elif s.harden_webrtc_enabled: + self._notify({"type": "log", "text": "WebRTC hardening enabled but not Admin — skipped."}) + + def _restore_privacy(self) -> None: + if self._mac_originals: + for ln in restore_macs(self._mac_originals): + self._notify({"type": "log", "text": f"MAC restore: {ln}"}) + self._mac_originals.clear() + if self._hostname_original and is_admin(): + ok, msg = set_computer_name(self._hostname_original) + self._notify({"type": "log", "text": f"Hostname restore: {msg}"}) + self._hostname_original = None + if self._ipv6_adapters: + for ln in enable_ipv6_on_adapters(self._ipv6_adapters): + self._notify({"type": "log", "text": ln}) + self._ipv6_adapters.clear() + if self._webrtc_was_applied and is_admin(): + apply_webrtc_hardening(False) + self._webrtc_was_applied = False + self._notify({"type": "log", "text": "WebRTC policy restored."}) + def _run_thread(self) -> None: try: asyncio.run(self._async_main()) @@ -165,13 +211,34 @@ class ChainService: self._notify({"type": "log", "text": f"GOST re-download failed: {e}"}) return - # ── Real IP ────────────────────────────────────────────────────────── + # ── VPN + direct IP ─────────────────────────────────────────────────── + self._vpn = detect_vpn() + mode = "VPN-aware (/16)" if self._vpn.active else "strict (exact IP)" + self._notify({ + "type": "vpn", + "active": self._vpn.active, + "label": self._vpn.label, + "adapter": self._vpn.adapter, + "leak_mode": mode, + }) + self._notify({ + "type": "log", + "text": ( + f"VPN: {self._vpn.label}" + + (f" ({self._vpn.adapter})" if self._vpn.adapter else "") + + f" — leak check: {mode}" + ), + }) + real_ip = await get_direct_ip(self._settings.ip_check_url) if real_ip: self._notify({"type": "real_ip", "ip": real_ip}) - self._notify({"type": "log", "text": f"Your real IP: {real_ip}"}) + label = "direct/VPN IP" if self._vpn.active else "your real IP" + self._notify({"type": "log", "text": f"{label.capitalize()}: {real_ip}"}) else: - self._notify({"type": "log", "text": "Could not determine real IP — leak detection disabled."}) + self._notify({"type": "log", "text": "Could not determine direct IP — leak detection disabled."}) + + self._apply_privacy() # ── Firewall kill-switch ────────────────────────────────────────────── if self._settings.kill_switch_enabled: @@ -301,6 +368,10 @@ class ChainService: self._current_chain = list(chain) self._notify({"type": "hops", "hops": chain, "status": "connecting"}) self._notify({"type": "phase", "phase": "gost_start"}) + if self._settings.flush_dns_on_rotate: + ok, msg = flush_dns_cache() + if ok: + log.debug("DNS cache flushed before chain run") listen = self._settings.listen_addr() cmd = build_gost_cmd(gost, listen, chain) self._notify({ @@ -358,19 +429,11 @@ class ChainService: self._proc = None return False - if real_ip and _is_same_network(exit_ip, real_ip): + if is_chain_leak(exit_ip, real_ip, self._vpn.active): + reason = leak_reason(exit_ip, real_ip, self._vpn.active) self._notify({"type": "hops", "hops": chain, "status": "dead", "exit_ip": exit_ip}) - self._notify({ - "type": "log", - "text": ( - f"Leak detected! Exit={exit_ip} shares network with real IP {real_ip} " - f"(same /16 subnet — chain not forwarding, likely exiting via VPN directly). Rotating." - ), - }) - log.warning( - "Subnet leak: exit=%s real=%s — chain proxy not forwarding. Blacklisting chain.", - exit_ip, real_ip, - ) + self._notify({"type": "log", "text": f"Leak detected — {reason}. Rotating."}) + log.warning("Chain leak: exit=%s real=%s vpn=%s — %s", exit_ip, real_ip, self._vpn.active, reason) # Blacklist the whole chain so we don't reuse broken proxies fixed = self._manual_exit_url() for h in chain: @@ -410,11 +473,8 @@ class ChainService: exit_ip = await check_chain_exit_ip(local_proxy, self._settings.ip_check_url, timeout) log.debug("Periodic exit IP check %.2fs → %s", time.monotonic() - t1, exit_ip or "none") - if not exit_ip or (real_ip and _is_same_network(exit_ip, real_ip)): - reason = ( - "exit IP gone" if not exit_ip - else f"exit {exit_ip} matches real network {real_ip} (VPN leak)" - ) + if is_chain_leak(exit_ip, real_ip, self._vpn.active): + reason = leak_reason(exit_ip, real_ip, self._vpn.active) self._notify({"type": "hops", "hops": chain, "status": "dead", "exit_ip": exit_ip}) self._notify({"type": "log", "text": f"Health check failed ({reason}) — rotating."}) break @@ -627,6 +687,9 @@ class ChainService: if self._force_rotate.is_set(): self._force_rotate.clear() self._notify({"type": "log", "text": "Manual rotate triggered."}) + if self._settings.flush_dns_on_rotate: + ok, msg = flush_dns_cache() + self._notify({"type": "log", "text": f"DNS flush: {msg}" if ok else f"DNS flush failed: {msg}"}) return "rotate" remaining = end - time.monotonic() self._notify({"type": "countdown", "secs": max(0, int(remaining))}) diff --git a/proxy_chain_manager/tray.py b/proxy_chain_manager/tray.py index 68b4c02..065df17 100644 --- a/proxy_chain_manager/tray.py +++ b/proxy_chain_manager/tray.py @@ -1,36 +1,73 @@ -"""System-tray icon: green = healthy chain, red = broken/stopped, yellow = connecting.""" +"""System tray: LED-style proxy on/off + hover tooltip with exit IP.""" from __future__ import annotations import threading from typing import Any, Callable import pystray -from PIL import Image, ImageDraw +from PIL import Image, ImageDraw, ImageFilter + +_SIZE = 64 -def _circle_icon(color: str, size: int = 64) -> Image.Image: - img = Image.new("RGBA", (size, size), (0, 0, 0, 0)) - draw = ImageDraw.Draw(img) - pad = 4 - draw.ellipse([pad, pad, size - pad, size - pad], fill=color) - return img +def _hex_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 _led_icon(led: str, glow: str, rim: str = "#2d3748") -> Image.Image: + """Traffic-light style icon: dark housing + glowing LED.""" + base = Image.new("RGBA", (_SIZE, _SIZE), (0, 0, 0, 0)) + draw = ImageDraw.Draw(base) + cx, cy = _SIZE // 2, _SIZE // 2 + + # Outer housing (rounded rect feel via ellipse) + draw.ellipse([4, 4, _SIZE - 4, _SIZE - 4], fill="#0d0d1a", outline=rim, width=2) + draw.ellipse([10, 10, _SIZE - 10, _SIZE - 10], fill="#12122a", outline="#1a1a3e", width=1) + + glow_layer = Image.new("RGBA", (_SIZE, _SIZE), (0, 0, 0, 0)) + g = ImageDraw.Draw(glow_layer) + lr, lg, lb = _hex_rgb(glow) + for radius, alpha in ((22, 35), (16, 70), (11, 120)): + g.ellipse( + [cx - radius, cy - radius, cx + radius, cy + radius], + fill=(lr, lg, lb, alpha), + ) + glow_layer = glow_layer.filter(ImageFilter.GaussianBlur(radius=2)) + base = Image.alpha_composite(base, glow_layer) + + draw = ImageDraw.Draw(base) + dr, dg, db = _hex_rgb(led) + draw.ellipse([cx - 9, cy - 9, cx + 9, cy + 9], fill=(dr, dg, db, 255)) + draw.ellipse([cx - 5, cy - 6, cx + 2, cy - 1], fill=(255, 255, 255, 90)) + + return base ICONS = { - "green": _circle_icon("#00e676"), - "red": _circle_icon("#ff1744"), - "yellow": _circle_icon("#ffab00"), - "gray": _circle_icon("#6c757d"), + "green": _led_icon("#00e676", "#00e676"), + "red": _led_icon("#ff1744", "#ff1744"), + "yellow": _led_icon("#ffab00", "#ffab00"), + "gray": _led_icon("#4a5568", "#3d4a5c", rim="#374151"), } -TIPS = { - "green": "Proxy Chain: healthy", - "red": "Proxy Chain: broken / stopped", - "yellow": "Proxy Chain: connecting…", - "gray": "Proxy Chain: idle", +_STATE_LABEL = { + "green": "PROXY ON", + "red": "PROXY OFF / ERROR", + "yellow": "CONNECTING…", + "gray": "STOPPED", } +def _tooltip(state: str, exit_ip: str | None) -> str: + label = _STATE_LABEL.get(state, "Proxy God") + if exit_ip: + return f"Proxy God — {label}\nExit IP: {exit_ip}" + if state == "green": + return f"Proxy God — {label}\nExit IP: (checking…)" + return f"Proxy God — {label}\nExit IP: —" + + class TrayIcon: def __init__( self, @@ -44,22 +81,26 @@ class TrayIcon: self._icon: pystray.Icon | None = None self._thread: threading.Thread | None = None self._state = "gray" + self._exit_ip: str | None = None + self._lock = threading.Lock() def start(self) -> None: if self._thread and self._thread.is_alive(): return menu = pystray.Menu( - pystray.MenuItem("Show", self._show, default=True), - pystray.MenuItem("Rotate now", self._rotate), + pystray.MenuItem("Show Proxy God", self._show, default=True), + pystray.MenuItem("Rotate chain now", self._rotate), pystray.Menu.SEPARATOR, pystray.MenuItem("Quit", self._quit), ) - self._icon = pystray.Icon( - "ProxyChainManager", - icon=ICONS[self._state], - title=TIPS[self._state], - menu=menu, - ) + with self._lock: + tip = _tooltip(self._state, self._exit_ip) + self._icon = pystray.Icon( + "ProxyGod", + icon=ICONS[self._state], + title=tip, + menu=menu, + ) self._thread = threading.Thread(target=self._icon.run, daemon=True) self._thread.start() @@ -70,14 +111,27 @@ class TrayIcon: except Exception: pass - def set_state(self, state: str) -> None: - """state: 'green', 'red', 'yellow', 'gray'.""" + def set_state(self, state: str, exit_ip: str | None = None) -> None: + """state: green (on), red (error), yellow (connecting), gray (stopped). + + Pass exit_ip to refresh the hover tooltip. Use exit_ip=None to keep the last IP. + """ if state not in ICONS: state = "gray" - self._state = state - if self._icon: - self._icon.icon = ICONS[state] - self._icon.title = TIPS[state] + with self._lock: + self._state = state + if exit_ip is not None: + self._exit_ip = exit_ip.strip() if exit_ip else None + if self._icon: + self._icon.icon = ICONS[state] + self._icon.title = _tooltip(state, self._exit_ip) + + def set_exit_ip(self, exit_ip: str | None) -> None: + """Update tooltip only (e.g. after health check, same LED color).""" + with self._lock: + self._exit_ip = exit_ip.strip() if exit_ip else None + if self._icon: + self._icon.title = _tooltip(self._state, self._exit_ip) def _show(self, icon: Any = None, item: Any = None) -> None: self._on_show() diff --git a/proxy_chain_manager/vpn_detect.py b/proxy_chain_manager/vpn_detect.py new file mode 100644 index 0000000..848fb5f --- /dev/null +++ b/proxy_chain_manager/vpn_detect.py @@ -0,0 +1,156 @@ +"""Detect whether a VPN tunnel is active on Windows (any provider).""" +from __future__ import annotations + +import glob +import logging +import re +import subprocess +from dataclasses import dataclass, field +from pathlib import Path + +log = logging.getLogger(__name__) + +# Adapter description / name substrings (case-insensitive) +_VPN_ADAPTER_HINTS = ( + "nordlynx", "nordvpn", "openvpn", "wireguard", "wintun", "tap-windows", + "tailscale", "zerotier", "cisco anyconnect", "fortinet", "pulse secure", + "globalprotect", "softether", "proton", "mullvad", "expressvpn", + "surfshark", "private internet", "pia ", "windscribe", "hotspot shield", + "tunnel", "vpn", +) + +# Executables to whitelist in kill-switch when present +_VPN_EXE_GLOBS: list[str] = [ + r"C:\Program Files\NordVPN\*.exe", + r"C:\Program Files\NordUpdater\*.exe", + r"C:\Program Files\NordVPN\NordSec ThreatProtection\*.exe", + r"C:\Program Files\OpenVPN\bin\*.exe", + r"C:\Program Files\OpenVPN Connect\*.exe", + r"C:\Program Files\WireGuard\*.exe", + r"C:\Program Files\Proton\VPN\*.exe", + r"C:\Program Files\Mullvad VPN\*.exe", + r"C:\Program Files\ExpressVPN\*.exe", + r"C:\Program Files\Surfshark\*.exe", + r"C:\Program Files\Private Internet Access\*.exe", + r"C:\Program Files\Tailscale\*.exe", + r"C:\Program Files\ZeroTier\One\*.exe", +] + +_PROVIDER_FROM_ADAPTER: list[tuple[str, str]] = [ + ("nordlynx", "NordVPN"), + ("nordvpn", "NordVPN"), + ("wireguard", "WireGuard"), + ("wintun", "WireGuard"), + ("openvpn", "OpenVPN"), + ("proton", "Proton VPN"), + ("mullvad", "Mullvad"), + ("expressvpn", "ExpressVPN"), + ("surfshark", "Surfshark"), + ("tailscale", "Tailscale"), + ("zerotier", "ZeroTier"), + ("tap-windows", "OpenVPN/TAP"), + ("globalprotect", "GlobalProtect"), + ("fortinet", "FortiClient"), + ("cisco", "Cisco VPN"), +] + + +@dataclass +class VpnStatus: + active: bool = False + label: str = "Direct (no VPN)" + adapter: str = "" + adapters: list[str] = field(default_factory=list) + + def short_label(self) -> str: + if not self.active: + return "Direct" + return self.label + + +def _run_ps(script: str, timeout: float = 12.0) -> str: + try: + r = subprocess.run( + ["powershell", "-NoProfile", "-NonInteractive", "-Command", script], + capture_output=True, + text=True, + timeout=timeout, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + return (r.stdout or "").strip() + except Exception as e: + log.debug("vpn_detect powershell failed: %s", e) + return "" + + +def _match_provider(name: str) -> str: + low = name.lower() + for hint, label in _PROVIDER_FROM_ADAPTER: + if hint in low: + return label + if "vpn" in low or "tunnel" in low: + return "VPN" + return "VPN" + + +def detect_vpn() -> VpnStatus: + """Inspect up network adapters for VPN/tunnel interfaces.""" + out = _run_ps( + "Get-NetAdapter | Where-Object { $_.Status -eq 'Up' } | " + "Select-Object -ExpandProperty Name" + ) + if not out: + # Fallback: netsh + try: + r = subprocess.run( + ["netsh", "interface", "show", "interface"], + capture_output=True, + text=True, + timeout=10, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + lines = (r.stdout or "").splitlines() + names = [] + for ln in lines[3:]: + parts = ln.split() + if len(parts) >= 4 and parts[0] == "Enabled": + names.append(" ".join(parts[3:])) + out = "\n".join(names) + except Exception: + return VpnStatus() + + adapters: list[str] = [] + for line in out.splitlines(): + name = line.strip() + if name: + adapters.append(name) + + hits: list[str] = [] + for name in adapters: + low = name.lower() + if any(h in low for h in _VPN_ADAPTER_HINTS): + hits.append(name) + + if not hits: + return VpnStatus(active=False, label="Direct (no VPN)", adapters=adapters) + + primary = hits[0] + return VpnStatus( + active=True, + label=_match_provider(primary), + adapter=primary, + adapters=adapters, + ) + + +def expand_vpn_executables() -> list[str]: + """Paths to VPN client binaries for firewall allow rules.""" + seen: set[str] = set() + out: list[str] = [] + for pattern in _VPN_EXE_GLOBS: + for p in glob.glob(pattern): + rp = str(Path(p).resolve()) + if rp not in seen: + seen.add(rp) + out.append(rp) + return out diff --git a/tests/test_proxy_god.py b/tests/test_proxy_god.py index 1b1850f..2e9ef3b 100644 --- a/tests/test_proxy_god.py +++ b/tests/test_proxy_god.py @@ -4,7 +4,7 @@ from __future__ import annotations import asyncio import unittest -from proxy_chain_manager.service import _is_same_network +from proxy_chain_manager.leak_detect import is_chain_leak, is_same_subnet from proxy_chain_manager.config import ( Settings, @@ -24,20 +24,34 @@ from proxy_chain_manager.validator import ( class TestSameNetwork(unittest.TestCase): def test_exact_match(self) -> None: - self.assertTrue(_is_same_network("1.2.3.4", "1.2.3.4")) + self.assertTrue(is_same_subnet("1.2.3.4", "1.2.3.4")) def test_same_slash16(self) -> None: - self.assertTrue(_is_same_network("185.220.101.5", "185.220.200.9")) + self.assertTrue(is_same_subnet("185.220.101.5", "185.220.200.9")) def test_different_slash16(self) -> None: - self.assertFalse(_is_same_network("185.220.101.5", "104.28.3.99")) + self.assertFalse(is_same_subnet("185.220.101.5", "104.28.3.99")) def test_none_safe(self) -> None: - self.assertFalse(_is_same_network(None, "1.2.3.4")) - self.assertFalse(_is_same_network("1.2.3.4", None)) + self.assertFalse(is_same_subnet(None, "1.2.3.4")) + self.assertFalse(is_same_subnet("1.2.3.4", None)) def test_non_ipv4_safe(self) -> None: - self.assertFalse(_is_same_network("not-an-ip", "1.2.3.4")) + self.assertFalse(is_same_subnet("not-an-ip", "1.2.3.4")) + + +class TestLeakDetect(unittest.TestCase): + def test_strict_no_vpn_same_ip(self) -> None: + self.assertTrue(is_chain_leak("1.2.3.4", "1.2.3.4", vpn_active=False)) + + def test_strict_no_vpn_different_ip(self) -> None: + self.assertFalse(is_chain_leak("5.6.7.8", "1.2.3.4", vpn_active=False)) + + def test_vpn_subnet_leak(self) -> None: + self.assertTrue(is_chain_leak("185.220.101.5", "185.220.200.9", vpn_active=True)) + + def test_vpn_different_subnet_ok(self) -> None: + self.assertFalse(is_chain_leak("104.28.3.99", "185.220.101.5", vpn_active=True)) class TestConfig(unittest.TestCase):