proxy-god/proxy_chain_manager/vpn_detect.py

"""Detect whether a VPN tunnel is active on Windows (any provider)."""
from __future__ import annotations

import glob
import logging
import re
import subprocess
from dataclasses import dataclass, field
from pathlib import Path

log = logging.getLogger(__name__)

# Adapter description / name substrings (case-insensitive)
_VPN_ADAPTER_HINTS = (
    "nordlynx", "nordvpn", "openvpn", "wireguard", "wintun", "tap-windows",
    "tailscale", "zerotier", "cisco anyconnect", "fortinet", "pulse secure",
    "globalprotect", "softether", "proton", "mullvad", "expressvpn",
    "surfshark", "private internet", "pia ", "windscribe", "hotspot shield",
    "tunnel", "vpn",
)

# Executables to whitelist in kill-switch when present
_VPN_EXE_GLOBS: list[str] = [
    r"C:\Program Files\NordVPN\*.exe",
    r"C:\Program Files\NordUpdater\*.exe",
    r"C:\Program Files\NordVPN\NordSec ThreatProtection\*.exe",
    r"C:\Program Files\OpenVPN\bin\*.exe",
    r"C:\Program Files\OpenVPN Connect\*.exe",
    r"C:\Program Files\WireGuard\*.exe",
    r"C:\Program Files\Proton\VPN\*.exe",
    r"C:\Program Files\Mullvad VPN\*.exe",
    r"C:\Program Files\ExpressVPN\*.exe",
    r"C:\Program Files\Surfshark\*.exe",
    r"C:\Program Files\Private Internet Access\*.exe",
    r"C:\Program Files\Tailscale\*.exe",
    r"C:\Program Files\ZeroTier\One\*.exe",
]

_PROVIDER_FROM_ADAPTER: list[tuple[str, str]] = [
    ("nordlynx", "NordVPN"),
    ("nordvpn", "NordVPN"),
    ("wireguard", "WireGuard"),
    ("wintun", "WireGuard"),
    ("openvpn", "OpenVPN"),
    ("proton", "Proton VPN"),
    ("mullvad", "Mullvad"),
    ("expressvpn", "ExpressVPN"),
    ("surfshark", "Surfshark"),
    ("tailscale", "Tailscale"),
    ("zerotier", "ZeroTier"),
    ("tap-windows", "OpenVPN/TAP"),
    ("globalprotect", "GlobalProtect"),
    ("fortinet", "FortiClient"),
    ("cisco", "Cisco VPN"),
]


@dataclass
class VpnStatus:
    active: bool = False
    label: str = "Direct (no VPN)"
    adapter: str = ""
    adapters: list[str] = field(default_factory=list)

    def short_label(self) -> str:
        if not self.active:
            return "Direct"
        return self.label


def _run_ps(script: str, timeout: float = 12.0) -> str:
    try:
        r = subprocess.run(
            ["powershell", "-NoProfile", "-NonInteractive", "-Command", script],
            capture_output=True,
            text=True,
            timeout=timeout,
            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
        )
        return (r.stdout or "").strip()
    except Exception as e:
        log.debug("vpn_detect powershell failed: %s", e)
        return ""


def _match_provider(name: str) -> str:
    low = name.lower()
    for hint, label in _PROVIDER_FROM_ADAPTER:
        if hint in low:
            return label
    if "vpn" in low or "tunnel" in low:
        return "VPN"
    return "VPN"


def detect_vpn() -> VpnStatus:
    """Inspect up network adapters for VPN/tunnel interfaces."""
    out = _run_ps(
        "Get-NetAdapter | Where-Object { $_.Status -eq 'Up' } | "
        "Select-Object -ExpandProperty Name"
    )
    if not out:
        # Fallback: netsh
        try:
            r = subprocess.run(
                ["netsh", "interface", "show", "interface"],
                capture_output=True,
                text=True,
                timeout=10,
                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
            )
            lines = (r.stdout or "").splitlines()
            names = []
            for ln in lines[3:]:
                parts = ln.split()
                if len(parts) >= 4 and parts[0] == "Enabled":
                    names.append(" ".join(parts[3:]))
            out = "\n".join(names)
        except Exception:
            return VpnStatus()

    adapters: list[str] = []
    for line in out.splitlines():
        name = line.strip()
        if name:
            adapters.append(name)

    hits: list[str] = []
    for name in adapters:
        low = name.lower()
        if any(h in low for h in _VPN_ADAPTER_HINTS):
            hits.append(name)

    if not hits:
        return VpnStatus(active=False, label="Direct (no VPN)", adapters=adapters)

    primary = hits[0]
    return VpnStatus(
        active=True,
        label=_match_provider(primary),
        adapter=primary,
        adapters=adapters,
    )


def expand_vpn_executables() -> list[str]:
    """Paths to VPN client binaries for firewall allow rules."""
    seen: set[str] = set()
    out: list[str] = []
    for pattern in _VPN_EXE_GLOBS:
        for p in glob.glob(pattern):
            rp = str(Path(p).resolve())
            if rp not in seen:
                seen.add(rp)
                out.append(rp)
    return out