v2: tabs, chain builder, obfuscation modes, hop slider, Defender exclusion, firewall kept
Made-with: Cursor
This commit is contained in:
175
proxy_chain_manager/firewall.py
Normal file
175
proxy_chain_manager/firewall.py
Normal file
@@ -0,0 +1,175 @@
|
||||
"""
|
||||
Windows Firewall enforcement — kill-switch style.
|
||||
|
||||
When engaged:
|
||||
- Default outbound policy → BLOCK (all profiles)
|
||||
- Allow rules for: GOST, NordVPN stack, this app, loopback, DHCP
|
||||
- Everything else is denied outbound — apps that don't go through the
|
||||
local proxy simply can't reach the internet.
|
||||
|
||||
When disengaged:
|
||||
- Remove our rules, restore default outbound → ALLOW
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import ctypes
|
||||
import glob
|
||||
import logging
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from .paths import gost_exe_path
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
RULE_PREFIX = "PCM_"
|
||||
|
||||
NORD_GLOBS = [
|
||||
r"C:\Program Files\NordVPN\*.exe",
|
||||
r"C:\Program Files\NordUpdater\*.exe",
|
||||
r"C:\Program Files\NordVPN\NordSec ThreatProtection\*.exe",
|
||||
]
|
||||
|
||||
|
||||
def is_admin() -> bool:
|
||||
try:
|
||||
return ctypes.windll.shell32.IsUserAnAdmin() != 0 # type: ignore[attr-defined]
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def request_admin_relaunch() -> bool:
|
||||
"""Re-launch the current process elevated. Returns True if launched, False on error."""
|
||||
if is_admin():
|
||||
return False
|
||||
try:
|
||||
exe = sys.executable
|
||||
args = " ".join(f'"{a}"' for a in sys.argv)
|
||||
ret = ctypes.windll.shell32.ShellExecuteW( # type: ignore[attr-defined]
|
||||
None, "runas", exe, args, None, 1
|
||||
)
|
||||
return ret > 32
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def _run(args: list[str], check: bool = False) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
args, capture_output=True, text=True, timeout=30,
|
||||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||||
)
|
||||
|
||||
|
||||
def _add_rule(name: str, **kw: str) -> bool:
|
||||
full = RULE_PREFIX + name
|
||||
args = ["netsh", "advfirewall", "firewall", "add", "rule", f"name={full}"]
|
||||
for k, v in kw.items():
|
||||
args.append(f"{k}={v}")
|
||||
r = _run(args)
|
||||
ok = r.returncode == 0
|
||||
if not ok:
|
||||
log.warning("Firewall rule %s failed: %s", full, (r.stdout or "") + (r.stderr or ""))
|
||||
return ok
|
||||
|
||||
|
||||
def _delete_rules() -> None:
|
||||
r = _run(["netsh", "advfirewall", "firewall", "show", "rule", "name=all", "dir=out"])
|
||||
lines = (r.stdout or "").splitlines()
|
||||
names: set[str] = set()
|
||||
for ln in lines:
|
||||
if ln.startswith("Rule Name:"):
|
||||
n = ln.split(":", 1)[1].strip()
|
||||
if n.startswith(RULE_PREFIX):
|
||||
names.add(n)
|
||||
for n in names:
|
||||
_run(["netsh", "advfirewall", "firewall", "delete", "rule", f"name={n}"])
|
||||
log.info("Deleted %d PCM firewall rules", len(names))
|
||||
|
||||
|
||||
def _set_outbound_policy(action: str) -> None:
|
||||
"""action = 'blockinbound,blockoutbound' or 'blockinbound,allowoutbound'."""
|
||||
for profile in ("domainprofile", "privateprofile", "publicprofile"):
|
||||
_run(["netsh", "advfirewall", "set", profile, "firewallpolicy", action])
|
||||
|
||||
|
||||
def _resolve_self_exe() -> Path:
|
||||
if getattr(sys, "frozen", False):
|
||||
return Path(sys.executable).resolve()
|
||||
return Path(sys.executable).resolve()
|
||||
|
||||
|
||||
def _expand_nord_exes() -> list[str]:
|
||||
out: list[str] = []
|
||||
for pattern in NORD_GLOBS:
|
||||
out.extend(glob.glob(pattern))
|
||||
return out
|
||||
|
||||
|
||||
def engage(gost_path: Path | None = None) -> tuple[bool, str]:
|
||||
"""Activate kill-switch firewall. Returns (success, message)."""
|
||||
if not is_admin():
|
||||
return False, "Need admin privileges for firewall enforcement."
|
||||
|
||||
gost = gost_path or gost_exe_path()
|
||||
self_exe = _resolve_self_exe()
|
||||
nord_exes = _expand_nord_exes()
|
||||
|
||||
_delete_rules()
|
||||
|
||||
# Allow loopback (apps → GOST on 127.0.0.1)
|
||||
_add_rule("Loopback", dir="out", action="allow",
|
||||
remoteip="127.0.0.0/8", protocol="any")
|
||||
|
||||
# Allow GOST outbound (it connects to the proxy hops)
|
||||
_add_rule("GOST", dir="out", action="allow",
|
||||
program=f'"{gost}"', protocol="any")
|
||||
|
||||
# Allow this application outbound (fetches proxy lists, validates)
|
||||
_add_rule("Self", dir="out", action="allow",
|
||||
program=f'"{self_exe}"', protocol="any")
|
||||
|
||||
# If running from python.exe, also allow pythonw.exe
|
||||
if self_exe.name.lower() in ("python.exe", "pythonw.exe"):
|
||||
for sibling in ("python.exe", "pythonw.exe"):
|
||||
p = self_exe.with_name(sibling)
|
||||
if p.is_file():
|
||||
_add_rule(f"Python_{sibling}", dir="out", action="allow",
|
||||
program=f'"{p}"', protocol="any")
|
||||
|
||||
# Allow all NordVPN executables
|
||||
for i, npath in enumerate(nord_exes):
|
||||
_add_rule(f"Nord_{i}", dir="out", action="allow",
|
||||
program=f'"{npath}"', protocol="any")
|
||||
|
||||
# Allow DHCP (or you lose your adapter)
|
||||
_add_rule("DHCP", dir="out", action="allow",
|
||||
protocol="udp", remoteport="67,68")
|
||||
|
||||
# Allow DNS (GOST needs to resolve proxy hostnames)
|
||||
_add_rule("DNS_UDP", dir="out", action="allow",
|
||||
protocol="udp", remoteport="53")
|
||||
_add_rule("DNS_TCP", dir="out", action="allow",
|
||||
protocol="tcp", remoteport="53")
|
||||
|
||||
# Set default outbound to BLOCK
|
||||
_set_outbound_policy("blockinbound,blockoutbound")
|
||||
|
||||
log.info("Firewall kill-switch engaged. %d Nord exes whitelisted.", len(nord_exes))
|
||||
return True, f"Kill-switch ON. {len(nord_exes)} Nord processes whitelisted."
|
||||
|
||||
|
||||
def disengage() -> tuple[bool, str]:
|
||||
"""Remove kill-switch, restore normal outbound."""
|
||||
if not is_admin():
|
||||
return False, "Need admin to remove firewall rules."
|
||||
_set_outbound_policy("blockinbound,allowoutbound")
|
||||
_delete_rules()
|
||||
log.info("Firewall kill-switch disengaged.")
|
||||
return True, "Kill-switch OFF. Normal outbound restored."
|
||||
|
||||
|
||||
def is_engaged() -> bool:
|
||||
"""Quick check: are our rules present?"""
|
||||
r = _run(["netsh", "advfirewall", "firewall", "show", "rule", f"name={RULE_PREFIX}GOST", "dir=out"])
|
||||
return RULE_PREFIX in (r.stdout or "")
|
||||
Reference in New Issue
Block a user