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

optimize: shuffle-drain pool (no repeat chains), executor for fetchers, DEVNULL pipes, verify=False validator, IP fallbacks, +/- hop topbar, timestamps, countdown, rotation counter, blacklist bad proxies

Browse Source 
Made-with: Cursor
This commit is contained in:
Dr Jones
2026-04-14 17:21:10 -07:00
parent 4bb626498f
commit 45798464e4
16 changed files with 8208 additions and 7935 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
proxy_chain_manager/__pycache__/app.cpython-311.pyc
View File

Binary file not shown.

BIN
proxy_chain_manager/__pycache__/gost_util.cpython-311.pyc
View File

Binary file not shown.

BIN
proxy_chain_manager/__pycache__/service.cpython-311.pyc
View File

Binary file not shown.

BIN
proxy_chain_manager/__pycache__/validator.cpython-311.pyc
View File

Binary file not shown.

573
proxy_chain_manager/app.py
View File

@@ -1,5 +1,5 @@
"""
Proxy Chain Manager — GUI
Proxy God — GUI
Tabs: Live | Chain Builder | Settings
"""
from __future__ import annotations
@@ -7,6 +7,7 @@ from __future__ import annotations
import logging
import queue
import sys
from datetime import datetime
from typing import Any
import customtkinter as ctk
@@ -19,7 +20,12 @@ from .config import (
load_settings,
save_settings,
)
from .firewall import disengage as fw_disengage, is_admin, is_engaged as fw_is_engaged, request_admin_relaunch
from .firewall import (
disengage as fw_disengage,
is_admin,
is_engaged as fw_is_engaged,
request_admin_relaunch,
)
from .paths import app_data_dir
from .service import ChainService
from .sysproxy import clear_system_proxy, is_system_proxy_set
@@ -45,6 +51,7 @@ ACCENT2 = "#2563eb"
GREEN = "#00e676"
RED = "#ff1744"
YELLOW = "#ffab00"
ORANGE = "#ff6d00"
PURPLE = "#bb86fc"
BLUE = "#448aff"
CYAN = "#00bcd4"
@@ -54,14 +61,18 @@ TEXT2 = "#94a3b8"
FONT = "Segoe UI"
def _ts() -> str:
return datetime.now().strftime("%H:%M:%S")
def main() -> None:
ctk.set_appearance_mode("dark")
ctk.set_default_color_theme("dark-blue")
root = ctk.CTk()
root.title("Proxy God v2")
root.geometry("1060x720")
root.minsize(900, 600)
root.geometry("1080x740")
root.minsize(900, 620)
root.configure(fg_color=BG)
ui_q: queue.Queue[dict[str, Any]] = queue.Queue()
@@ -69,126 +80,187 @@ def main() -> None:
s = load_settings()
# ── tray ─────────────────────────────────────────────────────────────────
def _tray_show() -> None:
root.after(0, lambda: (root.deiconify(), root.lift()))
def _tray_quit() -> None:
root.after(0, _close)
tray = TrayIcon(on_show=_tray_show, on_quit=_tray_quit, on_rotate=svc.rotate_now)
tray = TrayIcon(
on_show=lambda: root.after(0, lambda: (root.deiconify(), root.lift())),
on_quit=lambda: root.after(0, _close),
on_rotate=svc.rotate_now,
)
tray.start()
# ─────────────────────────────────────────────────────────────────────────
# SHARED STATE
# ─────────────────────────────────────────────────────────────────────────
_hop_count = [s.chain_length] # mutable ref so topbar buttons can mutate it
pulse_job: dict[str, str | None] = {"id": None}
def _cancel_pulse() -> None:
if pulse_job["id"]:
try:
root.after_cancel(pulse_job["id"]) # type: ignore[arg-type]
except Exception:
pass
pulse_job["id"] = None
def _short(u: str) -> str:
u = u.replace("http://", "").replace("socks5://", "s5://").replace("socks4://", "s4://").replace("https://", "")
return u[:28] + "…" if len(u) > 30 else u
# ─────────────────────────────────────────────────────────────────────────
# HELPERS
# ─────────────────────────────────────────────────────────────────────────
def _btn(parent: Any, label: str, cmd: Any, w: int = 80, h: int = 28, **kw: Any) -> ctk.CTkButton:
cfg = {"fg_color": ACCENT, "hover_color": ACCENT2, "text_color": TEXT}
cfg: dict[str, Any] = {"fg_color": ACCENT, "hover_color": ACCENT2, "text_color": TEXT}
cfg.update(kw)
return ctk.CTkButton(
parent, text=label, command=cmd, width=w, height=h,
font=(FONT, 11), corner_radius=6, **cfg,
)
pulse_job: dict[str, Any] = {"id": None}
def _cancel_pulse() -> None:
jid = pulse_job.get("id")
if jid:
try:
root.after_cancel(jid)
except Exception:
pass
pulse_job["id"] = None
def _short(u: str) -> str:
u = u.replace("http://", "").replace("socks5://", "s5://").replace("socks4://", "s4://")
return u[:28] + "…" if len(u) > 30 else u
# ─────────────────────────────────────────────────────────────────────────
# TOP BAR
# ─────────────────────────────────────────────────────────────────────────
topbar = ctk.CTkFrame(root, fg_color=CARD, corner_radius=0, height=46)
topbar = ctk.CTkFrame(root, fg_color=CARD, corner_radius=0, height=48)
topbar.pack(fill="x", side="top")
topbar.pack_propagate(False)
# Status
status_dot = ctk.CTkLabel(topbar, text="●", font=(FONT, 20), text_color=RED)
status_dot.pack(side="left", padx=(10, 2))
status_txt = ctk.CTkLabel(topbar, text="STOPPED", font=(FONT, 12, "bold"), text_color=TEXT)
status_txt.pack(side="left", padx=(0, 10))
status_txt.pack(side="left", padx=(0, 8))
def _start() -> None:
_save_settings()
svc.start()
_btn(topbar, "▶ Start", _start, w=90).pack(side="left", padx=2)
_btn(topbar, "■ Stop", svc.stop, w=80).pack(side="left", padx=2)
_btn(topbar, "↻ Rotate", svc.rotate_now, w=80).pack(side="left", padx=2)
_btn(topbar, "Quit", lambda: root.after(0, _close), w=55,
fg_color="#7f1d1d", hover_color="#991b1b").pack(side="left", padx=(8, 2))
_btn(topbar, "▶ Start", _start, w=86).pack(side="left", padx=2)
_btn(topbar, "■ Stop", svc.stop, w=76).pack(side="left", padx=2)
_btn(topbar, "↻ Rotate", svc.rotate_now, w=76).pack(side="left", padx=2)
_btn(topbar, "Quit", lambda: root.after(0, _close), w=52,
fg_color="#7f1d1d", hover_color="#991b1b").pack(side="left", padx=(10, 2))
# Hop count +/- control
ctk.CTkLabel(topbar, text="│", text_color=DIM).pack(side="left", padx=6)
ctk.CTkLabel(topbar, text="Hops:", font=(FONT, 11), text_color=TEXT2).pack(side="left", padx=(0, 2))
hop_count_lbl = ctk.CTkLabel(topbar, text=str(_hop_count[0]),
font=(FONT, 15, "bold"), text_color=CYAN, width=24)
hop_count_lbl.pack(side="left")
def _hop_adjust(delta: int) -> None:
new = max(1, min(8, _hop_count[0] + delta))
_hop_count[0] = new
hop_count_lbl.configure(text=str(new))
# Update hop slider in Chain Builder if it exists
try:
hop_slider.set(new)
hop_val_lbl.configure(text=str(new))
except Exception:
pass
# Immediately apply to settings
_apply_hop_count(new)
_btn(topbar, "−", lambda: _hop_adjust(-1), w=26, h=26,
fg_color=PANEL, hover_color=ACCENT).pack(side="left", padx=1)
_btn(topbar, "+", lambda: _hop_adjust(+1), w=26, h=26,
fg_color=PANEL, hover_color=ACCENT).pack(side="left", padx=(1, 6))
def _apply_hop_count(n: int) -> None:
"""Save hop count change to live settings immediately."""
cur = svc.settings
ns = Settings(
local_host=cur.local_host,
local_port=cur.local_port,
chain_length=n,
obfuscation_mode=cur.obfuscation_mode,
use_pinned_chain=cur.use_pinned_chain,
pinned_chain=cur.pinned_chain,
health_check_seconds=cur.health_check_seconds,
full_refresh_seconds=cur.full_refresh_seconds,
validation_concurrency=cur.validation_concurrency,
max_candidates=cur.max_candidates,
validation_timeout_seconds=cur.validation_timeout_seconds,
prefer_elite=cur.prefer_elite,
kill_switch_enabled=cur.kill_switch_enabled,
proxy_bypass=cur.proxy_bypass,
sources=cur.sources,
ip_check_url=cur.ip_check_url,
)
svc.update_settings(ns)
# Boot buttons
ctk.CTkLabel(topbar, text="│", text_color=DIM).pack(side="left", padx=4)
boot_lbl = ctk.CTkLabel(topbar, text="Boot:?", font=(FONT, 10), text_color=TEXT2)
boot_lbl.pack(side="left", padx=2)
def _refresh_boot() -> None:
on = task_exists()
boot_lbl.configure(text="Boot: ON" if on else "Boot: OFF",
boot_lbl.configure(text="Boot:ON" if on else "Boot:OFF",
text_color=GREEN if on else DIM)
def _inst_boot() -> None:
ok, msg = install_logon_task()
_log("Boot task installed." if ok else f"Boot install fail: {msg}")
_log("Boot task installed." if ok else f"Boot install: {msg}")
_refresh_boot()
def _rm_boot() -> None:
ok, msg = uninstall_logon_task()
_log("Boot task removed." if ok else f"Boot remove fail: {msg}")
_log("Boot task removed." if ok else f"Boot remove: {msg}")
_refresh_boot()
_btn(topbar, "Boot+", _inst_boot, w=55).pack(side="left", padx=2)
_btn(topbar, "Boot-", _rm_boot, w=55).pack(side="left", padx=2)
_btn(topbar, "Boot+", _inst_boot, w=54).pack(side="left", padx=2)
_btn(topbar, "Boot−", _rm_boot, w=54).pack(side="left", padx=2)
# Right side status indicators
proxy_lbl = ctk.CTkLabel(topbar, text=f"proxy: {LISTEN_HOST}:{s.local_port}",
# Right side indicators
rotation_lbl = ctk.CTkLabel(topbar, text="rot: 0", font=(FONT, 10), text_color=TEXT2)
rotation_lbl.pack(side="right", padx=(4, 12))
countdown_lbl = ctk.CTkLabel(topbar, text="next: —", font=(FONT, 10), text_color=TEXT2)
countdown_lbl.pack(side="right", padx=4)
proxy_lbl = ctk.CTkLabel(topbar, text=f"{LISTEN_HOST}:{s.local_port}",
font=(FONT, 11, "bold"), text_color=BLUE)
proxy_lbl.pack(side="right", padx=(4, 12))
proxy_lbl.pack(side="right", padx=(4, 4))
ctk.CTkLabel(topbar, text="proxy:", font=(FONT, 10), text_color=TEXT2).pack(side="right")
fw_lbl = ctk.CTkLabel(topbar, text="FW:—", font=(FONT, 10), text_color=DIM)
fw_lbl.pack(side="right", padx=4)
fw_lbl.pack(side="right", padx=6)
sys_lbl = ctk.CTkLabel(topbar, text="SYS:—", font=(FONT, 10), text_color=DIM)
sys_lbl.pack(side="right", padx=4)
admin_lbl = ctk.CTkLabel(topbar,
text="⚡ADMIN" if is_admin() else "👤USER",
font=(FONT, 10, "bold"),
text_color=GREEN if is_admin() else YELLOW)
admin_lbl.pack(side="right", padx=4)
admin_badge = ctk.CTkLabel(
topbar,
text="⚡ ADMIN" if is_admin() else "👤 USER",
font=(FONT, 10, "bold"),
text_color=GREEN if is_admin() else YELLOW,
)
admin_badge.pack(side="right", padx=4)
if not is_admin():
def _elevate() -> None:
if request_admin_relaunch():
_close()
_btn(topbar, "Run as Admin", _elevate, w=100,
_btn(topbar, "Run as Admin", _elevate, w=104,
fg_color="#7f1d1d", hover_color="#991b1b").pack(side="right", padx=4)
# ─────────────────────────────────────────────────────────────────────────
# CHAIN VIZ BAR
# ─────────────────────────────────────────────────────────────────────────
viz_bar = ctk.CTkFrame(root, fg_color=PANEL, corner_radius=0, height=54)
viz_bar = ctk.CTkFrame(root, fg_color=PANEL, corner_radius=0, height=52)
viz_bar.pack(fill="x")
viz_bar.pack_propagate(False)
your_ip_lbl = ctk.CTkLabel(viz_bar, text="Your IP: —", font=(FONT, 10), text_color=TEXT2)
your_ip_lbl.pack(side="left", padx=(12, 6), pady=4)
your_ip_lbl.pack(side="left", padx=(12, 8))
chain_canvas = ctk.CTkFrame(viz_bar, fg_color="transparent")
chain_canvas.pack(side="left", fill="both", expand=True, padx=4)
chain_canvas.pack(side="left", fill="both", expand=True)
exit_ip_lbl = ctk.CTkLabel(viz_bar, text="Exit: —",
font=(FONT, 13, "bold"), text_color=YELLOW)
exit_ip_lbl.pack(side="right", padx=(6, 14))
exit_ip_lbl.pack(side="right", padx=(8, 14))
hop_dot_labels: list[ctk.CTkLabel] = []
@@ -199,7 +271,7 @@ def main() -> None:
c = YELLOW if step % 2 == 0 else DIM
for d in hop_dot_labels:
d.configure(text_color=c)
pulse_job["id"] = root.after(380, lambda: _pulse(step + 1))
pulse_job["id"] = root.after(380, lambda s=step: _pulse(s + 1)) # type: ignore[arg-type]
def _render_chain(hops: list[str], status: str, exit_ip: str | None) -> None:
_cancel_pulse()
@@ -207,12 +279,14 @@ def main() -> None:
w.destroy()
hop_dot_labels.clear()
color_map = {"healthy": GREEN, "dead": RED, "connecting": YELLOW}
hop_c = color_map.get(status, DIM)
hop_c = {"healthy": GREEN, "dead": RED, "connecting": YELLOW}.get(status, DIM)
def _node(text: str, color: str, bold: bool = False) -> None:
f = (FONT, 11, "bold") if bold else (FONT, 10)
ctk.CTkLabel(chain_canvas, text=text, font=f, text_color=color).pack(side="left", padx=1)
ctk.CTkLabel(
chain_canvas, text=text,
font=(FONT, 11, "bold") if bold else (FONT, 10),
text_color=color,
).pack(side="left", padx=1)
def _arr() -> None:
ctk.CTkLabel(chain_canvas, text="→", font=(FONT, 11), text_color=DIM).pack(side="left", padx=2)
@@ -233,14 +307,10 @@ def main() -> None:
_arr()
_node("WEB", BLUE, bold=True)
if exit_ip:
exit_ip_lbl.configure(
text=f"Exit: {exit_ip}",
text_color=GREEN if status == "healthy" else RED,
)
else:
exit_ip_lbl.configure(text="Exit: —",
text_color=YELLOW if status == "connecting" else DIM)
exit_ip_lbl.configure(
text=f"Exit: {exit_ip}" if exit_ip else "Exit: —",
text_color=GREEN if status == "healthy" else (RED if exit_ip else YELLOW),
)
if status == "connecting" and hop_dot_labels:
_pulse()
@@ -250,153 +320,175 @@ def main() -> None:
# ─────────────────────────────────────────────────────────────────────────
prog_strip = ctk.CTkFrame(root, fg_color=BG, height=22)
prog_strip.pack(fill="x", padx=8, pady=(4, 0))
phase_lbl = ctk.CTkLabel(prog_strip, text="idle", font=(FONT, 10), text_color=TEXT2)
prog_strip.pack_propagate(False)
phase_lbl = ctk.CTkLabel(prog_strip, text="idle", font=(FONT, 10), text_color=TEXT2, width=70)
phase_lbl.pack(side="left", padx=4)
prog_bar = ctk.CTkProgressBar(prog_strip, height=8, corner_radius=4,
fg_color=CARD, progress_color=BLUE)
prog_bar.pack(side="left", fill="x", expand=True, padx=4)
prog_bar.set(0)
pool_lbl = ctk.CTkLabel(prog_strip, text="pool: 0", font=(FONT, 10), text_color=TEXT2)
pool_lbl = ctk.CTkLabel(prog_strip, text="pool: 0", font=(FONT, 10), text_color=TEXT2, width=70)
pool_lbl.pack(side="right", padx=4)
# ─────────────────────────────────────────────────────────────────────────
# TABVIEW
# ─────────────────────────────────────────────────────────────────────────
tabs = ctk.CTkTabview(root, fg_color=BG, segmented_button_fg_color=CARD,
segmented_button_selected_color=ACCENT2,
segmented_button_unselected_color=CARD,
segmented_button_selected_hover_color=ACCENT,
border_width=0)
tabs = ctk.CTkTabview(
root,
fg_color=BG,
segmented_button_fg_color=CARD,
segmented_button_selected_color=ACCENT2,
segmented_button_unselected_color=CARD,
segmented_button_selected_hover_color=ACCENT,
border_width=0,
)
tabs.pack(fill="both", expand=True, padx=8, pady=(4, 8))
tab_live = tabs.add(" Live ")
tab_chain = tabs.add(" Chain Builder ")
tab_settings = tabs.add(" Settings ")
for tab in (tab_live, tab_chain, tab_settings):
tab.configure(fg_color=BG)
# ─── TAB: LIVE ───────────────────────────────────────────────────────────
tab_live.configure(fg_color=BG)
# Stats row
# Stats cards row
stats_row = ctk.CTkFrame(tab_live, fg_color="transparent")
stats_row.pack(fill="x", pady=(4, 6))
def _stat_card(parent: Any, title: str, var: ctk.StringVar) -> None:
def _stat_card(parent: Any, title: str, var: ctk.StringVar, color: str = TEXT) -> None:
f = ctk.CTkFrame(parent, fg_color=CARD, corner_radius=8)
f.pack(side="left", fill="x", expand=True, padx=4)
ctk.CTkLabel(f, text=title, font=(FONT, 9), text_color=TEXT2).pack(anchor="w", padx=8, pady=(6, 0))
ctk.CTkLabel(f, textvariable=var, font=(FONT, 15, "bold"), text_color=TEXT).pack(anchor="w", padx=8, pady=(0, 6))
ctk.CTkLabel(f, text=title, font=(FONT, 9), text_color=TEXT2).pack(
anchor="w", padx=8, pady=(6, 0))
ctk.CTkLabel(f, textvariable=var, font=(FONT, 14, "bold"), text_color=color).pack(
anchor="w", padx=8, pady=(0, 6))
v_real_ip = ctk.StringVar(value="—")
v_exit_ip = ctk.StringVar(value="—")
v_pool = ctk.StringVar(value="0")
v_hops = ctk.StringVar(value="—")
_stat_card(stats_row, "Your IP (Nord/ISP)", v_real_ip)
_stat_card(stats_row, "Exit IP (last proxy)", v_exit_ip)
_stat_card(stats_row, "Valid proxy pool", v_pool)
_stat_card(stats_row, "Active hops", v_hops)
v_rotation = ctk.StringVar(value="0")
_stat_card(stats_row, "Your IP (through Nord)", v_real_ip)
_stat_card(stats_row, "Exit IP (last hop)", v_exit_ip)
_stat_card(stats_row, "Valid pool size", v_pool)
_stat_card(stats_row, "Rotations this session", v_rotation)
# Log box
log_frame = ctk.CTkFrame(tab_live, fg_color=CARD, corner_radius=8)
log_frame.pack(fill="both", expand=True, padx=0)
log_box = ctk.CTkTextbox(log_frame, font=("Consolas", 11), fg_color=BG,
text_color=TEXT, scrollbar_button_color=ACCENT)
log_frame.pack(fill="both", expand=True)
log_box = ctk.CTkTextbox(
log_frame, font=("Consolas", 11),
fg_color=BG, text_color=TEXT,
scrollbar_button_color=ACCENT,
)
log_box.pack(fill="both", expand=True, padx=4, pady=4)
def _log(line: str) -> None:
log_box.insert("end", line + "\n")
log_box.insert("end", f"[{_ts()}] {line}\n")
log_box.see("end")
# ─── TAB: CHAIN BUILDER ──────────────────────────────────────────────────
tab_chain.configure(fg_color=BG)
cb_split = ctk.CTkFrame(tab_chain, fg_color="transparent")
cb_split.pack(fill="x", pady=(4, 8))
cb_top = ctk.CTkFrame(tab_chain, fg_color="transparent")
cb_top.pack(fill="x", pady=(4, 8))
# Left col — mode + hops
cb_left = ctk.CTkFrame(cb_top, fg_color=CARD, corner_radius=8)
# Left — mode + hop slider
cb_left = ctk.CTkFrame(cb_split, fg_color=CARD, corner_radius=8)
cb_left.pack(side="left", fill="both", expand=True, padx=(0, 6))
ctk.CTkLabel(cb_left, text="Obfuscation Mode", font=(FONT, 12, "bold"), text_color=TEXT).pack(
anchor="w", padx=10, pady=(10, 4))
ctk.CTkLabel(cb_left, text="Obfuscation Mode", font=(FONT, 12, "bold"),
text_color=TEXT).pack(anchor="w", padx=12, pady=(12, 4))
mode_var = ctk.StringVar(value=s.obfuscation_mode)
for mode_key in OBFUSCATION_MODES:
for mk in OBFUSCATION_MODES:
ctk.CTkRadioButton(
cb_left,
text=OBFUSCATION_LABELS[mode_key],
text=OBFUSCATION_LABELS[mk],
variable=mode_var,
value=mode_key,
value=mk,
font=(FONT, 11),
fg_color=ACCENT2,
hover_color=ACCENT,
text_color=TEXT,
).pack(anchor="w", padx=14, pady=2)
).pack(anchor="w", padx=16, pady=3)
ctk.CTkLabel(cb_left, text="", height=6).pack()
ctk.CTkLabel(cb_left, text="Hop Count", font=(FONT, 12, "bold"), text_color=TEXT).pack(
anchor="w", padx=10)
ctk.CTkFrame(cb_left, fg_color="transparent", height=8).pack()
ctk.CTkLabel(cb_left, text="Hop Count (also: topbar +/−)",
font=(FONT, 12, "bold"), text_color=TEXT).pack(anchor="w", padx=12)
hop_val_lbl = ctk.CTkLabel(cb_left, text=str(s.chain_length),
font=(FONT, 22, "bold"), text_color=CYAN)
hop_val_lbl = ctk.CTkLabel(cb_left, text=str(_hop_count[0]),
font=(FONT, 26, "bold"), text_color=CYAN)
hop_val_lbl.pack(pady=(2, 0))
def _on_hop_slider(val: float) -> None:
hop_val_lbl.configure(text=str(int(val)))
n = int(val)
_hop_count[0] = n
hop_val_lbl.configure(text=str(n))
hop_count_lbl.configure(text=str(n))
_apply_hop_count(n)
hop_slider = ctk.CTkSlider(cb_left, from_=2, to=8, number_of_steps=6,
command=_on_hop_slider,
fg_color=CARD, progress_color=ACCENT2,
button_color=CYAN, button_hover_color=BLUE)
hop_slider = ctk.CTkSlider(
cb_left, from_=1, to=8, number_of_steps=7,
command=_on_hop_slider,
fg_color=PANEL, progress_color=ACCENT2,
button_color=CYAN, button_hover_color=BLUE,
)
hop_slider.set(s.chain_length)
hop_slider.pack(fill="x", padx=14, pady=(2, 12))
hop_slider.pack(fill="x", padx=16, pady=(2, 6))
elite_var2 = ctk.BooleanVar(value=s.prefer_elite)
ctk.CTkCheckBox(cb_left, text="Elite proxies only",
variable=elite_var2, font=(FONT, 11),
fg_color=ACCENT2, hover_color=ACCENT, text_color=TEXT).pack(
anchor="w", padx=14, pady=(0, 10))
hop_legend = ctk.CTkFrame(cb_left, fg_color="transparent")
hop_legend.pack(fill="x", padx=16, pady=(0, 8))
ctk.CTkLabel(hop_legend, text="1", font=(FONT, 9), text_color=DIM).pack(side="left")
ctk.CTkLabel(hop_legend, text="8", font=(FONT, 9), text_color=DIM).pack(side="right")
# Right col — manual chain editor
cb_right = ctk.CTkFrame(cb_top, fg_color=CARD, corner_radius=8, width=380)
elite_var = ctk.BooleanVar(value=s.prefer_elite)
ctk.CTkCheckBox(
cb_left, text="Elite proxies only (slower but more anonymous)",
variable=elite_var, font=(FONT, 11),
fg_color=ACCENT2, hover_color=ACCENT, text_color=TEXT,
).pack(anchor="w", padx=16, pady=(0, 12))
# Right — manual chain editor
cb_right = ctk.CTkFrame(cb_split, fg_color=CARD, corner_radius=8, width=400)
cb_right.pack(side="left", fill="both", expand=True)
hdr_row = ctk.CTkFrame(cb_right, fg_color="transparent")
hdr_row.pack(fill="x", padx=10, pady=(10, 4))
ctk.CTkLabel(hdr_row, text="Manual Chain", font=(FONT, 12, "bold"), text_color=TEXT).pack(side="left")
hdr_row.pack(fill="x", padx=12, pady=(12, 4))
ctk.CTkLabel(hdr_row, text="Manual Chain",
font=(FONT, 12, "bold"), text_color=TEXT).pack(side="left")
use_manual_var = ctk.BooleanVar(value=s.use_pinned_chain)
ctk.CTkCheckBox(hdr_row, text="Use this chain",
variable=use_manual_var,
font=(FONT, 10), fg_color=ACCENT2,
hover_color=ACCENT, text_color=TEXT2).pack(side="right")
# Listbox-style chain using a scrollable frame with rows
chain_list_frame = ctk.CTkScrollableFrame(cb_right, fg_color=BG,
height=160, corner_radius=6,
scrollbar_button_color=ACCENT)
chain_list_frame.pack(fill="x", padx=10, pady=(0, 6))
ctk.CTkCheckBox(
hdr_row, text="Pin & use this chain",
variable=use_manual_var, font=(FONT, 10),
fg_color=ACCENT2, hover_color=ACCENT, text_color=TEXT2,
).pack(side="right")
manual_chain: list[str] = list(s.pinned_chain)
row_frames: list[ctk.CTkFrame] = []
chain_list_frame = ctk.CTkScrollableFrame(
cb_right, fg_color=BG, height=170, corner_radius=6,
scrollbar_button_color=ACCENT,
)
chain_list_frame.pack(fill="x", padx=12, pady=(0, 6))
def _rebuild_chain_ui() -> None:
for w in list(chain_list_frame.winfo_children()):
w.destroy()
row_frames.clear()
for i, hop in enumerate(manual_chain):
_chain_row(i, hop)
def _chain_row(idx: int, hop: str) -> None:
fr = ctk.CTkFrame(chain_list_frame, fg_color=PANEL, corner_radius=6, height=30)
fr = ctk.CTkFrame(chain_list_frame, fg_color=PANEL, corner_radius=6, height=32)
fr.pack(fill="x", pady=2)
fr.pack_propagate(False)
row_frames.append(fr)
# Drag indicator + index
ctk.CTkLabel(fr, text=f" {idx+1}.", font=(FONT, 10, "bold"),
text_color=CYAN, width=24).pack(side="left")
text_color=CYAN, width=26).pack(side="left")
# Protocol badge
proto = hop.split("://")[0].upper() if "://" in hop else "?"
proto = hop.split("://")[0].upper() if "://" in hop else "HTTP"
badge_c = {"HTTP": "#1e3a8a", "SOCKS5": "#064e3b", "SOCKS4": "#3b1a64"}.get(proto, DIM)
ctk.CTkLabel(fr, text=proto, font=(FONT, 9, "bold"),
fg_color=badge_c, corner_radius=4, padx=4,
@@ -419,67 +511,64 @@ def main() -> None:
del manual_chain[i]
_rebuild_chain_ui()
_btn(fr, "↑", _up, w=24, h=22, fg_color=DIM, hover_color=ACCENT).pack(side="right", padx=1)
_btn(fr, "↓", _dn, w=24, h=22, fg_color=DIM, hover_color=ACCENT).pack(side="right", padx=1)
_btn(fr, "✕", _rm, w=24, h=22, fg_color="#7f1d1d", hover_color=RED).pack(side="right", padx=(1, 4))
_btn(fr, "↑", _up, w=26, h=24, fg_color=DIM, hover_color=ACCENT).pack(side="right", padx=1)
_btn(fr, "↓", _dn, w=26, h=24, fg_color=DIM, hover_color=ACCENT).pack(side="right", padx=1)
_btn(fr, "✕", _rm, w=26, h=24, fg_color="#7f1d1d", hover_color=RED).pack(side="right", padx=(1, 6))
_rebuild_chain_ui()
# Add proxy row
# Add proxy input
add_row = ctk.CTkFrame(cb_right, fg_color="transparent")
add_row.pack(fill="x", padx=10, pady=(0, 4))
add_entry = ctk.CTkEntry(add_row, placeholder_text="http://1.2.3.4:8080 or socks5://...",
font=(FONT, 10), fg_color=BG, border_color=ACCENT, height=28)
add_row.pack(fill="x", padx=12, pady=(0, 4))
add_entry = ctk.CTkEntry(
add_row, placeholder_text="http://ip:port or socks5://ip:port",
font=(FONT, 10), fg_color=BG, border_color=ACCENT, height=28,
)
add_entry.pack(side="left", fill="x", expand=True, padx=(0, 4))
def _add_proxy() -> None:
v = add_entry.get().strip()
if v and ("://" in v or ":" in v):
if "://" not in v:
v = "http://" + v
manual_chain.append(v)
_rebuild_chain_ui()
add_entry.delete(0, "end")
else:
_log("Invalid proxy format — use http://ip:port or socks5://ip:port")
if not v:
return
if "://" not in v:
v = "http://" + v
manual_chain.append(v)
_rebuild_chain_ui()
add_entry.delete(0, "end")
_btn(add_row, "+ Add", _add_proxy, w=70, h=28).pack(side="left")
_btn(add_row, "+ Add", _add_proxy, w=68, h=28).pack(side="left")
def _paste_from_current() -> None:
def _paste_current() -> None:
for h in svc.current_chain:
if h not in manual_chain:
manual_chain.append(h)
_rebuild_chain_ui()
_log("Pasted active chain into manual chain editor.")
_log("Active chain pasted into manual editor.")
_btn(cb_right, "↙ Paste current chain", _paste_from_current, w=200, h=26,
_btn(cb_right, "↙ Paste active chain", _paste_current, w=180, h=26,
fg_color=DIM, hover_color=ACCENT).pack(pady=(0, 10))
# Sources section
# Sources
src_frame = ctk.CTkFrame(tab_chain, fg_color=CARD, corner_radius=8)
src_frame.pack(fill="x", pady=(0, 6))
src_hdr = ctk.CTkFrame(src_frame, fg_color="transparent")
src_hdr.pack(fill="x", padx=10, pady=(8, 4))
ctk.CTkLabel(src_hdr, text="Proxy Sources (one URL per line)",
font=(FONT, 12, "bold"), text_color=TEXT).pack(side="left")
sources_box = ctk.CTkTextbox(src_frame, height=80, font=("Consolas", 10),
fg_color=BG, text_color=TEXT,
scrollbar_button_color=ACCENT)
sources_box.pack(fill="x", padx=10, pady=(0, 8))
ctk.CTkLabel(src_frame, text="Proxy Sources (one JSON URL per line)",
font=(FONT, 11, "bold"), text_color=TEXT).pack(
anchor="w", padx=12, pady=(8, 4))
sources_box = ctk.CTkTextbox(
src_frame, height=72, font=("Consolas", 10),
fg_color=BG, text_color=TEXT, scrollbar_button_color=ACCENT,
)
sources_box.pack(fill="x", padx=12, pady=(0, 8))
sources_box.insert("end", "\n".join(s.sources))
# ─── TAB: SETTINGS ───────────────────────────────────────────────────────
tab_settings.configure(fg_color=BG)
sf_outer = ctk.CTkScrollableFrame(tab_settings, fg_color=BG,
scrollbar_button_color=ACCENT)
sf_outer.pack(fill="both", expand=True)
def _section(title: str) -> ctk.CTkFrame:
ctk.CTkLabel(sf_outer, text=title, font=(FONT, 12, "bold"), text_color=CYAN).pack(
anchor="w", padx=6, pady=(14, 4))
ctk.CTkLabel(sf_outer, text=title, font=(FONT, 12, "bold"),
text_color=CYAN).pack(anchor="w", padx=6, pady=(14, 4))
f = ctk.CTkFrame(sf_outer, fg_color=CARD, corner_radius=8)
f.pack(fill="x", padx=6, pady=(0, 4))
return f
@@ -488,44 +577,50 @@ def main() -> None:
def _fld(parent: Any, label: str, key: str, val: str, tip: str = "") -> None:
row = ctk.CTkFrame(parent, fg_color="transparent")
row.pack(fill="x", padx=10, pady=4)
ctk.CTkLabel(row, text=label, font=(FONT, 11), text_color=TEXT, width=220,
anchor="w").pack(side="left")
e = ctk.CTkEntry(row, height=26, font=(FONT, 11), fg_color=BG,
border_color=ACCENT, width=120)
row.pack(fill="x", padx=12, pady=5)
ctk.CTkLabel(row, text=label, font=(FONT, 11), text_color=TEXT,
width=230, anchor="w").pack(side="left")
e = ctk.CTkEntry(row, height=26, font=(FONT, 11),
fg_color=BG, border_color=ACCENT, width=130)
e.insert(0, val)
e.pack(side="left")
if tip:
ctk.CTkLabel(row, text=f" {tip}", font=(FONT, 10), text_color=TEXT2).pack(side="left")
ctk.CTkLabel(row, text=f" {tip}", font=(FONT, 10),
text_color=TEXT2).pack(side="left")
entries[key] = e
net = _section("Network")
_fld(net, "Local port", "port", str(s.local_port), "(apps point here)")
_fld(net, "Proxy bypass list", "bypass", s.proxy_bypass, "(semicolon separated)")
_fld(net, "Local listener port", "port", str(s.local_port), "(point apps here)")
_fld(net, "Proxy bypass list", "bypass", s.proxy_bypass, "(semicolons)")
_fld(net, "IP check URL", "check_url", s.ip_check_url)
timing = _section("Timing")
_fld(timing, "Health check interval (sec)", "health", str(s.health_check_seconds), "(0 = manual only)")
_fld(timing, "Full list refresh (sec)", "refresh", str(s.full_refresh_seconds))
_fld(timing, "Health check interval (sec)", "health", str(s.health_check_seconds),
"(how often exit IP is re-verified)")
_fld(timing, "Full pool refresh (sec)", "refresh", str(s.full_refresh_seconds),
"(re-fetch + re-validate)")
validation = _section("Validation")
_fld(validation, "Concurrent proxy tests", "conc", str(s.validation_concurrency))
_fld(validation, "Max candidates per cycle", "maxc", str(s.max_candidates))
_fld(validation, "Per-proxy timeout (sec)", "timeout", str(s.validation_timeout_seconds))
val_sec = _section("Validation")
_fld(val_sec, "Concurrent validators", "conc", str(s.validation_concurrency))
_fld(val_sec, "Max candidates per cycle", "maxc", str(s.max_candidates),
"(random sample from fetch)")
_fld(val_sec, "Per-proxy timeout (sec)", "timeout", str(s.validation_timeout_seconds))
security = _section("Security")
ks_row = ctk.CTkFrame(security, fg_color="transparent")
ks_row.pack(fill="x", padx=10, pady=6)
sec_sec = _section("Security")
ks_row = ctk.CTkFrame(sec_sec, fg_color="transparent")
ks_row.pack(fill="x", padx=12, pady=8)
ks_var = ctk.BooleanVar(value=s.kill_switch_enabled)
ctk.CTkCheckBox(ks_row, text="Firewall kill-switch (block all traffic if proxy chain dies)",
variable=ks_var, font=(FONT, 11),
fg_color=ACCENT2, hover_color=ACCENT, text_color=TEXT).pack(side="left")
ctk.CTkCheckBox(
ks_row,
text="Firewall kill-switch (block ALL traffic if chain is down — requires Admin)",
variable=ks_var, font=(FONT, 11),
fg_color=ACCENT2, hover_color=ACCENT, text_color=TEXT,
).pack(side="left")
save_btn_frame = ctk.CTkFrame(sf_outer, fg_color="transparent")
save_btn_frame.pack(fill="x", padx=6, pady=8)
_btn(save_btn_frame, "💾 Save All Settings", lambda: _save_settings(verbose=True),
w=220, h=36, font=(FONT, 13)).pack(side="left")
ctk.CTkFrame(sf_outer, fg_color="transparent", height=4).pack()
_btn(sf_outer, "💾 Save All Settings",
lambda: _save_settings(verbose=True), w=220, h=36,
font=(FONT, 13)).pack(anchor="w", padx=6, pady=8)
# ─────────────────────────────────────────────────────────────────────────
# SAVE SETTINGS
@@ -533,21 +628,27 @@ def main() -> None:
def _save_settings(verbose: bool = False) -> None:
try:
src_raw = sources_box.get("1.0", "end").strip()
src_list = [ln.strip() for ln in src_raw.splitlines() if ln.strip().startswith("http")]
src_list = [ln.strip() for ln in src_raw.splitlines()
if ln.strip().startswith("http")]
n_hops = _hop_count[0]
hop_slider.set(n_hops)
hop_val_lbl.configure(text=str(n_hops))
hop_count_lbl.configure(text=str(n_hops))
ns = Settings(
local_host=LISTEN_HOST,
local_port=int(entries["port"].get().strip()),
chain_length=int(hop_slider.get()),
chain_length=n_hops,
obfuscation_mode=mode_var.get(),
use_pinned_chain=bool(use_manual_var.get()),
pinned_chain=list(manual_chain),
health_check_seconds=int(entries["health"].get().strip()),
full_refresh_seconds=int(entries["refresh"].get().strip()),
validation_concurrency=int(entries["conc"].get().strip()),
max_candidates=int(entries["maxc"].get().strip()),
validation_timeout_seconds=float(entries["timeout"].get().strip()),
prefer_elite=bool(elite_var2.get()),
health_check_seconds=max(10, int(entries["health"].get().strip())),
full_refresh_seconds=max(60, int(entries["refresh"].get().strip())),
validation_concurrency=max(1, int(entries["conc"].get().strip())),
max_candidates=max(10, int(entries["maxc"].get().strip())),
validation_timeout_seconds=max(2.0, float(entries["timeout"].get().strip())),
prefer_elite=bool(elite_var.get()),
kill_switch_enabled=bool(ks_var.get()),
proxy_bypass=entries["bypass"].get().strip() or Settings().proxy_bypass,
sources=src_list or Settings().sources,
@@ -557,11 +658,11 @@ def main() -> None:
raise ValueError("Port must be 1-65535")
svc.update_settings(ns)
save_settings(ns)
proxy_lbl.configure(text=f"proxy: {LISTEN_HOST}:{ns.local_port}")
proxy_lbl.configure(text=f"{LISTEN_HOST}:{ns.local_port}")
if verbose:
_log("✓ Settings saved.")
except Exception as e:
_log(f"Settings error: {e!s}")
_log(f"⚠ Settings error: {e!s}")
# ─────────────────────────────────────────────────────────────────────────
# STATUS REFRESHERS
@@ -582,8 +683,10 @@ def main() -> None:
# ─────────────────────────────────────────────────────────────────────────
def _handle(m: dict[str, Any]) -> None:
t = m.get("type")
if t == "log":
_log(str(m.get("text", "")))
elif t == "state":
running = bool(m.get("running"))
status_dot.configure(text_color=GREEN if running else RED)
@@ -592,35 +695,68 @@ def main() -> None:
if not running:
prog_bar.set(0)
phase_lbl.configure(text="idle")
countdown_lbl.configure(text="next: —", text_color=TEXT2)
tray.set_state("gray")
v_exit_ip.set("—")
v_hops.set("—")
v_rotation.set("0")
elif t == "phase":
phase_lbl.configure(text=str(m.get("phase", "")))
if m.get("phase") != "validate":
ph = str(m.get("phase", ""))
phase_lbl.configure(text=ph)
if ph == "validate":
prog_bar.configure(progress_color=YELLOW)
elif ph == "fetch":
prog_bar.configure(progress_color=ORANGE)
prog_bar.set(0.15)
elif ph == "running":
prog_bar.configure(progress_color=GREEN)
prog_bar.set(1.0)
else:
prog_bar.set(0)
elif t == "validate_progress":
d, tot = int(m.get("done", 0)), max(1, int(m.get("total", 1)))
prog_bar.set(d / tot)
elif t == "pool":
cnt = int(m.get("count", 0))
pool_lbl.configure(text=f"pool: {cnt}")
v_pool.set(str(cnt))
elif t == "real_ip":
ip = str(m.get("ip", "—"))
your_ip_lbl.configure(text=f"Your IP: {ip}")
v_real_ip.set(ip)
elif t == "firewall":
_refresh_fw(m.get("engaged"))
_refresh_fw(bool(m.get("engaged")))
elif t == "rotation":
n = str(m.get("n", 0))
rotation_lbl.configure(text=f"rot: {n}")
v_rotation.set(n)
elif t == "countdown":
secs = int(m.get("secs", 0))
if secs > 0:
m_s, s_s = divmod(secs, 60)
countdown_lbl.configure(
text=f"next: {m_s}m{s_s:02d}s",
text_color=YELLOW if secs < 30 else TEXT2,
)
else:
countdown_lbl.configure(text="next: checking…", text_color=YELLOW)
elif t == "hops":
hops = [str(x) for x in (m.get("hops") or [])]
status = str(m.get("status", "connecting"))
exit_ip = m.get("exit_ip")
_render_chain(hops, status, exit_ip)
_refresh_sysproxy()
tray.set_state({"healthy": "green", "dead": "red", "connecting": "yellow"}.get(status, "yellow"))
tray.set_state(
{"healthy": "green", "dead": "red", "connecting": "yellow"}.get(status, "yellow")
)
v_exit_ip.set(str(exit_ip) if exit_ip else "—")
v_hops.set(str(len(hops)))
def _pump() -> None:
try:
@@ -628,7 +764,7 @@ def main() -> None:
_handle(ui_q.get_nowait())
except queue.Empty:
pass
root.after(100, _pump)
root.after(80, _pump)
# ─────────────────────────────────────────────────────────────────────────
# CLOSE / MINIMIZE
@@ -638,7 +774,7 @@ def main() -> None:
tray.stop()
svc.stop()
clear_system_proxy()
if is_admin():
if is_admin() and fw_is_engaged():
fw_disengage()
root.destroy()
@@ -651,10 +787,11 @@ def main() -> None:
_refresh_sysproxy()
_refresh_fw()
_log("─" * 60)
_log("Proxy God v2 — ready.")
_log(f"Admin: {'YES' if is_admin() else 'NO (Run as Admin for kill-switch)'}")
_log(f"Log: {LOG_PATH}")
_log("Proxy God v2 — ready.")
_log(f"Admin: {'YES — kill-switch available' if is_admin() else 'NO — run as Admin for kill-switch'}")
_log(f"Log file: {LOG_PATH}")
_log("Press START to fetch, validate and chain proxies.")
_log("Use +/− in top bar to change hop count instantly.")
_log("─" * 60)
_pump()
root.mainloop()

6
proxy_chain_manager/gost_util.py
View File

@@ -68,12 +68,14 @@ def build_gost_cmd(gost: Path, listen_http: str, forwards: list[str]) -> list[st
def popen_no_window(args: list[str]) -> subprocess.Popen:
# Use DEVNULL for both pipes — GOST writes logs to stderr, leaving pipes
# unread would fill the OS buffer and deadlock the child process.
cr = getattr(subprocess, "CREATE_NO_WINDOW", 0)
return subprocess.Popen(
args,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
creationflags=cr,
)

339
proxy_chain_manager/service.py
View File

@@ -6,6 +6,7 @@ import random
import threading
import time
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from typing import Any
from .config import Settings, load_settings, save_settings
@@ -19,18 +20,18 @@ log = logging.getLogger(__name__)
Notify = Callable[[dict[str, Any]], None]
# Thread pool for blocking I/O (proxy list fetching) so we don't stall asyncio
_FETCH_POOL = ThreadPoolExecutor(max_workers=4, thread_name_prefix="fetcher")
def _filter_by_mode(pool: list[str], mode: str) -> list[str]:
"""Filter pool by obfuscation mode."""
"""Filter pool by obfuscation/protocol mode."""
if mode == "http_only":
return [u for u in pool if u.startswith("http://")]
if mode == "socks5_only":
return [u for u in pool if u.startswith("socks5://")]
if mode == "random_mix":
random.shuffle(pool)
return pool
# auto — keep all
return pool
# "random_mix" and "auto" — keep all, caller will shuffle
return list(pool)
class ChainService:
@@ -45,6 +46,12 @@ class ChainService:
self._settings = load_settings()
self._current_chain: list[str] = []
# Shuffle-and-drain pool state — never repeat a proxy within a cycle
self._available: list[str] = [] # proxies not yet used this cycle
self._used: set[str] = set() # proxies used this cycle
# Per-session blacklist: proxies that crashed GOST immediately
self._blacklist: set[str] = set()
@property
def settings(self) -> Settings:
return self._settings
@@ -61,7 +68,9 @@ class ChainService:
if self._thread and self._thread.is_alive():
return
self._stop.clear()
self._thread = threading.Thread(target=self._run_thread, name="ChainService", daemon=True)
self._thread = threading.Thread(
target=self._run_thread, name="ChainService", daemon=True
)
self._thread.start()
def stop(self) -> None:
@@ -76,6 +85,8 @@ class ChainService:
def rotate_now(self) -> None:
self._force_rotate.set()
# ─────────────────────────────────────────────────────────────────────────
def _teardown_network(self) -> None:
clear_system_proxy()
self._notify({"type": "log", "text": "System proxy cleared."})
@@ -96,19 +107,23 @@ class ChainService:
self._teardown_network()
self._notify({"type": "state", "running": False})
# ─────────────────────────────────────────────────────────────────────────
# MAIN ASYNC LOOP
# ─────────────────────────────────────────────────────────────────────────
async def _async_main(self) -> None:
self._notify({"type": "state", "running": True})
# ── GOST setup ───────────────────────────────────────────────────────
try:
gost = ensure_gost()
self._notify({"type": "log", "text": f"GOST ready: {gost}"})
except Exception as e:
self._notify({"type": "log", "text": f"GOST setup failed: {e} — check internet connection."})
self._notify({"type": "log", "text": f"GOST setup failed: {e}"})
return
# Verify GOST isn't quarantined by checking file size
if gost.stat().st_size < 10_000:
self._notify({"type": "log", "text": "GOST exe looks invalid (may be quarantined). Re-downloading..."})
self._notify({"type": "log", "text": "GOST appears quarantined. Re-downloading..."})
gost.unlink(missing_ok=True)
try:
gost = ensure_gost()
@@ -116,121 +131,143 @@ class ChainService:
self._notify({"type": "log", "text": f"GOST re-download failed: {e}"})
return
# ── Real IP ──────────────────────────────────────────────────────────
real_ip = await get_direct_ip(self._settings.ip_check_url)
if real_ip:
self._notify({"type": "real_ip", "ip": real_ip})
self._notify({"type": "log", "text": f"Your IP (without chain): {real_ip}"})
self._notify({"type": "log", "text": f"Your real IP: {real_ip}"})
else:
self._notify({"type": "log", "text": "Could not determine real IP — exit IP comparison disabled."})
self._notify({"type": "log", "text": "Could not determine real IP — leak detection disabled."})
# Engage firewall kill-switch if enabled
# ── Firewall kill-switch ──────────────────────────────────────────────
if self._settings.kill_switch_enabled:
if is_admin():
ok, msg = fw_engage(gost)
self._notify({"type": "log", "text": msg})
self._notify({"type": "firewall", "engaged": ok})
else:
self._notify({"type": "log", "text": "Kill-switch skipped — not running as Administrator."})
self._notify({"type": "log", "text": "Kill-switch skipped (not Admin)."})
self._notify({"type": "firewall", "engaged": False})
else:
self._notify({"type": "log", "text": "Kill-switch disabled in settings."})
self._notify({"type": "firewall", "engaged": False})
# ── Main rotation loop ────────────────────────────────────────────────
full_pool: list[str] = []
last_full = 0.0
pool: list[str] = []
rotation_num = 0
while not self._stop.is_set():
now = time.monotonic()
need_refresh = (
not pool
not full_pool
or now - last_full >= float(self._settings.full_refresh_seconds)
)
# If using pinned chain, skip pool management
if self._settings.use_pinned_chain and len(self._settings.pinned_chain) >= 1:
# Pinned (manual) chain mode — skip pool management
if self._settings.use_pinned_chain and self._settings.pinned_chain:
chain = list(self._settings.pinned_chain)
await self._run_chain(gost, chain, real_ip)
rotation_num += 1
self._notify({"type": "rotation", "n": rotation_num})
ok = await self._run_chain(gost, chain, real_ip)
if not ok:
await self._sleep_interruptible(10)
if self._stop.is_set():
break
continue
# ── Pool refresh ─────────────────────────────────────────────────
if need_refresh:
self._notify({"type": "phase", "phase": "fetch"})
pool = await self._build_pool()
full_pool = await self._build_pool()
last_full = time.monotonic()
self._notify({"type": "pool", "count": len(pool)})
self._available = list(full_pool)
random.shuffle(self._available)
self._used.clear()
self._notify({"type": "pool", "count": len(full_pool)})
filtered = _filter_by_mode(list(pool), self._settings.obfuscation_mode)
if len(filtered) < 2:
self._notify({
"type": "log",
"text": (
f"Pool has {len(filtered)} proxies for mode '{self._settings.obfuscation_mode}'. "
"Try 'auto' mode or increase max candidates. Retrying..."
)
})
await asyncio.sleep(30)
if not full_pool:
self._notify({"type": "log", "text": "Empty pool — retrying in 60s..."})
await self._sleep_interruptible(60)
last_full = 0.0
continue
chain = self._pick_chain(filtered)
# ── Pick next chain (shuffle-and-drain, no repeats per cycle) ────
chain = self._pick_chain()
if not chain:
# Pool exhausted for this cycle — reshuffle and restart
self._notify({"type": "log", "text": "Pool cycle complete — reshuffling for next round."})
self._available = list(full_pool)
random.shuffle(self._available)
self._used.clear()
chain = self._pick_chain()
if not chain:
await self._sleep_interruptible(15)
continue
rotation_num += 1
self._notify({"type": "rotation", "n": rotation_num})
await self._run_chain(gost, chain, real_ip)
if self._stop.is_set():
break
terminate_process(self._proc)
self._proc = None
async def _run_chain(self, gost: Any, chain: list[str], real_ip: str | None) -> None:
"""Spin up GOST with the given chain, monitor it, return when chain dies or rotation triggered."""
# ─────────────────────────────────────────────────────────────────────────
# CHAIN RUNNER
# ─────────────────────────────────────────────────────────────────────────
async def _run_chain(
self, gost: Any, chain: list[str], real_ip: str | None
) -> bool:
"""Start GOST with chain, verify exit IP, monitor until rotation/stop.
Returns True if chain ran successfully, False if it immediately failed."""
self._current_chain = list(chain)
self._notify({"type": "hops", "hops": chain, "status": "connecting"})
listen = self._settings.listen_addr()
cmd = build_gost_cmd(gost, listen, chain)
self._notify({"type": "log", "text": "Starting: " + " → ".join(self._short(h) for h in chain)})
self._notify({
"type": "log",
"text": f"Chain #{len(self._used) // max(1, self._settings.chain_length)}: "
+ " → ".join(self._short(h) for h in chain),
})
terminate_process(self._proc)
self._proc = popen_no_window(cmd)
# Brief settle time
await asyncio.sleep(2.0)
# Check GOST didn't immediately die
if self._proc.poll() is not None:
stderr = b""
try:
_, stderr = self._proc.communicate(timeout=2)
except Exception:
pass
err_msg = stderr.decode(errors="replace").strip() if stderr else "unknown error"
# GOST died immediately — blacklist these proxies
for h in chain:
self._blacklist.add(h)
self._available = [x for x in self._available if x != h]
self._notify({"type": "hops", "hops": chain, "status": "dead", "exit_ip": None})
self._notify({"type": "log", "text": f"GOST exited immediately: {err_msg}"})
return
self._notify({"type": "log", "text": "GOST exited immediately — proxies blacklisted."})
return False
local_proxy = f"http://{listen}"
exit_ip = await check_chain_exit_ip(
local_proxy,
self._settings.ip_check_url,
min(25.0, self._settings.validation_timeout_seconds + 10.0),
)
timeout = min(30.0, self._settings.validation_timeout_seconds + 12.0)
exit_ip = await check_chain_exit_ip(local_proxy, self._settings.ip_check_url, timeout)
if not exit_ip:
self._notify({"type": "hops", "hops": chain, "status": "dead", "exit_ip": None})
self._notify({"type": "log", "text": "Chain failed IP check. Rotating."})
self._notify({"type": "log", "text": "Chain IP check failed. Rotating."})
terminate_process(self._proc)
self._proc = None
return
return False
if real_ip and exit_ip == real_ip:
self._notify({"type": "hops", "hops": chain, "status": "dead", "exit_ip": exit_ip})
self._notify({"type": "log", "text": f"Exit IP {exit_ip} == real IP! Chain leaking. Rotating."})
self._notify({"type": "log", "text": f"Leak! Exit={exit_ip} == real IP. Rotating."})
terminate_process(self._proc)
self._proc = None
return
return False
self._notify({"type": "hops", "hops": chain, "status": "healthy", "exit_ip": exit_ip})
self._notify({"type": "log", "text": f"Chain healthy — Exit IP: {exit_ip}"})
self._notify({"type": "log", "text": f"✓ Chain healthy — Exit IP: {exit_ip}"})
set_system_proxy(
self._settings.local_host,
self._settings.local_port,
@@ -238,32 +275,127 @@ class ChainService:
)
self._notify({"type": "log", "text": f"System proxy → {self._settings.listen_addr()}"})
# Monitor loop
refresh_deadline = time.monotonic() + float(self._settings.full_refresh_seconds)
while not self._stop.is_set() and time.monotonic() < refresh_deadline:
w = await self._wait_health_interval()
if w in ("stop", "rotate"):
# ── Health monitor loop ───────────────────────────────────────────────
while not self._stop.is_set():
result = await self._wait_health_interval()
if result in ("stop", "rotate"):
break
if self._proc.poll() is not None:
self._notify({"type": "log", "text": "GOST process died; rebuilding."})
if self._proc is None or self._proc.poll() is not None:
self._notify({"type": "log", "text": "GOST process died — rebuilding."})
break
exit_ip = await check_chain_exit_ip(
local_proxy,
self._settings.ip_check_url,
min(25.0, self._settings.validation_timeout_seconds + 10.0),
)
self._notify({"type": "log", "text": "Health check..."})
exit_ip = await check_chain_exit_ip(local_proxy, self._settings.ip_check_url, timeout)
if not exit_ip or (real_ip and exit_ip == real_ip):
self._notify({"type": "hops", "hops": chain, "status": "dead", "exit_ip": exit_ip})
self._notify({"type": "log", "text": "Health check failed; rotating."})
self._notify({"type": "log", "text": "Health check failed — rotating."})
break
self._notify({"type": "hops", "hops": chain, "status": "healthy", "exit_ip": exit_ip})
self._notify({"type": "log", "text": f"✓ Still healthy — Exit IP: {exit_ip}"})
terminate_process(self._proc)
self._proc = None
return True
# ─────────────────────────────────────────────────────────────────────────
# POOL MANAGEMENT
# ─────────────────────────────────────────────────────────────────────────
def _pick_chain(self) -> list[str]:
"""Pick chain_length proxies from _available (shuffle-and-drain).
Filters out blacklisted proxies. Marks picked ones as used."""
s = self._settings
k = max(1, s.chain_length)
# Apply mode filter and remove blacklisted entries
candidates = [
u for u in self._available
if u not in self._blacklist
]
# Apply obfuscation mode filter
if s.obfuscation_mode == "http_only":
candidates = [u for u in candidates if u.startswith("http://")]
elif s.obfuscation_mode == "socks5_only":
candidates = [u for u in candidates if u.startswith("socks5://")]
elif s.obfuscation_mode == "random_mix":
random.shuffle(candidates)
if len(candidates) < k:
return []
picked = candidates[:k]
# Remove picked from available
picked_set = set(picked)
self._available = [u for u in self._available if u not in picked_set]
self._used.update(picked)
return picked
async def _build_pool(self) -> list[str]:
"""Fetch and validate proxies. Runs blocking I/O in thread pool."""
s = self._settings
loop = asyncio.get_event_loop()
# Fetch all sources concurrently in thread pool (they are blocking)
async def _fetch_one(url: str) -> list[str]:
try:
rows = await loop.run_in_executor(
_FETCH_POOL, lambda: fetch_proxy_json(url)
)
entries = normalize_entries(rows, s.prefer_elite)
self._notify({"type": "log", "text": f"Fetched {len(entries)} from source."})
return entries
except Exception as e:
self._notify({"type": "log", "text": f"Fetch error: {e!s}"})
return []
results = await asyncio.gather(*(_fetch_one(u) for u in s.sources))
raw_urls: list[str] = []
for chunk in results:
raw_urls.extend(chunk)
if not raw_urls:
self._notify({"type": "log", "text": "No proxies fetched from any source!"})
return []
# Deduplicate, remove blacklisted, shuffle before capping
seen: set[str] = set()
unique: list[str] = []
for u in raw_urls:
if u not in seen and u not in self._blacklist:
seen.add(u)
unique.append(u)
random.shuffle(unique)
if len(unique) > s.max_candidates:
unique = unique[: s.max_candidates]
self._notify({"type": "phase", "phase": "validate"})
self._notify({"type": "log", "text": f"Validating {len(unique)} candidates..."})
def on_prog(done: int, total: int) -> None:
self._notify({"type": "validate_progress", "done": done, "total": total})
good = await validate_proxies(
unique,
s.ip_check_url,
s.validation_concurrency,
s.validation_timeout_seconds,
on_progress=on_prog,
)
random.shuffle(good)
self._notify({"type": "log", "text": f"Valid proxies: {len(good)} / {len(unique)}"})
self._notify({"type": "phase", "phase": "running"})
return good
# ─────────────────────────────────────────────────────────────────────────
# UTILITIES
# ─────────────────────────────────────────────────────────────────────────
async def _wait_health_interval(self) -> str | None:
"""Sleep for health_check_seconds. Returns 'stop', 'rotate', or None."""
total = float(self._settings.health_check_seconds)
end = time.monotonic() + total
while time.monotonic() < end:
@@ -271,68 +403,27 @@ class ChainService:
return "stop"
if self._force_rotate.is_set():
self._force_rotate.clear()
self._notify({"type": "log", "text": "Manual rotate."})
self._notify({"type": "log", "text": "Manual rotate triggered."})
return "rotate"
await asyncio.sleep(0.2)
remaining = end - time.monotonic()
self._notify({"type": "countdown", "secs": max(0, int(remaining))})
await asyncio.sleep(1.0)
self._notify({"type": "countdown", "secs": 0})
return None
def _pick_chain(self, pool: list[str]) -> list[str]:
k = min(self._settings.chain_length, len(pool))
return random.sample(pool, k=k)
async def _sleep_interruptible(self, seconds: float) -> None:
end = time.monotonic() + seconds
while time.monotonic() < end:
if self._stop.is_set():
return
await asyncio.sleep(0.2)
@staticmethod
def _short(url: str) -> str:
return (
url = (
url.replace("http://", "")
.replace("socks5://", "s5://")
.replace("socks4://", "s4://")
.replace("https://", "https://")
.replace("https://", "")
)
async def _build_pool(self) -> list[str]:
s = self._settings
raw_urls: list[str] = []
for url in s.sources:
if self._stop.is_set():
break
try:
rows = fetch_proxy_json(url)
entries = normalize_entries(rows, s.prefer_elite)
raw_urls.extend(entries)
self._notify({"type": "log", "text": f"Fetched {len(rows)} entries from source."})
except Exception as e:
self._notify({"type": "log", "text": f"Fetch error: {e!s}"})
if not raw_urls:
self._notify({"type": "log", "text": "No proxies fetched from any source!"})
return []
# Deduplicate
seen: set[str] = set()
unique: list[str] = []
for u in raw_urls:
if u not in seen:
seen.add(u)
unique.append(u)
raw_urls = unique
if len(raw_urls) > s.max_candidates:
raw_urls = random.sample(raw_urls, k=s.max_candidates)
self._notify({"type": "phase", "phase": "validate"})
self._notify({"type": "log", "text": f"Validating {len(raw_urls)} candidates..."})
def on_prog(done: int, total: int) -> None:
self._notify({"type": "validate_progress", "done": done, "total": total})
good = await validate_proxies(
raw_urls,
s.ip_check_url,
s.validation_concurrency,
s.validation_timeout_seconds,
on_progress=on_prog,
)
random.shuffle(good)
self._notify({"type": "log", "text": f"Valid: {len(good)} / {len(raw_urls)}"})
self._notify({"type": "phase", "phase": "running"})
return good
return url[:30] + "…" if len(url) > 32 else url

107
proxy_chain_manager/validator.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import json
import logging
from typing import Callable
@@ -8,6 +9,33 @@ import httpx
log = logging.getLogger(__name__)
# Secondary fallback endpoints for IP resolution
_IP_FALLBACKS = [
"https://api.ipify.org?format=json",
"https://httpbin.org/ip",
"https://ifconfig.me/ip",
]
def _parse_ip(text: str) -> str | None:
"""Parse an IP string from JSON or plain-text response body."""
body = text.strip()
if not body:
return None
if body.startswith("{"):
try:
d = json.loads(body)
ip = d.get("ip") or d.get("origin") or d.get("query")
return str(ip).split(",")[0].strip() if ip else None
except Exception:
return None
# Plain-text: must look like an IP (short, no HTML)
if len(body) < 50 and "<" not in body:
candidate = body.split()[0].strip()
if candidate.count(".") >= 3 or ":" in candidate:
return candidate
return None
async def validate_proxies(
proxy_urls: list[str],
@@ -38,9 +66,14 @@ async def validate_proxies(
async def _check_one(proxy_url: str, check_url: str, timeout_seconds: float) -> bool:
t = httpx.Timeout(timeout_seconds, connect=min(8.0, timeout_seconds))
try:
t = httpx.Timeout(timeout_seconds, connect=min(8.0, timeout_seconds))
async with httpx.AsyncClient(proxy=proxy_url, timeout=t, verify=True, follow_redirects=True) as c:
async with httpx.AsyncClient(
proxy=proxy_url,
timeout=t,
verify=False, # many proxies have self-signed or no TLS
follow_redirects=True,
) as c:
r = await c.get(check_url)
return r.status_code == 200 and len(r.content) > 0
except Exception:
@@ -48,49 +81,55 @@ async def _check_one(proxy_url: str, check_url: str, timeout_seconds: float) ->
async def check_chain_exit_ip(
listen_proxy: str, check_url: str, timeout_seconds: float
listen_proxy: str,
check_url: str,
timeout_seconds: float,
) -> str | None:
"""Query the IP-check URL through the local chain proxy.
Returns the exit IP string on success, or None on failure."""
Returns the exit IP string on success, or None on failure.
Tries the configured URL first, then falls back to alternatives."""
urls = [check_url] + [u for u in _IP_FALLBACKS if u != check_url]
t = httpx.Timeout(timeout_seconds, connect=min(10.0, timeout_seconds))
try:
t = httpx.Timeout(timeout_seconds, connect=min(10.0, timeout_seconds))
async with httpx.AsyncClient(proxy=listen_proxy, timeout=t, verify=True, follow_redirects=True) as c:
r = await c.get(check_url)
if r.status_code != 200:
return None
body = r.text.strip()
# ipify returns {"ip":"x.x.x.x"} in json mode
if body.startswith("{"):
import json
async with httpx.AsyncClient(
proxy=listen_proxy,
timeout=t,
verify=False,
follow_redirects=True,
) as c:
for url in urls:
try:
return json.loads(body).get("ip") or json.loads(body).get("origin")
r = await c.get(url)
if r.status_code == 200:
ip = _parse_ip(r.text)
if ip:
return ip
except Exception:
return None
# plain text mode
if body and len(body) < 50:
return body
return None
continue
except Exception:
return None
pass
return None
async def get_direct_ip(check_url: str, timeout_seconds: float = 15.0) -> str | None:
"""Get our own exit IP without the proxy chain (so we can compare)."""
urls = [check_url] + [u for u in _IP_FALLBACKS if u != check_url]
t = httpx.Timeout(timeout_seconds)
try:
t = httpx.Timeout(timeout_seconds)
async with httpx.AsyncClient(timeout=t, verify=True, follow_redirects=True) as c:
r = await c.get(check_url)
if r.status_code != 200:
return None
body = r.text.strip()
if body.startswith("{"):
import json
async with httpx.AsyncClient(
timeout=t,
verify=False,
follow_redirects=True,
) as c:
for url in urls:
try:
return json.loads(body).get("ip") or json.loads(body).get("origin")
r = await c.get(url)
if r.status_code == 200:
ip = _parse_ip(r.text)
if ip:
return ip
except Exception:
return None
if body and len(body) < 50:
return body
return None
continue
except Exception:
return None
pass
return None