proxy-god/proxy_chain_manager/config.py

from __future__ import annotations

import json
from dataclasses import asdict, dataclass, field
from pathlib import Path

from .paths import app_data_dir

# 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("/")


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:
    # ── 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 = False        # use manually ordered chain
    pinned_chain: list[str] = field(default_factory=list)  # user-ordered hop list
    # Fixed last hop only (ignored when use_pinned_chain is True — full manual chain wins)
    manual_exit_proxy: str = ""

    # ── 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 = 400
    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>"

    # ── 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 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:
        if int(s.health_check_seconds) < 10:
            s.health_check_seconds = 10
            changed = True
    except (TypeError, ValueError):
        s.health_check_seconds = 180
        changed = True
    try:
        if int(s.full_refresh_seconds) < 60:
            s.full_refresh_seconds = 60
            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 = 400
        changed = True
    try:
        if float(s.validation_timeout_seconds) < 2.0:
            s.validation_timeout_seconds = 2.0
            changed = True
    except (TypeError, ValueError):
        s.validation_timeout_seconds = 12.0
        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):
        return Settings()
    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):
        s.sources = 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"
    p.write_text(json.dumps(asdict(s), indent=2), encoding="utf-8")