herdr-session-switch/herdr_hub/proxy_env.py

"""Read literal remote proxy settings without executing shell startup files.

This standalone helper runs on the remote host; proxy values never leave it.
"""
from __future__ import annotations

import os
from pathlib import Path
import re
import shlex
import socket
import sys


PROXY_KEYS = {"http_proxy", "https_proxy", "all_proxy", "no_proxy"}


def proxy_assignments(text: str) -> dict[str, str]:
    values = {}
    for line in text.splitlines():
        # Only standalone, unindented literal assignments. Do not evaluate
        # functions, shell commands, substitutions, or conditional expressions.
        if not re.match(r"^(?:export\s+)?(?:https?_proxy|all_proxy|no_proxy)=", line, re.I):
            continue
        if any(marker in line for marker in ("$", "`", ";", "&&", "||")):
            continue
        try:
            words = shlex.split(line, comments=True)
        except ValueError:
            continue
        if words and words[0] == "export":
            words = words[1:]
        assignments = [word.split("=", 1) for word in words]
        if not assignments or any(len(item) != 2 or item[0].lower() not in PROXY_KEYS for item in assignments):
            continue
        for key, value in assignments:
            if "\0" not in value:
                values[key.lower()] = value
    return values


def config_paths(home: Path, shell: str) -> list[Path]:
    paths = [Path("/etc/environment"), Path("/etc/profile")]
    paths.extend(sorted(Path("/etc/profile.d").glob("*.sh")))
    paths.extend(sorted((home / ".config/environment.d").glob("*.conf")))
    paths.append(home / ".profile")
    if Path(shell).name in ("zsh",):
        paths.extend([home / ".zshenv", home / ".zprofile", home / ".zshrc"])
    else:
        paths.extend([home / ".bash_profile", home / ".bash_login", home / ".bashrc"])
    return paths


def configure_proxy(environment: dict[str, str], paths: list[Path]) -> list[str]:
    # An existing remote SSH environment (even an explicit empty value) wins.
    inherited = {key.lower() for key in environment if key.lower() in PROXY_KEYS}
    loaded = {}
    sources = []
    for path in paths:
        try:
            values = proxy_assignments(path.read_text(encoding="utf-8"))
        except (OSError, UnicodeError):
            continue
        values = {key: value for key, value in values.items() if key not in inherited}
        if values:
            loaded.update(values)
            sources.append(str(path))
    environment.update(loaded)
    # curl ignores uppercase HTTP_PROXY; normalize remote values, preserving
    # explicit lowercase precedence and NO_PROXY exclusions.
    for key in PROXY_KEYS:
        if key not in environment and key.upper() in environment:
            environment[key] = environment[key.upper()]
    return sources


def application_proxy_paths(home: Path) -> list[Path]:
    """Known runtime config locations; never inspect downloaded subscriptions."""
    return running_proxy_paths() + [
        home / ".local/share/io.github.clash-verge-rev.clash-verge-rev/clash-verge.yaml",
        home / ".config/io.github.clash-verge-rev.clash-verge-rev/clash-verge.yaml",
        home / ".config/clash-verge/clash-verge.yaml",
        home / ".config/mihomo/config.yaml",
        home / ".config/clash/config.yaml",
    ]


def running_proxy_paths(proc_root: Path = Path("/proc")) -> list[Path]:
    """Find the config explicitly used by a running Clash/Mihomo core."""
    paths = []
    try:
        processes = list(proc_root.iterdir())
    except OSError:
        return paths
    for process in processes:
        if not process.name.isdigit():
            continue
        try:
            arguments = (process / "cmdline").read_bytes().decode().rstrip("\0").split("\0")
        except (OSError, UnicodeError):
            continue
        if not arguments or not re.fullmatch(r"(?:mihomo|clash)(?:[-.][A-Za-z0-9_.-]+)?", Path(arguments[0]).name):
            continue
        config = None
        directory = None
        for index, argument in enumerate(arguments[1:], 1):
            flag, separator, value = argument.partition("=")
            if not separator and index + 1 < len(arguments):
                value = arguments[index + 1]
            if flag in ("-f", "--config") and value:
                config = Path(value)
            elif flag in ("-d", "--config-dir") and value:
                directory = Path(value)
        if config is None and directory is not None:
            config = directory / "config.yaml"
        if config is None:
            continue
        if not config.is_absolute():
            try:
                config = (process / "cwd").resolve(strict=True) / config
            except OSError:
                continue
        if config not in paths:
            paths.append(config)
    return paths


def configure_application_proxy(environment: dict[str, str], paths: list[Path]) -> str | None:
    # An explicit environment setting (including an empty/disabled value)
    # takes precedence over a proxy application's independent listener.
    if any(key.lower() in {"http_proxy", "https_proxy", "all_proxy"} for key in environment):
        return None
    for path in paths:
        try:
            text = path.read_text(encoding="utf-8")
        except (OSError, UnicodeError):
            continue
        # This deliberately reads only simple top-level listener settings,
        # not arbitrary YAML, provider credentials or subscription profiles.
        fields = dict(re.findall(
            r"^(mixed-port|port|socks-port|bind-address|authentication):[ \t]*(.*)$",
            text, re.M,
        ))
        if "authentication" in fields and fields["authentication"].strip() != "[]":
            continue
        try:
            bind = shlex.split(fields.get("bind-address", "'*'"), comments=True)
        except ValueError:
            continue
        if len(bind) != 1 or bind[0] not in ("*", "0.0.0.0", "127.0.0.1", "::", "::1", "localhost"):
            continue
        host = "::1" if bind[0] in ("::", "::1") else "127.0.0.1"
        for key, scheme in (("mixed-port", "http"), ("port", "http"), ("socks-port", "socks5h")):
            match = re.fullmatch(r"[\"']?(\d+)[\"']?\s*(?:#.*)?", fields.get(key, ""))
            if not match:
                continue
            port = int(match[1])
            if not 1 <= port <= 65535:
                continue
            try:
                with socket.create_connection((host, port), timeout=1):
                    pass
            except OSError:
                continue
            address = f"[{host}]" if host == "::1" else host
            proxy = f"{scheme}://{address}:{port}"
            environment.update(http_proxy=proxy, https_proxy=proxy, all_proxy=proxy)
            return str(path)
    return None


if __name__ == "__main__":
    environment = dict(os.environ)
    sources = configure_proxy(environment, config_paths(Path.home(), environment.get("SHELL", "bash")))
    application_source = configure_application_proxy(environment, application_proxy_paths(Path.home()))
    if application_source:
        sources.append(application_source + "（应用代理端口可连接）")
    active = any(environment.get(key) for key in ("https_proxy", "all_proxy", "http_proxy"))
    if active:
        source = "、".join(sources) if sources else "SSH 环境变量"
        print("使用远端自身代理，配置来源：" + source, flush=True)
    else:
        print("网络方式：远端直连（未发现可用代理配置；此行不是错误）。", flush=True)
    os.execve("/bin/sh", ["sh", "-c", sys.argv[1]], environment)