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


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)
    _broadcast()
    log.info("System proxy set %s (bypass=%s) — HKCU + HKLM(adm) + Connections + WinHTTP", proxy, bypass)


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()
    _broadcast()
    log.info("System proxy cleared — HKCU + HKLM(adm) + Connections + WinHTTP")


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