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",
)

_NON_VPN_HINTS = (
    "wan miniport",
    "microsoft kernel debug",
    "isatap",
    "teredo",
    "loopback",
    "pseudo-interface",
)

# 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' } | "
        "ForEach-Object { \"$($_.Name)|$($_.InterfaceDescription)\" }"
    )
    if not out:
        # Fallback for PowerShell 2.0 / older Windows Server: parse "wmic nic"
        # then "netsh interface show interface". Both are locale-tolerant.
        try:
            r = subprocess.run(
                ["wmic", "nic", "where", "NetEnabled=true",
                 "get", "NetConnectionID,Name", "/format:list"],
                capture_output=True,
                text=True,
                timeout=12,
                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
            )
            blob = (r.stdout or "")
            names: list[str] = []
            cur_id = ""
            cur_name = ""
            for ln in blob.splitlines():
                ln = ln.strip()
                if not ln:
                    if cur_id:
                        names.append(f"{cur_id}|{cur_name}")
                    cur_id, cur_name = "", ""
                    continue
                if ln.startswith("NetConnectionID="):
                    cur_id = ln.split("=", 1)[1].strip()
                elif ln.startswith("Name="):
                    cur_name = ln.split("=", 1)[1].strip()
            if cur_id:
                names.append(f"{cur_id}|{cur_name}")
            out = "\n".join(names)
        except Exception:
            out = ""

    if not out:
        # Last-ditch: netsh. Locale-tolerant — match the connected/enabled state
        # by extracting the interface name from the rightmost column. Older
        # localized Server SKUs (es, de, fr, etc.) don't print literal "Enabled".
        try:
            r = subprocess.run(
                ["netsh", "interface", "show", "interface"],
                capture_output=True,
                text=True,
                timeout=10,
                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
            )
            raw = (r.stdout or "")
            names = []
            for ln in raw.splitlines():
                s = ln.rstrip()
                if not s or s.startswith("-") or ":" in s.split(" ")[0]:
                    continue
                parts = re.split(r"\s{2,}", s.strip())
                if len(parts) >= 4:
                    nm = parts[-1].strip()
                    if nm and nm.lower() not in ("interface name", "nombre de interfaz"):
                        names.append(f"{nm}|")
            out = "\n".join(names)
        except Exception:
            return VpnStatus()

    adapters: list[str] = []
    adapter_blob: list[str] = []
    for line in out.splitlines():
        raw = line.strip()
        if not raw:
            continue
        if "|" in raw:
            name, desc = raw.split("|", 1)
        else:
            name, desc = raw, ""
        name = name.strip()
        desc = desc.strip()
        if name:
            adapters.append(name)
            adapter_blob.append((name + " " + desc).strip())

    hits: list[str] = []
    for i, name in enumerate(adapters):
        low = adapter_blob[i].lower()
        if any(h in low for h in _NON_VPN_HINTS):
            continue
        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