234 lines
7.7 KiB
Python
234 lines
7.7 KiB
Python
"""Detect whether a VPN tunnel is active on Windows (any provider)."""
|
|
from __future__ import annotations
|
|
|
|
import glob
|
|
import logging
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
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."""
|
|
if sys.platform == "darwin":
|
|
try:
|
|
r = subprocess.run(["ifconfig"], capture_output=True, text=True, timeout=8)
|
|
names: list[str] = []
|
|
for line in (r.stdout or "").splitlines():
|
|
if line and not line.startswith("\t") and ":" in line:
|
|
name = line.split(":", 1)[0].strip()
|
|
if name:
|
|
names.append(name)
|
|
hits = [n for n in names if n.startswith(("utun", "tun", "tap", "wg"))]
|
|
if not hits:
|
|
return VpnStatus(active=False, label="Direct (no VPN)", adapters=names)
|
|
primary = hits[0]
|
|
return VpnStatus(active=True, label=_match_provider(primary), adapter=primary, adapters=names)
|
|
except Exception:
|
|
return VpnStatus()
|
|
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
|