Files
PROXY_GOD_MAC/proxy_chain_manager/webrtc_check.py
2026-05-23 22:09:43 -07:00

242 lines
10 KiB
Python

"""WebRTC leak detection and hardening checks.
WebRTC leaks expose your real IP through STUN/ICE even when a proxy is set.
This module checks every controllable surface:
1. OS-level Chrome / Edge Group Policy (prevents non-proxied UDP)
2. Firefox managed profile prefs (user.js must disable PeerConnection)
3. STUN UDP reachability (if STUN is reachable, WebRTC CAN leak)
4. Firefox process prefs (confirm our user.js was accepted)
"""
from __future__ import annotations
import logging
import socket
import subprocess
import sys
import winreg
from dataclasses import dataclass, field
from pathlib import Path
log = logging.getLogger(__name__)
# STUN servers commonly used by WebRTC
_STUN_HOSTS: list[tuple[str, int]] = [
("stun.l.google.com", 19302),
("stun1.l.google.com", 19302),
("stun.cloudflare.com", 3478),
("stun.stunprotocol.org", 3478),
]
# Firefox user.js prefs that block WebRTC
_REQUIRED_WEBRTC_PREFS: dict[str, str] = {
"media.peerconnection.enabled": "false",
"media.peerconnection.ice.default_address_only": "true",
"media.peerconnection.ice.no_host": "true",
}
# Chrome / Edge HKLM policy keys
_CHROMIUM_POLICY_PATHS = (
r"SOFTWARE\Policies\Google\Chrome",
r"SOFTWARE\Policies\Microsoft\Edge",
r"SOFTWARE\Policies\Chromium",
)
_WEBRTC_POLICY_VALUE = "DefaultWebRtcIpHandlingPolicy"
_WEBRTC_BLOCK_VALUE = 3 # disable_non_proxied_udp (was 2 = public+private only — wrong)
_WEBRTC_POLICY_STR_KEY = "WebRtcIPHandling"
_WEBRTC_BLOCK_STR_VALUE = "disable_non_proxied_udp"
@dataclass
class WebRtcCheckResult:
"""Aggregated WebRTC leak surface scan."""
# Policy / config checks
chrome_edge_policy_set: bool = False
chrome_edge_policy_detail: str = ""
firefox_prefs_ok: bool = False
firefox_prefs_detail: str = ""
# Network reachability
stun_reachable: bool = False
stun_detail: str = ""
# Convenience flags
any_leak_risk: bool = True
issues: list[str] = field(default_factory=list)
checks: list[tuple[str, bool, str]] = field(default_factory=list) # (name, ok, detail)
def add(self, name: str, ok: bool, detail: str) -> None:
self.checks.append((name, ok, detail))
if not ok:
self.issues.append(f"{name}: {detail}")
# ─────────────────────────────────────────────────────────────────────────────
# 1. Chrome / Edge Group Policy
# ─────────────────────────────────────────────────────────────────────────────
def check_chromium_webrtc_policy() -> tuple[bool, str]:
"""Return (policy_set, detail_string).
Accepts either the legacy DWORD DefaultWebRtcIpHandlingPolicy==3
OR the modern REG_SZ WebRtcIPHandling=="disable_non_proxied_udp".
"""
if sys.platform != "win32":
return True, "Chrome/Edge registry policy not applicable on macOS"
found: list[str] = []
missing: list[str] = []
for path in _CHROMIUM_POLICY_PATHS:
try:
with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, path, 0, winreg.KEY_QUERY_VALUE) as k:
name = path.split("\\")[-1]
dword_ok = False
str_ok = False
# Check legacy DWORD
try:
val = int(winreg.QueryValueEx(k, _WEBRTC_POLICY_VALUE)[0])
dword_ok = val == _WEBRTC_BLOCK_VALUE
if not dword_ok:
missing.append(f"{name} DWORD={val} (need {_WEBRTC_BLOCK_VALUE})")
except OSError:
pass
# Check modern REG_SZ
try:
val_str = str(winreg.QueryValueEx(k, _WEBRTC_POLICY_STR_KEY)[0])
str_ok = val_str == _WEBRTC_BLOCK_STR_VALUE
if not str_ok:
missing.append(f"{name} REG_SZ={val_str!r} (need {_WEBRTC_BLOCK_STR_VALUE!r})")
except OSError:
pass
if dword_ok or str_ok:
found.append(name)
elif not dword_ok and not str_ok:
missing.append(f"{name}: no WebRTC policy keys present")
except OSError:
continue # key not present at all — browser not installed or not policy-managed
if found:
return True, "Policy set: " + ", ".join(found)
if missing:
return False, "Not set: " + "; ".join(missing)
return False, "No Chrome/Edge/Chromium policy keys found (browsers may not be installed)"
# ─────────────────────────────────────────────────────────────────────────────
# 2. Firefox managed profile prefs
# ─────────────────────────────────────────────────────────────────────────────
def check_firefox_webrtc_prefs(profile_dir: Path) -> tuple[bool, str]:
"""Verify our written user.js contains the required WebRTC-disabling prefs."""
user_js = profile_dir / "user.js"
if not user_js.is_file():
return False, f"user.js not found at {profile_dir} — profile not yet initialized"
try:
content = user_js.read_text(encoding="utf-8", errors="replace")
except Exception as e:
return False, f"Cannot read user.js: {e}"
missing: list[str] = []
for pref, expected_val in _REQUIRED_WEBRTC_PREFS.items():
# Look for: user_pref("pref.name", value);
needle = f'"{pref}", {expected_val}'
if needle not in content:
missing.append(pref)
if missing:
return False, "Missing in user.js: " + ", ".join(missing)
return True, "All WebRTC prefs present in managed profile"
# ─────────────────────────────────────────────────────────────────────────────
# 3. STUN server UDP reachability
# ─────────────────────────────────────────────────────────────────────────────
def _probe_stun_udp(host: str, port: int, timeout: float = 2.5) -> bool:
"""
Send a minimal STUN Binding Request over UDP and wait for a response.
Returns True if we get *any* response (STUN or ICMP unreachable).
"""
# Minimal STUN Binding Request (RFC 5389)
msg = (
b"\x00\x01" # Message Type: Binding Request
b"\x00\x00" # Message Length: 0
b"\x21\x12\xa4\x42" # Magic Cookie
+ b"\x00" * 12 # Transaction ID (12 bytes)
)
try:
ip = socket.gethostbyname(host)
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(timeout)
s.sendto(msg, (ip, port))
data, _ = s.recvfrom(512)
s.close()
return len(data) > 0
except socket.timeout:
return False
except Exception:
return False
def check_stun_reachability(timeout: float = 3.0) -> tuple[bool, str]:
"""
Returns (reachable, detail).
reachable = True means STUN servers ARE reachable → WebRTC COULD leak.
"""
reachable: list[str] = []
for host, port in _STUN_HOSTS[:3]:
try:
if _probe_stun_udp(host, port, timeout):
reachable.append(f"{host}:{port}")
except Exception:
continue
if reachable:
return True, "STUN reachable: " + ", ".join(reachable) + " — WebRTC UDP is NOT blocked"
return False, "STUN servers unreachable — WebRTC UDP appears blocked (good)"
# ─────────────────────────────────────────────────────────────────────────────
# Composite scan
# ─────────────────────────────────────────────────────────────────────────────
def run_webrtc_check(profile_dir: Path | None = None) -> WebRtcCheckResult:
"""
Run all WebRTC leak surface checks.
profile_dir: path to the managed Firefox profile (to verify user.js).
"""
res = WebRtcCheckResult()
# 1. Chrome/Edge policy
policy_ok, policy_detail = check_chromium_webrtc_policy()
res.chrome_edge_policy_set = policy_ok
res.chrome_edge_policy_detail = policy_detail
if sys.platform == "win32":
res.add("Chrome/Edge WebRTC policy (HKLM)", policy_ok, policy_detail)
# 2. Firefox profile prefs
if profile_dir is not None:
fx_ok, fx_detail = check_firefox_webrtc_prefs(profile_dir)
res.firefox_prefs_ok = fx_ok
res.firefox_prefs_detail = fx_detail
res.add("Firefox managed profile WebRTC prefs", fx_ok, fx_detail)
else:
res.add("Firefox managed profile WebRTC prefs", False,
"Profile not specified — not checked")
# 3. STUN UDP reachability
stun_reachable, stun_detail = check_stun_reachability()
res.stun_reachable = stun_reachable
res.stun_detail = stun_detail
# STUN reachable is only a leak risk when the active browser profile has
# not disabled WebRTC. On macOS the clean port is Firefox-profile based,
# not OS registry based.
stun_ok = (not stun_reachable) or (sys.platform != "win32" and res.firefox_prefs_ok)
detail = stun_detail
if stun_reachable and stun_ok:
detail += " — browser profile disables WebRTC, so UDP reachability is informational"
res.add("STUN UDP reachability", stun_ok, detail)
# Overall verdict
res.any_leak_risk = bool(res.issues)
return res