proxy-god/proxy_chain_manager/sysproxy.py

"""Set / clear the Windows system-wide HTTP proxy.

Layers configured so the proxy actually applies system-wide:

1. ``HKCU\\…\\Internet Settings``       — simple values WinINet reads first
2. ``Internet Settings\\Connections\\DefaultConnectionSettings`` binary blob —
   overrides #1 for any process that opens the per-connection structure.
3. ``HKLM\\…\\Internet Settings`` mirror (Admin only) — required on Server when
   ``ProxySettingsPerUser=0`` is set by Group Policy / baseline image.
4. ``netsh winhttp set proxy``         — .NET, Windows Update, services and
   headless browser components that bypass WinINet.

We also actively detect and warn about:
- ``ProxySettingsPerUser=0``  → HKCU is ignored; we fix it when Admin.
- Group Policy proxy keys     → those override us; we surface the path so the
  user can verify or remove them.
- ``AutoConfigURL`` / ``AutoDetect`` (WPAD) → wiped on engage.
"""
from __future__ import annotations

import ctypes
import logging
import struct
import subprocess
import winreg

log = logging.getLogger(__name__)

_KEY_PATH = r"Software\Microsoft\Windows\CurrentVersion\Internet Settings"
_CONN_KEY_PATH = (
    r"Software\Microsoft\Windows\CurrentVersion\Internet Settings\Connections"
)
_HKLM_INET_PATH = r"SOFTWARE\Microsoft\Windows\CurrentVersion\Internet Settings"
_HKLM_CONN_PATH = (
    r"SOFTWARE\Microsoft\Windows\CurrentVersion\Internet Settings\Connections"
)
_POLICY_PATHS = (
    (winreg.HKEY_LOCAL_MACHINE,
     r"SOFTWARE\Policies\Microsoft\Windows\CurrentVersion\Internet Settings"),
    (winreg.HKEY_CURRENT_USER,
     r"SOFTWARE\Policies\Microsoft\Windows\CurrentVersion\Internet Settings"),
    (winreg.HKEY_LOCAL_MACHINE,
     r"SOFTWARE\Policies\Microsoft\Internet Explorer\Control Panel"),
)
_SETTINGS_CHANGED = 39
_REFRESH = 37

_FLAG_DIRECT       = 0x01
_FLAG_MANUAL_PROXY = 0x02
_FLAG_AUTO_CONFIG  = 0x04
_FLAG_AUTO_DETECT  = 0x08


def _is_admin() -> bool:
    try:
        return ctypes.windll.shell32.IsUserAnAdmin() != 0  # type: ignore[attr-defined]
    except Exception:
        return False


def _broadcast() -> None:
    """Tell running apps (browsers, etc.) that proxy settings changed."""
    try:
        inet = ctypes.windll.wininet  # type: ignore[attr-defined]
        inet.InternetSetOptionW(0, _SETTINGS_CHANGED, 0, 0)
        inet.InternetSetOptionW(0, _REFRESH, 0, 0)
    except Exception:
        pass


def _bump_counter(prev: bytes | None) -> int:
    if not prev or len(prev) < 8:
        return 1
    try:
        return int.from_bytes(prev[4:8], "little") + 1
    except Exception:
        return 1


def _build_blob(
    proxy: str,
    bypass: str,
    auto_config_url: str = "",
    counter: int = 1,
) -> bytes:
    """REG_BINARY layout for DefaultConnectionSettings (little-endian)."""
    flags = _FLAG_DIRECT
    if proxy:
        flags |= _FLAG_MANUAL_PROXY
    if auto_config_url:
        flags |= _FLAG_AUTO_CONFIG

    p = proxy.encode("utf-8")
    b = bypass.encode("utf-8")
    a = auto_config_url.encode("utf-8")

    return (
        struct.pack("<III", 0x46, counter, flags)
        + struct.pack("<I", len(p)) + p
        + struct.pack("<I", len(b)) + b
        + struct.pack("<I", len(a)) + a
        + b"\x00" * 32
    )


def _read_blob_counter(hive: int, path: str) -> int:
    try:
        with winreg.OpenKey(hive, path, 0, winreg.KEY_QUERY_VALUE) as key:
            val, _ = winreg.QueryValueEx(key, "DefaultConnectionSettings")
            return _bump_counter(val)
    except OSError:
        return 1


def _write_conn_blob(hive: int, path: str, proxy: str, bypass: str) -> bool:
    counter = _read_blob_counter(hive, path)
    blob = _build_blob(proxy, bypass, "", counter)
    try:
        key = winreg.CreateKeyEx(hive, path, 0, winreg.KEY_SET_VALUE)
        with key:
            winreg.SetValueEx(key, "DefaultConnectionSettings", 0, winreg.REG_BINARY, blob)
            winreg.SetValueEx(key, "SavedLegacySettings", 0, winreg.REG_BINARY, blob)
        return True
    except OSError as e:
        log.debug("Connections blob write failed (%s): %s", path, e)
        return False


def _clear_conn_blob(hive: int, path: str) -> None:
    counter = _read_blob_counter(hive, path)
    blob = _build_blob("", "", "", counter)
    try:
        with winreg.OpenKey(hive, path, 0, winreg.KEY_SET_VALUE) as key:
            winreg.SetValueEx(key, "DefaultConnectionSettings", 0, winreg.REG_BINARY, blob)
            winreg.SetValueEx(key, "SavedLegacySettings", 0, winreg.REG_BINARY, blob)
    except OSError:
        pass


def _wipe_autoconfig(key: winreg.HKEYType) -> None:
    for name in ("AutoConfigURL", "AutoDetect"):
        try:
            winreg.DeleteValue(key, name)
        except OSError:
            pass


def _set_winhttp_proxy(host: str, port: int, bypass: str) -> None:
    try:
        bypass_list = (
            bypass.replace(";", " ").replace("<local>", "<local>").strip()
        ) or "<local>"
        subprocess.run(
            ["netsh", "winhttp", "set", "proxy",
             f"{host}:{port}", f"bypass-list={bypass_list}"],
            capture_output=True, text=True, timeout=15,
            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
        )
    except Exception as e:
        log.debug("netsh winhttp set proxy failed: %s", e)


def _reset_winhttp_proxy() -> None:
    try:
        subprocess.run(
            ["netsh", "winhttp", "reset", "proxy"],
            capture_output=True, text=True, timeout=15,
            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
        )
    except Exception:
        pass


# ── browser-policy proxy (Chrome / Edge) ─────────────────────────────────────
#
# Why this exists: when you set WebRTC policy on Chrome/Edge, the browser sees
# itself as "managed". Some managed-Chrome builds then *ignore* the system
# proxy unless an explicit ``ProxyServer`` policy is also set. They also keep
# trying HTTP/3 over QUIC (UDP/443), which bypasses HTTP proxies entirely —
# under the kill-switch those UDP packets get dropped and pages hang forever.
#
# Setting these policy values pins the browser to our chain (HTTP proxy) and
# forces it off QUIC. Cleared cleanly on disengage.
_BROWSER_POLICY_PATHS = (
    r"SOFTWARE\Policies\Google\Chrome",
    r"SOFTWARE\Policies\Microsoft\Edge",
    r"SOFTWARE\Policies\Chromium",
)

def _set_browser_proxy_policy(host: str, port: int, bypass: str) -> int:
    """Pin Chromium-based browsers (Chrome, Edge, Chromium) to our local proxy
    via Group Policy registry, and disable QUIC. Returns count of keys updated.

    Requires Admin to write under HKLM\\SOFTWARE\\Policies. Silently skips when
    not elevated — the user gets a warning in the service log instead.
    """
    if not _is_admin():
        return 0
    proxy_value = f"http={host}:{port};https={host}:{port}"
    bypass_value = bypass.replace(";", ",")
    written = 0
    for path in _BROWSER_POLICY_PATHS:
        try:
            with winreg.CreateKeyEx(
                winreg.HKEY_LOCAL_MACHINE, path, 0, winreg.KEY_SET_VALUE
            ) as key:
                winreg.SetValueEx(key, "ProxyMode", 0, winreg.REG_SZ, "fixed_servers")
                winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, proxy_value)
                winreg.SetValueEx(key, "ProxyBypassList", 0, winreg.REG_SZ, bypass_value)
                winreg.SetValueEx(key, "QuicAllowed", 0, winreg.REG_DWORD, 0)
                written += 1
        except OSError:
            continue
    return written


def _clear_browser_proxy_policy() -> None:
    if not _is_admin():
        return
    for path in _BROWSER_POLICY_PATHS:
        try:
            with winreg.OpenKey(
                winreg.HKEY_LOCAL_MACHINE, path, 0, winreg.KEY_SET_VALUE
            ) as key:
                for name in ("ProxyMode", "ProxyServer", "ProxyBypassList", "QuicAllowed"):
                    try:
                        winreg.DeleteValue(key, name)
                    except OSError:
                        pass
        except OSError:
            continue


def _read_proxy_settings_per_user() -> int | None:
    """``HKLM\\…\\Internet Settings\\ProxySettingsPerUser``.

    When 0, Windows ignores HKCU proxy values and reads only HKLM. Set on many
    Server images and AD policy baselines — the #1 cause of "set the proxy but
    the browser still goes direct" on Server.
    """
    try:
        with winreg.OpenKey(
            winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, 0, winreg.KEY_QUERY_VALUE
        ) as key:
            val, _ = winreg.QueryValueEx(key, "ProxySettingsPerUser")
            return int(val)
    except OSError:
        return None


def _fix_proxy_settings_per_user() -> bool:
    """Force ``ProxySettingsPerUser=1`` (per-user proxy honored). Admin only."""
    if not _is_admin():
        return False
    try:
        with winreg.CreateKeyEx(
            winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, 0, winreg.KEY_SET_VALUE
        ) as key:
            winreg.SetValueEx(key, "ProxySettingsPerUser", 0, winreg.REG_DWORD, 1)
        log.info("HKLM\\…\\ProxySettingsPerUser → 1 (was %s)", _read_proxy_settings_per_user())
        return True
    except OSError as e:
        log.warning("Could not fix ProxySettingsPerUser: %s", e)
        return False


def _set_hklm_root(proxy: str, bypass: str) -> bool:
    """Mirror the per-user keys at HKLM so machine-scope WinINet reads honor us."""
    if not _is_admin():
        return False
    try:
        with winreg.CreateKeyEx(
            winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, 0, winreg.KEY_SET_VALUE
        ) as key:
            winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 1)
            winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, proxy)
            winreg.SetValueEx(key, "ProxyOverride", 0, winreg.REG_SZ, bypass)
            _wipe_autoconfig(key)
        return True
    except OSError as e:
        log.warning("HKLM root proxy write failed: %s", e)
        return False


def _clear_hklm_root() -> None:
    if not _is_admin():
        return
    try:
        with winreg.OpenKey(
            winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, 0, winreg.KEY_SET_VALUE
        ) as key:
            winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 0)
            _wipe_autoconfig(key)
    except OSError:
        pass


def detect_policy_overrides() -> list[str]:
    """Return human-readable paths of Group Policy proxy locks that beat us.

    Anything we find here will silently override our manual proxy. We do *not*
    delete policy keys (admins set those deliberately) — we just surface them.
    """
    hits: list[str] = []
    proxy_value_names = ("ProxyServer", "ProxyEnable", "AutoConfigURL", "ProxyOverride")
    for hive, path in _POLICY_PATHS:
        try:
            with winreg.OpenKey(hive, path, 0, winreg.KEY_QUERY_VALUE) as key:
                for n in proxy_value_names:
                    try:
                        winreg.QueryValueEx(key, n)
                        hive_name = "HKLM" if hive == winreg.HKEY_LOCAL_MACHINE else "HKCU"
                        hits.append(f"{hive_name}\\{path}\\{n}")
                    except OSError:
                        continue
        except OSError:
            continue
    return hits


def _read_str(hive: int, path: str, name: str) -> str:
    try:
        with winreg.OpenKey(hive, path, 0, winreg.KEY_QUERY_VALUE) as key:
            val, _ = winreg.QueryValueEx(key, name)
            return str(val)
    except OSError:
        return ""


def _read_dword(hive: int, path: str, name: str) -> int | None:
    try:
        with winreg.OpenKey(hive, path, 0, winreg.KEY_QUERY_VALUE) as key:
            val, _ = winreg.QueryValueEx(key, name)
            return int(val)
    except OSError:
        return None


def _read_winhttp_proxy() -> str:
    try:
        r = subprocess.run(
            ["netsh", "winhttp", "show", "proxy"],
            capture_output=True, text=True, timeout=10,
            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
        )
        for ln in (r.stdout or "").splitlines():
            if "proxy" in ln.lower() and ":" in ln:
                return ln.strip()
        return (r.stdout or "").strip().splitlines()[-1] if r.stdout else ""
    except Exception:
        return ""


def diagnose_system_proxy() -> list[str]:
    """Return a list of human-readable lines covering every place proxy can
    be configured. Logged on each engage so "chain green but browser leaks"
    is debuggable without RegEdit gymnastics.
    """
    lines: list[str] = []
    per_user = _read_proxy_settings_per_user()
    per_user_msg = (
        "1 (HKCU honored)" if per_user == 1
        else "0 (HKCU IGNORED — Server lock!)" if per_user == 0
        else "(unset → defaults to HKCU)"
    )
    lines.append(f"ProxySettingsPerUser = {per_user_msg}")

    en_u = _read_dword(winreg.HKEY_CURRENT_USER, _KEY_PATH, "ProxyEnable")
    ps_u = _read_str(winreg.HKEY_CURRENT_USER, _KEY_PATH, "ProxyServer")
    lines.append(f"HKCU  Enable={en_u}  Server='{ps_u}'")

    en_m = _read_dword(winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, "ProxyEnable")
    ps_m = _read_str(winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, "ProxyServer")
    lines.append(f"HKLM  Enable={en_m}  Server='{ps_m}'")

    ac_u = _read_str(winreg.HKEY_CURRENT_USER, _KEY_PATH, "AutoConfigURL")
    ad_u = _read_dword(winreg.HKEY_CURRENT_USER, _KEY_PATH, "AutoDetect")
    if ac_u or ad_u:
        lines.append(f"WPAD/PAC: AutoConfigURL='{ac_u}' AutoDetect={ad_u}")

    winhttp = _read_winhttp_proxy()
    if winhttp:
        lines.append(f"WinHTTP: {winhttp}")

    pol = detect_policy_overrides()
    if pol:
        lines.append(f"Policy overrides (these BEAT us): {len(pol)} entries")
        for p in pol[:5]:
            lines.append(f"  ! {p}")
        if len(pol) > 5:
            lines.append(f"  ! …and {len(pol) - 5} more")
    else:
        lines.append("Policy overrides: none")

    return lines


def set_system_proxy(
    host: str,
    port: int,
    bypass: str = "localhost;127.*;10.*;192.168.*;<local>",
) -> None:
    proxy = f"{host}:{port}"

    # HKCU root
    try:
        with winreg.OpenKey(
            winreg.HKEY_CURRENT_USER, _KEY_PATH, 0, winreg.KEY_SET_VALUE
        ) as key:
            winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 1)
            winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, proxy)
            winreg.SetValueEx(key, "ProxyOverride", 0, winreg.REG_SZ, bypass)
            _wipe_autoconfig(key)
    except OSError:
        log.exception("Failed to set system proxy (HKCU root)")
        return

    _write_conn_blob(winreg.HKEY_CURRENT_USER, _CONN_KEY_PATH, proxy, bypass)

    # If ProxySettingsPerUser=0 (Server / GPO baseline), HKCU is ignored.
    # When Admin, fix the flag AND mirror to HKLM so the proxy actually applies.
    per_user = _read_proxy_settings_per_user()
    if per_user == 0:
        log.warning(
            "ProxySettingsPerUser=0 detected — Windows is ignoring per-user "
            "proxy. Attempting Admin fix (HKLM mirror)."
        )
        if _is_admin():
            _fix_proxy_settings_per_user()
            _set_hklm_root(proxy, bypass)
            _write_conn_blob(winreg.HKEY_LOCAL_MACHINE, _HKLM_CONN_PATH, proxy, bypass)
        else:
            log.warning(
                "Not running as Admin — cannot fix ProxySettingsPerUser. "
                "Re-launch elevated for proxy to apply on this Server image."
            )

    # On Admin runs always mirror HKLM root too. Harmless when PerUser=1 and
    # correct when PerUser=0 or absent.
    if _is_admin():
        _set_hklm_root(proxy, bypass)
        _write_conn_blob(winreg.HKEY_LOCAL_MACHINE, _HKLM_CONN_PATH, proxy, bypass)

    _set_winhttp_proxy(host, port, bypass)
    n_browser = _set_browser_proxy_policy(host, port, bypass)
    _broadcast()
    log.info(
        "System proxy set %s (bypass=%s) — HKCU + HKLM(adm) + Connections + WinHTTP + %d browser policy",
        proxy, bypass, n_browser,
    )


def clear_system_proxy() -> None:
    try:
        with winreg.OpenKey(
            winreg.HKEY_CURRENT_USER, _KEY_PATH, 0, winreg.KEY_SET_VALUE
        ) as key:
            winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 0)
            _wipe_autoconfig(key)
    except OSError:
        log.exception("Failed to clear system proxy")

    _clear_conn_blob(winreg.HKEY_CURRENT_USER, _CONN_KEY_PATH)
    if _is_admin():
        _clear_hklm_root()
        _clear_conn_blob(winreg.HKEY_LOCAL_MACHINE, _HKLM_CONN_PATH)
    _reset_winhttp_proxy()
    _clear_browser_proxy_policy()
    _broadcast()
    log.info("System proxy cleared — HKCU + HKLM(adm) + Connections + WinHTTP + browser policy")


def is_system_proxy_set() -> bool:
    try:
        with winreg.OpenKey(
            winreg.HKEY_CURRENT_USER, _KEY_PATH, 0, winreg.KEY_QUERY_VALUE
        ) as key:
            val, _ = winreg.QueryValueEx(key, "ProxyEnable")
            return val == 1
    except OSError:
        return False