system_proxy.py

"""macOS system proxy control via networksetup.

Stateless: callers track desired/applied state and call apply()/clear() to converge.
"""
import logging
import subprocess

logger = logging.getLogger("magic-proxy.system_proxy")

DEFAULT_BYPASS = ["*.local", "169.254/16", "127.0.0.1", "localhost"]
_TIMEOUT = 5


def _run(args):
    """Run a networksetup command. Returns (ok, stderr). Never raises."""
    try:
        cp = subprocess.run(
            args, capture_output=True, text=True, timeout=_TIMEOUT,
        )
    except (subprocess.TimeoutExpired, OSError) as e:
        return False, str(e)
    if cp.returncode != 0:
        return False, (cp.stderr or "").strip()
    return True, ""


def _active_services():
    """Return list of enabled network service names (skips disabled '*' lines)."""
    # Bypass _run here: we need stdout, which _run discards on success.
    try:
        cp = subprocess.run(
            ["networksetup", "-listallnetworkservices"],
            capture_output=True, text=True, timeout=_TIMEOUT,
        )
    except (subprocess.TimeoutExpired, OSError):
        return []
    if cp.returncode != 0:
        return []
    services = []
    for i, line in enumerate(cp.stdout.splitlines()):
        if i == 0:
            continue  # header/notice line
        line = line.strip()
        if not line or line.startswith("*"):
            continue
        services.append(line)
    return services


def apply(host, port, bypass):
    """Set HTTP+HTTPS proxy + bypass for every active service.

    Returns (ok, err). ok=True if at least one service was fully configured.
    """
    services = _active_services()
    if not services:
        return False, "no active network services"

    port_s = str(port)
    successes = 0
    errors = []
    # networksetup -setproxybypassdomains requires the literal "Empty" to clear.
    bypass_args = bypass if bypass else ["Empty"]
    for svc in services:
        svc_ok = True
        # No break: keep going so errors[] reflects all failures for this service.
        for cmd in (
            ["networksetup", "-setwebproxy", svc, host, port_s],
            ["networksetup", "-setsecurewebproxy", svc, host, port_s],
            ["networksetup", "-setproxybypassdomains", svc, *bypass_args],
        ):
            ok, err = _run(cmd)
            if not ok:
                svc_ok = False
                errors.append(f"{svc}: {err}")
                logger.warning("system_proxy apply failed: %s -> %s", cmd, err)
        if svc_ok:
            successes += 1

    if successes > 0:
        return True, ""
    return False, "; ".join(errors)


def clear():
    """Turn off HTTP+HTTPS proxy on every active service. Returns (ok, err).

    No-op success when there are no active services.
    """
    services = _active_services()
    if not services:
        return True, ""

    successes = 0
    errors = []
    for svc in services:
        svc_ok = True
        for cmd in (
            ["networksetup", "-setwebproxystate", svc, "off"],
            ["networksetup", "-setsecurewebproxystate", svc, "off"],
        ):
            ok, err = _run(cmd)
            if not ok:
                svc_ok = False
                errors.append(f"{svc}: {err}")
                logger.warning("system_proxy clear failed: %s -> %s", cmd, err)
        if svc_ok:
            successes += 1

    if successes > 0:
        return True, ""
    return False, "; ".join(errors)


def is_active():
    """True if any active service has HTTP proxy enabled."""
    for svc in _active_services():
        # Bypass _run here: we need stdout, which _run discards on success.
        try:
            cp = subprocess.run(
                ["networksetup", "-getwebproxy", svc],
                capture_output=True, text=True, timeout=_TIMEOUT,
            )
        except (subprocess.TimeoutExpired, OSError):
            continue
        if cp.returncode == 0 and "Enabled: Yes" in cp.stdout:
            return True
    return False