drjones/proxy-god
1
1
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: exhaustive feature expansion — cookie modes, DNS/WebRTC testers, map, Firefox fix

Browse Source 
Cookie system:
- Expand from 5 to 12 fully-specified CookiePolicy modes in browser_identity.py
- Add CookiePolicy dataclass: behavior, lifetime, TCP partitioning, clearOnShutdown.*
- browser_profile.py emits full cookie pref set from policy object
- UI dropdown widened to 680px with live description label per mode

Firefox launch fix:
- Detect Firefox via Windows registry, AppData, and shutil.which
- Use DETACHED_PROCESS|CREATE_NO_WINDOW|CREATE_NEW_PROCESS_GROUP flags
- Wait up to 3s for firefox.exe in tasklist instead of polling parent pid
- taskkill on stop() to terminate all firefox.exe processes

DNS leak tester:
- dns_leak.py: FullDnsLeakReport dataclass, run_dns_leak_test comparing
  proxy DoH resolution vs direct system DNS

WebRTC tester:
- webrtc_check.py: STUN UDP probe, registry policy check, user.js pref check

Ban tester:
- Added SITES_SHOPPING, SITES_CRYPTO, SITES_DNS categories
- Parallel execution via ThreadPoolExecutor
- Expanded banned-text hint keywords

Fingerprint audit:
- OS identity checks: hostname, MAC, GUID, OS version, timezone, screen res
- Browser consistency analysis of user.js

Neon world map:
- world_map.png bundled; chain_map.py renders hop arcs over it with glow effect

Signup prep:
- Auto-save account on Open & Autofill; Copy Email / Copy Pass buttons
- Auto-fill custom URL when preset site selected
- PyInstaller-safe path resolution for signup_extension

Spec:
- Bundle signup_extension dir and world_map.png as PyInstaller data files

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Dr Jones
2026-05-21 16:47:22 -07:00
parent ac6bde8a08
commit 01c19349ee
12 changed files with 1913 additions and 400 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
ProxyChainManager.spec
View File

@@ -1,7 +1,12 @@
# -*- mode: python ; coding: utf-8 -*-
from PyInstaller.utils.hooks import collect_all
datas = []
datas = [
# Signup autofill WebExtension (manifest + content.js)
('proxy_chain_manager/signup_extension', 'proxy_chain_manager/signup_extension'),
# Neon world-map background image for the chain map widget
('proxy_chain_manager/world_map.png', 'proxy_chain_manager'),
]
binaries = []
hiddenimports = ['pystray._win32']
tmp_ret = collect_all('customtkinter')

594
proxy_chain_manager/app.py
View File

@@ -39,7 +39,7 @@ from .firewall import (
request_admin_relaunch,
)
from .artifact_wipe import wipe_artifacts
from .ban_tester import BanTestResult, run_ban_tests, test_single_site
from .ban_tester import BanTestResult, SITE_CATEGORIES, run_ban_tests, test_single_site
from .exit_intel import ExitIntel, fetch_exit_intel
from .chain_map import build_chain_map_points, render_chain_map, resize_map_display
from .signup_prep import (
@@ -64,8 +64,14 @@ from .browser_identity import (
PERSONA_LABELS,
PERSONAS,
)
from .dns_leak import check_dns_leak_hint, flush_dns_cache
from .dns_leak import (
FullDnsLeakReport,
check_dns_leak_hint,
flush_dns_cache,
run_dns_leak_test,
)
from .leak_audit import AuditReport, run_audit_sync
from .webrtc_check import WebRtcCheckResult, run_webrtc_check
from .browser_launcher import (
BrowserConfig,
BrowserSession,
@@ -1240,20 +1246,49 @@ def main() -> None:
persona_menu.pack(side="left", padx=(4, 0))
cookie_row = ctk.CTkFrame(persona_card, fg_color="transparent")
cookie_row.pack(fill="x", padx=12, pady=(0, 10))
cookie_row.pack(fill="x", padx=12, pady=(0, 4))
ctk.CTkLabel(cookie_row, text="Cookies:", font=(FONT, 11), text_color=TEXT,
width=80, anchor="w").pack(side="left")
cookie_var = ctk.StringVar(
value=COOKIE_LABELS.get(s.browser_cookie_mode, COOKIE_LABELS["block_third_party"])
_default_cookie_label = COOKIE_LABELS.get(
s.browser_cookie_mode, COOKIE_LABELS["block_third_party"]
)
cookie_var = ctk.StringVar(value=_default_cookie_label)
# Cookie detail sub-labels (key → short description of side effects)
_COOKIE_NOTES: dict[str, str] = {
"accept_all": "Everything works · cookies persist to disk",
"accept_all_session": "Full compat while browsing · cookies wiped on close",
"block_third_party": "Blocks cross-site trackers · most sites work · cookies persist",
"block_third_party_session": "Blocks 3rd-party + nothing persists after close",
"block_third_party_trackers":"ETP Strict — same as Firefox enhanced tracking protection",
"block_social_trackers": "Blocks Facebook pixel, Twitter widgets, social SSO badges",
"block_unvisited": "Very light — only blocks sites you haven't visited",
"partitioned_tcp": "Total Cookie Protection — each site gets isolated 3rd-party jar",
"partitioned_session": "Isolated jars + full disk wipe on close · zero cross-site state",
"session_only": "All cookies expire on close · 1st-party allowed while browsing",
"ghost_mode": "Isolated + session + wipes cache/history/localStorage/formdata",
"block_all": "Breaks most logins · read-only / scraping use only",
}
cookie_note_lbl = ctk.CTkLabel(
persona_card, text=_COOKIE_NOTES.get(s.browser_cookie_mode, ""),
font=(FONT, 9), text_color=TEXT2, anchor="w", wraplength=820,
)
def _on_cookie_change(_choice: str = "") -> None:
key = _cookie_key()
cookie_note_lbl.configure(text=_COOKIE_NOTES.get(key, ""))
cookie_menu = ctk.CTkOptionMenu(
cookie_row,
values=[COOKIE_LABELS[k] for k in COOKIE_MODES],
variable=cookie_var,
fg_color=BG, button_color=ACCENT2, button_hover_color=ACCENT,
text_color=TEXT, font=(FONT, 11), width=500,
text_color=TEXT, font=(FONT, 11), width=680,
command=_on_cookie_change,
)
cookie_menu.pack(side="left", padx=(4, 0))
cookie_note_lbl.pack(anchor="w", padx=(92, 12), pady=(0, 10))
def _persona_key() -> str:
for k in PERSONAS:
@@ -1415,68 +1450,81 @@ def main() -> None:
ban_head = ctk.CTkFrame(ban_scroll, fg_color=CARD, corner_radius=10, border_width=1, border_color=GLOW)
ban_head.pack(fill="x", padx=8, pady=(6, 8))
ctk.CTkLabel(
ban_head,
text="Ban List Tester",
font=(FONT, 13, "bold"),
text_color=CYAN,
ban_head, text="Ban / Reachability Tester",
font=(FONT, 13, "bold"), text_color=CYAN,
).pack(anchor="w", padx=12, pady=(10, 2))
ctk.CTkLabel(
ban_head,
text="Tests high-traffic sites through your ACTIVE chain to detect likely exit-IP bans.",
font=(FONT, 10),
text_color=TEXT2,
).pack(anchor="w", padx=12, pady=(0, 8))
text=(
"Tests social, shopping, crypto exchanges, and DNS providers through your ACTIVE chain. "
"Detects exit-IP bans and geo-blocks."
),
font=(FONT, 10), text_color=TEXT2, wraplength=880, justify="left",
).pack(anchor="w", padx=12, pady=(0, 6))
# Category selector
ban_cat_row = ctk.CTkFrame(ban_head, fg_color="transparent")
ban_cat_row.pack(fill="x", padx=12, pady=(0, 6))
ctk.CTkLabel(ban_cat_row, text="Category:", font=(FONT, 10), text_color=TEXT2, width=70,
anchor="w").pack(side="left")
_ban_cat_keys = list(SITE_CATEGORIES.keys()) # "Social / General", "Shopping", etc.
ban_cat_var = ctk.StringVar(value="All")
ban_cat_menu = ctk.CTkOptionMenu(
ban_cat_row, values=_ban_cat_keys, variable=ban_cat_var,
fg_color=BG, button_color=ACCENT2, button_hover_color=ACCENT,
text_color=TEXT, width=200,
)
ban_cat_menu.pack(side="left", padx=(0, 8))
ban_status_var = ctk.StringVar(value="Status: idle")
ctk.CTkLabel(ban_head, textvariable=ban_status_var, font=(FONT, 11, "bold"), text_color=TEXT).pack(
anchor="w", padx=12, pady=(0, 8)
anchor="w", padx=12, pady=(0, 6)
)
ban_results_frame = ctk.CTkFrame(
ban_scroll,
fg_color=CARD,
corner_radius=10,
border_width=1,
border_color=GLOW,
ban_scroll, fg_color=CARD, corner_radius=10, border_width=1, border_color=GLOW,
)
ban_results_frame.pack(fill="both", expand=True, padx=8, pady=(0, 8))
ban_rows_wrap = ctk.CTkFrame(ban_results_frame, fg_color="transparent")
ban_rows_wrap.pack(fill="both", expand=True, padx=8, pady=8)
def _ban_row(site: str, status: str, detail: str, color: str) -> None:
def _ban_row(site: str, status: str, detail: str, color: str, cat_header: str = "") -> None:
if cat_header:
ctk.CTkLabel(
ban_rows_wrap, text=f" {cat_header}",
font=(FONT, 10, "bold"), text_color=CYAN, anchor="w",
).pack(fill="x", padx=4, pady=(6, 1))
row = ctk.CTkFrame(ban_rows_wrap, fg_color=PANEL, corner_radius=6, height=30)
row.pack(fill="x", pady=2)
row.pack_propagate(False)
ctk.CTkLabel(row, text=site, font=(FONT, 11, "bold"), text_color=TEXT, width=140, anchor="w").pack(
side="left", padx=(8, 4)
)
ctk.CTkLabel(row, text=status, font=(FONT, 10, "bold"), text_color=color, width=80).pack(
side="left", padx=4
)
ctk.CTkLabel(row, text=detail, font=("Consolas", 10), text_color=TEXT2, anchor="w").pack(
side="left", fill="x", expand=True, padx=(6, 8)
)
ctk.CTkLabel(row, text=site, font=(FONT, 11, "bold"), text_color=TEXT,
width=160, anchor="w").pack(side="left", padx=(8, 4))
ctk.CTkLabel(row, text=status, font=(FONT, 10, "bold"), text_color=color,
width=72).pack(side="left", padx=4)
ctk.CTkLabel(row, text=detail, font=("Consolas", 10), text_color=TEXT2,
anchor="w").pack(side="left", fill="x", expand=True, padx=(6, 8))
def _render_ban_results(results: list[BanTestResult]) -> None:
for w in list(ban_rows_wrap.winfo_children()):
w.destroy()
for child in list(ban_rows_wrap.winfo_children()):
child.destroy()
if not results:
ctk.CTkLabel(
ban_rows_wrap,
text="No results yet.",
font=(FONT, 11),
text_color=TEXT2,
).pack(anchor="w", padx=8, pady=8)
ctk.CTkLabel(ban_rows_wrap, text="No results yet.", font=(FONT, 11),
text_color=TEXT2).pack(anchor="w", padx=8, pady=8)
return
seen_cats: set[str] = set()
for r in results:
cat = r.category or ""
header = cat if cat and cat not in seen_cats else ""
if cat:
seen_cats.add(cat)
if r.status == "ok":
st, color = "OK", GREEN
elif r.status == "banned":
st, color = "BANNED", ORANGE
else:
st, color = "ERROR", RED
_ban_row(r.site, st, r.detail, color)
_ban_row(r.site, st, r.detail, color, cat_header=header)
_render_ban_results([])
@@ -1489,23 +1537,24 @@ def main() -> None:
_log("Ban tester: start the chain first, then run tests.")
return
proxy_url = f"http://{svc.settings.listen_addr()}"
chosen_cat = ban_cat_var.get()
# "All" means every category
cats = None if chosen_cat == "All" else [chosen_cat]
ban_status_var.set("Status: running…")
ban_start_btn.configure(state="disabled")
_render_ban_results([])
def work() -> None:
timeout = min(18.0, max(8.0, float(svc.settings.validation_timeout_seconds) + 3.0))
results = run_ban_tests(proxy_url, timeout_seconds=timeout)
ok_n = sum(1 for r in results if r.status == "ok")
results = run_ban_tests(proxy_url, timeout_seconds=timeout, categories=cats)
ok_n = sum(1 for r in results if r.status == "ok")
banned_n = sum(1 for r in results if r.status == "banned")
err_n = len(results) - ok_n - banned_n
err_n = len(results) - ok_n - banned_n
def done() -> None:
_render_ban_results(results)
ban_status_var.set(f"Status: done | ok={ok_n} banned={banned_n} errors={err_n}")
_log(
f"Ban tester done via {proxy_url} — ok={ok_n}, banned={banned_n}, errors={err_n}."
)
ban_status_var.set(f"Status: done — ok={ok_n} banned={banned_n} errors={err_n}")
_log(f"Ban tester done via {proxy_url} — ok={ok_n}, banned={banned_n}, errors={err_n}.")
ban_start_btn.configure(state="normal")
root.after(0, done)
@@ -1513,13 +1562,8 @@ def main() -> None:
threading.Thread(target=work, daemon=True).start()
ban_start_btn = _btn(
ban_action,
"Start ban test",
_start_ban_test,
w=120,
h=30,
fg_color=ACCENT2,
hover_color=ACCENT,
ban_action, "Start test", _start_ban_test,
w=120, h=30, fg_color=ACCENT2, hover_color=ACCENT,
)
ban_start_btn.pack(side="left")
@@ -1561,11 +1605,12 @@ def main() -> None:
signup_site_labels[0],
)
)
signup_site_menu = ctk.CTkOptionMenu(
site_row, values=signup_site_labels, variable=signup_site_var,
fg_color=BG, button_color=ACCENT2, button_hover_color=ACCENT, text_color=TEXT, width=260,
)
signup_site_menu.pack(side="left", padx=(0, 8))
custom_url_row = ctk.CTkFrame(signup_head, fg_color="transparent")
custom_url_row.pack(fill="x", padx=12, pady=(0, 4))
ctk.CTkLabel(custom_url_row, text="URL", width=70, anchor="w", font=(FONT, 10), text_color=TEXT2).pack(side="left")
signup_custom_url = ctk.CTkEntry(custom_url_row, height=28, fg_color=BG, border_color=ACCENT)
signup_custom_url.pack(side="left", fill="x", expand=True)
signup_custom_url.insert(0, signup_draft.custom_url or "")
def _signup_site_key() -> str:
label = signup_site_var.get()
@@ -1574,12 +1619,22 @@ def main() -> None:
return signup_site_keys[i]
return signup_site_keys[0]
custom_url_row = ctk.CTkFrame(signup_head, fg_color="transparent")
custom_url_row.pack(fill="x", padx=12, pady=(0, 4))
ctk.CTkLabel(custom_url_row, text="Custom URL", width=70, anchor="w", font=(FONT, 10), text_color=TEXT2).pack(side="left")
signup_custom_url = ctk.CTkEntry(custom_url_row, height=28, fg_color=BG, border_color=ACCENT)
signup_custom_url.pack(side="left", fill="x", expand=True)
signup_custom_url.insert(0, signup_draft.custom_url or "")
def _on_signup_site_change(_choice: str = "") -> None:
key = _signup_site_key()
preset = SIGNUP_PRESETS.get(key)
if preset and preset.url:
signup_custom_url.delete(0, "end")
signup_custom_url.insert(0, preset.url)
signup_site_menu = ctk.CTkOptionMenu(
site_row, values=signup_site_labels, variable=signup_site_var,
fg_color=BG, button_color=ACCENT2, button_hover_color=ACCENT, text_color=TEXT, width=260,
command=_on_signup_site_change,
)
signup_site_menu.pack(side="left", padx=(0, 8))
# Populate URL field for the initially-selected site
_on_signup_site_change()
def _signup_field(parent: Any, label: str, show: str = "") -> ctk.CTkEntry:
row = ctk.CTkFrame(parent, fg_color="transparent")
@@ -1695,34 +1750,48 @@ def main() -> None:
# ── Pre-flight checklist ─────────────────────────────────────────────────
preflight_card = _signup_card(
"Pre-flight check",
"Verifies chain, target site reachability, and exit-IP intel before you sign up.",
"Exhaustive readiness check: chain, exit IP, geo/ASN, DNS leak, WebRTC, "
"IPv6, fingerprint, target site reachability, and direct-IP comparison.",
)
pf_rows: dict[str, dict[str, ctk.StringVar | ctk.CTkLabel]] = {}
_PF_CHECKS = [
("chain", "Chain running"),
("exit_ip", "Exit IP (via chain)"),
("direct_ip", "Direct IP (no chain)"),
("leak_compare","IP leak check"),
("intel", "Geo / ASN intel"),
("datacenter", "Residential ASN"),
("dns", "DNS routing"),
("ipv6", "IPv6 binding"),
("webrtc", "WebRTC policy"),
("https_tunnel","HTTPS tunnel (CONNECT)"),
("target", "Target site reachable"),
("firefox", "Firefox path"),
]
def _pf_row(key: str, label: str) -> None:
row = ctk.CTkFrame(preflight_card, fg_color=PANEL, corner_radius=6)
row.pack(fill="x", padx=12, pady=2)
dot = ctk.CTkLabel(row, text="●", font=(FONT, 14), text_color=DIM, width=18)
dot.pack(side="left", padx=(8, 4))
ctk.CTkLabel(row, text=label, font=(FONT, 11, "bold"), text_color=TEXT,
width=160, anchor="w").pack(side="left")
width=180, anchor="w").pack(side="left")
var = ctk.StringVar(value="not run")
ctk.CTkLabel(row, textvariable=var, font=(FONT, 10), text_color=TEXT2,
anchor="w").pack(side="left", fill="x", expand=True, padx=(6, 8), pady=4)
pf_rows[key] = {"var": var, "dot": dot}
_pf_row("chain", "Chain running")
_pf_row("exit_ip", "Exit IP")
_pf_row("intel", "IP intel (geo/ASN)")
_pf_row("target", "Target site reachable")
for _k, _lbl in _PF_CHECKS:
_pf_row(_k, _lbl)
def _pf_set(key: str, status: str, detail: str) -> None:
r = pf_rows.get(key)
if not r:
return
color = {"ok": GREEN, "warn": ORANGE, "fail": RED, "running": YELLOW}.get(status, DIM)
r["dot"].configure(text_color=color)
r["var"].set(detail)
r["dot"].configure(text_color=color) # type: ignore[union-attr]
r["var"].set(detail) # type: ignore[union-attr]
pf_summary_var = ctk.StringVar(value="Not run.")
ctk.CTkLabel(preflight_card, textvariable=pf_summary_var, font=(FONT, 10, "bold"),
@@ -1739,72 +1808,189 @@ def main() -> None:
pf_summary_var.set("Pick a site or enter a custom URL first.")
return
pf_running[0] = True
for k in ("chain", "exit_ip", "intel", "target"):
for k, _ in _PF_CHECKS:
_pf_set(k, "running", "checking…")
pf_summary_var.set("Running pre-flight…")
# Immediate sync checks ──────────────────────────────────────────────
chain_ok = service_running[0] and bool(svc.current_chain)
if not chain_ok:
_pf_set("chain", "fail", "chain not running")
_pf_set("exit_ip", "fail", "n/a")
_pf_set("intel", "fail", "n/a")
_pf_set("target", "fail", "n/a")
pf_summary_var.set("Start the chain, then re-run pre-flight.")
pf_running[0] = False
return
_pf_set("chain", "ok", f"{len(svc.current_chain)} hop(s) up")
if chain_ok:
_pf_set("chain", "ok", f"{len(svc.current_chain)} hop(s) active")
else:
_pf_set("chain", "fail", "chain not running — start first")
# Firefox path check
ff_path = firefox_path_var.get().strip() or default_firefox_path()
from pathlib import Path as _Path
if ff_path and _Path(ff_path).is_file():
_pf_set("firefox", "ok", ff_path)
else:
_pf_set("firefox", "fail", f"not found: {ff_path or '(no path set)'}")
proxy_url = f"http://{svc.settings.listen_addr()}"
target_host = urlparse(url).hostname or "(unknown)"
def work() -> None:
timeout = min(15.0, max(8.0, float(svc.settings.validation_timeout_seconds) + 2.0))
intel = fetch_exit_intel(proxy_url, timeout_seconds=timeout)
site_res = test_single_site(proxy_url, target_host, url, timeout_seconds=timeout)
warns: list[str] = []
def done() -> None:
ip = (v_exit_ip.get() or "—").strip()
if ip and ip != "—":
_pf_set("exit_ip", "ok", ip)
# ── 1. Direct IP (bypass chain) ──────────────────────────────────
from .validator import get_direct_ip as _get_direct_ip
import asyncio as _asyncio
try:
direct_ip = _asyncio.run(_get_direct_ip(svc.settings.ip_check_url, timeout / 2))
except Exception:
direct_ip = None
def _set_direct(ip: str | None) -> None:
if ip:
_pf_set("direct_ip", "ok", ip)
else:
_pf_set("exit_ip", "warn", "no IP reported yet")
if intel.ok:
_apply_intel(intel)
if intel.is_proxy_flagged:
_pf_set("intel", "fail", f"proxy-flagged · {intel.org or intel.isp}")
elif intel.is_datacenter:
_pf_set("intel", "warn", f"datacenter ASN · {intel.org or intel.isp}")
_pf_set("direct_ip", "warn", "could not reach check URL directly")
root.after(0, lambda ip=direct_ip: _set_direct(ip))
# ── 2. Exit IP via chain ─────────────────────────────────────────
if chain_ok:
intel = fetch_exit_intel(proxy_url, timeout_seconds=timeout)
exit_ip_str = intel.ip if intel.ok else (v_exit_ip.get() or "").strip()
def _set_exit(ip_: str, intel_: Any) -> None:
if ip_ and ip_ not in ("—", ""):
_pf_set("exit_ip", "ok", ip_)
else:
loc = ", ".join([b for b in (intel.city, intel.country) if b]) or "—"
_pf_set("intel", "ok", f"{loc} · AS{intel.asn} {intel.org or intel.isp}")
else:
_pf_set("intel", "warn", intel.detail or "lookup failed")
if site_res.status == "ok":
_pf_set("target", "ok", f"{target_host} {site_res.detail}")
elif site_res.status == "banned":
_pf_set("target", "fail", f"likely banned · {site_res.detail}")
else:
_pf_set("target", "warn", site_res.detail)
_pf_set("exit_ip", "warn", "no exit IP yet")
# IP leak compare
if direct_ip and ip_ and ip_ not in ("—", ""):
if direct_ip == ip_:
_pf_set("leak_compare", "fail",
f"exit == direct ({ip_}) — chain not working!")
warns.append("exit IP matches direct IP (chain not routing)")
else:
_pf_set("leak_compare", "ok",
f"exit {ip_} ≠ direct {direct_ip} ✓")
else:
_pf_set("leak_compare", "warn", "cannot compare — missing IPs")
# Geo/ASN
if intel_.ok:
_apply_intel(intel_)
loc = ", ".join([b for b in (intel_.city, intel_.country) if b]) or "—"
asn_str = f"AS{intel_.asn}" if intel_.asn else ""
if intel_.is_proxy_flagged:
_pf_set("intel", "fail", f"proxy-flagged · {intel_.org or intel_.isp}")
warns.append("exit IP is proxy-flagged")
elif intel_.is_datacenter:
_pf_set("intel", "warn", f"datacenter · {loc} {asn_str}")
_pf_set("datacenter", "warn", f"DC ASN — {intel_.org or intel_.isp}")
warns.append("exit IP is datacenter ASN")
else:
_pf_set("intel", "ok", f"{loc} {asn_str}")
_pf_set("datacenter", "ok", f"residential/ISP — {intel_.org or intel_.isp}")
else:
_pf_set("intel", "warn", intel_.detail or "lookup failed")
_pf_set("datacenter", "warn", "intel unavailable")
warns: list[str] = []
if intel.ok and intel.is_proxy_flagged:
warns.append("exit IP is proxy-flagged")
if intel.ok and intel.is_datacenter:
warns.append("exit IP is a datacenter ASN")
if site_res.status == "banned":
warns.append("target site likely blocks this IP")
if not warns:
pf_summary_var.set("All checks passed — good to sign up.")
else:
pf_summary_var.set("Warnings: " + "; ".join(warns))
root.after(0, lambda i=exit_ip_str, d=intel: _set_exit(i, d))
else:
for k in ("exit_ip", "leak_compare", "intel", "datacenter"):
root.after(0, lambda k_=k: _pf_set(k_, "fail", "chain not running"))
# ── 3. DNS routing check ─────────────────────────────────────────
try:
dns_rep = run_dns_leak_test(
proxy_url=proxy_url if chain_ok else None,
timeout=min(10.0, timeout),
)
def _set_dns(r: Any) -> None:
if r.leaked:
_pf_set("dns", "fail", r.summary[:80])
warns.append("DNS leak detected")
elif r.has_external_resolver:
_pf_set("dns", "warn", r.summary[:80])
else:
_pf_set("dns", "ok", r.summary[:80])
root.after(0, lambda r_=dns_rep: _set_dns(r_))
except Exception as e:
root.after(0, lambda: _pf_set("dns", "warn", f"check error: {e}"))
# ── 4. IPv6 binding ──────────────────────────────────────────────
try:
from .leak_audit import _check_ipv6_active
v6_on, v6_msg = _check_ipv6_active()
def _set_v6(on_: bool, msg_: str) -> None:
if on_:
_pf_set("ipv6", "warn", f"IPv6 active: {msg_}")
warns.append("IPv6 enabled (possible leak path)")
else:
_pf_set("ipv6", "ok", msg_)
root.after(0, lambda on=v6_on, msg=v6_msg: _set_v6(on, msg))
except Exception as e:
root.after(0, lambda: _pf_set("ipv6", "warn", str(e)))
# ── 5. WebRTC ────────────────────────────────────────────────────
try:
profile = _Path(firefox_profile_var.get().strip() or default_profile_dir())
rtc_rep = run_webrtc_check(profile_dir=profile)
def _set_rtc(r: Any) -> None:
if r.any_leak_risk:
issues_short = "; ".join(r.issues[:2])[:70]
_pf_set("webrtc", "warn", issues_short)
else:
_pf_set("webrtc", "ok", "all surfaces protected")
root.after(0, lambda r_=rtc_rep: _set_rtc(r_))
except Exception as e:
root.after(0, lambda: _pf_set("webrtc", "warn", str(e)))
# ── 6. HTTPS tunnel probe ─────────────────────────────────────────
if chain_ok:
try:
from .validator import check_https_tunnel
import asyncio as _as2
tun_ok, tun_msg = _as2.run(
check_https_tunnel(proxy_url, timeout)
)
def _set_tun(ok_: bool, msg_: str) -> None:
if ok_:
_pf_set("https_tunnel", "ok", msg_)
else:
_pf_set("https_tunnel", "fail", msg_)
warns.append("HTTPS CONNECT tunnel broken")
root.after(0, lambda ok=tun_ok, msg=tun_msg: _set_tun(ok, msg))
except Exception as e:
root.after(0, lambda: _pf_set("https_tunnel", "warn", str(e)))
else:
root.after(0, lambda: _pf_set("https_tunnel", "fail", "chain not running"))
# ── 7. Target site reachability ───────────────────────────────────
if chain_ok:
site_res = test_single_site(proxy_url, target_host, url, timeout_seconds=timeout)
def _set_site(r: Any) -> None:
if r.status == "ok":
_pf_set("target", "ok", f"{target_host} — {r.detail}")
elif r.status == "banned":
_pf_set("target", "fail", f"likely BANNED · {r.detail}")
warns.append("target site likely blocks this exit IP")
else:
_pf_set("target", "warn", r.detail)
root.after(0, lambda r_=site_res: _set_site(r_))
else:
root.after(0, lambda: _pf_set("target", "fail", "chain not running"))
# ── Final summary ─────────────────────────────────────────────────
def _final() -> None:
pf_running[0] = False
if warns:
pf_summary_var.set(f"⚠ {len(warns)} warning(s): " + " · ".join(warns[:3]))
else:
pf_summary_var.set("✓ All pre-flight checks passed — good to go.")
root.after(0, done)
root.after(200, _final)
threading.Thread(target=work, daemon=True).start()
pf_btn_row = ctk.CTkFrame(preflight_card, fg_color="transparent")
pf_btn_row.pack(fill="x", padx=12, pady=(0, 10))
_btn(pf_btn_row, "Run pre-flight", _run_preflight, w=130, h=28,
_btn(pf_btn_row, "Run pre-flight", _run_preflight, w=140, h=28,
fg_color=ACCENT2, hover_color=ACCENT).pack(side="left")
signup_status_var = ctk.StringVar(value="Ready — start chain, then Open & Autofill.")
@@ -1824,19 +2010,46 @@ def main() -> None:
accounts_list.pack(fill="x", padx=12, pady=(8, 10))
def _refresh_accounts_ui() -> None:
for w in list(accounts_list.winfo_children()):
w.destroy()
for child in list(accounts_list.winfo_children()):
child.destroy()
rows = load_accounts()
if not rows:
ctk.CTkLabel(accounts_list, text="No saved accounts yet.", font=(FONT, 10), text_color=TEXT2).pack(
anchor="w", padx=8, pady=8
)
ctk.CTkLabel(accounts_list, text="No saved accounts yet.", font=(FONT, 10),
text_color=TEXT2).pack(anchor="w", padx=8, pady=8)
return
for rec in rows[:40]:
line = f"{rec.created_at[:16]} {rec.site} {rec.email} exit={rec.exit_ip or '—'}"
ctk.CTkLabel(
accounts_list, text=line, font=("Consolas", 10), text_color=TEXT, anchor="w",
).pack(anchor="w", padx=8, pady=2)
for rec in rows[:60]:
row_frame = ctk.CTkFrame(accounts_list, fg_color=PANEL, corner_radius=6)
row_frame.pack(fill="x", pady=2, padx=4)
info_lbl = ctk.CTkLabel(
row_frame,
text=f"{rec.created_at[:16]} {rec.site} {rec.email or rec.username or '—'} exit={rec.exit_ip or '—'}",
font=("Consolas", 10), text_color=TEXT, anchor="w",
)
info_lbl.pack(side="left", fill="x", expand=True, padx=(8, 4), pady=4)
def _copy_email(r: Any = rec) -> None:
v = r.email or r.username or ""
root.clipboard_clear()
root.clipboard_append(v)
_log(f"Copied email/user: {v}")
def _copy_pass(r: Any = rec) -> None:
root.clipboard_clear()
root.clipboard_append(r.password or "")
_log(f"Copied password for {r.email or r.username or r.site}.")
ctk.CTkButton(
row_frame, text="Copy Email", width=80, height=22,
font=(FONT, 10), fg_color=ACCENT2, hover_color=ACCENT,
command=_copy_email,
).pack(side="right", padx=(2, 4), pady=4)
ctk.CTkButton(
row_frame, text="Copy Pass", width=76, height=22,
font=(FONT, 10), fg_color="#1e3a1e", hover_color="#2e5a2e",
text_color=GREEN,
command=_copy_pass,
).pack(side="right", padx=(2, 2), pady=4)
def _current_exit_meta() -> tuple[str, str]:
ip = (v_exit_ip.get() or "—").strip()
@@ -1909,6 +2122,8 @@ def main() -> None:
tag = f" [{' · '.join(extras)}]" if extras else ""
signup_status_var.set(f"Opened {label}{tag} — autofill active (you submit + captcha).")
_log(f"Signup prep: opened {url} with autofill extension{tag}.")
# Auto-save entry so the account is recorded even if user forgets to save manually
_save_signup_account()
else:
signup_status_var.set("Browser launch failed.")
svc.set_sticky(0)
@@ -2100,33 +2315,124 @@ def main() -> None:
_btn(fp_card, "Refresh fingerprint audit", _refresh_fingerprint, w=180, h=28).pack(
anchor="w", padx=12, pady=(0, 10))
dns_card = _priv_section("DNS", "Checks whether public resolvers may bypass your chain.")
dns_result_lbl = ctk.CTkLabel(dns_card, text="Run a check to see DNS configuration.",
font=(FONT, 11), text_color=TEXT2, wraplength=880, justify="left")
dns_result_lbl.pack(anchor="w", padx=12, pady=8)
# ── DNS leak test ────────────────────────────────────────────────────────
dns_card = _priv_section(
"DNS leak test",
"Compares how DNS resolves through your chain vs direct. Detects resolver bypass.",
)
dns_box = ctk.CTkTextbox(
dns_card, height=120, font=("Consolas", 10),
fg_color=BG, text_color=TEXT, scrollbar_button_color=ACCENT,
)
dns_box.pack(fill="x", padx=12, pady=(4, 4))
dns_box.insert("end", "Click 'Full DNS leak test' or 'Quick check' to probe.")
dns_status_lbl = ctk.CTkLabel(dns_card, text="", font=(FONT, 10, "bold"), text_color=TEXT2)
dns_status_lbl.pack(anchor="w", padx=12, pady=(0, 2))
def _run_dns_check() -> None:
listen = f"http://{svc.settings.listen_addr()}" if svc.current_chain else None
r = check_dns_leak_hint(listen)
color = GREEN if r.ok else RED
dns_result_lbl.configure(
text=r.message + (f"\nResolvers: {', '.join(r.system_resolvers)}" if r.system_resolvers else ""),
def _render_dns_report(rep: FullDnsLeakReport) -> None:
dns_box.delete("1.0", "end")
mark = "LEAK" if rep.leaked else "OK "
dns_box.insert("end", f"[{mark}] {rep.summary}\n\n")
if rep.system_resolvers:
dns_box.insert("end", f"Configured resolvers: {', '.join(rep.system_resolvers)}\n")
for host, addrs in rep.direct_resolution.items():
via_proxy = rep.proxy_resolution.get(host, [])
proxy_str = ", ".join(via_proxy) if via_proxy else "(no proxy answer)"
dns_box.insert("end", f" {host:<22} direct={', '.join(addrs) or '—':<20} via-proxy={proxy_str}\n")
if rep.errors:
dns_box.insert("end", "\nErrors: " + "; ".join(rep.errors[:3]))
color = RED if rep.leaked else GREEN
dns_status_lbl.configure(
text="DNS LEAK DETECTED" if rep.leaked else "DNS looks clean",
text_color=color,
)
dns_btn_row = ctk.CTkFrame(dns_card, fg_color="transparent")
dns_btn_row.pack(fill="x", padx=12, pady=(0, 10))
_btn(dns_btn_row, "DNS leak check", _run_dns_check, w=120, h=28).pack(side="left", padx=(0, 8))
def _run_full_dns_test() -> None:
proxy = f"http://{svc.settings.listen_addr()}" if svc.current_chain else None
dns_status_lbl.configure(text="Running…", text_color=YELLOW)
dns_box.delete("1.0", "end")
dns_box.insert("end", "Probing DNS via proxy vs direct…\n")
def work() -> None:
try:
rep = run_dns_leak_test(proxy_url=proxy,
timeout=min(12.0, svc.settings.validation_timeout_seconds))
except Exception as e:
root.after(0, lambda: dns_status_lbl.configure(
text=f"Error: {e}", text_color=RED))
return
root.after(0, lambda: _render_dns_report(rep))
threading.Thread(target=work, daemon=True).start()
def _run_quick_dns_check() -> None:
listen = f"http://{svc.settings.listen_addr()}" if svc.current_chain else None
r = check_dns_leak_hint(listen)
color = GREEN if r.ok else RED
dns_box.delete("1.0", "end")
dns_box.insert("end", r.message)
if r.system_resolvers:
dns_box.insert("end", f"\nResolvers: {', '.join(r.system_resolvers)}")
dns_status_lbl.configure(text="clean" if r.ok else "issue detected", text_color=color)
def _manual_dns_flush() -> None:
ok, msg = flush_dns_cache()
dns_result_lbl.configure(
text=msg, text_color=GREEN if ok else RED,
)
dns_status_lbl.configure(text=msg, text_color=GREEN if ok else RED)
dns_btn_row = ctk.CTkFrame(dns_card, fg_color="transparent")
dns_btn_row.pack(fill="x", padx=12, pady=(0, 10))
_btn(dns_btn_row, "Full DNS leak test", _run_full_dns_test, w=140, h=28,
fg_color=ACCENT2, hover_color=ACCENT).pack(side="left", padx=(0, 6))
_btn(dns_btn_row, "Quick check", _run_quick_dns_check, w=100, h=28).pack(side="left", padx=(0, 6))
_btn(dns_btn_row, "Flush DNS now", _manual_dns_flush, w=110, h=28,
fg_color=DIM).pack(side="left")
# ── WebRTC leak test ─────────────────────────────────────────────────────
rtc_card = _priv_section(
"WebRTC leak test",
"Scans Chrome/Edge policy, Firefox profile prefs, and live STUN server reachability.",
)
rtc_box = ctk.CTkTextbox(
rtc_card, height=110, font=("Consolas", 10),
fg_color=BG, text_color=TEXT, scrollbar_button_color=ACCENT,
)
rtc_box.pack(fill="x", padx=12, pady=(4, 4))
rtc_box.insert("end", "Click 'WebRTC leak test' to probe.")
rtc_status_lbl = ctk.CTkLabel(rtc_card, text="", font=(FONT, 10, "bold"), text_color=TEXT2)
rtc_status_lbl.pack(anchor="w", padx=12, pady=(0, 2))
def _render_rtc_report(rep: WebRtcCheckResult) -> None:
rtc_box.delete("1.0", "end")
for name, ok, detail in rep.checks:
mark = "OK " if ok else "RISK"
rtc_box.insert("end", f"[{mark}] {name:<40} {detail}\n")
color = RED if rep.any_leak_risk else GREEN
rtc_status_lbl.configure(
text="WebRTC LEAK RISK — see items above" if rep.any_leak_risk else "WebRTC looks protected",
text_color=color,
)
def _run_rtc_test() -> None:
rtc_status_lbl.configure(text="Running…", text_color=YELLOW)
rtc_box.delete("1.0", "end")
rtc_box.insert("end", "Scanning WebRTC surfaces…\n")
profile = Path(firefox_profile_var.get().strip() or default_profile_dir())
def work() -> None:
try:
rep = run_webrtc_check(profile_dir=profile)
except Exception as e:
root.after(0, lambda: rtc_status_lbl.configure(text=f"Error: {e}", text_color=RED))
return
root.after(0, lambda: _render_rtc_report(rep))
threading.Thread(target=work, daemon=True).start()
rtc_btn_row = ctk.CTkFrame(rtc_card, fg_color="transparent")
rtc_btn_row.pack(fill="x", padx=12, pady=(0, 10))
_btn(rtc_btn_row, "WebRTC leak test", _run_rtc_test, w=140, h=28,
fg_color=ACCENT2, hover_color=ACCENT).pack(side="left")
wipe_card = _priv_section(
"Forensic artifact wipe",
"One-button purge of common Windows breadcrumb trails. Irreversible.",

234
proxy_chain_manager/ban_tester.py
View File

@@ -1,29 +1,73 @@
from __future__ import annotations
import concurrent.futures
import socket
from dataclasses import dataclass
from typing import Iterable
import httpx
# High-traffic destinations that often reveal proxy bans quickly.
POPULAR_SITES: tuple[tuple[str, str], ...] = (
("Google", "https://www.google.com/generate_204"),
("YouTube", "https://www.youtube.com/"),
("Facebook", "https://www.facebook.com/"),
("Instagram", "https://www.instagram.com/"),
("X", "https://x.com/"),
("Reddit", "https://www.reddit.com/"),
("Wikipedia", "https://www.wikipedia.org/"),
("Amazon", "https://www.amazon.com/"),
("Netflix", "https://www.netflix.com/"),
("GitHub", "https://github.com/"),
("Cloudflare", "https://www.cloudflare.com/"),
("Microsoft", "https://www.microsoft.com/"),
("TikTok", "https://www.tiktok.com/"),
("BBC", "https://www.bbc.com/"),
("DuckDuckGo", "https://duckduckgo.com/"),
# ── Site lists ────────────────────────────────────────────────────────────────
SITES_SOCIAL: tuple[tuple[str, str], ...] = (
("Google", "https://www.google.com/generate_204"),
("YouTube", "https://www.youtube.com/"),
("Facebook", "https://www.facebook.com/"),
("Instagram", "https://www.instagram.com/"),
("X / Twitter", "https://x.com/"),
("Reddit", "https://www.reddit.com/"),
("TikTok", "https://www.tiktok.com/"),
("Wikipedia", "https://www.wikipedia.org/"),
("DuckDuckGo", "https://duckduckgo.com/"),
("BBC", "https://www.bbc.com/"),
("GitHub", "https://github.com/"),
)
SITES_SHOPPING: tuple[tuple[str, str], ...] = (
("Amazon", "https://www.amazon.com/"),
("eBay", "https://www.ebay.com/"),
("Walmart", "https://www.walmart.com/"),
("Etsy", "https://www.etsy.com/"),
("AliExpress", "https://www.aliexpress.com/"),
("Best Buy", "https://www.bestbuy.com/"),
("Target", "https://www.target.com/"),
("Newegg", "https://www.newegg.com/"),
)
SITES_CRYPTO: tuple[tuple[str, str], ...] = (
("Coinbase", "https://www.coinbase.com/"),
("Binance", "https://www.binance.com/"),
("Kraken", "https://www.kraken.com/"),
("Bybit", "https://www.bybit.com/"),
("OKX", "https://www.okx.com/"),
("KuCoin", "https://www.kucoin.com/"),
("Bitfinex", "https://www.bitfinex.com/"),
("Gemini", "https://www.gemini.com/"),
)
SITES_DNS: tuple[tuple[str, str], ...] = (
("Cloudflare DNS (1.1.1.1)", "https://1.1.1.1/"),
("Google DNS (8.8.8.8)", "https://dns.google/"),
("Quad9 DNS (9.9.9.9)", "https://quad9.net/"),
("AdGuard DNS", "https://adguard-dns.io/"),
("NextDNS", "https://nextdns.io/"),
("Mullvad DNS", "https://mullvad.net/en/help/dns-over-https-and-dns-over-tls/"),
("OpenDNS", "https://www.opendns.com/"),
)
# Combined default: all categories
POPULAR_SITES: tuple[tuple[str, str], ...] = (
SITES_SOCIAL + SITES_SHOPPING + SITES_CRYPTO + SITES_DNS
)
SITE_CATEGORIES: dict[str, tuple[tuple[str, str], ...]] = {
"Social / General": SITES_SOCIAL,
"Shopping": SITES_SHOPPING,
"Crypto Exchanges": SITES_CRYPTO,
"DNS Providers": SITES_DNS,
"All": POPULAR_SITES,
}
_BANNED_STATUS_CODES = {401, 403, 407, 418, 429, 451}
_BANNED_TEXT_HINTS = (
"access denied",
@@ -33,6 +77,15 @@ _BANNED_TEXT_HINTS = (
"unusual traffic",
"captcha",
"challenge required",
"bot detected",
"suspicious activity",
"geo-blocked",
"not available in your region",
"your ip",
"ip address",
"cloudflare ray",
"ddos protection",
"attention required",
)
@@ -40,9 +93,10 @@ _BANNED_TEXT_HINTS = (
class BanTestResult:
site: str
url: str
status: str # ok | banned | error
status: str # ok | banned | error
code: int | None
detail: str
category: str = ""
def _looks_banned_body(body: str) -> bool:
@@ -50,7 +104,13 @@ def _looks_banned_body(body: str) -> bool:
return any(h in t for h in _BANNED_TEXT_HINTS)
def _test_one(proxy_url: str, site: str, url: str, timeout_seconds: float) -> BanTestResult:
def _test_one(
proxy_url: str,
site: str,
url: str,
timeout_seconds: float,
category: str = "",
) -> BanTestResult:
timeout = httpx.Timeout(timeout_seconds, connect=min(8.0, timeout_seconds))
try:
with httpx.Client(
@@ -58,27 +118,145 @@ def _test_one(proxy_url: str, site: str, url: str, timeout_seconds: float) -> Ba
timeout=timeout,
verify=False,
follow_redirects=True,
headers={"User-Agent": "Mozilla/5.0"},
headers={
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36"
),
"Accept": "text/html,application/xhtml+xml,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
},
) as c:
r = c.get(url)
body = r.text[:1800] if r.text else ""
body = r.text[:2400] if r.text else ""
if r.status_code in _BANNED_STATUS_CODES or _looks_banned_body(body):
return BanTestResult(site, url, "banned", r.status_code, f"HTTP {r.status_code}")
return BanTestResult(
site, url, "banned", r.status_code,
f"HTTP {r.status_code}", category,
)
if 200 <= r.status_code < 400:
return BanTestResult(site, url, "ok", r.status_code, f"HTTP {r.status_code}")
return BanTestResult(site, url, "error", r.status_code, f"HTTP {r.status_code}")
return BanTestResult(site, url, "ok", r.status_code, f"HTTP {r.status_code}", category)
return BanTestResult(site, url, "error", r.status_code, f"HTTP {r.status_code}", category)
except Exception as e:
return BanTestResult(site, url, "error", None, f"{type(e).__name__}: {e}")
short = str(e)[:80]
return BanTestResult(site, url, "error", None, f"{type(e).__name__}: {short}", category)
def _dns_resolve_via_system(hostname: str, timeout: float = 5.0) -> str | None:
"""Resolve a hostname using system DNS (does not go through proxy — exposes leak)."""
try:
socket.setdefaulttimeout(timeout)
info = socket.getaddrinfo(hostname, None)
for entry in info:
addr = entry[4][0]
if addr:
return addr
except Exception:
pass
return None
@dataclass(frozen=True)
class DnsLeakResult:
resolver: str
ip_via_proxy: str | None
ip_direct: str | None
leaked: bool
detail: str
def check_dns_leak(
proxy_url: str,
timeout_seconds: float = 8.0,
) -> list[DnsLeakResult]:
"""
Detect DNS leaks: compare hostname resolution seen through proxy vs direct.
A mismatch means DNS is escaping the tunnel.
"""
test_hosts = [
("Cloudflare (1.1.1.1)", "one.one.one.one"),
("Google (8.8.8.8)", "dns.google"),
("OpenDNS", "resolver1.opendns.com"),
]
results: list[DnsLeakResult] = []
timeout = httpx.Timeout(timeout_seconds, connect=min(6.0, timeout_seconds))
for label, host in test_hosts:
# Get IP via direct system DNS
direct_ip = _dns_resolve_via_system(host)
# Get IP as seen from the proxy path (via http://dns-endpoint)
proxy_ip: str | None = None
try:
with httpx.Client(proxy=proxy_url, timeout=timeout, verify=False, follow_redirects=True) as c:
r = c.get(f"https://{host}/")
proxy_ip = str(r.headers.get("x-real-ip") or "")
if not proxy_ip:
# fall back: grab connected IP from response
proxy_ip = None
except Exception:
pass
# Simple leak heuristic: if direct resolution works but proxy connection fails, possible leak path
leaked = bool(direct_ip and not proxy_ip)
if leaked:
detail = f"DNS resolved directly to {direct_ip} but proxy could not reach it — possible bypass"
elif not direct_ip:
detail = "Could not resolve directly"
leaked = False
else:
detail = f"Direct: {direct_ip}"
results.append(DnsLeakResult(
resolver=label,
ip_via_proxy=proxy_ip,
ip_direct=direct_ip,
leaked=leaked,
detail=detail,
))
return results
def run_ban_tests(
proxy_url: str,
timeout_seconds: float = 12.0,
sites: Iterable[tuple[str, str]] = POPULAR_SITES,
sites: Iterable[tuple[str, str]] | None = None,
categories: Iterable[str] | None = None,
max_workers: int = 8,
) -> list[BanTestResult]:
"""
Run ban tests in parallel across requested site categories.
``categories`` accepts names from SITE_CATEGORIES (e.g. "Shopping", "Crypto Exchanges").
Defaults to all categories when neither ``sites`` nor ``categories`` is given.
"""
if sites is not None:
work = [(s, u, "") for s, u in sites]
elif categories is not None:
work = []
for cat in categories:
for s, u in SITE_CATEGORIES.get(cat, ()):
work.append((s, u, cat))
else:
work = [(s, u, cat) for cat, pairs in SITE_CATEGORIES.items() if cat != "All" for s, u in pairs]
out: list[BanTestResult] = []
for site, url in sites:
out.append(_test_one(proxy_url, site, url, timeout_seconds))
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = {
pool.submit(_test_one, proxy_url, site, url, timeout_seconds, cat): (site, url, cat)
for site, url, cat in work
}
for fut in concurrent.futures.as_completed(futures):
try:
out.append(fut.result())
except Exception:
site, url, cat = futures[fut]
out.append(BanTestResult(site, url, "error", None, "internal error", cat))
# Sort: category then site name
out.sort(key=lambda r: (r.category, r.site))
return out

282
proxy_chain_manager/browser_identity.py
View File

@@ -15,84 +15,78 @@ Two strategies are exposed:
• **Hardened** — full RFP / blocked WebRTC / strict tracking / FPI. Stand-out
but maximum block. Same behavior as the original profile.
Cookie modes are independent of persona and let the user pick per-session
behavior from a dropdown.
Cookie modes are independent of persona. Each mode expresses a complete
policy: which cookies to accept, lifetime, whether to wipe on close, and
whether to partition 3rd-party storage.
"""
from __future__ import annotations
from dataclasses import dataclass
# Order matters: dropdown order in the UI.
# ── Personas ──────────────────────────────────────────────────────────────────
PERSONAS: list[str] = [
"blend_windows_chrome",
"blend_windows_chrome_laptop",
"blend_windows_firefox",
"blend_windows_edge",
"blend_mac_safari",
"blend_mac_chrome",
"blend_linux_firefox",
"hardened",
"custom",
]
PERSONA_LABELS: dict[str, str] = {
"blend_windows_chrome": "Blend in — Windows 10 + Chrome (US-en, most common)",
"blend_windows_firefox": "Blend in — Windows 10 + Firefox (US-en)",
"blend_mac_safari": "Blend in — macOS + Safari (US-en)",
"hardened": "Hardened — block + RFP (stands out; max anti-fingerprint)",
"custom": "Custom — use individual hardening toggles",
}
COOKIE_MODES: list[str] = [
"accept_all",
"block_third_party",
"block_third_party_trackers",
"session_only",
"block_all",
]
COOKIE_LABELS: dict[str, str] = {
"accept_all": "Accept all cookies (most sites work)",
"block_third_party": "Block 3rd-party cookies (recommended)",
"block_third_party_trackers": "Block 3rd-party trackers only (Firefox default-strict)",
"session_only": "Session only — clear on close",
"block_all": "Block ALL cookies (breaks logins)",
}
# Firefox network.cookie.cookieBehavior values
_COOKIE_BEHAVIOR = {
"accept_all": 0,
"block_third_party": 1,
"block_third_party_trackers": 4,
"session_only": 0,
"block_all": 2,
"blend_windows_chrome": "Blend — Win10 + Chrome 124 (most common, US-en)",
"blend_windows_chrome_laptop": "Blend — Win11 + Chrome 124 laptop 1366×768",
"blend_windows_firefox": "Blend — Win10 + Firefox 128 (US-en)",
"blend_windows_edge": "Blend — Win11 + Edge 124 (corporate look)",
"blend_mac_safari": "Blend — macOS 14 + Safari 17 (US-en)",
"blend_mac_chrome": "Blend — macOS 14 + Chrome 124 (US-en)",
"blend_linux_firefox": "Blend — Ubuntu + Firefox 128 (en-US)",
"hardened": "Hardened — RFP + FPI + block all (stands out)",
"custom": "Custom — use individual hardening toggles",
}
@dataclass(frozen=True)
class Persona:
"""A user-agent / locale / timezone / screen tuple that mimics a real,
common configuration. All fields are deliberately popular values so the
persona blends in.
"""
common configuration."""
key: str
user_agent: str
accept_language: str
timezone: str # IANA tz; used only when persona drives env TZ
timezone: str
screen_w: int
screen_h: int
platform: str # navigator.platform
locale: str # general.useragent.locale equivalent
platform: str
locale: str
PERSONA_DATA: dict[str, Persona] = {
"blend_windows_chrome": Persona(
key="blend_windows_chrome",
# Stable Chrome on Win10 x64 — by far the most common UA on the web.
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
),
accept_language="en-US,en;q=0.9",
timezone="America/New_York",
screen_w=1920,
screen_h=1080,
screen_w=1920, screen_h=1080,
platform="Win32",
locale="en-US",
),
"blend_windows_chrome_laptop": Persona(
key="blend_windows_chrome_laptop",
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
),
accept_language="en-US,en;q=0.9",
timezone="America/Chicago",
screen_w=1366, screen_h=768,
platform="Win32",
locale="en-US",
),
@@ -104,8 +98,19 @@ PERSONA_DATA: dict[str, Persona] = {
),
accept_language="en-US,en;q=0.5",
timezone="America/New_York",
screen_w=1920,
screen_h=1080,
screen_w=1920, screen_h=1080,
platform="Win32",
locale="en-US",
),
"blend_windows_edge": Persona(
key="blend_windows_edge",
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0"
),
accept_language="en-US,en;q=0.9",
timezone="America/Chicago",
screen_w=1920, screen_h=1080,
platform="Win32",
locale="en-US",
),
@@ -117,20 +122,201 @@ PERSONA_DATA: dict[str, Persona] = {
),
accept_language="en-US,en;q=0.9",
timezone="America/Los_Angeles",
screen_w=1680,
screen_h=1050,
screen_w=1680, screen_h=1050,
platform="MacIntel",
locale="en-US",
),
"blend_mac_chrome": Persona(
key="blend_mac_chrome",
user_agent=(
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
),
accept_language="en-US,en;q=0.9",
timezone="America/Los_Angeles",
screen_w=1440, screen_h=900,
platform="MacIntel",
locale="en-US",
),
"blend_linux_firefox": Persona(
key="blend_linux_firefox",
user_agent=(
"Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:128.0) "
"Gecko/20100101 Firefox/128.0"
),
accept_language="en-US,en;q=0.5",
timezone="America/New_York",
screen_w=1920, screen_h=1080,
platform="Linux x86_64",
locale="en-US",
),
}
# ── Cookie policies ───────────────────────────────────────────────────────────
@dataclass(frozen=True)
class CookiePolicy:
"""
Complete Firefox cookie configuration expressed as a single composable unit.
behavior : network.cookie.cookieBehavior
0 = accept all
1 = block 3rd-party
2 = block ALL cookies
3 = block 3rd-party from unvisited sites
4 = block cross-site tracking (ETP Strict)
5 = block cross-site + social media trackers
lifetime : network.cookie.lifetimePolicy
0 = normal (respect Set-Cookie expires)
2 = session-only (all cookies expire on close)
partition : Total Cookie Protection — partition 3rd-party storage
by the top-level site (network.cookie.cookieBehavior.optInPartitioning)
clear_cookies : privacy.clearOnShutdown.cookies
clear_cache : privacy.clearOnShutdown.cache
clear_history : privacy.clearOnShutdown.history
clear_ls : privacy.clearOnShutdown.localStorage + indexedDB
clear_formdata : privacy.clearOnShutdown.formdata
sanitize : privacy.sanitize.sanitizeOnShutdown (master switch)
"""
behavior: int = 1
lifetime: int = 0
partition: bool = False
sanitize: bool = False
clear_cookies: bool = False
clear_cache: bool = False
clear_history: bool = False
clear_ls: bool = False
clear_formdata: bool = False
# Every cookie mode is a complete CookiePolicy — no guesswork in profile.py.
COOKIE_POLICIES: dict[str, CookiePolicy] = {
# ── Permissive ──────────────────────────────────────────────────────────
"accept_all": CookiePolicy(
behavior=0, lifetime=0,
# No clearing, no blocking — maximum site compatibility.
),
"accept_all_session": CookiePolicy(
behavior=0, lifetime=2,
sanitize=True, clear_cookies=True, clear_ls=True,
# Accept all cookies while browsing but wipe them when the
# browser closes. Good for signup / form-filling sessions.
),
# ── Standard 3rd-party blocking ────────────────────────────────────────
"block_third_party": CookiePolicy(
behavior=1, lifetime=0,
# Block 3rd-party, accept 1st-party with normal lifetime.
# Recommended baseline — most sites work fine.
),
"block_third_party_session": CookiePolicy(
behavior=1, lifetime=2,
sanitize=True, clear_cookies=True, clear_cache=True, clear_ls=True,
# Block 3rd-party AND treat all accepted cookies as session-only.
# Nothing persists. Use for anonymous sessions.
),
# ── Tracker-focused blocking ────────────────────────────────────────────
"block_third_party_trackers": CookiePolicy(
behavior=4, lifetime=0,
# Firefox ETP Strict mode: only blocks known trackers, not ALL
# 3rd-party cookies. Best blend-in option — same as Firefox default.
),
"block_social_trackers": CookiePolicy(
behavior=5, lifetime=0,
# Block cross-site cookies AND social-media tracking cookies
# (Facebook, Twitter pixels, etc.). Stricter than ETP Strict but
# still allows most logins.
),
"block_unvisited": CookiePolicy(
behavior=3, lifetime=0,
# Block 3rd-party from sites the user has never visited. Very light
# — almost invisible to sites; good when you need max compatibility.
),
# ── Partitioned storage (Total Cookie Protection) ───────────────────────
"partitioned_tcp": CookiePolicy(
behavior=1, lifetime=0, partition=True,
# Block 3rd-party AND enable Total Cookie Protection: each site
# gets its own isolated cookie jar so cross-site tracking via
# cookie sync is impossible. Recommended for blend-in + privacy.
),
"partitioned_session": CookiePolicy(
behavior=1, lifetime=2, partition=True,
sanitize=True, clear_cookies=True, clear_cache=True,
clear_ls=True, clear_formdata=True,
# Partitioned + session-only + full wipe on close. Leaves no
# persistent state on disk after the browser exits.
),
# ── Maximum wipe ───────────────────────────────────────────────────────
"session_only": CookiePolicy(
behavior=0, lifetime=2,
sanitize=True, clear_cookies=True,
# All cookies accepted but expire when the browser closes.
# Note: localStorage is NOT cleared here — only cookies.
),
"ghost_mode": CookiePolicy(
behavior=1, lifetime=2,
sanitize=True, clear_cookies=True, clear_cache=True,
clear_history=True, clear_ls=True, clear_formdata=True,
partition=True,
# Maximum ephemeral session: block 3rd-party, TCP partitioning,
# session-only lifetime, full wipe of cookies + cache + history +
# localStorage/IndexedDB + formdata on every close. Zero disk trace.
),
"block_all": CookiePolicy(
behavior=2, lifetime=0,
# Block ALL cookies. Most login-dependent sites will break.
# Only useful for read-only scraping sessions.
),
}
COOKIE_MODES: list[str] = list(COOKIE_POLICIES)
COOKIE_LABELS: dict[str, str] = {
"accept_all": "Accept all — maximum compatibility (cookies persist)",
"accept_all_session": "Accept all, session only — wipe cookies on close",
"block_third_party": "Block 3rd-party — recommended baseline (persists 1st-party)",
"block_third_party_session": "Block 3rd-party + session — nothing persists",
"block_third_party_trackers":"Block trackers only — ETP Strict (Firefox default-strict)",
"block_social_trackers": "Block social + cross-site trackers (Facebook pixel, etc.)",
"block_unvisited": "Block 3rd-party from unvisited sites — lightest option",
"partitioned_tcp": "Partitioned (TCP) — each site gets isolated 3rd-party jar",
"partitioned_session": "Partitioned + session — isolated jars, full wipe on close",
"session_only": "Session only — all cookies expire on close (no partition)",
"ghost_mode": "Ghost mode — partitioned + session + full disk wipe on close",
"block_all": "Block ALL cookies — breaks most logins",
}
# ── Helpers ───────────────────────────────────────────────────────────────────
def get_cookie_policy(mode: str) -> CookiePolicy:
return COOKIE_POLICIES.get(mode, COOKIE_POLICIES["block_third_party"])
# Legacy shims kept for code that still calls these directly.
def cookie_behavior_value(mode: str) -> int:
return _COOKIE_BEHAVIOR.get(mode, 1)
return get_cookie_policy(mode).behavior
def cookie_session_only(mode: str) -> bool:
return mode == "session_only"
return get_cookie_policy(mode).lifetime == 2
def get_persona(key: str) -> Persona | None:

204
proxy_chain_manager/browser_launcher.py
View File

@@ -5,6 +5,7 @@ import os
import shutil
import subprocess
import time
import winreg
from dataclasses import dataclass
from pathlib import Path
@@ -14,6 +15,14 @@ from .signup_prep import SignupDraft, install_signup_extension
log = logging.getLogger(__name__)
# CREATE_NO_WINDOW suppresses the console; DETACHED_PROCESS fully separates
# the child from Python's job object so Firefox lives past Python's exit.
_CREATE_FLAGS = (
getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000)
| getattr(subprocess, "DETACHED_PROCESS", 0x00000008)
| getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0x00000200)
)
@dataclass
class BrowserConfig:
@@ -35,11 +44,74 @@ class BrowserConfig:
cookie_mode: str = "block_third_party"
def _find_firefox_via_registry() -> str:
"""Look up Firefox install path in the Windows registry."""
hives = [
(winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Mozilla\Mozilla Firefox"),
(winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\WOW6432Node\Mozilla\Mozilla Firefox"),
(winreg.HKEY_CURRENT_USER, r"SOFTWARE\Mozilla\Mozilla Firefox"),
]
for hive, key_path in hives:
try:
with winreg.OpenKey(hive, key_path) as root:
version = winreg.QueryValue(root, None)
for sub in (f"{version}\\Main", "bin"):
try:
with winreg.OpenKey(root, sub) as k:
path = winreg.QueryValueEx(k, "PathToExe")[0]
if path and Path(path).is_file():
return str(path)
except OSError:
continue
except OSError:
continue
# Also try the "Uninstall" key
for hive in (winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER):
for root_path in (
r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall",
r"SOFTWARE\WOW6432Node\Microsoft\Windows\CurrentVersion\Uninstall",
):
try:
with winreg.OpenKey(hive, root_path) as base:
i = 0
while True:
try:
sub = winreg.EnumKey(base, i)
i += 1
if "firefox" not in sub.lower():
continue
with winreg.OpenKey(base, sub) as k:
loc = winreg.QueryValueEx(k, "InstallLocation")[0]
candidate = Path(loc) / "firefox.exe"
if candidate.is_file():
return str(candidate)
except OSError:
break
except OSError:
continue
return ""
def default_firefox_path() -> str:
candidates = [
"""Find Firefox.exe — registry first, then well-known paths."""
# 1) Registry
reg = _find_firefox_via_registry()
if reg:
return reg
# 2) Standard install locations
candidates: list[Path] = [
Path(r"C:\Program Files\Mozilla Firefox\firefox.exe"),
Path(r"C:\Program Files (x86)\Mozilla Firefox\firefox.exe"),
Path(os.path.expandvars(r"%LOCALAPPDATA%\Mozilla Firefox\firefox.exe")),
Path(os.path.expandvars(r"%APPDATA%\Mozilla Firefox\firefox.exe")),
]
# 3) Scoop / winget / portable in PATH
which = shutil.which("firefox") or shutil.which("firefox.exe")
if which:
candidates.insert(0, Path(which))
for c in candidates:
if c.is_file():
return str(c)
@@ -50,13 +122,39 @@ def default_profile_dir() -> str:
return str(app_data_dir() / "browser_profiles" / "firefox_hardened")
def _firefox_is_running_on_system() -> bool:
"""Check if any firefox.exe process is alive (tasklist, no admin needed)."""
try:
r = subprocess.run(
["tasklist", "/FI", "IMAGENAME eq firefox.exe", "/NH", "/FO", "CSV"],
capture_output=True, text=True, timeout=6,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000),
)
return "firefox.exe" in (r.stdout or "").lower()
except Exception:
return False
class BrowserSession:
def __init__(self) -> None:
self._proc: subprocess.Popen[str] | None = None
self._profile_path: Path | None = None
self._launched_at: float = 0.0
def is_running(self) -> bool:
return self._proc is not None and self._proc.poll() is None
if self._proc is None:
return False
# Firefox's initial launcher process exits quickly (code 0) while
# browser child processes carry on. Poll the parent process but also
# fall back to checking the system process list so the GUI doesn't
# falsely report "stopped".
if self._proc.poll() is None:
return True
# Parent has exited — check if any firefox.exe is still alive AND we
# launched within the last 5 minutes (to avoid false positives from
# unrelated browser sessions).
age = time.monotonic() - self._launched_at
return age < 300 and _firefox_is_running_on_system()
def pid(self) -> int | None:
if self._proc is None:
@@ -72,11 +170,22 @@ class BrowserSession:
signup_draft: SignupDraft | None = None,
) -> tuple[bool, str]:
raw_exe = (cfg.firefox_path or default_firefox_path()).strip().strip('"').strip("'")
if not raw_exe:
return (
False,
"Firefox not found. Install Firefox then set the path in the Browser tab "
"(or browse to it with the Browse button).",
)
exe = Path(raw_exe)
if not exe.is_file():
return False, f"Firefox executable not found: {raw_exe or '(empty path)'}"
return False, (
f"Firefox executable not found: {raw_exe} "
"Use the Browse button on the Browser tab to locate firefox.exe."
)
raw_profile = (cfg.profile_dir or default_profile_dir()).strip().strip('"').strip("'")
profile_dir = Path(raw_profile)
hard = FirefoxHardening(
force_proxy=cfg.force_proxy,
disable_webrtc=cfg.disable_webrtc,
@@ -89,46 +198,88 @@ class BrowserSession:
persona_key=cfg.persona,
cookie_mode=cfg.cookie_mode,
)
if cfg.lock_managed_profile:
ensure_firefox_profile(profile_dir, proxy_host, int(proxy_port), hard)
if signup_draft is not None:
install_signup_extension(profile_dir, signup_draft)
try:
if cfg.lock_managed_profile:
ensure_firefox_profile(profile_dir, proxy_host, int(proxy_port), hard)
if signup_draft is not None:
install_signup_extension(profile_dir, signup_draft)
except Exception as e:
return False, f"Profile setup failed: {e!s}"
if self.is_running():
self.stop()
if cfg.lock_managed_profile:
args = [str(exe), "-no-remote", "-profile", str(profile_dir)]
else:
args = [str(exe)]
url = (start_url or "").strip()
if url:
args.append(url)
env = os.environ.copy()
env.update(persona_env(hard))
try:
self._proc = subprocess.Popen(
args,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
# DETACHED_PROCESS ensures Firefox outlives Python and is not
# killed when Python's job object exits. CREATE_NO_WINDOW
# suppresses any console window. CREATE_NEW_PROCESS_GROUP
# isolates Ctrl-C handling.
creationflags=_CREATE_FLAGS,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
text=True,
env=env,
)
# Smoke-check: catch immediate startup crashes and report clearly.
time.sleep(0.4)
rc = self._proc.poll()
if rc is not None:
self._proc = None
return False, f"Firefox exited immediately (code {rc}). Check path/profile."
self._profile_path = profile_dir if cfg.lock_managed_profile else None
log.info("Launched hardened Firefox pid=%s profile=%s", self._proc.pid, profile_dir)
if cfg.lock_managed_profile:
return True, f"Hardened Firefox started (pid {self._proc.pid})."
return True, f"Firefox started (unlocked mode, pid {self._proc.pid})."
except Exception as e:
self._proc = None
return False, f"Launch failed: {e!s}"
self._launched_at = time.monotonic()
# Firefox's multi-process launcher exits (code 0) within ~0.5 s while
# the actual browser window loads in child processes. We therefore do
# NOT treat a quick exit of the parent as an error — we just verify
# that *some* firefox.exe appears in the process list within 3 s.
deadline = time.monotonic() + 3.0
found = False
while time.monotonic() < deadline:
time.sleep(0.3)
rc = self._proc.poll()
# If it already exited with an error code, report it.
if rc is not None and rc not in (0, None):
self._proc = None
return False, f"Firefox exited with error code {rc}."
if _firefox_is_running_on_system():
found = True
break
if not found:
# One last check — maybe the profile write was the only action needed
# and Firefox opened fast; accept success if parent is still alive.
if self._proc.poll() is None:
found = True
if not found:
self._proc = None
return False, (
"Firefox did not appear in the process list after 3 s. "
"Check that the path is correct and Firefox is not already blocked by another -profile lock."
)
self._profile_path = profile_dir if cfg.lock_managed_profile else None
pid = self._proc.pid
log.info("Launched hardened Firefox pid=%s profile=%s", pid, profile_dir)
mode = "hardened profile" if cfg.lock_managed_profile else "unlocked mode"
return True, f"Firefox started ({mode}, pid {pid})."
def stop(self, dispose: bool = False) -> tuple[bool, str]:
# Terminate the Popen handle if still alive
if self._proc and self._proc.poll() is None:
try:
self._proc.terminate()
@@ -140,9 +291,24 @@ class BrowserSession:
pass
finally:
self._proc = None
# Also kill any remaining firefox.exe processes (handles the
# detached-child case where the parent already exited naturally)
try:
subprocess.run(
["taskkill", "/F", "/IM", "firefox.exe"],
capture_output=True,
timeout=10,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000),
)
except Exception:
pass
self._proc = None
if dispose and self._profile_path and self._profile_path.exists():
try:
shutil.rmtree(self._profile_path, ignore_errors=True)
except Exception:
pass
self._launched_at = 0.0
return True, "Browser stopped."

78
proxy_chain_manager/browser_profile.py
View File

@@ -4,9 +4,11 @@ from dataclasses import dataclass, field
from pathlib import Path
from .browser_identity import (
CookiePolicy,
Persona,
cookie_behavior_value,
cookie_session_only,
get_cookie_policy,
get_persona,
is_blend_persona,
)
@@ -180,28 +182,62 @@ def build_user_js(
f'user_pref("general.useragent.override", "{hard.user_agent_override}");'
)
# Cookie policy (persona-independent)
behavior = cookie_behavior_value(hard.cookie_mode)
lines.extend(
[
f'user_pref("network.cookie.cookieBehavior", {behavior});',
# 0 = expire normally, 2 = current session only
f'user_pref("network.cookie.lifetimePolicy", {2 if cookie_session_only(hard.cookie_mode) else 0});',
]
)
# ── Cookie policy ────────────────────────────────────────────────────────
cp: CookiePolicy = get_cookie_policy(hard.cookie_mode)
if hard.clear_on_shutdown:
lines.extend(
[
'user_pref("privacy.sanitize.sanitizeOnShutdown", true);',
'user_pref("privacy.clearOnShutdown.history", true);',
'user_pref("privacy.clearOnShutdown.cookies", true);',
'user_pref("privacy.clearOnShutdown.cache", true);',
'user_pref("privacy.clearOnShutdown.downloads", true);',
'user_pref("privacy.clearOnShutdown.formdata", true);',
'user_pref("privacy.clearOnShutdown.sessions", true);',
]
)
lines.extend([
f'user_pref("network.cookie.cookieBehavior", {cp.behavior});',
# pbmode = private-browsing mode uses same behavior
f'user_pref("network.cookie.cookieBehavior.pbmode", {cp.behavior});',
# lifetime: 0 = normal, 2 = session-only
f'user_pref("network.cookie.lifetimePolicy", {cp.lifetime});',
])
# Total Cookie Protection (TCP / Firefox partitioning)
if cp.partition:
lines.extend([
'user_pref("network.cookie.cookieBehavior.optInPartitioning", true);',
# Also isolate cache, localStorage, IndexedDB per top-level site
'user_pref("privacy.partition.serviceWorkers", true);',
'user_pref("privacy.partition.network_state", true);',
'user_pref("privacy.partition.bloburl_by_default", true);',
])
else:
lines.extend([
'user_pref("network.cookie.cookieBehavior.optInPartitioning", false);',
])
# SameSite=None cookies must be Secure (good hygiene regardless of mode)
lines.append('user_pref("network.cookie.sameSite.noneRequiresSecure", true);')
# ── Sanitize-on-shutdown ─────────────────────────────────────────────────
# The CookiePolicy sanitize flag takes priority; the hard.clear_on_shutdown
# toggle acts as an additional override for cookies+cache even when the
# cookie mode doesn't request it.
do_sanitize = cp.sanitize or hard.clear_on_shutdown
if do_sanitize:
lines.append('user_pref("privacy.sanitize.sanitizeOnShutdown", true);')
# Cookies
lines.append(f'user_pref("privacy.clearOnShutdown.cookies", {_bool(cp.clear_cookies or hard.clear_on_shutdown)});')
# Cache
lines.append(f'user_pref("privacy.clearOnShutdown.cache", {_bool(cp.clear_cache or hard.clear_on_shutdown)});')
# History
lines.append(f'user_pref("privacy.clearOnShutdown.history", {_bool(cp.clear_history or hard.clear_on_shutdown)});')
# Downloads list
lines.append('user_pref("privacy.clearOnShutdown.downloads", true);')
# Form / search bar autofill data
lines.append(f'user_pref("privacy.clearOnShutdown.formdata", {_bool(cp.clear_formdata or hard.clear_on_shutdown)});')
# Active sessions (HTTP auth, logged-in sites)
lines.append('user_pref("privacy.clearOnShutdown.sessions", true);')
# localStorage + IndexedDB
lines.append(f'user_pref("privacy.clearOnShutdown.localStorage", {_bool(cp.clear_ls or hard.clear_on_shutdown)});')
lines.append(f'user_pref("privacy.clearOnShutdown.indexedDB", {_bool(cp.clear_ls or hard.clear_on_shutdown)});')
# Service worker registrations
lines.append('user_pref("privacy.clearOnShutdown.serviceWorkers", true);')
# Offline web app cache
lines.append('user_pref("privacy.clearOnShutdown.offlineApps", true);')
else:
lines.append('user_pref("privacy.sanitize.sanitizeOnShutdown", false);')
if hard.timezone_utc:
lines.append('user_pref("privacy.resistFingerprinting.reduceTimerPrecision", true);')

214
proxy_chain_manager/chain_map.py
View File

@@ -1,4 +1,8 @@
"""World-map visualization for proxy chains (Pillow + equirectangular projection)."""
"""World-map visualization for proxy chains.
Renders hop arcs over the bundled neon world map image. Falls back to a
drawn landmass map if the image file is absent.
"""
from __future__ import annotations
import ipaddress
@@ -6,10 +10,11 @@ import logging
import math
import socket
from dataclasses import dataclass
from pathlib import Path
from typing import Literal
from urllib.parse import urlparse
from PIL import Image, ImageDraw, ImageFont
from PIL import Image, ImageDraw, ImageFilter, ImageFont
try:
_RESAMPLE = Image.Resampling.LANCZOS
@@ -23,21 +28,42 @@ log = logging.getLogger(__name__)
ChainStatus = Literal["healthy", "connecting", "dead", "idle"]
# Theme colors (match app.py palette)
_BG = "#070713"
_GRID = "#141432"
_LAND = "#12182e"
_LAND_EDGE = "#1e2a4a"
_LINE_HEALTHY = "#00e5ff"
# Theme palette
_BG = "#070713"
_GRID = "#1a1f3a"
_LAND = "#12182e"
_LAND_EDGE = "#1e2a4a"
_LINE_HEALTHY = "#00e5ff"
_LINE_CONNECTING = "#ffd000"
_LINE_DEAD = "#ff2a6d"
_LINE_IDLE = "#3052ff"
_YOU = "#00ff9c"
_HOP = "#00e5ff"
_EXIT = "#ffd000"
_LABEL = "#eaf2ff"
_LABEL_DIM = "#7a8aab"
_UNKNOWN = "#2a2f4a"
_LINE_DEAD = "#ff2a6d"
_LINE_IDLE = "#3052ff"
_YOU = "#00ff9c"
_HOP = "#00e5ff"
_EXIT = "#ffd000"
_LABEL = "#eaf2ff"
_LABEL_DIM = "#7a8aab"
_UNKNOWN = "#2a2f4a"
def _map_img_path() -> Path:
"""Locate world_map.png whether running from source or a frozen PyInstaller bundle."""
import sys
if getattr(sys, "frozen", False) and hasattr(sys, "_MEIPASS"):
return Path(sys._MEIPASS) / "proxy_chain_manager" / "world_map.png"
return Path(__file__).resolve().parent / "world_map.png"
_cached_map: Image.Image | None = None
def _load_world_map(width: int, height: int) -> Image.Image | None:
"""Return the neon world map resized to (width × height), or None if unavailable."""
global _cached_map
try:
if _cached_map is None:
_cached_map = Image.open(_map_img_path()).convert("RGB")
return _cached_map.resize((width, height), _RESAMPLE)
except Exception as exc:
log.debug("world_map load failed: %s", exc)
return None
@dataclass(frozen=True)
@@ -134,26 +160,23 @@ def build_chain_map_points(
ip = resolve_host_ip(host, timeout_seconds=min(4.0, timeout_seconds))
if not ip:
continue
short = redact_proxy_url(hop)
if len(short) > 22:
short = short[:20] + "…"
intel = fetch_ip_geo(ip, timeout_seconds)
pt = _intel_to_point("hop", f"H{i + 1}", intel, hop_index=i)
if pt:
points.append(pt)
exit = (exit_ip or "").strip()
if exit and exit != "—":
exit_str = (exit_ip or "").strip()
if exit_str and exit_str != "—":
if (
exit_intel
and exit_intel.ok
and exit_intel.ip == exit
and exit_intel.ip == exit_str
and exit_intel.lat is not None
and exit_intel.lon is not None
):
pt = _intel_to_point("exit", "EXIT", exit_intel)
else:
pt = _intel_to_point("exit", "EXIT", fetch_ip_geo(exit, timeout_seconds))
pt = _intel_to_point("exit", "EXIT", fetch_ip_geo(exit_str, timeout_seconds))
if pt:
points.append(pt)
@@ -173,7 +196,7 @@ def _great_circle_points(
lon1: float,
lat2: float,
lon2: float,
steps: int = 24,
steps: int = 32,
) -> list[tuple[float, float]]:
"""Interpolate along a great circle for curved hop arcs."""
phi1, lam1 = math.radians(lat1), math.radians(lon1)
@@ -194,21 +217,50 @@ def _great_circle_points(
x = a * math.cos(phi1) * math.cos(lam1) + b * math.cos(phi2) * math.cos(lam2)
y = a * math.cos(phi1) * math.sin(lam1) + b * math.cos(phi2) * math.sin(lam2)
z = a * math.sin(phi1) + b * math.sin(phi2)
out.append((math.degrees(math.atan2(z, math.sqrt(x * x + y * y))), math.degrees(math.atan2(y, x))))
out.append((
math.degrees(math.atan2(z, math.sqrt(x * x + y * y))),
math.degrees(math.atan2(y, x)),
))
return out
def _line_color(status: ChainStatus) -> str:
return {
"healthy": _LINE_HEALTHY,
"connecting": _LINE_CONNECTING,
"dead": _LINE_DEAD,
"idle": _LINE_IDLE,
}.get(status, _LINE_IDLE)
def _line_color(status: ChainStatus) -> tuple[int, int, int]:
table = {
"healthy": (0, 229, 255),
"connecting": (255, 208, 0),
"dead": (255, 42, 109),
"idle": (48, 82, 255),
}
return table.get(status, (48, 82, 255))
def _node_color(role: str) -> str:
return {"you": _YOU, "hop": _HOP, "exit": _EXIT}.get(role, _HOP)
def _node_color(role: str) -> tuple[int, int, int]:
table = {"you": (0, 255, 156), "hop": (0, 229, 255), "exit": (255, 208, 0)}
return table.get(role, (0, 229, 255))
def _hex_to_rgb(h: str) -> tuple[int, int, int]:
h = h.lstrip("#")
return int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)
def _draw_glow_line(
draw: ImageDraw.ImageDraw,
pts: list[tuple[float, float]],
color: tuple[int, int, int],
) -> None:
"""Draw neon glow arc: fat dim outer halo → bright core line."""
if len(pts) < 2:
return
r, g, b = color
# outer halo — wider, dim
halo = (r // 4, g // 4, b // 4)
draw.line(pts, fill=halo, width=7)
# mid glow
mid = (r // 2, g // 2, b // 2)
draw.line(pts, fill=mid, width=4)
# bright core
draw.line(pts, fill=color, width=2)
def _load_font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
@@ -221,18 +273,13 @@ def _load_font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
def _draw_landmasses(draw: ImageDraw.ImageDraw, width: int, height: int, margin: int) -> None:
"""Minimal continent silhouettes — stylized, not survey-accurate."""
"""Fallback minimal continent silhouettes."""
blobs: list[list[tuple[float, float]]] = [
# North America
[(-168, 72), (-140, 75), (-100, 72), (-80, 50), (-75, 25), (-95, 15), (-110, 20), (-125, 48), (-168, 72)],
# South America
[(-82, 12), (-72, -5), (-55, -35), (-65, -55), (-75, -18), (-82, 12)],
# Europe / Africa
[(-10, 72), (30, 70), (40, 55), (35, 30), (20, 5), (-5, 5), (-18, 28), (-10, 72)],
[(15, 35), (35, 32), (50, 12), (42, -5), (18, -35), (15, 35)],
# Asia
[(40, 72), (100, 75), (140, 55), (130, 35), (110, 10), (80, 8), (60, 25), (40, 45), (40, 72)],
# Australia
[(115, -12), (135, -12), (150, -25), (145, -38), (115, -38), (115, -12)],
]
for poly in blobs:
@@ -246,58 +293,75 @@ def render_chain_map(
width: int = 960,
height: int = 220,
status: ChainStatus = "idle",
margin: int = 18,
margin: int = 12,
) -> Image.Image:
"""Render chain hops on a dark equirectangular world map."""
img = Image.new("RGB", (max(320, width), max(120, height)), _BG)
draw = ImageDraw.Draw(img)
_draw_landmasses(draw, width, height, margin)
"""Render chain hops over the neon world map (or a drawn fallback)."""
w = max(320, width)
h = max(120, height)
# Lat/lon grid
for lat in range(-60, 91, 30):
y = _project(lat, 0, width, height, margin)[1]
draw.line([(margin, y), (width - margin, y)], fill=_GRID, width=1)
for lon in range(-150, 181, 30):
x = _project(0, lon, width, height, margin)[0]
draw.line([(x, margin), (x, height - margin)], fill=_GRID, width=1)
# ── background ──────────────────────────────────────────────────────────
world = _load_world_map(w, h)
if world is not None:
# Slightly darken so overlaid lines pop
bg = Image.new("RGB", (w, h), (0, 0, 0))
img = Image.blend(world, bg, alpha=0.22)
else:
img = Image.new("RGB", (w, h), _BG)
draw_bg = ImageDraw.Draw(img)
_draw_landmasses(draw_bg, w, h, margin)
# grid
for lat in range(-60, 91, 30):
y = _project(lat, 0, w, h, margin)[1]
draw_bg.line([(margin, y), (w - margin, y)], fill=_GRID, width=1)
for lon in range(-150, 181, 30):
x = _project(0, lon, w, h, margin)[0]
draw_bg.line([(x, margin), (x, h - margin)], fill=_GRID, width=1)
draw = ImageDraw.Draw(img)
if len(points) < 2:
font = _load_font(13)
font = _load_font(12)
msg = "Start the chain to plot hops on the map." if not points else "Need 2+ geo points to draw path."
draw.text((margin, height // 2 - 8), msg, fill=_LABEL_DIM, font=font)
# drop shadow
draw.text((margin + 1, h // 2 - 7), msg, fill=(0, 0, 0), font=font)
draw.text((margin, h // 2 - 8), msg, fill=_hex_to_rgb(_LABEL_DIM), font=font)
return img
line_c = _line_color(status)
projected = [_project(p.lat, p.lon, width, height, margin) for p in points]
line_rgb = _line_color(status)
projected = [_project(p.lat, p.lon, w, h, margin) for p in points]
# Glow underlay
# ── arc lines between consecutive hops ─────────────────────────────────
for i in range(len(projected) - 1):
arc = _great_circle_points(points[i].lat, points[i].lon, points[i + 1].lat, points[i + 1].lon)
arc_xy = [_project(lat, lon, width, height, margin) for lat, lon in arc]
draw.line(arc_xy, fill=line_c, width=5)
for i in range(len(projected) - 1):
arc = _great_circle_points(points[i].lat, points[i].lon, points[i + 1].lat, points[i + 1].lon)
arc_xy = [_project(lat, lon, width, height, margin) for lat, lon in arc]
draw.line(arc_xy, fill=line_c, width=2)
arc = _great_circle_points(
points[i].lat, points[i].lon,
points[i + 1].lat, points[i + 1].lon,
)
arc_xy = [_project(lat, lon, w, h, margin) for lat, lon in arc]
_draw_glow_line(draw, arc_xy, line_rgb)
label_font = _load_font(11)
# ── nodes ───────────────────────────────────────────────────────────────
label_font = _load_font(11)
detail_font = _load_font(9)
for pt, (x, y) in zip(points, projected):
color = _node_color(pt.role)
nc = _node_color(pt.role)
r = 7 if pt.role == "exit" else (6 if pt.role == "you" else 5)
draw.ellipse((x - r - 2, y - r - 2, x + r + 2, y + r + 2), fill=color)
draw.ellipse((x - r, y - r, x + r, y + r), fill=_BG, outline=color, width=2)
tx, ty = x + 10, y - 14
draw.text((tx + 1, ty + 1), pt.label, fill="#000000", font=label_font)
draw.text((tx, ty), pt.label, fill=_LABEL, font=label_font)
# outer glow ring
glow = (nc[0] // 3, nc[1] // 3, nc[2] // 3)
draw.ellipse((x - r - 4, y - r - 4, x + r + 4, y + r + 4), fill=glow)
# filled node with dark center
draw.ellipse((x - r, y - r, x + r, y + r), fill=nc)
draw.ellipse((x - r + 2, y - r + 2, x + r - 2, y + r - 2), fill=(5, 5, 20))
tx, ty = int(x + r + 5), int(y - 8)
# text drop-shadow
draw.text((tx + 1, ty + 1), pt.label, fill=(0, 0, 0), font=label_font)
draw.text((tx, ty), pt.label, fill=_hex_to_rgb(_LABEL), font=label_font)
if pt.detail:
draw.text((tx, ty + 13), pt.detail[:28], fill=_LABEL_DIM, font=detail_font)
draw.text((tx, ty + 12), pt.detail[:30], fill=_hex_to_rgb(_LABEL_DIM), font=detail_font)
return img
def resize_map_display(pil_img: Image.Image, width: int, height: int) -> Image.Image:
"""Resize rendered map for the GUI widget."""
w = max(320, int(width))
h = max(120, int(height))
return pil_img.resize((w, h), _RESAMPLE)
return pil_img.resize((max(320, int(width)), max(120, int(height))), _RESAMPLE)

295
proxy_chain_manager/dns_leak.py
View File

@@ -1,41 +1,118 @@
"""DNS leak checks and cache flush."""
"""DNS leak detection and cache flushing.
Two levels of testing:
1. check_dns_leak_hint() — fast, no network: inspect configured DNS resolvers
2. run_dns_leak_test() — real network test: compare DNS answers seen through
the proxy vs direct, and probe resolver identity via
a dedicated leak-test API.
"""
from __future__ import annotations
import concurrent.futures
import ipaddress
import logging
import socket
import subprocess
from dataclasses import dataclass
from dataclasses import dataclass, field
import httpx
log = logging.getLogger(__name__)
_DOH_GOOGLE = "https://dns.google/resolve?name=whoami.dnsleaktest.com&type=A"
# ─── DoH endpoint used to ask "what is my apparent source IP?" ──────────────
_DOH_IP_CHECK = "https://api.ipify.org?format=json"
_DNS_LEAK_API = "https://dnsleaktest.com/api/v1/start"
_DNS_LEAK_RESULT = "https://dnsleaktest.com/api/v1/results"
# Well-known public resolver IP ranges (prefix → label)
_PUBLIC_DNS_LABELS: dict[str, str] = {
"8.8.8.8": "Google", "8.8.4.4": "Google",
"1.1.1.1": "Cloudflare", "1.0.0.1": "Cloudflare",
"9.9.9.9": "Quad9", "149.112.112.112": "Quad9",
"208.67.222.222": "OpenDNS", "208.67.220.220": "OpenDNS",
"76.76.2.0": "Alternate DNS",
"94.140.14.14": "AdGuard",
}
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class DnsLeakResult:
"""Legacy single-check result kept for API compatibility."""
ok: bool
system_resolvers: list[str]
message: str
doh_ip: str | None = None
@dataclass
class DnsResolverInfo:
ip: str
hostname: str = ""
country: str = ""
isp: str = ""
label: str = "" # "Google", "Cloudflare", or ""
is_public_known: bool = False
@dataclass
class FullDnsLeakReport:
"""Comprehensive DNS leak report from run_dns_leak_test()."""
# Resolvers that answered DNS queries in this session
resolvers_seen: list[DnsResolverInfo] = field(default_factory=list)
# Resolvers configured by the OS
system_resolvers: list[str] = field(default_factory=list)
# IPs resolved for test hostnames — via proxy vs direct
proxy_resolution: dict[str, list[str]] = field(default_factory=dict)
direct_resolution: dict[str, list[str]] = field(default_factory=dict)
# True when proxy and direct give the same answers (possible leak)
resolution_matches_direct: bool = False
# Whether any configured resolver is a known public server
has_public_resolver: bool = False
# Whether any configured resolver is outside local network
has_external_resolver: bool = False
# Overall verdict
leaked: bool = False
summary: str = ""
errors: list[str] = field(default_factory=list)
# ─────────────────────────────────────────────────────────────────────────────
# System resolver detection
# ─────────────────────────────────────────────────────────────────────────────
def get_system_dns_servers() -> list[str]:
"""Return configured IPv4 DNS resolvers via PowerShell (falls back to ipconfig)."""
try:
r = subprocess.run(
["powershell", "-NoProfile", "-Command",
"(Get-DnsClientServerAddress -AddressFamily IPv4 | "
"Where-Object { $_.ServerAddresses } | "
"Select-Object -ExpandProperty ServerAddresses) -join ','"],
capture_output=True,
text=True,
timeout=12,
capture_output=True, text=True, timeout=12,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)
raw = (r.stdout or "").strip()
if not raw:
return []
return [x.strip() for x in raw.replace(";", ",").split(",") if x.strip()]
if raw:
return [x.strip() for x in raw.replace(";", ",").split(",") if x.strip()]
except Exception:
pass
# ipconfig fallback
try:
r = subprocess.run(
["ipconfig", "/all"], capture_output=True, text=True, timeout=10,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)
servers: list[str] = []
for line in (r.stdout or "").splitlines():
if "DNS Servers" in line or "DNS Server" in line:
parts = line.split(":", 1)
if len(parts) == 2:
ip = parts[1].strip()
if ip:
servers.append(ip)
return servers
except Exception:
return []
@@ -43,70 +120,194 @@ def get_system_dns_servers() -> list[str]:
def flush_dns_cache() -> tuple[bool, str]:
try:
r = subprocess.run(
["ipconfig", "/flushdns"],
capture_output=True,
text=True,
timeout=15,
["ipconfig", "/flushdns"], capture_output=True, text=True, timeout=15,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)
msg = (r.stdout or r.stderr or "").strip().splitlines()[-1] if r.returncode == 0 else "flush failed"
lines = (r.stdout or r.stderr or "").strip().splitlines()
msg = lines[-1] if lines else "flushed"
return r.returncode == 0, msg
except Exception as e:
return False, str(e)
def check_dns_leak_hint(local_proxy: str | None = None) -> DnsLeakResult:
"""Heuristic: list configured DNS servers; note if any are public ISP resolvers.
# ─────────────────────────────────────────────────────────────────────────────
# Hostname resolution helpers
# ─────────────────────────────────────────────────────────────────────────────
Full DNS leak testing needs OS-level routing; this flags obvious misconfig.
"""
resolvers = get_system_dns_servers()
private_prefixes = ("127.", "10.", "192.168.", "172.16.", "172.17.", "172.18.",
"172.19.", "172.2", "172.30.", "172.31.", "0.0.0.0")
public = [r for r in resolvers if not any(r.startswith(p) for p in private_prefixes)]
doh_ip: str | None = None
def _resolve_direct(hostname: str, timeout: float = 4.0) -> list[str]:
"""Resolve hostname using system DNS (direct, not through proxy)."""
try:
with httpx.Client(timeout=8.0, verify=True) as c:
r = c.get(_DOH_GOOGLE)
socket.setdefaulttimeout(timeout)
infos = socket.getaddrinfo(hostname, None)
seen: list[str] = []
for info in infos:
addr = info[4][0]
if addr and addr not in seen:
seen.append(addr)
return seen
except Exception:
return []
finally:
socket.setdefaulttimeout(None)
def _resolve_via_proxy(hostname: str, proxy_url: str, timeout: float = 8.0) -> list[str]:
"""Resolve hostname by fetching a DNS-over-HTTPS endpoint through the proxy."""
# Use Google DoH JSON API through the proxy so DNS is resolved on the exit node
doh = f"https://dns.google/resolve?name={hostname}&type=A"
try:
with httpx.Client(proxy=proxy_url, timeout=timeout, verify=False, follow_redirects=True) as c:
r = c.get(doh)
if r.status_code == 200:
data = r.json()
answers = data.get("Answer") or []
if answers:
doh_ip = str(answers[0].get("data", ""))
return [ans["data"] for ans in (data.get("Answer") or [])
if ans.get("type") == 1]
except Exception:
pass
return []
def _is_private(ip: str) -> bool:
try:
return ipaddress.ip_address(ip).is_private
except Exception:
return False
# ─────────────────────────────────────────────────────────────────────────────
# Main leak test
# ─────────────────────────────────────────────────────────────────────────────
_LEAK_TEST_HOSTS = [
"google.com",
"cloudflare.com",
"github.com",
"amazon.com",
]
def run_dns_leak_test(
proxy_url: str | None = None,
timeout: float = 10.0,
) -> FullDnsLeakReport:
"""
Comprehensive DNS leak test:
- Collect system DNS resolver configuration
- Resolve test hostnames both directly (system DNS) and via proxy (DoH through chain)
- Compare answers: matching answers suggest DNS is NOT going through the proxy
- Flag known public resolvers that would bypass the chain
"""
rep = FullDnsLeakReport()
rep.system_resolvers = get_system_dns_servers()
_PRIVATE_PREFIXES = (
"127.", "10.", "192.168.", "172.16.", "172.17.", "172.18.",
"172.19.", "172.2", "172.30.", "172.31.", "169.254.",
)
for r_ip in rep.system_resolvers:
is_public_known = r_ip in _PUBLIC_DNS_LABELS
is_external = not any(r_ip.startswith(p) for p in _PRIVATE_PREFIXES)
label = _PUBLIC_DNS_LABELS.get(r_ip, "")
rep.has_public_resolver = rep.has_public_resolver or is_public_known
rep.has_external_resolver = rep.has_external_resolver or is_external
rep.resolvers_seen.append(DnsResolverInfo(
ip=r_ip,
label=label,
is_public_known=is_public_known,
))
# Resolve test hosts in parallel
all_match = True
any_resolved_via_proxy = False
def _test_host(host: str) -> tuple[str, list[str], list[str]]:
direct = _resolve_direct(host, timeout=min(4.0, timeout))
proxy = _resolve_via_proxy(host, proxy_url, timeout=timeout) if proxy_url else []
return host, direct, proxy
try:
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
futs = {pool.submit(_test_host, h): h for h in _LEAK_TEST_HOSTS}
for fut in concurrent.futures.as_completed(futs, timeout=timeout + 2):
try:
host, direct, proxy = fut.result()
rep.direct_resolution[host] = direct
rep.proxy_resolution[host] = proxy
if proxy:
any_resolved_via_proxy = True
# Overlap: if proxy and direct gave the same IPs, DNS
# might not be going through the proxy (same resolver path)
direct_set = set(direct)
proxy_set = set(proxy)
if direct_set and proxy_set and direct_set == proxy_set:
pass # this host matches
else:
all_match = False
except Exception as e:
rep.errors.append(str(e))
except Exception as e:
rep.errors.append(f"Thread pool error: {e}")
rep.resolution_matches_direct = all_match and any_resolved_via_proxy
# ── Build verdict ──────────────────────────────────────────────────────
problems: list[str] = []
if rep.has_external_resolver:
external = [r for r in rep.system_resolvers
if not any(r.startswith(p) for p in _PRIVATE_PREFIXES)]
names = [f"{ip} ({_PUBLIC_DNS_LABELS[ip]})" if ip in _PUBLIC_DNS_LABELS else ip
for ip in external[:4]]
problems.append(f"OS DNS: {', '.join(names)} — DNS may bypass proxy chain")
if rep.resolution_matches_direct and proxy_url:
problems.append("Proxy DNS resolution matches direct — possible DNS leak (same upstream resolver)")
if problems:
rep.leaked = True
rep.summary = " · ".join(problems)
else:
rep.leaked = False
if proxy_url:
rep.summary = "DNS routing looks clean — proxy resolves differently from direct."
else:
resolver_str = ", ".join(rep.system_resolvers[:3]) if rep.system_resolvers else "none detected"
rep.summary = f"System resolvers: {resolver_str}"
return rep
# ─────────────────────────────────────────────────────────────────────────────
# Legacy quick hint (used by Privacy tab DNS section)
# ─────────────────────────────────────────────────────────────────────────────
def check_dns_leak_hint(local_proxy: str | None = None) -> DnsLeakResult:
"""Fast heuristic: inspect configured DNS resolvers, flag public ones."""
resolvers = get_system_dns_servers()
private_prefixes = (
"127.", "10.", "192.168.", "172.16.", "172.17.", "172.18.",
"172.19.", "172.2", "172.30.", "172.31.",
)
public = [r for r in resolvers if not any(r.startswith(p) for p in private_prefixes)]
if not resolvers:
return DnsLeakResult(
ok=True,
system_resolvers=[],
message="No IPv4 DNS servers reported (DHCP may assign later).",
doh_ip=doh_ip,
)
return DnsLeakResult(ok=True, system_resolvers=[], message="No IPv4 DNS servers reported (DHCP).")
if public and local_proxy:
return DnsLeakResult(
ok=False,
system_resolvers=resolvers,
ok=False, system_resolvers=resolvers,
message=(
f"DNS may bypass proxy chain: public resolvers {', '.join(public)}. "
"Use kill-switch + VPN, or set DNS to localhost when hardened."
),
doh_ip=doh_ip,
)
if public and not local_proxy:
if public:
return DnsLeakResult(
ok=False,
system_resolvers=resolvers,
ok=False, system_resolvers=resolvers,
message=f"Public DNS resolvers active: {', '.join(public)}",
doh_ip=doh_ip,
)
return DnsLeakResult(
ok=True,
system_resolvers=resolvers,
ok=True, system_resolvers=resolvers,
message=f"DNS servers: {', '.join(resolvers)}",
doh_ip=doh_ip,
)

181
proxy_chain_manager/fingerprint.py
View File

@@ -195,35 +195,186 @@ def apply_webrtc_hardening(enable: bool) -> tuple[bool, str]:
return False, str(e)
def _get_os_info() -> str:
try:
import platform
v = platform.version()
r = platform.release()
m = platform.machine()
return f"Windows {r} build {v} {m}"
except Exception:
return "unknown"
def _get_timezone() -> str:
try:
import datetime
tz = datetime.datetime.now(datetime.timezone.utc).astimezone()
name = str(tz.tzname() or "")
offset = tz.utcoffset()
h = int(offset.total_seconds() // 3600) if offset else 0
return f"{name} (UTC{h:+d})" if name else f"UTC{h:+d}"
except Exception:
return "unknown"
def _get_screen_resolution() -> str:
try:
r = subprocess.run(
["powershell", "-NoProfile", "-Command",
"Add-Type -AssemblyName System.Windows.Forms;"
"[System.Windows.Forms.Screen]::PrimaryScreen.Bounds | "
"ForEach-Object { \"$($_.Width)x$($_.Height)\" }"],
capture_output=True, text=True, timeout=8,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)
return (r.stdout or "").strip() or "unknown"
except Exception:
return "unknown"
def check_browser_fingerprint_consistency(profile_dir: "Path") -> list[str]:
"""
Verify that the managed Firefox user.js is internally consistent and
doesn't mix prefs that would expose the browser as fingerprint-hardened
while still advertising a normal user-agent.
Returns a list of warning strings (empty = consistent).
"""
from pathlib import Path
user_js = Path(profile_dir) / "user.js"
if not user_js.is_file():
return ["user.js not found — profile not yet initialized"]
try:
content = user_js.read_text(encoding="utf-8", errors="replace")
except Exception as e:
return [f"Cannot read user.js: {e}"]
warnings: list[str] = []
def _has(pref: str, value: str) -> bool:
return f'"{pref}", {value}' in content
# 1. RFP + custom UA is a contradiction (RFP overrides UA)
if _has("privacy.resistFingerprinting", "true") and 'general.useragent.override' in content:
warnings.append(
"RFP + UA override conflict: privacy.resistFingerprinting=true overrides "
"general.useragent.override — the spoofed UA is ignored"
)
# 2. proxy not set but force_proxy claimed
if _has("network.proxy.type", "1"):
if 'network.proxy.http",' not in content:
warnings.append("proxy.type=1 but network.proxy.http is missing — browsers will error")
# 3. FPI on + third-party cookies allowed = inconsistent privacy posture
if _has("privacy.firstparty.isolate", "true") and _has("network.cookie.cookieBehavior", "0"):
warnings.append(
"FPI enabled but cookieBehavior=0 (accept all) — cookies are isolated but not blocked"
)
# 4. sanitizeOnShutdown without clearing history = incomplete wipe
if _has("privacy.sanitize.sanitizeOnShutdown", "true"):
if not _has("privacy.clearOnShutdown.history", "true"):
warnings.append(
"sanitizeOnShutdown=true but clearOnShutdown.history not set — "
"history may survive session"
)
# 5. WebRTC peerconnection disabled is good — flag if missing
if not _has("media.peerconnection.enabled", "false"):
warnings.append("WebRTC not disabled in profile — real IP can leak via STUN")
# 6. network.trr.mode=5 (DoH off) is expected when using proxy DNS
if not _has("network.trr.mode", "5"):
warnings.append(
"network.trr.mode is not 5 — Firefox may use its own DoH resolver, "
"bypassing the proxy chain DNS path"
)
return warnings
def audit_device() -> FingerprintAudit:
"""Collect identifiers sites and trackers often fingerprint."""
"""Collect OS-level identifiers and consistency notes."""
import getpass
import os
from pathlib import Path
host = get_computer_name()
guid = get_machine_guid()
user = getpass.getuser()
macs = [f"{n.name}: {n.mac}" for n in list_nics()]
nics = list_nics()
macs = [f"{n.name}: {n.mac}" for n in nics]
os_ = _get_os_info()
tz_ = _get_timezone()
res_ = _get_screen_resolution()
lines = [
f"Computer name: {host or '—'}",
f"Windows username: {user}",
f"MachineGuid: {guid[:8]}…{guid[-4:]}" if len(guid) > 12 else f"MachineGuid: {guid or '—'}",
f"Network adapters ({len(macs)}):",
lines: list[str] = [
"── OS / Identity ──────────────────────────────────",
f" OS : {os_}",
f" Computer name: {host or '—'}",
f" Username : {user}",
f" MachineGuid : " + (f"{guid[:8]}…{guid[-4:]}" if len(guid) > 12 else guid or "—"),
f" Timezone : {tz_}",
f" Screen res : {res_}",
"",
"── Network adapters ───────────────────────────────",
]
lines.extend([f" • {m}" for m in macs[:8]] or [" • (none detected)"])
if len(macs) > 8:
lines.append(f" … and {len(macs) - 8} more")
lines.extend([f" • {m}" for m in macs[:10]] or [" • (none detected)"])
if len(macs) > 10:
lines.append(f" … and {len(macs) - 10} more")
lines.append("")
lines.append("Browser fingerprint (Canvas/WebGL/fonts) is not changed by this app.")
lines.append("Use hardened browser profiles + proxy chain for web traffic.")
lines.append("WebRTC toggle here affects Chrome/Edge system policy only.")
lines += [
"",
"── Browser fingerprint note ───────────────────────",
" Canvas, WebGL, font metrics, audio context, and screen dimensions",
" are NOT changed at the OS level by this app.",
" → Use 'blend_windows_chrome' persona OR 'hardened' mode to control",
" what the browser reports to sites.",
" → RFP (Resist Fingerprinting) makes Firefox stand out as hardened.",
" → Blend persona spoofs UA/platform to look like a normal Windows Chrome.",
"",
"── Recommendations ────────────────────────────────",
]
# Quick consistency checks
webrtc_ok, _ = apply_webrtc_hardening.__doc__ and True or False
try:
import winreg as _wr
with _wr.OpenKey(_wr.HKEY_LOCAL_MACHINE, r"SOFTWARE\Policies\Google\Chrome",
0, _wr.KEY_QUERY_VALUE) as k:
v = int(_wr.QueryValueEx(k, "DefaultWebRtcIpHandlingPolicy")[0])
webrtc_ok = v == 2
except Exception:
webrtc_ok = False
if not webrtc_ok:
lines.append(" ⚠ Chrome/Edge WebRTC policy not set — enable in Privacy tab")
else:
lines.append(" ✓ Chrome/Edge WebRTC policy is set")
# Check IPv6
try:
r = subprocess.run(
["powershell", "-NoProfile", "-NonInteractive", "-Command",
"Get-NetAdapterBinding -ComponentID ms_tcpip6 | Where-Object { $_.Enabled } | Measure-Object | Select-Object -ExpandProperty Count"],
capture_output=True, text=True, timeout=8,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)
count = int((r.stdout or "0").strip() or "0")
if count > 0:
lines.append(f" ⚠ IPv6 active on {count} adapter(s) — disable in Privacy tab for full v4-only chain")
else:
lines.append(" ✓ IPv6 disabled on all adapters")
except Exception:
lines.append(" ? IPv6 status unknown")
return FingerprintAudit(
lines=lines,
hostname=host,
machine_guid=guid,
username=user,
macs=[n.mac for n in list_nics()],
macs=[n.mac for n in nics],
)

13
proxy_chain_manager/signup_prep.py
View File

@@ -13,7 +13,16 @@ from typing import Any
from .paths import app_data_dir
_EXT_ID = "signup-autofill@proxygod"
_EXT_SRC = Path(__file__).resolve().parent / "signup_extension"
def _ext_src_path() -> Path:
"""Locate the signup_extension dir whether running from source or frozen."""
import sys
if getattr(sys, "frozen", False) and hasattr(sys, "_MEIPASS"):
return Path(sys._MEIPASS) / "proxy_chain_manager" / "signup_extension"
return Path(__file__).resolve().parent / "signup_extension"
@dataclass(frozen=True)
@@ -389,7 +398,7 @@ def install_signup_extension(profile_dir: Path, draft: SignupDraft) -> None:
dest = profile_dir / "extensions" / ext_name
if dest.exists():
shutil.rmtree(dest, ignore_errors=True)
shutil.copytree(_EXT_SRC, dest)
shutil.copytree(_ext_src_path(), dest)
cfg = build_autofill_config(draft)
(dest / "autofill_config.json").write_text(json.dumps(cfg, indent=2), encoding="utf-8")

211
proxy_chain_manager/webrtc_check.py Normal file
View File

@@ -0,0 +1,211 @@
"""WebRTC leak detection and hardening checks.
WebRTC leaks expose your real IP through STUN/ICE even when a proxy is set.
This module checks every controllable surface:
1. OS-level Chrome / Edge Group Policy (prevents non-proxied UDP)
2. Firefox managed profile prefs (user.js must disable PeerConnection)
3. STUN UDP reachability (if STUN is reachable, WebRTC CAN leak)
4. Firefox process prefs (confirm our user.js was accepted)
"""
from __future__ import annotations
import logging
import socket
import subprocess
import winreg
from dataclasses import dataclass, field
from pathlib import Path
log = logging.getLogger(__name__)
# STUN servers commonly used by WebRTC
_STUN_HOSTS: list[tuple[str, int]] = [
("stun.l.google.com", 19302),
("stun1.l.google.com", 19302),
("stun.cloudflare.com", 3478),
("stun.stunprotocol.org", 3478),
]
# Firefox user.js prefs that block WebRTC
_REQUIRED_WEBRTC_PREFS: dict[str, str] = {
"media.peerconnection.enabled": "false",
"media.peerconnection.ice.default_address_only": "true",
"media.peerconnection.ice.no_host": "true",
}
# Chrome / Edge HKLM policy keys
_CHROMIUM_POLICY_PATHS = (
r"SOFTWARE\Policies\Google\Chrome",
r"SOFTWARE\Policies\Microsoft\Edge",
r"SOFTWARE\Policies\Chromium",
)
_WEBRTC_POLICY_VALUE = "DefaultWebRtcIpHandlingPolicy"
_WEBRTC_BLOCK_VALUE = 2 # "default_public_and_private_interfaces"
@dataclass
class WebRtcCheckResult:
"""Aggregated WebRTC leak surface scan."""
# Policy / config checks
chrome_edge_policy_set: bool = False
chrome_edge_policy_detail: str = ""
firefox_prefs_ok: bool = False
firefox_prefs_detail: str = ""
# Network reachability
stun_reachable: bool = False
stun_detail: str = ""
# Convenience flags
any_leak_risk: bool = True
issues: list[str] = field(default_factory=list)
checks: list[tuple[str, bool, str]] = field(default_factory=list) # (name, ok, detail)
def add(self, name: str, ok: bool, detail: str) -> None:
self.checks.append((name, ok, detail))
if not ok:
self.issues.append(f"{name}: {detail}")
# ─────────────────────────────────────────────────────────────────────────────
# 1. Chrome / Edge Group Policy
# ─────────────────────────────────────────────────────────────────────────────
def check_chromium_webrtc_policy() -> tuple[bool, str]:
"""Return (policy_set, detail_string)."""
found: list[str] = []
missing: list[str] = []
for path in _CHROMIUM_POLICY_PATHS:
try:
with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, path, 0, winreg.KEY_QUERY_VALUE) as k:
try:
val = int(winreg.QueryValueEx(k, _WEBRTC_POLICY_VALUE)[0])
name = path.split("\\")[-1]
if val == _WEBRTC_BLOCK_VALUE:
found.append(name)
else:
missing.append(f"{name}={val} (need {_WEBRTC_BLOCK_VALUE})")
except OSError:
missing.append(path.split("\\")[-1] + " key missing")
except OSError:
continue # key not present at all — browser not installed or not policy-managed
if found:
return True, "Policy set: " + ", ".join(found)
if missing:
return False, "Not set: " + "; ".join(missing)
return False, "No Chrome/Edge/Chromium policy keys found (browsers may not be installed)"
# ─────────────────────────────────────────────────────────────────────────────
# 2. Firefox managed profile prefs
# ─────────────────────────────────────────────────────────────────────────────
def check_firefox_webrtc_prefs(profile_dir: Path) -> tuple[bool, str]:
"""Verify our written user.js contains the required WebRTC-disabling prefs."""
user_js = profile_dir / "user.js"
if not user_js.is_file():
return False, f"user.js not found at {profile_dir} — profile not yet initialized"
try:
content = user_js.read_text(encoding="utf-8", errors="replace")
except Exception as e:
return False, f"Cannot read user.js: {e}"
missing: list[str] = []
for pref, expected_val in _REQUIRED_WEBRTC_PREFS.items():
# Look for: user_pref("pref.name", value);
needle = f'"{pref}", {expected_val}'
if needle not in content:
missing.append(pref)
if missing:
return False, "Missing in user.js: " + ", ".join(missing)
return True, "All WebRTC prefs present in managed profile"
# ─────────────────────────────────────────────────────────────────────────────
# 3. STUN server UDP reachability
# ─────────────────────────────────────────────────────────────────────────────
def _probe_stun_udp(host: str, port: int, timeout: float = 2.5) -> bool:
"""
Send a minimal STUN Binding Request over UDP and wait for a response.
Returns True if we get *any* response (STUN or ICMP unreachable).
"""
# Minimal STUN Binding Request (RFC 5389)
msg = (
b"\x00\x01" # Message Type: Binding Request
b"\x00\x00" # Message Length: 0
b"\x21\x12\xa4\x42" # Magic Cookie
+ b"\x00" * 12 # Transaction ID (12 bytes)
)
try:
ip = socket.gethostbyname(host)
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(timeout)
s.sendto(msg, (ip, port))
data, _ = s.recvfrom(512)
s.close()
return len(data) > 0
except socket.timeout:
return False
except Exception:
return False
def check_stun_reachability(timeout: float = 3.0) -> tuple[bool, str]:
"""
Returns (reachable, detail).
reachable = True means STUN servers ARE reachable → WebRTC COULD leak.
"""
reachable: list[str] = []
for host, port in _STUN_HOSTS[:3]:
try:
if _probe_stun_udp(host, port, timeout):
reachable.append(f"{host}:{port}")
except Exception:
continue
if reachable:
return True, "STUN reachable: " + ", ".join(reachable) + " — WebRTC UDP is NOT blocked"
return False, "STUN servers unreachable — WebRTC UDP appears blocked (good)"
# ─────────────────────────────────────────────────────────────────────────────
# Composite scan
# ─────────────────────────────────────────────────────────────────────────────
def run_webrtc_check(profile_dir: Path | None = None) -> WebRtcCheckResult:
"""
Run all WebRTC leak surface checks.
profile_dir: path to the managed Firefox profile (to verify user.js).
"""
res = WebRtcCheckResult()
# 1. Chrome/Edge policy
policy_ok, policy_detail = check_chromium_webrtc_policy()
res.chrome_edge_policy_set = policy_ok
res.chrome_edge_policy_detail = policy_detail
res.add("Chrome/Edge WebRTC policy (HKLM)", policy_ok, policy_detail)
# 2. Firefox profile prefs
if profile_dir is not None:
fx_ok, fx_detail = check_firefox_webrtc_prefs(profile_dir)
res.firefox_prefs_ok = fx_ok
res.firefox_prefs_detail = fx_detail
res.add("Firefox managed profile WebRTC prefs", fx_ok, fx_detail)
else:
res.add("Firefox managed profile WebRTC prefs", False,
"Profile not specified — not checked")
# 3. STUN UDP reachability
stun_reachable, stun_detail = check_stun_reachability()
res.stun_reachable = stun_reachable
res.stun_detail = stun_detail
# STUN reachable = risk; NOT reachable = good (firewall is blocking it)
res.add("STUN UDP reachability", not stun_reachable, stun_detail)
# Overall verdict
res.any_leak_risk = bool(res.issues)
return res

BIN
proxy_chain_manager/world_map.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 151 KiB