network_settings.py

"""Persist application proxy preferences and apply them to Git child processes."""
import json
import os
from pathlib import Path
import threading
from urllib.parse import urlsplit

DEFAULT_PROXY = "http://127.0.0.1:17891"


def validate_settings(value):
    enabled = value.get("proxy_enabled", False)
    url = value.get("proxy_url", DEFAULT_PROXY)
    if not isinstance(enabled, bool) or not isinstance(url, str):
        raise ValueError("代理设置格式无效。")
    url = url.strip()
    if not url and not enabled:
        url = DEFAULT_PROXY
    try:
        parsed = urlsplit(url)
        valid = (parsed.scheme in {"http", "https", "socks5", "socks5h"}
                 and parsed.hostname and parsed.port
                 and not parsed.username and not parsed.password
                 and parsed.path in {"", "/"} and not parsed.query and not parsed.fragment
                 and not any(c.isspace() or ord(c) < 32 for c in url))
    except ValueError:
        valid = False
    if not valid:
        raise ValueError("请输入含协议和端口的代理地址，例如 http://127.0.0.1:17891；不要包含账号或密码。")
    return {"proxy_enabled": enabled, "proxy_url": url}


class NetworkSettings:
    def __init__(self, path):
        self.path = Path(path)
        self.lock = threading.RLock()
        self.value = validate_settings(json.loads(self.path.read_text(encoding="utf-8"))
                                       if self.path.exists() else {})

    def snapshot(self):
        with self.lock:
            return dict(self.value)

    def update(self, value):
        validated = validate_settings(value)
        with self.lock:
            temp = self.path.with_suffix(".tmp")
            temp.write_text(json.dumps(validated, ensure_ascii=False, indent=2), encoding="utf-8")
            os.replace(temp, self.path)
            self.value = validated
            return dict(self.value)


def git_environment(settings=None, url=""):
    settings = validate_settings(settings or {})
    env = {key: value for key, value in os.environ.items()
           if key.lower() not in {"http_proxy", "https_proxy", "all_proxy", "no_proxy"}}
    proxy = settings["proxy_url"] if settings["proxy_enabled"] else ""
    if proxy:
        env.update(HTTP_PROXY=proxy, HTTPS_PROXY=proxy, ALL_PROXY=proxy)
    # NO_PROXY=* also bypasses proxies configured in existing Git config files.
    env["NO_PROXY"] = "localhost,127.0.0.1,::1" if proxy else "*"
    overrides = {"http.proxy": proxy, "remote.origin.proxy": proxy}
    if url.startswith(("http://", "https://")):
        overrides[f"http.{url}.proxy"] = proxy
    # Environment config is inherited by Git's LFS and submodule subprocesses.
    count = int(env.get("GIT_CONFIG_COUNT", "0"))
    for key, value in overrides.items():
        env[f"GIT_CONFIG_KEY_{count}"] = key
        env[f"GIT_CONFIG_VALUE_{count}"] = value
        count += 1
    env["GIT_CONFIG_COUNT"] = str(count)
    return env