diff --git a/aibox/tests/test_b_tools.py b/aibox/tests/test_b_tools.py index e81b1e9..6563c50 100644 --- a/aibox/tests/test_b_tools.py +++ b/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(): diff --git a/aibox/tools/proxy_tools.py b/aibox/tools/proxy_tools.py index 426543c..350c8f8 100644 --- a/aibox/tools/proxy_tools.py +++ b/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)