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

Add ban tester, signup prep, exit IP intel, and sticky-exit hold.

Browse Source 
New GUI tabs probe popular sites for exit-IP bans and prep signup autofill through the chain; Live tab shows geo/ASN/datacenter flags, and sticky-exit keeps the same egress IP during signup flows.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Indiana Holmes
2026-05-19 15:37:28 -07:00
parent 8258ca0997
commit 26863fc2a5
8 changed files with 1587 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

709
proxy_chain_manager/app.py
View File

@@ -1,10 +1,11 @@
""" """
Proxy God — GUI Proxy God — GUI
Tabs: Live | Chain Builder | Browser | Privacy | Settings Tabs: Live | Chain Builder | Browser | Ban Tester | Signup Prep | Privacy | Settings
""" """
from __future__ import annotations from __future__ import annotations
import logging import logging
import ipaddress
import queue import queue
import sys import sys
import threading import threading
@@ -13,6 +14,7 @@ from dataclasses import asdict
from datetime import datetime from datetime import datetime
from typing import Any, Literal from typing import Any, Literal
from tkinter import filedialog from tkinter import filedialog
from urllib.parse import urlparse
import customtkinter as ctk import customtkinter as ctk
@@ -36,6 +38,24 @@ from .firewall import (
request_admin_relaunch, request_admin_relaunch,
) )
from .artifact_wipe import wipe_artifacts from .artifact_wipe import wipe_artifacts
from .ban_tester import BanTestResult, run_ban_tests, test_single_site
from .exit_intel import ExitIntel, fetch_exit_intel
from .signup_prep import (
SIGNUP_PRESET_KEYS,
SIGNUP_PRESETS,
AccountRecord,
SignupDraft,
append_account,
generate_password,
load_accounts,
load_draft,
random_first_name,
random_full_name,
random_last_name,
random_username,
resolve_signup_url,
save_draft,
)
from .browser_identity import ( from .browser_identity import (
COOKIE_LABELS, COOKIE_LABELS,
COOKIE_MODES, COOKIE_MODES,
@@ -175,6 +195,7 @@ def main() -> None:
browser = BrowserSession() browser = BrowserSession()
browser_should_run = [False] browser_should_run = [False]
browser_last_launch_ts = [0.0] browser_last_launch_ts = [0.0]
service_running = [False]
# ── tray ───────────────────────────────────────────────────────────────── # ── tray ─────────────────────────────────────────────────────────────────
tray = TrayIcon( tray = TrayIcon(
@@ -200,6 +221,17 @@ def main() -> None:
u = u.replace("http://", "").replace("socks5://", "s5://").replace("socks4://", "s4://").replace("https://", "") u = u.replace("http://", "").replace("socks5://", "s5://").replace("socks4://", "s4://").replace("https://", "")
return u[:28] + "…" if len(u) > 30 else u return u[:28] + "…" if len(u) > 30 else u
def _extract_ip_host(proxy_url: str) -> str | None:
p = urlparse(normalize_proxy_url(proxy_url))
host = (p.hostname or "").strip()
if not host:
return None
try:
ipaddress.ip_address(host)
return host
except ValueError:
return None
# ───────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────────────────────────────────
# HELPERS # HELPERS
# ───────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────────────────────────────────
@@ -422,9 +454,11 @@ def main() -> None:
tab_live = tabs.add(" Live ") tab_live = tabs.add(" Live ")
tab_chain = tabs.add(" Chain Builder ") tab_chain = tabs.add(" Chain Builder ")
tab_browser = tabs.add(" Browser ") tab_browser = tabs.add(" Browser ")
tab_ban = tabs.add(" Ban Tester ")
tab_signup = tabs.add(" Signup Prep ")
tab_privacy = tabs.add(" Privacy ") tab_privacy = tabs.add(" Privacy ")
tab_settings = tabs.add(" Settings ") tab_settings = tabs.add(" Settings ")
for tab in (tab_live, tab_chain, tab_browser, tab_privacy, tab_settings): for tab in (tab_live, tab_chain, tab_browser, tab_ban, tab_signup, tab_privacy, tab_settings):
tab.configure(fg_color=BG) tab.configure(fg_color=BG)
# ─── TAB: LIVE ─────────────────────────────────────────────────────────── # ─── TAB: LIVE ───────────────────────────────────────────────────────────
@@ -450,6 +484,99 @@ def main() -> None:
_stat_card(stats_row, "Valid pool size", v_pool) _stat_card(stats_row, "Valid pool size", v_pool)
_stat_card(stats_row, "Rotations this session", v_rotation) _stat_card(stats_row, "Rotations this session", v_rotation)
# ── Exit IP intel card (geo / ASN / datacenter) ──────────────────────────
intel_card = ctk.CTkFrame(tab_live, fg_color=CARD, corner_radius=8)
intel_card.pack(fill="x", pady=(0, 6))
intel_head = ctk.CTkFrame(intel_card, fg_color="transparent")
intel_head.pack(fill="x", padx=10, pady=(8, 2))
ctk.CTkLabel(
intel_head, text="Exit IP intel", font=(FONT, 11, "bold"), text_color=CYAN,
).pack(side="left")
v_intel_loc = ctk.StringVar(value="Location: —")
v_intel_asn = ctk.StringVar(value="ASN / ISP: —")
v_intel_tags = ctk.StringVar(value="Flags: —")
intel_body = ctk.CTkFrame(intel_card, fg_color="transparent")
intel_body.pack(fill="x", padx=10, pady=(0, 8))
ctk.CTkLabel(intel_body, textvariable=v_intel_loc, font=(FONT, 11), text_color=TEXT,
anchor="w").pack(anchor="w")
ctk.CTkLabel(intel_body, textvariable=v_intel_asn, font=(FONT, 11), text_color=TEXT2,
anchor="w").pack(anchor="w")
intel_tags_lbl = ctk.CTkLabel(intel_body, textvariable=v_intel_tags, font=(FONT, 11, "bold"),
text_color=GREEN, anchor="w")
intel_tags_lbl.pack(anchor="w")
intel_running = [False]
last_intel: dict[str, ExitIntel | None] = {"intel": None}
def _apply_intel(intel: ExitIntel | None) -> None:
last_intel["intel"] = intel
if intel is None or not intel.ok:
v_intel_loc.set("Location: —")
v_intel_asn.set("ASN / ISP: —")
v_intel_tags.set("Flags: " + (intel.detail if intel and intel.detail else "—"))
intel_tags_lbl.configure(text_color=TEXT2)
return
loc_parts = [p for p in (intel.city, intel.region, intel.country) if p]
v_intel_loc.set("Location: " + (", ".join(loc_parts) if loc_parts else "—"))
asn_str = f"AS{intel.asn} " if intel.asn else ""
org_str = intel.org or intel.isp or "—"
v_intel_asn.set(f"ASN / ISP: {asn_str}{org_str}")
tags: list[str] = []
color = GREEN
if intel.is_datacenter:
tags.append("DATACENTER")
color = ORANGE
if intel.is_proxy_flagged:
tags.append("PROXY-FLAGGED")
color = RED
if intel.is_mobile:
tags.append("MOBILE")
if not tags:
tags.append("residential-like")
color = GREEN
v_intel_tags.set("Flags: " + " · ".join(tags))
intel_tags_lbl.configure(text_color=color)
def _refresh_exit_intel(silent: bool = False) -> None:
if intel_running[0]:
return
if not service_running[0] or not svc.current_chain:
if not silent:
_log("Exit intel: chain not running.")
_apply_intel(None)
return
intel_running[0] = True
v_intel_tags.set("Flags: looking up…")
intel_tags_lbl.configure(text_color=YELLOW)
proxy_url = f"http://{svc.settings.listen_addr()}"
def work() -> None:
intel = fetch_exit_intel(
proxy_url,
timeout_seconds=min(15.0, svc.settings.validation_timeout_seconds + 3.0),
)
def done() -> None:
_apply_intel(intel)
intel_running[0] = False
if not silent:
if intel.ok:
_log(f"Exit intel: {intel.summary()} (via {intel.source})")
else:
_log(f"Exit intel: lookup failed — {intel.detail}")
root.after(0, done)
threading.Thread(target=work, daemon=True).start()
intel_btn_row = ctk.CTkFrame(intel_card, fg_color="transparent")
intel_btn_row.pack(fill="x", padx=10, pady=(0, 8))
_btn(intel_btn_row, "Refresh intel", lambda: _refresh_exit_intel(False), w=110, h=24,
fg_color=ACCENT2, hover_color=ACCENT).pack(side="left")
# Log box # Log box
log_frame = ctk.CTkFrame(tab_live, fg_color=CARD, corner_radius=8) log_frame = ctk.CTkFrame(tab_live, fg_color=CARD, corner_radius=8)
log_frame.pack(fill="both", expand=True) log_frame.pack(fill="both", expand=True)
@@ -1178,6 +1305,9 @@ def main() -> None:
ok, msg = browser.stop(dispose=cfg.disposable_profile) ok, msg = browser.stop(dispose=cfg.disposable_profile)
_log(msg) _log(msg)
_refresh_browser_status() _refresh_browser_status()
if svc.sticky_remaining() > 0:
svc.set_sticky(0)
_log("Sticky exit released (browser stopped).")
launch_browser_btn = _btn( launch_browser_btn = _btn(
browser_action_row, "Launch Hardened Firefox", _launch_browser, w=190, h=34, font=(FONT, 12) browser_action_row, "Launch Hardened Firefox", _launch_browser, w=190, h=34, font=(FONT, 12)
@@ -1189,6 +1319,568 @@ def main() -> None:
) )
stop_browser_btn.pack(side="left") stop_browser_btn.pack(side="left")
# ─── TAB: BAN TESTER ───────────────────────────────────────────────────────
ban_scroll = ctk.CTkScrollableFrame(tab_ban, fg_color=BG, scrollbar_button_color=ACCENT)
ban_scroll.pack(fill="both", expand=True, padx=4, pady=4)
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,
).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))
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)
)
ban_results_frame = ctk.CTkFrame(
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:
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)
)
def _render_ban_results(results: list[BanTestResult]) -> None:
for w in list(ban_rows_wrap.winfo_children()):
w.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)
return
for r in results:
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)
_render_ban_results([])
ban_action = ctk.CTkFrame(ban_head, fg_color="transparent")
ban_action.pack(fill="x", padx=12, pady=(0, 10))
def _start_ban_test() -> None:
if not service_running[0] or not svc.current_chain:
ban_status_var.set("Status: chain not running")
_log("Ban tester: start the chain first, then run tests.")
return
proxy_url = f"http://{svc.settings.listen_addr()}"
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")
banned_n = sum(1 for r in results if r.status == "banned")
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_start_btn.configure(state="normal")
root.after(0, done)
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_start_btn.pack(side="left")
# ─── TAB: SIGNUP PREP ──────────────────────────────────────────────────────
signup_scroll = ctk.CTkScrollableFrame(tab_signup, fg_color=BG, scrollbar_button_color=ACCENT)
signup_scroll.pack(fill="both", expand=True, padx=4, pady=4)
signup_draft = load_draft()
def _signup_card(title: str, subtitle: str = "") -> ctk.CTkFrame:
ctk.CTkLabel(signup_scroll, text=title, font=(FONT, 13, "bold"), text_color=CYAN).pack(
anchor="w", padx=8, pady=(12, 0)
)
if subtitle:
ctk.CTkLabel(
signup_scroll, text=subtitle, font=(FONT, 10), text_color=TEXT2,
wraplength=900, justify="left",
).pack(anchor="w", padx=8, pady=(0, 4))
f = ctk.CTkFrame(signup_scroll, fg_color=CARD, corner_radius=10, border_width=1, border_color=GLOW)
f.pack(fill="x", padx=8, pady=(4, 8))
return f
signup_head = _signup_card(
"Signup autofill (manual submit)",
"Opens signup pages through your chain and autofills your fields. You solve captchas and click submit.",
)
site_row = ctk.CTkFrame(signup_head, fg_color="transparent")
site_row.pack(fill="x", padx=12, pady=(10, 4))
ctk.CTkLabel(site_row, text="Site", width=70, anchor="w", font=(FONT, 10), text_color=TEXT2).pack(side="left")
signup_site_keys = list(SIGNUP_PRESET_KEYS) + ["custom"]
signup_site_labels = [
(SIGNUP_PRESETS[k].label if k in SIGNUP_PRESETS else "Custom URL")
for k in signup_site_keys
]
signup_site_var = ctk.StringVar(
value=next(
(signup_site_labels[i] for i, k in enumerate(signup_site_keys) if k == signup_draft.site_key),
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))
def _signup_site_key() -> str:
label = signup_site_var.get()
for i, lbl in enumerate(signup_site_labels):
if lbl == label:
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 _signup_field(parent: Any, label: str, show: str = "") -> ctk.CTkEntry:
row = ctk.CTkFrame(parent, fg_color="transparent")
row.pack(fill="x", padx=12, pady=3)
ctk.CTkLabel(row, text=label, width=70, anchor="w", font=(FONT, 10), text_color=TEXT2).pack(side="left")
e = ctk.CTkEntry(row, height=28, fg_color=BG, border_color=ACCENT, show=show)
e.pack(side="left", fill="x", expand=True)
return e
signup_email = _signup_field(signup_head, "Email")
signup_email.insert(0, signup_draft.email or "")
signup_pass = _signup_field(signup_head, "Password", show="*")
signup_pass.insert(0, signup_draft.password or "")
signup_user = _signup_field(signup_head, "Username")
signup_user.insert(0, signup_draft.username or "")
signup_fname = _signup_field(signup_head, "First name")
signup_fname.insert(0, signup_draft.first_name or "")
signup_lname = _signup_field(signup_head, "Last name")
signup_lname.insert(0, signup_draft.last_name or "")
signup_notes = _signup_field(signup_head, "Notes")
signup_notes.insert(0, signup_draft.notes or "")
def _clipboard_copy(text: str, label: str) -> None:
t = (text or "").strip()
if not t:
_log(f"Signup prep: nothing to copy for {label}.")
return
root.clipboard_clear()
root.clipboard_append(t)
root.update()
_log(f"Copied {label} to clipboard.")
def _collect_signup_draft() -> SignupDraft:
return SignupDraft(
site_key=_signup_site_key(),
custom_url=signup_custom_url.get().strip(),
email=signup_email.get().strip(),
password=signup_pass.get().strip(),
username=signup_user.get().strip(),
first_name=signup_fname.get().strip(),
last_name=signup_lname.get().strip(),
notes=signup_notes.get().strip(),
)
def _persist_signup_draft() -> SignupDraft:
d = _collect_signup_draft()
save_draft(d)
return d
signup_btn_row = ctk.CTkFrame(signup_head, fg_color="transparent")
signup_btn_row.pack(fill="x", padx=12, pady=(8, 6))
def _gen_signup_password() -> None:
pw = generate_password()
signup_pass.delete(0, "end")
signup_pass.insert(0, pw)
_persist_signup_draft()
_log("Generated signup password.")
def _gen_signup_names() -> None:
f, l = random_full_name()
signup_fname.delete(0, "end"); signup_fname.insert(0, f)
signup_lname.delete(0, "end"); signup_lname.insert(0, l)
_persist_signup_draft()
_log(f"Generated name: {f} {l}")
def _gen_signup_username() -> None:
f = signup_fname.get().strip() or random_first_name()
l = signup_lname.get().strip() or random_last_name()
u = random_username(f, l)
signup_user.delete(0, "end"); signup_user.insert(0, u)
_persist_signup_draft()
_log(f"Generated username: {u}")
def _gen_signup_everything() -> None:
f, l = random_full_name()
signup_fname.delete(0, "end"); signup_fname.insert(0, f)
signup_lname.delete(0, "end"); signup_lname.insert(0, l)
signup_user.delete(0, "end"); signup_user.insert(0, random_username(f, l))
signup_pass.delete(0, "end"); signup_pass.insert(0, generate_password())
_persist_signup_draft()
_log(f"Generated identity: {f} {l}")
_btn(signup_btn_row, "Gen password", _gen_signup_password, w=100, h=28,
fg_color=DIM, hover_color=ACCENT).pack(side="left", padx=(0, 6))
_btn(signup_btn_row, "Gen name", _gen_signup_names, w=80, h=28,
fg_color=DIM, hover_color=ACCENT).pack(side="left", padx=(0, 6))
_btn(signup_btn_row, "Gen username", _gen_signup_username, w=100, h=28,
fg_color=DIM, hover_color=ACCENT).pack(side="left", padx=(0, 6))
_btn(signup_btn_row, "Gen ALL", _gen_signup_everything, w=80, h=28,
fg_color=ACCENT2, hover_color=ACCENT).pack(side="left", padx=(0, 12))
_btn(signup_btn_row, "Copy email", lambda: _clipboard_copy(signup_email.get(), "email"),
w=88, h=28, fg_color=DIM, hover_color=ACCENT).pack(side="left", padx=(0, 6))
_btn(signup_btn_row, "Copy pass", lambda: _clipboard_copy(signup_pass.get(), "password"),
w=88, h=28, fg_color=DIM, hover_color=ACCENT).pack(side="left", padx=(0, 6))
# Sticky exit + paranoid mode toggles
signup_opts_row = ctk.CTkFrame(signup_head, fg_color="transparent")
signup_opts_row.pack(fill="x", padx=12, pady=(0, 6))
sticky_exit_var = ctk.BooleanVar(value=True)
paranoid_signup_var = ctk.BooleanVar(value=False)
ctk.CTkCheckBox(
signup_opts_row, text="Sticky exit (hold final hop while signup is open)",
variable=sticky_exit_var, font=(FONT, 10),
fg_color=ACCENT2, hover_color=ACCENT, text_color=TEXT,
).pack(side="left", padx=(0, 12))
ctk.CTkCheckBox(
signup_opts_row, text="Paranoid mode (disposable profile + DNS flush)",
variable=paranoid_signup_var, font=(FONT, 10),
fg_color=ACCENT2, hover_color=ACCENT, text_color=TEXT,
).pack(side="left")
# ── Pre-flight checklist ─────────────────────────────────────────────────
preflight_card = _signup_card(
"Pre-flight check",
"Verifies chain, target site reachability, and exit-IP intel before you sign up.",
)
pf_rows: dict[str, dict[str, ctk.StringVar | ctk.CTkLabel]] = {}
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")
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")
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)
pf_summary_var = ctk.StringVar(value="Not run.")
ctk.CTkLabel(preflight_card, textvariable=pf_summary_var, font=(FONT, 10, "bold"),
text_color=TEXT).pack(anchor="w", padx=12, pady=(6, 8))
pf_running = [False]
def _run_preflight() -> None:
if pf_running[0]:
return
draft = _persist_signup_draft()
url = resolve_signup_url(draft)
if not url:
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"):
_pf_set(k, "running", "checking…")
pf_summary_var.set("Running pre-flight…")
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")
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)
def done() -> None:
ip = (v_exit_ip.get() or "—").strip()
if ip and ip != "—":
_pf_set("exit_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}")
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)
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))
pf_running[0] = False
root.after(0, done)
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,
fg_color=ACCENT2, hover_color=ACCENT).pack(side="left")
signup_status_var = ctk.StringVar(value="Ready — start chain, then Open & Autofill.")
ctk.CTkLabel(signup_head, textvariable=signup_status_var, font=(FONT, 10), text_color=TEXT2).pack(
anchor="w", padx=12, pady=(0, 8)
)
signup_action = ctk.CTkFrame(signup_scroll, fg_color="transparent")
signup_action.pack(fill="x", padx=8, pady=(0, 6))
signup_launch_retries = [0]
accounts_card = _signup_card("Saved accounts", "Logged locally with exit IP + hop used at save time.")
accounts_list = ctk.CTkScrollableFrame(
accounts_card, fg_color=BG, height=180, corner_radius=6, scrollbar_button_color=ACCENT,
)
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()
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
)
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)
def _current_exit_meta() -> tuple[str, str]:
ip = (v_exit_ip.get() or "—").strip()
if ip == "—":
ip = ""
hop = ""
chain = svc.current_chain
if chain:
hop = redact_proxy_url(chain[-1])
return ip, hop
def _launch_signup_autofill() -> None:
draft = _persist_signup_draft()
url = resolve_signup_url(draft)
if not url:
signup_status_var.set("Missing signup URL.")
_log("Signup prep: pick a site or enter a custom URL.")
return
if not draft.email and not draft.username:
signup_status_var.set("Enter email or username.")
_log("Signup prep: email or username required for autofill.")
return
if not draft.password:
signup_status_var.set("Enter or generate a password.")
_log("Signup prep: password required for autofill.")
return
cfg = _browser_cfg()
cfg.lock_managed_profile = True
paranoid = bool(paranoid_signup_var.get())
if paranoid:
cfg.disposable_profile = True
cfg.clear_on_close = True
cfg.kill_on_chain_drop = True
ok_dns, msg_dns = flush_dns_cache()
_log(f"Paranoid mode: DNS flush — {msg_dns}" if ok_dns else f"Paranoid mode: DNS flush failed — {msg_dns}")
if not service_running[0] or not svc.current_chain:
if signup_launch_retries[0] == 0:
_log("Signup prep: starting chain first…")
_start()
signup_launch_retries[0] += 1
if signup_launch_retries[0] > 12:
signup_status_var.set("Chain did not start in time.")
signup_launch_retries[0] = 0
_log("Signup prep: chain not ready — start chain manually.")
return
signup_status_var.set("Waiting for chain…")
root.after(2500, _launch_signup_autofill)
return
signup_launch_retries[0] = 0
if bool(sticky_exit_var.get()):
svc.set_sticky(1800)
_log("Sticky exit engaged — auto-rotation suspended for 30 min.")
ok, msg = browser.launch(
cfg, LISTEN_HOST, svc.settings.local_port,
start_url=url, signup_draft=draft,
)
_log(msg)
_refresh_browser_status()
if ok:
preset = SIGNUP_PRESETS.get(draft.site_key)
label = preset.label if preset else "Custom"
extras = []
if bool(sticky_exit_var.get()):
extras.append("sticky")
if paranoid:
extras.append("paranoid")
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}.")
else:
signup_status_var.set("Browser launch failed.")
svc.set_sticky(0)
def _save_signup_account() -> None:
draft = _persist_signup_draft()
url = resolve_signup_url(draft)
exit_ip, exit_hop = _current_exit_meta()
preset = SIGNUP_PRESETS.get(draft.site_key)
site_label = preset.label if preset else "Custom"
rec = AccountRecord(
id=datetime.now().strftime("%Y%m%d%H%M%S%f"),
site=site_label,
url=url,
email=draft.email,
password=draft.password,
username=draft.username,
first_name=draft.first_name,
last_name=draft.last_name,
exit_ip=exit_ip,
exit_hop=exit_hop,
notes=draft.notes,
status="created",
created_at=datetime.now().isoformat(timespec="seconds"),
)
append_account(rec)
_refresh_accounts_ui()
signup_status_var.set(f"Saved account for {site_label}.")
_log(f"Signup prep: saved account {draft.email or draft.username} ({site_label}).")
def _preflight_then_open() -> None:
_run_preflight()
def follow_up(attempts: int = 0) -> None:
if pf_running[0] and attempts < 30:
root.after(500, lambda: follow_up(attempts + 1))
return
_launch_signup_autofill()
root.after(500, follow_up)
signup_open_btn = _btn(
signup_action, "Open & Autofill", _launch_signup_autofill,
w=140, h=34, fg_color=ACCENT2, hover_color=ACCENT, font=(FONT, 12),
)
signup_open_btn.pack(side="left", padx=(0, 8))
_btn(
signup_action, "Pre-flight + Open", _preflight_then_open,
w=150, h=34, fg_color=ACCENT, hover_color=ACCENT2,
).pack(side="left", padx=(0, 8))
_btn(
signup_action, "Save account", _save_signup_account,
w=110, h=34, fg_color=GREEN, hover_color="#00b377",
).pack(side="left", padx=(0, 8))
_btn(
signup_action, "Refresh list", _refresh_accounts_ui,
w=100, h=34, fg_color=DIM, hover_color=ACCENT,
).pack(side="left")
_refresh_accounts_ui()
# ─── TAB: PRIVACY ────────────────────────────────────────────────────────── # ─── TAB: PRIVACY ──────────────────────────────────────────────────────────
priv_scroll = ctk.CTkScrollableFrame(tab_privacy, fg_color=BG, priv_scroll = ctk.CTkScrollableFrame(tab_privacy, fg_color=BG,
scrollbar_button_color=ACCENT) scrollbar_button_color=ACCENT)
@@ -1635,6 +2327,7 @@ def main() -> None:
elif t == "state": elif t == "state":
running = bool(m.get("running")) running = bool(m.get("running"))
service_running[0] = running
status_dot.configure(text_color=GREEN if running else RED) status_dot.configure(text_color=GREEN if running else RED)
status_txt.configure(text="RUNNING" if running else "STOPPED") status_txt.configure(text="RUNNING" if running else "STOPPED")
_refresh_sysproxy() _refresh_sysproxy()
@@ -1740,15 +2433,24 @@ def main() -> None:
status = str(m.get("status", "connecting")) status = str(m.get("status", "connecting"))
exit_ip = m.get("exit_ip") exit_ip = m.get("exit_ip")
exit_s = str(exit_ip) if exit_ip else None exit_s = str(exit_ip) if exit_ip else None
# If the live check hasn't resolved yet, show the explicit final-hop IP
# (when available) so tunnel/fixed-exit setups still expose the egress node.
if not exit_s and hops:
exit_s = _extract_ip_host(hops[-1])
fix = normalize_proxy_url(svc.settings.manual_exit_proxy) fix = normalize_proxy_url(svc.settings.manual_exit_proxy)
fixed_last = bool(fix) and not svc.settings.use_pinned_chain fixed_last = bool(fix) and not svc.settings.use_pinned_chain
_render_chain(hops, status, exit_ip, fixed_last=fixed_last) _render_chain(hops, status, exit_s, fixed_last=fixed_last)
_refresh_sysproxy() _refresh_sysproxy()
tray_color = {"healthy": "green", "dead": "red", "connecting": "yellow"}.get( tray_color = {"healthy": "green", "dead": "red", "connecting": "yellow"}.get(
status, "yellow" status, "yellow"
) )
tray.set_state(tray_color, exit_ip=exit_s) tray.set_state(tray_color, exit_ip=exit_s)
prev_exit = v_exit_ip.get()
v_exit_ip.set(exit_s if exit_s else "—") v_exit_ip.set(exit_s if exit_s else "—")
if status == "healthy" and exit_s and exit_s != prev_exit:
root.after(400, lambda: _refresh_exit_intel(silent=True))
elif status != "healthy":
_apply_intel(None)
if status == "dead" and browser.is_running() and bool(br_kill_drop_var.get()): if status == "dead" and browser.is_running() and bool(br_kill_drop_var.get()):
browser.stop(dispose=bool(br_disposable_var.get())) browser.stop(dispose=bool(br_disposable_var.get()))
_refresh_browser_status() _refresh_browser_status()
@@ -1800,6 +2502,7 @@ def main() -> None:
_log("Press START to fetch, validate and chain proxies.") _log("Press START to fetch, validate and chain proxies.")
_log("Chain Builder → build your chain, Test entire chain, then Start.") _log("Chain Builder → build your chain, Test entire chain, then Start.")
_log("Browser tab → launch hardened Firefox profile that follows your chain.") _log("Browser tab → launch hardened Firefox profile that follows your chain.")
_log("Signup Prep → open signup pages with autofill (Google / Proton / HydraProxy / custom).")
_log("Privacy tab → MAC, hostname, IPv6, WebRTC, fingerprint audit, DNS checks.") _log("Privacy tab → MAC, hostname, IPv6, WebRTC, fingerprint audit, DNS checks.")
_log("Works with or without VPN — leak detection adapts automatically.") _log("Works with or without VPN — leak detection adapts automatically.")
_log("─" * 60) _log("─" * 60)

92
proxy_chain_manager/ban_tester.py Normal file
View File

@@ -0,0 +1,92 @@
from __future__ import annotations
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/"),
)
_BANNED_STATUS_CODES = {401, 403, 407, 418, 429, 451}
_BANNED_TEXT_HINTS = (
"access denied",
"forbidden",
"temporarily blocked",
"request blocked",
"unusual traffic",
"captcha",
"challenge required",
)
@dataclass(frozen=True)
class BanTestResult:
site: str
url: str
status: str # ok | banned | error
code: int | None
detail: str
def _looks_banned_body(body: str) -> bool:
t = (body or "").lower()
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:
timeout = httpx.Timeout(timeout_seconds, connect=min(8.0, timeout_seconds))
try:
with httpx.Client(
proxy=proxy_url,
timeout=timeout,
verify=False,
follow_redirects=True,
headers={"User-Agent": "Mozilla/5.0"},
) as c:
r = c.get(url)
body = r.text[:1800] 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}")
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}")
except Exception as e:
return BanTestResult(site, url, "error", None, f"{type(e).__name__}: {e}")
def run_ban_tests(
proxy_url: str,
timeout_seconds: float = 12.0,
sites: Iterable[tuple[str, str]] = POPULAR_SITES,
) -> list[BanTestResult]:
out: list[BanTestResult] = []
for site, url in sites:
out.append(_test_one(proxy_url, site, url, timeout_seconds))
return out
def test_single_site(
proxy_url: str,
site: str,
url: str,
timeout_seconds: float = 10.0,
) -> BanTestResult:
"""Probe one URL through the proxy. Used by pre-flight checks."""
return _test_one(proxy_url, site, url, timeout_seconds)

8
proxy_chain_manager/browser_launcher.py
View File

@@ -10,6 +10,7 @@ from pathlib import Path
from .browser_profile import FirefoxHardening, ensure_firefox_profile, persona_env from .browser_profile import FirefoxHardening, ensure_firefox_profile, persona_env
from .paths import app_data_dir from .paths import app_data_dir
from .signup_prep import SignupDraft, install_signup_extension
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -67,6 +68,8 @@ class BrowserSession:
cfg: BrowserConfig, cfg: BrowserConfig,
proxy_host: str, proxy_host: str,
proxy_port: int, proxy_port: int,
start_url: str = "",
signup_draft: SignupDraft | None = None,
) -> tuple[bool, str]: ) -> tuple[bool, str]:
raw_exe = (cfg.firefox_path or default_firefox_path()).strip().strip('"').strip("'") raw_exe = (cfg.firefox_path or default_firefox_path()).strip().strip('"').strip("'")
exe = Path(raw_exe) exe = Path(raw_exe)
@@ -88,12 +91,17 @@ class BrowserSession:
) )
if cfg.lock_managed_profile: if cfg.lock_managed_profile:
ensure_firefox_profile(profile_dir, proxy_host, int(proxy_port), hard) ensure_firefox_profile(profile_dir, proxy_host, int(proxy_port), hard)
if signup_draft is not None:
install_signup_extension(profile_dir, signup_draft)
if self.is_running(): if self.is_running():
self.stop() self.stop()
if cfg.lock_managed_profile: if cfg.lock_managed_profile:
args = [str(exe), "-no-remote", "-profile", str(profile_dir)] args = [str(exe), "-no-remote", "-profile", str(profile_dir)]
else: else:
args = [str(exe)] args = [str(exe)]
url = (start_url or "").strip()
if url:
args.append(url)
env = os.environ.copy() env = os.environ.copy()
env.update(persona_env(hard)) env.update(persona_env(hard))
try: try:

164
proxy_chain_manager/exit_intel.py Normal file
View File

@@ -0,0 +1,164 @@
"""Exit IP intelligence: geo, ASN, datacenter heuristics.
Designed to be called through the local chain proxy so the lookup itself
flows over the active path (no out-of-band leak). Uses ip-api.com (free,
no key) with a fallback to ipwho.is. Both return geo + ASN.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Any
import httpx
log = logging.getLogger(__name__)
# Common datacenter / hosting ASN keywords. Used as a quick heuristic to
# warn the operator before signup (residential ASNs are far less likely to
# trip Google/Proton anti-fraud than DC ranges).
_DC_KEYWORDS = (
"amazon", "aws", "google", "microsoft", "azure", "digitalocean",
"linode", "ovh", "hetzner", "vultr", "choopa", "scaleway",
"leaseweb", "contabo", "hostinger", "godaddy", "namecheap",
"cloudflare", "fastly", "akamai", "datacamp", "m247",
"psychz", "quadranet", "colocrossing", "rackspace",
"alibaba", "tencent", "huawei", "online s.a.s",
"wholesale", "datacenter", "data center", "hosting",
"server", "cloud", "vps",
)
@dataclass
class ExitIntel:
ok: bool
ip: str = ""
country: str = ""
country_code: str = ""
city: str = ""
region: str = ""
timezone: str = ""
asn: str = ""
org: str = ""
isp: str = ""
is_datacenter: bool = False
is_mobile: bool = False
is_proxy_flagged: bool = False
source: str = ""
detail: str = ""
def summary(self) -> str:
if not self.ok:
return f"unknown — {self.detail}"
loc_bits = [b for b in (self.city, self.region, self.country) if b]
loc = ", ".join(loc_bits) if loc_bits else "—"
tags: list[str] = []
if self.is_datacenter:
tags.append("DATACENTER")
if self.is_mobile:
tags.append("MOBILE")
if self.is_proxy_flagged:
tags.append("PROXY-FLAGGED")
tag_str = f" [{' · '.join(tags)}]" if tags else ""
asn_str = f" AS{self.asn}" if self.asn else ""
return f"{self.ip} {loc}{asn_str}{tag_str}"
def _looks_dc(org: str, isp: str) -> bool:
haystack = f"{org} {isp}".lower()
return any(k in haystack for k in _DC_KEYWORDS)
def _parse_ipapi(payload: dict[str, Any]) -> ExitIntel:
if (payload.get("status") or "").lower() != "success":
return ExitIntel(ok=False, detail=str(payload.get("message") or "ip-api error"))
asn_raw = str(payload.get("as") or "")
asn = asn_raw.split()[0].lstrip("AS").strip() if asn_raw else ""
org = str(payload.get("org") or payload.get("isp") or "")
isp = str(payload.get("isp") or "")
return ExitIntel(
ok=True,
ip=str(payload.get("query") or ""),
country=str(payload.get("country") or ""),
country_code=str(payload.get("countryCode") or ""),
city=str(payload.get("city") or ""),
region=str(payload.get("regionName") or ""),
timezone=str(payload.get("timezone") or ""),
asn=asn,
org=org,
isp=isp,
is_datacenter=bool(payload.get("hosting")) or _looks_dc(org, isp),
is_mobile=bool(payload.get("mobile")),
is_proxy_flagged=bool(payload.get("proxy")),
source="ip-api.com",
detail="ok",
)
def _parse_ipwhois(payload: dict[str, Any]) -> ExitIntel:
if not payload.get("success", True):
return ExitIntel(ok=False, detail=str(payload.get("message") or "ipwho.is error"))
conn = payload.get("connection") or {}
asn_val = conn.get("asn")
asn = str(asn_val) if asn_val is not None else ""
org = str(conn.get("org") or "")
isp = str(conn.get("isp") or "")
return ExitIntel(
ok=True,
ip=str(payload.get("ip") or ""),
country=str(payload.get("country") or ""),
country_code=str(payload.get("country_code") or ""),
city=str(payload.get("city") or ""),
region=str(payload.get("region") or ""),
timezone=str((payload.get("timezone") or {}).get("id") or ""),
asn=asn,
org=org,
isp=isp,
is_datacenter=_looks_dc(org, isp),
is_mobile=False,
is_proxy_flagged=False,
source="ipwho.is",
detail="ok",
)
def fetch_exit_intel(proxy_url: str, timeout_seconds: float = 12.0) -> ExitIntel:
"""Look up exit IP geo/ASN/datacenter through the given proxy.
Returns an ExitIntel with ok=False and detail set on failure. Never raises.
"""
timeout = httpx.Timeout(timeout_seconds, connect=min(6.0, timeout_seconds))
sources = [
(
"http://ip-api.com/json/?fields=status,message,country,countryCode,"
"regionName,city,timezone,isp,org,as,mobile,proxy,hosting,query",
_parse_ipapi,
),
("https://ipwho.is/", _parse_ipwhois),
]
last_err = ""
try:
with httpx.Client(
proxy=proxy_url,
timeout=timeout,
verify=False,
follow_redirects=True,
headers={"User-Agent": "Mozilla/5.0"},
) as c:
for url, parser in sources:
try:
r = c.get(url)
if r.status_code != 200 or not r.content:
last_err = f"{url} -> HTTP {r.status_code}"
continue
data = r.json()
out = parser(data)
if out.ok:
return out
last_err = out.detail
except Exception as e:
last_err = f"{type(e).__name__}: {e}"
continue
except Exception as e:
return ExitIntel(ok=False, detail=f"{type(e).__name__}: {e}")
return ExitIntel(ok=False, detail=last_err or "no source responded")

29
proxy_chain_manager/service.py
View File

@@ -79,6 +79,9 @@ class ChainService:
self._proc = None self._proc = None
self._settings = load_settings() self._settings = load_settings()
self._current_chain: list[str] = [] self._current_chain: list[str] = []
# Sticky-exit: while time.monotonic() < self._sticky_until, leak-based
# auto-rotation is skipped so signup flows keep the same egress IP.
self._sticky_until: float = 0.0
# Shuffle-and-drain pool state — never repeat a proxy within a cycle # Shuffle-and-drain pool state — never repeat a proxy within a cycle
self._available: list[str] = [] # proxies not yet used this cycle self._available: list[str] = [] # proxies not yet used this cycle
@@ -142,6 +145,21 @@ class ChainService:
def rotate_now(self) -> None: def rotate_now(self) -> None:
self._force_rotate.set() self._force_rotate.set()
def set_sticky(self, seconds: float) -> None:
"""Hold current exit for ``seconds`` (suppresses leak-driven rotation).
Manual rotation still works. Pass 0 to clear sticky mode immediately.
"""
if seconds <= 0:
self._sticky_until = 0.0
return
self._sticky_until = time.monotonic() + float(seconds)
def sticky_remaining(self) -> float:
"""Seconds remaining on sticky-exit hold (0.0 if not sticky)."""
rem = self._sticky_until - time.monotonic()
return rem if rem > 0 else 0.0
# ───────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────────────────────────────────
def _teardown_network(self) -> None: def _teardown_network(self) -> None:
@@ -588,6 +606,17 @@ class ChainService:
if is_chain_leak(exit_ip, real_ip, self._vpn.active): if is_chain_leak(exit_ip, real_ip, self._vpn.active):
reason = leak_reason(exit_ip, real_ip, self._vpn.active) reason = leak_reason(exit_ip, real_ip, self._vpn.active)
rem = self.sticky_remaining()
if rem > 0:
self._notify({
"type": "log",
"text": (
f"Health check flagged ({reason}) — STICKY EXIT active "
f"({int(rem)}s left), skipping rotation."
),
})
self._notify({"type": "hops", "hops": chain, "status": "healthy", "exit_ip": exit_ip})
continue
self._notify({"type": "hops", "hops": chain, "status": "dead", "exit_ip": exit_ip}) self._notify({"type": "hops", "hops": chain, "status": "dead", "exit_ip": exit_ip})
self._notify({"type": "log", "text": f"Health check failed ({reason}) — rotating."}) self._notify({"type": "log", "text": f"Health check failed ({reason}) — rotating."})
break break

153
proxy_chain_manager/signup_extension/content.js Normal file
View File

@@ -0,0 +1,153 @@
(function () {
"use strict";
var cfg = null;
var filledKeys = {};
var startedAt = Date.now();
var MAX_MS = 120000;
function loadConfig(cb) {
try {
var rt = typeof browser !== "undefined" ? browser : chrome;
var url = rt.runtime.getURL("autofill_config.json");
fetch(url, { cache: "no-store" })
.then(function (r) { return r.json(); })
.then(function (j) { cb(j || null); })
.catch(function () { cb(null); });
} catch (e) {
cb(null);
}
}
function hostMatches(host, patterns) {
if (!patterns || !patterns.length) return true;
host = (host || "").toLowerCase();
for (var i = 0; i < patterns.length; i++) {
var p = String(patterns[i] || "").toLowerCase();
if (!p) continue;
if (p.indexOf("*") >= 0) {
var re = new RegExp("^" + p.replace(/\./g, "\\.").replace(/\*/g, ".*") + "$");
if (re.test(host)) return true;
} else if (host === p || host.endsWith("." + p)) {
return true;
}
}
return false;
}
function setValue(el, value) {
if (!el || value == null || value === "") return false;
try {
el.focus();
el.value = value;
el.dispatchEvent(new Event("input", { bubbles: true }));
el.dispatchEvent(new Event("change", { bubbles: true }));
return true;
} catch (e) {
return false;
}
}
function trySelectors(selectors, value) {
if (!selectors || !value) return false;
for (var i = 0; i < selectors.length; i++) {
var nodes = document.querySelectorAll(selectors[i]);
for (var j = 0; j < nodes.length; j++) {
var el = nodes[j];
if (!el || el.disabled || el.readOnly) continue;
if (el.type === "hidden") continue;
if (setValue(el, value)) return true;
}
}
return false;
}
function runFill() {
if (!cfg || !cfg.fields) return;
if (Date.now() - startedAt > MAX_MS) return;
var host = location.hostname || "";
if (cfg.hosts && cfg.hosts.length && !hostMatches(host, cfg.hosts)) return;
var fields = cfg.fields;
var keys = Object.keys(fields);
for (var i = 0; i < keys.length; i++) {
var key = keys[i];
var val = fields[key];
if (!val) continue;
if (filledKeys[key]) continue;
if (trySelectors(fields[key + "_selectors"] || defaultSelectors(key), val)) {
filledKeys[key] = true;
}
}
}
function defaultSelectors(key) {
var map = {
email: [
'input[type="email"]',
'input[name="email"]',
'input[name="Email"]',
'input[id="email"]',
'input[autocomplete="email"]',
'input[autocomplete="username"]',
'input[name="identifier"]'
],
password: [
'input[type="password"]:not([name*="Confirm"]):not([name*="confirm"]):not([name*="Again"]):not([name*="again"])',
'input[name="password"]',
'input[name="Passwd"]',
'input[id="password"]',
'input[autocomplete="new-password"]'
],
password_confirm: [
'input[name="PasswdAgain"]',
'input[name="password_confirm"]',
'input[name="confirmPassword"]',
'input[name="passwordConfirmation"]',
'input[autocomplete="new-password"]:nth-of-type(2)'
],
first_name: [
'input[name="firstName"]',
'input[name="firstname"]',
'input[name="first_name"]',
'input[autocomplete="given-name"]'
],
last_name: [
'input[name="lastName"]',
'input[name="lastname"]',
'input[name="last_name"]',
'input[autocomplete="family-name"]'
],
username: [
'input[name="username"]',
'input[name="Username"]',
'input[id="username"]',
'input[autocomplete="username"]'
]
};
return map[key] || [];
}
function boot() {
loadConfig(function (j) {
cfg = j;
if (!cfg) return;
runFill();
var obs = new MutationObserver(function () { runFill(); });
obs.observe(document.documentElement || document.body, {
childList: true,
subtree: true
});
var timer = setInterval(function () {
runFill();
if (Date.now() - startedAt > MAX_MS) clearInterval(timer);
}, 900);
});
}
if (document.readyState === "loading") {
document.addEventListener("DOMContentLoaded", boot);
} else {
boot();
}
})();

21
proxy_chain_manager/signup_extension/manifest.json Normal file
View File

@@ -0,0 +1,21 @@
{
"manifest_version": 2,
"name": "Proxy God Signup Autofill",
"version": "1.0.0",
"description": "Autofills signup forms from Proxy God (manual submit + captcha).",
"applications": {
"gecko": {
"id": "signup-autofill@proxygod"
}
},
"content_scripts": [
{
"matches": ["<all_urls>"],
"js": ["content.js"],
"run_at": "document_idle",
"all_frames": true
}
],
"permissions": ["<all_urls>"],
"web_accessible_resources": ["autofill_config.json"]
}

414
proxy_chain_manager/signup_prep.py Normal file
View File

@@ -0,0 +1,414 @@
from __future__ import annotations
import json
import secrets
import shutil
import string
import uuid
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from .paths import app_data_dir
_EXT_ID = "signup-autofill@proxygod"
_EXT_SRC = Path(__file__).resolve().parent / "signup_extension"
@dataclass(frozen=True)
class SignupSitePreset:
key: str
label: str
url: str
hosts: tuple[str, ...]
extra_selectors: dict[str, list[str]] = field(default_factory=dict)
SIGNUP_PRESETS: dict[str, SignupSitePreset] = {
"google": SignupSitePreset(
key="google",
label="Google",
url="https://accounts.google.com/signup/v2/createaccount?flowName=GlifWebSignIn&flowEntry=SignUp",
hosts=("accounts.google.com", "google.com"),
extra_selectors={
"first_name": ['input[name="firstName"]'],
"last_name": ['input[name="lastName"]'],
"email": ['input[name="Username"]', 'input[type="email"]'],
"password": ['input[name="Passwd"]'],
"password_confirm": ['input[name="PasswdAgain"]'],
},
),
"protonmail": SignupSitePreset(
key="protonmail",
label="Proton Mail",
url="https://account.proton.me/signup",
hosts=("account.proton.me", "proton.me"),
extra_selectors={
"email": ['input[id="email"]', 'input[name="email"]', 'input[type="email"]'],
"password": ['input[id="password"]', 'input[name="password"]'],
"password_confirm": ['input[id="password-confirm"]', 'input[name="passwordConfirmation"]'],
},
),
"hydraproxy": SignupSitePreset(
key="hydraproxy",
label="HydraProxy",
url="https://dashboard.hydraproxy.com/register",
hosts=("dashboard.hydraproxy.com", "hydraproxy.com"),
extra_selectors={
"email": ['input[name="email"]', 'input[type="email"]', 'input[id="email"]'],
"password": ['input[name="password"]', 'input[type="password"]'],
"password_confirm": ['input[name="password_confirmation"]', 'input[name="confirmPassword"]'],
"username": ['input[name="username"]', 'input[name="name"]'],
},
),
"outlook": SignupSitePreset(
key="outlook",
label="Outlook / Microsoft",
url="https://signup.live.com/signup",
hosts=("signup.live.com", "login.live.com", "account.microsoft.com"),
extra_selectors={
"email": ['input[name="MemberName"]', 'input[type="email"]'],
"password": ['input[name="Password"]', 'input[name="PasswordInput"]'],
"first_name": ['input[name="FirstName"]'],
"last_name": ['input[name="LastName"]'],
},
),
"github": SignupSitePreset(
key="github",
label="GitHub",
url="https://github.com/signup",
hosts=("github.com",),
extra_selectors={
"email": ['input[name="user[email]"]', 'input[autocomplete="email"]'],
"password": ['input[name="user[password]"]', 'input[autocomplete="new-password"]'],
"username": ['input[name="user[login]"]'],
},
),
"reddit": SignupSitePreset(
key="reddit",
label="Reddit",
url="https://www.reddit.com/register",
hosts=("www.reddit.com", "reddit.com"),
extra_selectors={
"email": ['input[name="email"]', 'input[type="email"]'],
"username": ['input[name="username"]'],
"password": ['input[name="password"]'],
},
),
"discord": SignupSitePreset(
key="discord",
label="Discord",
url="https://discord.com/register",
hosts=("discord.com",),
extra_selectors={
"email": ['input[name="email"]', 'input[type="email"]'],
"username": ['input[name="username"]'],
"password": ['input[name="password"]'],
},
),
"x_twitter": SignupSitePreset(
key="x_twitter",
label="X / Twitter",
url="https://x.com/i/flow/signup",
hosts=("x.com", "twitter.com"),
extra_selectors={
"email": ['input[name="email"]', 'input[type="email"]'],
"username": ['input[name="username"]'],
"password": ['input[name="password"]'],
"first_name": ['input[name="name"]'],
},
),
"telegram": SignupSitePreset(
key="telegram",
label="Telegram Web",
url="https://web.telegram.org/",
hosts=("web.telegram.org", "telegram.org"),
),
"tutanota": SignupSitePreset(
key="tutanota",
label="Tuta Mail",
url="https://app.tuta.com/signup",
hosts=("app.tuta.com", "tutanota.com", "tuta.com"),
extra_selectors={
"email": ['input[id="mailAddress"]', 'input[name="mailAddress"]'],
"password": ['input[id="password1"]', 'input[type="password"]'],
"password_confirm": ['input[id="password2"]'],
},
),
"mailcom": SignupSitePreset(
key="mailcom",
label="mail.com",
url="https://signup.mail.com/",
hosts=("signup.mail.com", "mail.com"),
extra_selectors={
"username": ['input[name="email_user"]', 'input[name="username"]'],
"password": ['input[name="password"]'],
"first_name": ['input[name="firstName"]'],
"last_name": ['input[name="lastName"]'],
},
),
}
# Public site-key list in display order. Useful for building dropdowns.
SIGNUP_PRESET_KEYS: tuple[str, ...] = (
"google", "protonmail", "outlook", "tutanota", "mailcom",
"github", "reddit", "discord", "x_twitter", "telegram",
"hydraproxy",
)
_FIRST_NAMES: tuple[str, ...] = (
"Alex", "Jordan", "Casey", "Morgan", "Taylor", "Riley", "Cameron",
"Sam", "Jamie", "Quinn", "Avery", "Drew", "Reese", "Skyler", "Logan",
"Hayden", "Parker", "Rowan", "Sage", "Emerson", "Finley", "Harper",
"Kai", "Micah", "Nico", "Phoenix", "River", "Shawn", "Sutton", "Wren",
)
_LAST_NAMES: tuple[str, ...] = (
"Reed", "Hayes", "Brooks", "Bennett", "Carter", "Foster", "Gray",
"Hughes", "Jensen", "Knight", "Lambert", "Mason", "Nash", "Owens",
"Porter", "Quincy", "Reyes", "Sutton", "Tate", "Underwood", "Vance",
"Walsh", "Yates", "Zimmerman", "Abbott", "Barker", "Cohen", "Dixon",
"Ellis", "Fitch",
)
def random_first_name() -> str:
return secrets.choice(_FIRST_NAMES)
def random_last_name() -> str:
return secrets.choice(_LAST_NAMES)
def random_full_name() -> tuple[str, str]:
return random_first_name(), random_last_name()
def random_username(first: str = "", last: str = "") -> str:
"""Generate a plausible-looking username.
If a name is provided we mix it with digits; otherwise we use two short
name fragments. Always lowercase, always 8-18 chars.
"""
f = (first or random_first_name()).lower()
l = (last or random_last_name()).lower()
seps = ("", ".", "_", "")
sep = secrets.choice(seps)
digits = "".join(secrets.choice(string.digits) for _ in range(secrets.choice([2, 3, 4])))
style = secrets.randbelow(4)
if style == 0:
base = f + sep + l + digits
elif style == 1:
base = f + digits
elif style == 2:
base = l + sep + f[0] + digits
else:
base = f[0] + sep + l + digits
base = base.strip(".-_")
return base[:18]
@dataclass
class SignupDraft:
site_key: str = "google"
custom_url: str = ""
email: str = ""
password: str = ""
username: str = ""
first_name: str = ""
last_name: str = ""
notes: str = ""
@dataclass
class AccountRecord:
id: str
site: str
url: str
email: str
password: str
username: str = ""
first_name: str = ""
last_name: str = ""
exit_ip: str = ""
exit_hop: str = ""
notes: str = ""
status: str = "created" # pending | created
created_at: str = ""
def accounts_path() -> Path:
return app_data_dir() / "signup_accounts.json"
def draft_path() -> Path:
return app_data_dir() / "signup_draft.json"
def generate_password(length: int = 18) -> str:
n = max(12, min(64, int(length)))
alphabet = string.ascii_letters + string.digits + "!@#$%^&*-_=+"
while True:
pw = "".join(secrets.choice(alphabet) for _ in range(n))
if (
any(c.islower() for c in pw)
and any(c.isupper() for c in pw)
and any(c.isdigit() for c in pw)
):
return pw
def preset_for(key: str) -> SignupSitePreset | None:
return SIGNUP_PRESETS.get(key)
def resolve_signup_url(draft: SignupDraft) -> str:
if draft.site_key == "custom":
return (draft.custom_url or "").strip()
p = preset_for(draft.site_key)
return p.url if p else ""
def load_draft() -> SignupDraft:
p = draft_path()
if not p.is_file():
return SignupDraft()
try:
raw = json.loads(p.read_text(encoding="utf-8"))
return SignupDraft(**{k: raw.get(k, "") for k in SignupDraft.__dataclass_fields__})
except (OSError, json.JSONDecodeError, TypeError):
return SignupDraft()
def save_draft(draft: SignupDraft) -> None:
draft_path().write_text(json.dumps(asdict(draft), indent=2), encoding="utf-8")
def load_accounts() -> list[AccountRecord]:
p = accounts_path()
if not p.is_file():
return []
try:
raw = json.loads(p.read_text(encoding="utf-8"))
if not isinstance(raw, list):
return []
out: list[AccountRecord] = []
for item in raw:
if not isinstance(item, dict):
continue
fields = AccountRecord.__dataclass_fields__
kwargs = {k: item.get(k, "") for k in fields}
if not kwargs.get("id"):
kwargs["id"] = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S%f")
out.append(AccountRecord(**kwargs))
return out
except (OSError, json.JSONDecodeError, TypeError):
return []
def save_accounts(rows: list[AccountRecord]) -> None:
accounts_path().write_text(
json.dumps([asdict(r) for r in rows], indent=2),
encoding="utf-8",
)
def append_account(record: AccountRecord) -> None:
rows = load_accounts()
rows.insert(0, record)
save_accounts(rows)
def build_autofill_config(draft: SignupDraft) -> dict[str, Any]:
preset = preset_for(draft.site_key)
hosts = list(preset.hosts) if preset else []
selectors: dict[str, list[str]] = {}
if preset:
selectors.update(preset.extra_selectors)
fields: dict[str, str] = {}
if draft.first_name:
fields["first_name"] = draft.first_name
if draft.last_name:
fields["last_name"] = draft.last_name
if draft.username:
fields["username"] = draft.username
if draft.email:
fields["email"] = draft.email
if draft.password:
fields["password"] = draft.password
fields["password_confirm"] = draft.password
out_fields: dict[str, Any] = {}
for key, val in fields.items():
out_fields[key] = val
sel_key = f"{key}_selectors"
if key in selectors:
out_fields[sel_key] = selectors[key]
return {
"site": draft.site_key,
"hosts": hosts,
"fields": out_fields,
"updated_at": datetime.now(timezone.utc).isoformat(),
}
def _extensions_json_entry(ext_dir_name: str) -> dict[str, Any]:
now = int(datetime.now(timezone.utc).timestamp()) * 1000
return {
"id": _EXT_ID,
"syncGUID": str(uuid.uuid4()),
"version": "1.0.0",
"type": "extension",
"location": {
"appProfile": {
"root": "app-profile",
"path": f"extensions/{ext_dir_name}",
}
},
"internalName": None,
"updateURL": None,
"webExtensionConverted": False,
"webExtensionId": _EXT_ID,
"active": True,
"userDisabled": False,
"appDisabled": False,
"installDate": now,
"scope": 1,
}
def install_signup_extension(profile_dir: Path, draft: SignupDraft) -> None:
"""Copy autofill WebExtension into the Firefox profile and write current field values."""
profile_dir.mkdir(parents=True, exist_ok=True)
ext_name = _EXT_ID
dest = profile_dir / "extensions" / ext_name
if dest.exists():
shutil.rmtree(dest, ignore_errors=True)
shutil.copytree(_EXT_SRC, dest)
cfg = build_autofill_config(draft)
(dest / "autofill_config.json").write_text(json.dumps(cfg, indent=2), encoding="utf-8")
ext_json_path = profile_dir / "extensions.json"
entry = _extensions_json_entry(ext_name)
data: dict[str, Any]
if ext_json_path.is_file():
try:
data = json.loads(ext_json_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
data = {"schemaVersion": 35, "addons": []}
else:
data = {"schemaVersion": 35, "addons": []}
addons = data.get("addons")
if not isinstance(addons, list):
addons = []
addons = [a for a in addons if isinstance(a, dict) and a.get("id") != _EXT_ID]
addons.append(entry)
data["addons"] = addons
ext_json_path.write_text(json.dumps(data, indent=2), encoding="utf-8")