feat(aibox): verify SOCKS5 connectivity before saving proxy config

- set_proxy 先做完整 SOCKS5 握手 + generate_204 探测（trust_env=False，超时 4s/6s），验证失败不落盘并按类型归因提示（auth/timeout/connect/missing_dependency/probe_failed）

- 异常类型优先分类，消息关键词兜底：Windows 本地化错误消息不影响归因

- 新增 skip_check 逃生口（代理未启动的预配置场景）；禁用代理不验证；password_env 指向空环境变量时提前精准提示

- 测试：+9 用例（mock 网络 + 真实异常类型），全量 383 passed 零回归

aibox/tests/test_b_tools.py

@@ -15,7 +15,7 @@ from datetime import datetime, timedelta
 
 import pytest
 
-from common import constants
+from common import ConfigManager, constants
 from subsystems import (
     MemoryDatabase, MemoDatabase, PromptManager, SkillManager,
     SubprocessCommandExecutor,
@@ -425,6 +425,160 @@ def test_get_proxy_when_disabled():
     assert "未启用" in r or "启用" in r
 
 
+# ---------- SetProxyTool 连通性验证（设置前验证，失败不落盘） ----------
+
+@pytest.fixture
+def fresh_config(tmp_path, monkeypatch):
+    """隔离 ConfigManager 单例：AI_CONFIG_PATH 指向临时文件，前后重置类状态"""
+    cfg_file = tmp_path / "config.json"
+    monkeypatch.setenv("AI_CONFIG_PATH", str(cfg_file))
+
+    def _reset():
+        ConfigManager._instance = None
+        ConfigManager._config = None
+        ConfigManager._config_path = None
+
+    _reset()
+    yield ConfigManager
+    _reset()
+
+
+class _FakeResp:
+    def __init__(self, status_code):
+        self.status_code = status_code
+
+
+def _fake_session_factory(exc=None, resp=None):
+    """构造替代 requests.Session 的假类：get() 抛异常或返回假响应"""
+    calls = []
+
+    class _FakeSession:
+        def __init__(self):
+            self.trust_env = True
+
+        def get(self, *args, **kwargs):
+            calls.append((args, kwargs))
+            if exc:
+                raise exc
+            return resp
+
+    _FakeSession.calls = calls
+    return _FakeSession
+
+
+def test_check_connectivity_ok(monkeypatch):
+    fake = _fake_session_factory(resp=_FakeResp(204))
+    monkeypatch.setattr("tools.proxy_tools.requests.Session", fake)
+    t = SetProxyTool()
+    ok, kind, detail = t._check_connectivity("127.0.0.1", 1080, None, None)
+    assert ok and kind == ""
+    # trust_env 必须关掉：排除系统 http_proxy 环境变量干扰
+    assert fake.calls, "应发出探测请求"
+
+
+def test_check_connectivity_classification(monkeypatch, tmp_path):
+    """异常类型/消息 → 错误类型分类（与系统语言无关）"""
+    import requests as requests_lib
+    cases = [
+        # Windows 本地化的连接拒绝消息：靠异常类型而非英文关键词分类
+        (requests_lib.exceptions.ConnectionError(
+            "[WinError 10061] 由于目标计算机积极拒绝，无法连接。"), "connect_failed"),
+        # ConnectTimeout 同时继承 ConnectionError/Timeout，必须按 Timeout 归类
+        (requests_lib.exceptions.ConnectTimeout("connection timeout"), "timeout"),
+        (requests_lib.exceptions.ConnectionError(
+            "SOCKS5 authentication failed"), "auth_failed"),
+        (requests_lib.exceptions.ConnectionError(
+            "Missing dependencies for SOCKS support."), "missing_dependency"),
+        (Exception("Read timed out"), "timeout"),          # 兜底：非 requests 异常按消息
+        (Exception("Connection refused"), "connect_failed"),
+        (Exception("something weird"), "error"),
+    ]
+    t = SetProxyTool()
+    for exc, expect_kind in cases:
+        monkeypatch.setattr(
+            "tools.proxy_tools.requests.Session",
+            _fake_session_factory(exc=exc))
+        ok, kind, _ = t._check_connectivity("127.0.0.1", 1080, "u", "p")
+        assert not ok
+        assert kind == expect_kind, f"{exc!r} 应分类为 {expect_kind}，实际 {kind}"
+
+
+def test_check_connectivity_bad_status(monkeypatch):
+    """探测端点返回非 200/204 → probe_failed"""
+    monkeypatch.setattr(
+        "tools.proxy_tools.requests.Session",
+        _fake_session_factory(resp=_FakeResp(502)))
+    t = SetProxyTool()
+    ok, kind, detail = t._check_connectivity("127.0.0.1", 1080, None, None)
+    assert not ok and kind == "probe_failed"
+
+
+def test_set_proxy_rejected_when_unreachable(fresh_config, monkeypatch):
+    """验证失败：返回 ❌ 且配置不落盘"""
+    monkeypatch.setattr(
+        SetProxyTool, "_check_connectivity",
+        lambda self, h, p, u, pw: (False, "connect_failed", "Connection refused"))
+    t = SetProxyTool()
+    r = t.execute(host="10.255.255.1", port=1080)
+    assert "❌" in r
+    assert "未保存" in r
+    # 配置未写入：代理仍未启用
+    assert fresh_config().get_proxy() is None
+
+
+def test_set_proxy_saved_when_check_passes(fresh_config, monkeypatch):
+    """验证通过：配置保存且结果带连通性详情"""
+    monkeypatch.setattr(
+        SetProxyTool, "_check_connectivity",
+        lambda self, h, p, u, pw: (True, "", "探测成功 HTTP 204"))
+    t = SetProxyTool()
+    r = t.execute(host="127.0.0.1", port=1080)
+    assert "✅" in r
+    assert "探测成功" in r
+    proxies = fresh_config().get_proxy()
+    assert proxies and "127.0.0.1:1080" in proxies["http"]
+
+
+def test_set_proxy_skip_check_bypasses_validation(fresh_config, monkeypatch):
+    """skip_check=True：不验证直接保存（代理未启动的预配置场景）"""
+    called = []
+
+    def _no_call(self, *a, **kw):
+        called.append(1)
+        raise AssertionError("skip_check=True 时不应调用验证")
+
+    monkeypatch.setattr(SetProxyTool, "_check_connectivity", _no_call)
+    t = SetProxyTool()
+    r = t.execute(host="127.0.0.1", port=1080, skip_check=True)
+    assert "✅" in r
+    assert not called
+    assert fresh_config().get_proxy() is not None
+
+
+def test_set_proxy_disabled_skips_check(fresh_config, monkeypatch):
+    """enabled=False：禁用代理无需验证"""
+    def _no_call(self, *a, **kw):
+        raise AssertionError("禁用代理不应触发验证")
+
+    monkeypatch.setattr(SetProxyTool, "_check_connectivity", _no_call)
+    t = SetProxyTool()
+    r = t.execute(host="127.0.0.1", port=1080, enabled=False)
+    assert "✅" in r
+    assert fresh_config().get_proxy() is None  # 禁用态 get_proxy 返回 None
+
+
+def test_set_proxy_empty_env_hint(fresh_config, monkeypatch):
+    """username + password_env 指向未设环境变量：精准提示而非裸 auth 失败"""
+    monkeypatch.delenv("AIBOX_TEST_SOCKS_PW", raising=False)
+    t = SetProxyTool()
+    r = t.execute(host="127.0.0.1", port=1080,
+                  username="u1", password_env="AIBOX_TEST_SOCKS_PW")
+    assert "❌" in r
+    assert "环境变量" in r
+    assert "未保存" in r
+    assert fresh_config().get_proxy() is None
+
+
 # ==================== AskUserTool ====================
 
 def test_ask_user_interface():

aibox/tools/proxy_tools.py

@@ -3,18 +3,26 @@
 tools.proxy_tools - 代理配置工具
 
 抽离自 aibox.py，包含 2 个工具：
-  - SetProxyTool : 设置 SOCKS5 代理
+  - SetProxyTool : 设置 SOCKS5 代理（设置前验证连通性）
   - GetProxyTool : 查看当前代理配置
 
 依赖：common (ConfigManager) + tools.base.Tool
 不反向依赖 aibox 任何符号。
 """
 
-from typing import Dict
+import os
+from typing import Dict, Optional, Tuple
+
+import requests
 
 from common import ConfigManager
 from .base import Tool
 
+# 探测端点：无内容的轻量响应，验证完整 SOCKS5 握手（含认证）与出网连通性
+_PROBE_URL = "http://www.gstatic.com/generate_204"
+# (连接超时, 读超时)：验证总耗时上限约 10s，避免阻塞工具调用过久
+_PROBE_TIMEOUT = (4, 6)
+
 
 class SetProxyTool(Tool):
     """设置 SOCKS5 代理的工具"""
@@ -25,6 +33,8 @@ class SetProxyTool(Tool):
     def get_description(self) -> str:
         return """设置智能体的 SOCKS5 代理配置。
 调用后所有 HTTP/HTTPS 请求（包括网页访问、API 调用）都会通过该代理。
+启用代理时会先验证连通性（完整 SOCKS5 握手 + 探测请求），验证失败不保存配置；
+代理暂未启动等场景可用 skip_check=true 跳过验证强制保存。
 可随时修改，即时生效。"""
 
     def get_parameters(self) -> Dict:
@@ -37,13 +47,104 @@ class SetProxyTool(Tool):
                 "password": {"type": "string", "description": "密码（可选，明文仅本次会话有效，不落盘）"},
                 "password_env": {"type": "string", "description": "存有密码的环境变量名（可选，跨重启持久的推荐方式）"},
                 "enabled": {"type": "boolean", "description": "是否启用代理，默认 true"},
-                "save": {"type": "boolean", "description": "是否保存到配置文件（持久化），默认 true"}
+                "save": {"type": "boolean", "description": "是否保存到配置文件（持久化），默认 true"},
+                "skip_check": {"type": "boolean", "description": "跳过连通性验证强制保存（默认 false）", "default": False}
             },
             "required": ["host", "port"]
         }
 
+    @staticmethod
+    def _resolve_password(password: Optional[str], password_env: Optional[str]) -> Optional[str]:
+        """对齐 ConfigManager._resolve_proxy_password：明文优先，否则查环境变量"""
+        if password:
+            return password
+        if password_env:
+            return os.environ.get(password_env) or None
+        return None
+
+    def _check_connectivity(self, host: str, port: int,
+                            username: Optional[str], password: Optional[str]) -> Tuple[bool, str, str]:
+        """验证代理连通性：走一次完整 SOCKS5 握手 + 探测请求。
+
+        只做 TCP connect 验不了用户名/密码，必须发真实请求。
+        返回 (ok, error_kind, detail)；error_kind 取值：
+        auth_failed / timeout / connect_failed / probe_failed / error
+        """
+        if username and password:
+            proxy_url = f"socks5://{username}:{password}@{host}:{port}"
+        else:
+            proxy_url = f"socks5://{host}:{port}"
+        try:
+            # trust_env=False：排除系统环境变量代理（http_proxy 等）干扰，只验证目标代理
+            session = requests.Session()
+            session.trust_env = False
+            resp = session.get(
+                _PROBE_URL,
+                proxies={"http": proxy_url, "https": proxy_url},
+                timeout=_PROBE_TIMEOUT,
+            )
+            # generate_204 正常返回 204；部分代理/目标返回 200 也视为连通
+            if resp.status_code in (200, 204):
+                return True, "", f"探测成功 HTTP {resp.status_code}"
+            return False, "probe_failed", f"探测端点返回 HTTP {resp.status_code}"
+        except requests.exceptions.Timeout as e:
+            # 含 ConnectTimeout（连接阶段超时）：异常类型与系统语言无关，优先于消息匹配
+            return False, "timeout", str(e)
+        except requests.exceptions.ConnectionError as e:
+            # 连接层失败（拒绝/不可达/中断）。PySocks 依赖缺失与认证失败
+            # 也会以 ConnectionError 形态冒出，按消息细分。
+            msg = str(e)
+            low = msg.lower()
+            if "missing dependencies for socks" in low:
+                return False, "missing_dependency", msg
+            if "authentication" in low or "auth" in low:
+                return False, "auth_failed", msg
+            return False, "connect_failed", msg
+        except Exception as e:
+            # 兜底：仅按库自身（非本地化）消息关键词分类
+            msg = str(e)
+            low = msg.lower()
+            if "missing dependencies for socks" in low:
+                return False, "missing_dependency", msg
+            if "authentication" in low or "auth" in low:
+                return False, "auth_failed", msg
+            if "timed out" in low or "timeout" in low:
+                return False, "timeout", msg
+            if "refused" in low or "unreachable" in low or "closed" in low:
+                return False, "connect_failed", msg
+            return False, "error", msg
+
     def execute(self, host: str, port: int, username: str = None, password: str = None,
-                password_env: str = None, enabled: bool = True, save: bool = True, **kwargs) -> str:
+                password_env: str = None, enabled: bool = True, save: bool = True,
+                skip_check: bool = False, **kwargs) -> str:
+        # 设置前连通性验证：坏地址/凭据在入口一次性暴露，避免静默落盘后在
+        # 后续 web/API 调用中离场景报错、LLM 难以归因。skip_check 留给
+        # "代理暂未启动但先写配置"的场景；禁用代理无需验证。
+        check_detail = ""
+        if enabled and not skip_check:
+            pw = self._resolve_password(password, password_env)
+            # 指定了 password_env 但环境变量为空：认证凭据缺失，提前给出精准提示
+            if username and not password and password_env and pw is None:
+                return self.format_result(False, "代理验证失败，未保存配置", {
+                    "错误类型": "环境变量为空",
+                    "详情": f"环境变量 {password_env} 未设置或为空",
+                    "提示": "请先设置该环境变量，或改用 password 参数（仅本次会话有效）",
+                })
+            ok, kind, detail = self._check_connectivity(host, port, username, pw)
+            if not ok:
+                hints = {
+                    "auth_failed": "用户名/密码错误，请核对凭据",
+                    "timeout": "代理不可达（超时），请检查地址与网络",
+                    "connect_failed": "连接被拒绝或不可达，请检查 host/port",
+                    "missing_dependency": "缺少 SOCKS 支持库，请安装：pip install pysocks",
+                }
+                return self.format_result(False, "代理不可达，未保存配置", {
+                    "错误类型": kind,
+                    "详情": detail[:300],
+                    "提示": hints.get(kind, "如确认配置无误可设置 skip_check=true 跳过验证"),
+                })
+            check_detail = detail
+
         config = ConfigManager()
         config.set_proxy(host, port, username, password, password_env, enabled, save)
         proxy_info = config.get_proxy_info()
@@ -60,6 +161,8 @@ class SetProxyTool(Tool):
             "密码来源": pw_source,
             "持久化": "已保存" if save else "仅内存"
         }
+        if check_detail:
+            data["连通性"] = check_detail
         return self.format_result(True, "代理设置成功", data)