460 lines
18 KiB
Python
460 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import shutil
|
|
from dataclasses import asdict, dataclass, field
|
|
from pathlib import Path
|
|
from urllib.parse import quote, unquote, urlparse, urlunparse
|
|
|
|
from .paths import app_data_dir
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# ── Settings schema version ───────────────────────────────────────────────────
|
|
# Bump this whenever a breaking field rename / type change is made so
|
|
# migrate() can upgrade old settings.json files safely.
|
|
SETTINGS_SCHEMA_VERSION = 1
|
|
|
|
# !! SECURITY NOTE !!
|
|
# settings.json is stored in plaintext at %LOCALAPPDATA%\ProxyChainManager\.
|
|
# It may contain proxy credentials (URLs with user:password). Do NOT log
|
|
# raw settings dicts — use _mask() to redact credential fields before logging.
|
|
|
|
|
|
def _mask(d: dict) -> dict:
|
|
"""Return a copy of settings dict with credential-bearing fields redacted."""
|
|
redact_keys = {"manual_exit_proxy", "pinned_chain", "sources"}
|
|
out: dict = {}
|
|
for k, v in d.items():
|
|
if k in redact_keys:
|
|
if isinstance(v, list):
|
|
out[k] = [redact_proxy_url(str(x)) for x in v]
|
|
else:
|
|
out[k] = redact_proxy_url(str(v))
|
|
else:
|
|
out[k] = v
|
|
return out
|
|
|
|
|
|
def _host_port_for_url(host: str, port: int | None) -> str:
|
|
"""``host`` and optional ``port`` as used in proxy URLs (bracket IPv6 literals)."""
|
|
if not host:
|
|
return ""
|
|
if ":" in host and "." not in host:
|
|
inner = f"[{host}]"
|
|
else:
|
|
inner = host
|
|
return f"{inner}:{port}" if port else inner
|
|
|
|
|
|
# Always bind the local forwarder to loopback. VM clones and NIC/IP changes (DHCP)
|
|
# do not affect 127.0.0.1; avoid storing LAN IPs or hostnames for the listener.
|
|
LISTEN_HOST = "127.0.0.1"
|
|
|
|
|
|
def normalize_proxy_url(raw: str) -> str:
|
|
"""Strip, add http:// if missing. Empty input → empty string."""
|
|
t = (raw or "").strip()
|
|
if not t:
|
|
return ""
|
|
if "://" not in t:
|
|
t = "http://" + t
|
|
return t.rstrip("/")
|
|
|
|
|
|
def split_proxy_for_edit(raw: str) -> tuple[str, str, str]:
|
|
"""Split stored proxy URL into (url_without_credentials, username, password).
|
|
|
|
Used to populate fixed-exit / form fields. Credentials are percent-decoded.
|
|
"""
|
|
t = normalize_proxy_url((raw or "").strip())
|
|
if not t:
|
|
return "", "", ""
|
|
p = urlparse(t)
|
|
if not p.hostname:
|
|
return t, "", ""
|
|
host = p.hostname
|
|
port = p.port
|
|
scheme = p.scheme or "http"
|
|
base = f"{scheme}://{_host_port_for_url(host, port)}"
|
|
u = unquote(p.username) if p.username else ""
|
|
pw = unquote(p.password) if p.password else ""
|
|
return base, u, pw
|
|
|
|
|
|
def merge_proxy_credentials(raw_url: str, username: str = "", password: str = "") -> str:
|
|
"""Build a single proxy URL: host/port from ``raw_url`` (any embedded auth ignored) plus optional auth.
|
|
|
|
User and password are URL-encoded for ``@`` and ``:`` inside values.
|
|
"""
|
|
base = normalize_proxy_url((raw_url or "").strip())
|
|
if not base:
|
|
return ""
|
|
p = urlparse(base)
|
|
if not p.hostname:
|
|
return base
|
|
host = p.hostname
|
|
port = p.port
|
|
scheme = p.scheme or "http"
|
|
user = (username or "").strip()
|
|
pw = (password or "").strip()
|
|
if not user and not pw:
|
|
# Leave credentials embedded in URL when form fields are blank (paste user:pass@host).
|
|
return base
|
|
hp = _host_port_for_url(host, port)
|
|
netloc = f"{quote(user, safe='')}:{quote(pw, safe='')}@{hp}"
|
|
return urlunparse((scheme, netloc, "", "", "", "")).rstrip("/")
|
|
|
|
|
|
def parse_exit_proxy_input(
|
|
raw: str, default_scheme: str = "socks5"
|
|
) -> tuple[str, str, str]:
|
|
"""Parse any sane paste format into ``(base_url, user, password)``.
|
|
|
|
Accepts (whitespace-trimmed, default scheme = SOCKS5 when none given):
|
|
|
|
``host:port``
|
|
``host:port:user:password``
|
|
``host:port:user:password:session_or_extra`` (extras folded into password)
|
|
``host:port user password`` (tab/space-separated)
|
|
``user:password@host:port``
|
|
``scheme://host:port``
|
|
``scheme://user:password@host:port``
|
|
|
|
User and password may contain ``@``, ``:`` and arbitrary "words" — those
|
|
are URL-encoded automatically when ``merge_proxy_credentials`` rebuilds
|
|
the URL, so pasting plain text is safe.
|
|
"""
|
|
t = (raw or "").strip()
|
|
if not t:
|
|
return "", "", ""
|
|
|
|
if "://" in t:
|
|
# Already a full URL — defer to the existing splitter.
|
|
return split_proxy_for_edit(t)
|
|
|
|
if "@" in t:
|
|
return split_proxy_for_edit(f"{default_scheme}://{t}")
|
|
|
|
if "\t" in t or any(c in t for c in (" ",)):
|
|
# "host:port user pass" or "host:port\tuser\tpass"
|
|
parts = [p for p in t.replace("\t", " ").split() if p]
|
|
if len(parts) >= 3:
|
|
hp, user, pw = parts[0], parts[1], " ".join(parts[2:])
|
|
return f"{default_scheme}://{hp}".rstrip("/"), user, pw
|
|
if len(parts) == 2:
|
|
return f"{default_scheme}://{parts[0]}".rstrip("/"), "", ""
|
|
t = parts[0]
|
|
|
|
bits = t.split(":")
|
|
if len(bits) >= 4:
|
|
host, port, user = bits[0], bits[1], bits[2]
|
|
password = ":".join(bits[3:]) # passwords containing ":" survive
|
|
return f"{default_scheme}://{host}:{port}", user, password
|
|
if len(bits) == 2:
|
|
return f"{default_scheme}://{bits[0]}:{bits[1]}", "", ""
|
|
# 1 token, or 3-token oddity — best-effort; treat as host[:port] only.
|
|
return f"{default_scheme}://{t}".rstrip("/"), "", ""
|
|
|
|
|
|
def redact_proxy_url(url: str) -> str:
|
|
"""Same URL shape for logs/UI, with credentials replaced by ``***``."""
|
|
t = (url or "").strip()
|
|
if not t:
|
|
return ""
|
|
if "://" not in t:
|
|
t = "http://" + t
|
|
p = urlparse(t.rstrip("/"))
|
|
if not p.hostname:
|
|
return t
|
|
has_auth = bool(p.username) or bool(p.password)
|
|
if not has_auth:
|
|
return t.rstrip("/")
|
|
scheme = p.scheme or "http"
|
|
hp = _host_port_for_url(p.hostname, p.port)
|
|
netloc = f"***:***@{hp}"
|
|
return urlunparse((scheme, netloc, "", "", "", "")).rstrip("/")
|
|
|
|
|
|
OBFUSCATION_MODES = ["auto", "http_only", "socks5_only", "random_mix"]
|
|
OBFUSCATION_LABELS = {
|
|
"auto": "Auto (best available)",
|
|
"http_only": "HTTP only",
|
|
"socks5_only": "SOCKS5 only",
|
|
"random_mix": "Random Mix (max noise)",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Settings:
|
|
# Schema version — used by migrate() to upgrade old settings.json files.
|
|
settings_version: int = SETTINGS_SCHEMA_VERSION
|
|
|
|
# ── network ──────────────────────────────────────────────────────────
|
|
local_host: str = LISTEN_HOST
|
|
local_port: int = 18888
|
|
|
|
# ── chain ─────────────────────────────────────────────────────────────
|
|
chain_length: int = 3 # number of hops (2-8)
|
|
obfuscation_mode: str = "auto" # see OBFUSCATION_MODES
|
|
use_pinned_chain: bool = True # use manually ordered chain from Chain Builder
|
|
pinned_chain: list[str] = field(default_factory=list) # user-ordered hop list
|
|
validate_pinned_on_start: bool = True # quick TCP probe of each pinned hop before chaining
|
|
max_browser_relaunches: int = 5 # cap auto-relaunch attempts so a broken profile doesn't loop
|
|
# Fixed last hop only (ignored when use_pinned_chain is True — full manual chain wins)
|
|
manual_exit_proxy: str = ""
|
|
|
|
# ── timing ────────────────────────────────────────────────────────────
|
|
health_check_seconds: int = 180 # how often to re-verify exit IP
|
|
full_refresh_seconds: int = 1800 # how often to re-fetch/re-validate pool
|
|
|
|
# ── validation ────────────────────────────────────────────────────────
|
|
validation_concurrency: int = 64
|
|
max_candidates: int = 200 # cap before validation (shuffled, so random sample)
|
|
min_pool_size: int = 20 # stop validating once this many good proxies found
|
|
validation_timeout_seconds: float = 12.0
|
|
prefer_elite: bool = False
|
|
|
|
# ── security ──────────────────────────────────────────────────────────
|
|
kill_switch_enabled: bool = True # engage Windows Firewall kill-switch when running
|
|
proxy_bypass: str = "localhost;127.*;10.*;192.168.*;<local>"
|
|
|
|
# ── privacy / device hardening (Privacy tab) ───────────────────────────
|
|
mac_spoof_enabled: bool = False # randomize NIC MAC while chain runs (Admin)
|
|
mac_rotate_on_chain_rotate: bool = False # re-randomize MAC on every chain rotation
|
|
spoof_hostname_enabled: bool = False # temporary computer name while chain runs (Admin)
|
|
flush_dns_on_rotate: bool = True # ipconfig /flushdns on each rotation
|
|
disable_ipv6_while_active: bool = False # disable IPv6 bindings while chain runs (Admin)
|
|
harden_webrtc_enabled: bool = False # Chrome/Edge WebRTC policy (Admin)
|
|
lan_lockdown_enabled: bool = False # kill LLMNR/NetBIOS/mDNS while chain runs
|
|
telemetry_kill_enabled: bool = False # disable DiagTrack + Activity History (Admin)
|
|
|
|
# ── hardened browser ────────────────────────────────────────────────────
|
|
firefox_path: str = ""
|
|
firefox_profile_dir: str = ""
|
|
browser_clear_on_close: bool = True
|
|
browser_disposable_profile: bool = False
|
|
browser_kill_on_chain_drop: bool = True
|
|
browser_auto_relaunch: bool = False
|
|
browser_lock_managed_profile: bool = True
|
|
browser_force_proxy: bool = True
|
|
browser_disable_webrtc: bool = True
|
|
browser_resist_fingerprinting: bool = True
|
|
browser_disable_telemetry: bool = True
|
|
browser_first_party_isolation: bool = True
|
|
browser_strict_tracking_protection: bool = True
|
|
browser_timezone_utc: bool = True
|
|
browser_persona: str = "blend_windows_chrome"
|
|
browser_cookie_mode: str = "block_third_party"
|
|
|
|
# ── sources ───────────────────────────────────────────────────────────
|
|
sources: list[str] = field(
|
|
default_factory=lambda: [
|
|
"https://cdn.jsdelivr.net/gh/proxifly/free-proxy-list@main/proxies/protocols/http/data.json",
|
|
"https://cdn.jsdelivr.net/gh/proxifly/free-proxy-list@main/proxies/protocols/socks5/data.json",
|
|
"https://cdn.jsdelivr.net/gh/proxifly/free-proxy-list@main/proxies/protocols/https/data.json",
|
|
]
|
|
)
|
|
ip_check_url: str = "https://api.ipify.org?format=json"
|
|
|
|
def listen_addr(self) -> str:
|
|
return f"{self.local_host}:{self.local_port}"
|
|
|
|
def settings_path(self) -> Path:
|
|
return app_data_dir() / "settings.json"
|
|
|
|
|
|
def _is_safe_https_url(url: str) -> bool:
|
|
"""Return True for HTTPS URLs pointing at public hosts (not RFC-1918/link-local)."""
|
|
try:
|
|
p = urlparse(url)
|
|
if p.scheme != "https":
|
|
return False
|
|
host = p.hostname or ""
|
|
if not host:
|
|
return False
|
|
import ipaddress as _ip
|
|
try:
|
|
addr = _ip.ip_address(host)
|
|
return not (addr.is_private or addr.is_loopback or addr.is_link_local)
|
|
except ValueError:
|
|
pass
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def migrate(raw: dict) -> dict:
|
|
"""Upgrade a raw settings dict from any schema version to the current one.
|
|
|
|
Each version block transforms ``raw`` in-place and bumps
|
|
``settings_version``. New fields are left absent so the dataclass
|
|
default takes effect.
|
|
"""
|
|
ver = int(raw.get("settings_version", 0))
|
|
# v0 → v1: no structural changes; just stamp the version
|
|
if ver < 1:
|
|
raw["settings_version"] = 1
|
|
return raw
|
|
|
|
|
|
def sanitize_settings(s: Settings) -> tuple[Settings, bool]:
|
|
"""Clamp invalid values from hand-edited JSON. Returns (settings, changed)."""
|
|
changed = False
|
|
try:
|
|
cl = int(s.chain_length)
|
|
if not (1 <= cl <= 8):
|
|
s.chain_length = max(1, min(8, cl))
|
|
changed = True
|
|
except (TypeError, ValueError):
|
|
s.chain_length = 3
|
|
changed = True
|
|
try:
|
|
p = int(s.local_port)
|
|
if not (1 <= p <= 65535):
|
|
s.local_port = max(1, min(65535, p))
|
|
changed = True
|
|
except (TypeError, ValueError):
|
|
s.local_port = 18888
|
|
changed = True
|
|
try:
|
|
hc = int(s.health_check_seconds)
|
|
if hc < 10:
|
|
s.health_check_seconds = 10
|
|
changed = True
|
|
elif hc > 3600:
|
|
s.health_check_seconds = 3600
|
|
changed = True
|
|
except (TypeError, ValueError):
|
|
s.health_check_seconds = 180
|
|
changed = True
|
|
try:
|
|
fr = int(s.full_refresh_seconds)
|
|
if fr < 60:
|
|
s.full_refresh_seconds = 60
|
|
changed = True
|
|
elif fr > 86400:
|
|
s.full_refresh_seconds = 86400
|
|
changed = True
|
|
except (TypeError, ValueError):
|
|
s.full_refresh_seconds = 1800
|
|
changed = True
|
|
try:
|
|
if int(s.validation_concurrency) < 1:
|
|
s.validation_concurrency = 1
|
|
changed = True
|
|
except (TypeError, ValueError):
|
|
s.validation_concurrency = 64
|
|
changed = True
|
|
try:
|
|
if int(s.max_candidates) < 10:
|
|
s.max_candidates = 10
|
|
changed = True
|
|
except (TypeError, ValueError):
|
|
s.max_candidates = 200
|
|
changed = True
|
|
try:
|
|
mps = int(s.min_pool_size)
|
|
if mps < 3:
|
|
s.min_pool_size = 3
|
|
changed = True
|
|
elif mps > 500:
|
|
s.min_pool_size = 500
|
|
changed = True
|
|
except (TypeError, ValueError):
|
|
s.min_pool_size = 20
|
|
changed = True
|
|
if not hasattr(s, "min_pool_size"):
|
|
s.min_pool_size = 20
|
|
changed = True
|
|
try:
|
|
vt = float(s.validation_timeout_seconds)
|
|
if vt < 2.0:
|
|
s.validation_timeout_seconds = 2.0
|
|
changed = True
|
|
elif vt > 120.0:
|
|
s.validation_timeout_seconds = 120.0
|
|
changed = True
|
|
except (TypeError, ValueError):
|
|
s.validation_timeout_seconds = 12.0
|
|
changed = True
|
|
# Empty list survives "all(isinstance…)" — would skip all fetches and yield an empty pool.
|
|
if not isinstance(s.sources, list) or not all(isinstance(x, str) for x in s.sources) or len(s.sources) == 0:
|
|
s.sources = list(Settings().sources)
|
|
changed = True
|
|
else:
|
|
safe = [u for u in s.sources if _is_safe_https_url(u)]
|
|
if len(safe) < len(s.sources):
|
|
log.warning("Removed %d unsafe source URL(s) from settings.", len(s.sources) - len(safe))
|
|
s.sources = safe if safe else list(Settings().sources)
|
|
changed = True
|
|
if s.ip_check_url and not _is_safe_https_url(s.ip_check_url):
|
|
log.warning("ip_check_url looks unsafe (%s) — reverting to default.", s.ip_check_url)
|
|
s.ip_check_url = Settings().ip_check_url
|
|
changed = True
|
|
return s, changed
|
|
|
|
|
|
def normalize_listen_host(s: Settings) -> tuple[Settings, bool]:
|
|
"""Force loopback bind so LAN/DHCP IP changes (e.g. cloned VMs) never break the listener."""
|
|
if s.local_host == LISTEN_HOST:
|
|
return s, False
|
|
s.local_host = LISTEN_HOST
|
|
return s, True
|
|
|
|
|
|
def load_settings() -> Settings:
|
|
p = app_data_dir() / "settings.json"
|
|
if not p.is_file():
|
|
return Settings()
|
|
try:
|
|
raw = json.loads(p.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
corrupt = p.with_suffix(".json.corrupt")
|
|
log.warning("settings.json unreadable (%s) — backing up to %s and using defaults.", exc, corrupt)
|
|
try:
|
|
shutil.copy2(p, corrupt)
|
|
except OSError:
|
|
pass
|
|
return Settings()
|
|
raw = migrate(raw)
|
|
s = Settings()
|
|
for k, v in raw.items():
|
|
if hasattr(s, k):
|
|
setattr(s, k, v)
|
|
if (
|
|
not isinstance(s.sources, list)
|
|
or not all(isinstance(x, str) for x in s.sources)
|
|
or len(s.sources) == 0
|
|
):
|
|
s.sources = list(Settings().sources)
|
|
if s.obfuscation_mode not in OBFUSCATION_MODES:
|
|
s.obfuscation_mode = "auto"
|
|
if not isinstance(s.pinned_chain, list):
|
|
s.pinned_chain = []
|
|
if not hasattr(s, "manual_exit_proxy") or not isinstance(s.manual_exit_proxy, str):
|
|
s.manual_exit_proxy = ""
|
|
s, ch1 = sanitize_settings(s)
|
|
s, ch2 = normalize_listen_host(s)
|
|
changed = ch1 or ch2
|
|
if changed:
|
|
try:
|
|
p.write_text(json.dumps(asdict(s), indent=2), encoding="utf-8")
|
|
except OSError:
|
|
pass
|
|
return s
|
|
|
|
|
|
def save_settings(s: Settings) -> None:
|
|
s, _ = sanitize_settings(s)
|
|
s, _ = normalize_listen_host(s)
|
|
p = app_data_dir() / "settings.json"
|
|
# Backup before overwriting so a crash mid-write doesn't corrupt settings.
|
|
if p.is_file():
|
|
try:
|
|
shutil.copy2(p, p.with_suffix(".json.bak"))
|
|
except OSError as exc:
|
|
log.warning("Could not back up settings.json: %s", exc)
|
|
p.write_text(json.dumps(asdict(s), indent=2), encoding="utf-8")
|