fix: harden proxy redaction and add regression tests

Author: whatevertogo

astrbot/core/provider/sources/openai_embedding_source.py

@@ -1,7 +1,5 @@
-import httpx
 from openai import AsyncOpenAI
 
-from astrbot import logger
 from astrbot.core.utils.network_utils import create_proxy_client
 
 from ..entities import ProviderType

astrbot/core/provider/sources/openai_tts_api_source.py

@@ -1,10 +1,8 @@
 import os
 import uuid
 
-import httpx
 from openai import NOT_GIVEN, AsyncOpenAI
 
-from astrbot import logger
 from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
 from astrbot.core.utils.network_utils import create_proxy_client
 

astrbot/core/utils/network_utils.py

@@ -73,9 +73,12 @@ def log_connection_failure(
 
     if effective_proxy:
         sanitized_proxy = _sanitize_proxy_url(effective_proxy)
+        error_text = str(error)
+        if effective_proxy:
+            error_text = error_text.replace(effective_proxy, sanitized_proxy)
         logger.error(
             f"[{provider_label}] 网络/代理连接失败 ({error_type})。"
-            f"代理地址: {sanitized_proxy}，错误: {error}"
+            f"代理地址: {sanitized_proxy}，错误: {error_text}"
         )
     else:
         logger.error(f"[{provider_label}] 网络连接失败 ({error_type})。错误: {error}")
@@ -94,27 +97,36 @@ def _is_socks_proxy(proxy: str) -> bool:
 
 
 def _sanitize_proxy_url(proxy: str) -> str:
-    """Sanitize proxy URL by masking password for safe logging.
+    """Sanitize proxy URL by masking credentials for safe logging.
 
     Args:
         proxy: The proxy URL string
 
     Returns:
-        Sanitized proxy URL with password masked (e.g., "http://user:****@host:port")
+        Sanitized proxy URL with credentials masked (e.g., "http://****@host:port")
     """
     try:
         from urllib.parse import urlparse, urlunparse
 
         parsed = urlparse(proxy)
-        if parsed.password:
-            # Replace password with asterisks
-            netloc = f"{parsed.username}:****@{parsed.hostname}"
+        # Any userinfo in netloc should be masked to avoid leaking tokens/passwords.
+        if "@" in parsed.netloc and parsed.hostname:
+            host = parsed.hostname
+            if ":" in host and not host.startswith("["):
+                host = f"[{host}]"
+            netloc = f"****@{host}"
             if parsed.port:
                 netloc += f":{parsed.port}"
-            sanitized = urlunparse(
-                (parsed.scheme, netloc, parsed.path, "", "", "")
+            return urlunparse(
+                (
+                    parsed.scheme,
+                    netloc,
+                    parsed.path,
+                    parsed.params,
+                    parsed.query,
+                    parsed.fragment,
+                )
             )
-            return sanitized
     except Exception:
         pass
     return proxy

tests/unit/test_network_utils.py

@@ -0,0 +1,44 @@
+"""Tests for network utility helpers."""
+
+from astrbot.core.utils import network_utils
+
+
+def test_sanitize_proxy_url_masks_password_credentials():
+    proxy = "http://user:secret@127.0.0.1:1080"
+    assert network_utils._sanitize_proxy_url(proxy) == "http://****@127.0.0.1:1080"
+
+
+def test_sanitize_proxy_url_masks_username_only_credentials():
+    proxy = "http://token@127.0.0.1:1080"
+    assert network_utils._sanitize_proxy_url(proxy) == "http://****@127.0.0.1:1080"
+
+
+def test_sanitize_proxy_url_masks_empty_password_credentials():
+    proxy = "http://user:@127.0.0.1:1080"
+    assert network_utils._sanitize_proxy_url(proxy) == "http://****@127.0.0.1:1080"
+
+
+def test_is_socks_proxy_detects_supported_schemes():
+    assert network_utils._is_socks_proxy("socks5://127.0.0.1:1080")
+    assert network_utils._is_socks_proxy("socks4://127.0.0.1:1080")
+    assert network_utils._is_socks_proxy("socks5h://127.0.0.1:1080")
+    assert not network_utils._is_socks_proxy("http://127.0.0.1:1080")
+
+
+def test_log_connection_failure_redacts_proxy_in_error_text(monkeypatch):
+    proxy = "http://token@127.0.0.1:1080"
+    captured = {}
+
+    def fake_error(message: str):
+        captured["message"] = message
+
+    monkeypatch.setattr(network_utils.logger, "error", fake_error)
+
+    network_utils.log_connection_failure(
+        provider_label="OpenAI",
+        error=RuntimeError(f"proxy connect failed: {proxy}"),
+        proxy=proxy,
+    )
+
+    assert "http://token@127.0.0.1:1080" not in captured["message"]
+    assert "http://****@127.0.0.1:1080" in captured["message"]