proxy-god/proxy_chain_manager/config.py

from __future__ import annotations

import json
from dataclasses import asdict, dataclass, field
from pathlib import Path
from urllib.parse import quote, unquote, urlparse, urlunparse

from .paths import app_data_dir


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