fix(telegram): expose proxy setup in WebUI (#5033)

This commit is contained in:
chengyongru
2026-07-23 14:18:49 +08:00
committed by GitHub
parent 3647875aba
commit 01cdfc8100
17 changed files with 472 additions and 24 deletions
+1
View File
@@ -8,6 +8,7 @@ from nanobot.channels.telegram.validation import validate
SETUP_SPEC = ChannelSetupSpec(
fields={
"token": field("secret"),
"proxy": field("secret"),
"allowFrom": field("list"),
"groupPolicy": field("enum", choices=GROUP_POLICIES, default="mention"),
},
@@ -4,11 +4,60 @@ import httpx
import pytest
from nanobot.channels.telegram import validation as telegram_validation
from nanobot.channels.telegram.manifest import SETUP_SPEC
from nanobot.channels.validation import validate_channel_config
from nanobot.config.loader import save_config
from nanobot.config.schema import Config
def test_telegram_setup_exposes_proxy_as_an_optional_secret() -> None:
proxy = SETUP_SPEC.fields["proxy"]
assert proxy.kind == "secret"
assert "proxy" not in SETUP_SPEC.simple_required_fields
def test_get_me_builds_http_client_with_explicit_proxy(
monkeypatch: pytest.MonkeyPatch,
) -> None:
token = "123456:abcdefghijklmnopqrstuvwxyz"
proxy = "socks5://proxy-user:proxy-pass@127.0.0.1:1080"
captured: dict[str, object] = {}
class FakeResponse:
def raise_for_status(self) -> None:
return None
def json(self) -> dict:
return {"ok": True, "result": {"id": 42}}
class FakeClient:
def __init__(self, **kwargs) -> None:
captured["kwargs"] = kwargs
def __enter__(self):
return self
def __exit__(self, *_args) -> None:
return None
def get(self, url: str) -> FakeResponse:
captured["url"] = url
return FakeResponse()
monkeypatch.setattr(telegram_validation.httpx, "Client", FakeClient)
result = telegram_validation._get_me(token, proxy)
assert result["ok"] is True
assert captured["kwargs"] == {
"timeout": 4.0,
"proxy": proxy,
"trust_env": False,
}
assert captured["url"] == f"https://api.telegram.org/bot{token}/getMe"
def test_validate_telegram_bad_token_is_invalid(tmp_path, monkeypatch: pytest.MonkeyPatch) -> None:
config_path = tmp_path / "config.json"
save_config(Config(), config_path)
@@ -21,7 +70,38 @@ def test_validate_telegram_bad_token_is_invalid(tmp_path, monkeypatch: pytest.Mo
assert result["missing_fields"] == []
def test_validate_telegram_does_not_expose_saved_token_in_http_errors(
@pytest.mark.parametrize("status_code", [401, 404])
def test_validate_telegram_rejects_denied_tokens_without_exposing_them(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
status_code: int,
) -> None:
token = "123456:abcdefghijklmnopqrstuvwxyz"
config_path = tmp_path / "config.json"
save_config(
Config.model_validate({"channels": {"telegram": {"token": token}}}),
config_path,
)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
def raise_http_error(token_value: str, _proxy: str | None) -> dict:
request = httpx.Request("GET", f"https://api.telegram.org/bot{token_value}/getMe")
response = httpx.Response(status_code, request=request)
raise httpx.HTTPStatusError("rejected", request=request, response=response)
monkeypatch.setattr(telegram_validation, "_get_me", raise_http_error)
result = validate_channel_config("telegram", {"channels.telegram.token": ""})
assert result["status"] == "invalid"
assert result["can_enable"] is False
assert token not in str(result)
assert any(
f"HTTP {status_code}" in check.get("message", "") for check in result["checks"]
)
def test_validate_telegram_keeps_transient_http_failures_retryable(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
@@ -33,14 +113,197 @@ def test_validate_telegram_does_not_expose_saved_token_in_http_errors(
)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
def raise_http_error(url: str, **_kwargs) -> dict:
request = httpx.Request("GET", url)
response = httpx.Response(401, request=request)
raise httpx.HTTPStatusError("unauthorized", request=request, response=response)
def raise_http_error(token_value: str, _proxy: str | None) -> dict:
request = httpx.Request("GET", f"https://api.telegram.org/bot{token_value}/getMe")
response = httpx.Response(503, request=request)
raise httpx.HTTPStatusError("unavailable", request=request, response=response)
monkeypatch.setattr(telegram_validation, "http_get", raise_http_error)
monkeypatch.setattr(telegram_validation, "_get_me", raise_http_error)
result = validate_channel_config("telegram", {"channels.telegram.token": ""})
assert result["status"] == "configured"
assert result["can_enable"] is True
assert token not in str(result)
assert any("HTTP 401" in check.get("message", "") for check in result["checks"])
assert any("HTTP 503" in check.get("message", "") for check in result["checks"])
def test_validate_telegram_marks_proxy_transport_failures_without_exposing_proxy(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
token = "123456:abcdefghijklmnopqrstuvwxyz"
proxy = "http://proxy-user:proxy-pass@127.0.0.1:7890"
config_path = tmp_path / "config.json"
save_config(
Config.model_validate(
{"channels": {"telegram": {"token": token, "proxy": proxy}}}
),
config_path,
)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
def raise_proxy_error(_token: str, _proxy: str | None) -> dict:
raise httpx.ProxyError("proxy credentials rejected")
monkeypatch.setattr(telegram_validation, "_get_me", raise_proxy_error)
result = validate_channel_config("telegram")
assert result["status"] == "configured"
assert result["can_enable"] is True
assert proxy not in str(result)
assert any(check["id"] == "proxy_connection" for check in result["checks"])
def test_validate_telegram_uses_saved_proxy_without_exposing_it(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
token = "123456:abcdefghijklmnopqrstuvwxyz"
proxy = "socks5://proxy-user:proxy-pass@127.0.0.1:1080"
config_path = tmp_path / "config.json"
save_config(
Config.model_validate(
{"channels": {"telegram": {"token": token, "proxy": proxy}}}
),
config_path,
)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
captured: dict[str, str | None] = {}
def fake_get_me(token_value: str, proxy_value: str | None) -> dict:
captured.update(token=token_value, proxy=proxy_value)
return {"ok": True, "result": {"id": 42, "username": "working_bot"}}
monkeypatch.setattr(telegram_validation, "_get_me", fake_get_me)
result = validate_channel_config("telegram")
assert result["status"] == "connected"
assert captured == {"token": token, "proxy": proxy}
assert proxy not in str(result)
def test_validate_telegram_resolves_saved_secret_env_refs_without_exposing_them(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
token = "123456:abcdefghijklmnopqrstuvwxyz"
token_ref = "${TELEGRAM_TOKEN_TEST}"
proxy_ref = "${TELEGRAM_PROXY_TEST}"
proxy = "http://proxy-user:proxy-pass@127.0.0.1:7890"
monkeypatch.setenv("TELEGRAM_TOKEN_TEST", token)
monkeypatch.setenv("TELEGRAM_PROXY_TEST", proxy)
config_path = tmp_path / "config.json"
save_config(
Config.model_validate(
{"channels": {"telegram": {"token": token_ref, "proxy": proxy_ref}}}
),
config_path,
)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
captured: dict[str, str | None] = {}
def fake_get_me(token_value: str, proxy_value: str | None) -> dict:
captured.update(token=token_value, proxy=proxy_value)
return {"ok": True, "result": {"id": 42, "username": "working_bot"}}
monkeypatch.setattr(telegram_validation, "_get_me", fake_get_me)
result = validate_channel_config("telegram")
assert result["status"] == "connected"
assert captured == {"token": token, "proxy": proxy}
assert token_ref not in str(result)
assert token not in str(result)
assert proxy_ref not in str(result)
assert proxy not in str(result)
def test_validate_telegram_rejects_unset_proxy_env_ref_without_connecting(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
token = "123456:abcdefghijklmnopqrstuvwxyz"
proxy_ref = "${TELEGRAM_MISSING_PROXY_TEST}"
monkeypatch.delenv("TELEGRAM_MISSING_PROXY_TEST", raising=False)
config_path = tmp_path / "config.json"
save_config(
Config.model_validate(
{"channels": {"telegram": {"token": token, "proxy": proxy_ref}}}
),
config_path,
)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
def fail_get_me(*_args) -> dict:
pytest.fail("an unresolved proxy reference must not fall back to direct access")
monkeypatch.setattr(telegram_validation, "_get_me", fail_get_me)
result = validate_channel_config("telegram")
assert result["status"] == "invalid"
assert result["can_enable"] is False
assert proxy_ref not in str(result)
assert any(check["id"] == "proxy_env" for check in result["checks"])
def test_validate_telegram_uses_proxy_submitted_with_new_token(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
token = "123456:abcdefghijklmnopqrstuvwxyz"
proxy = "http://127.0.0.1:7890"
config_path = tmp_path / "config.json"
save_config(Config(), config_path)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
captured: dict[str, str | None] = {}
def fake_get_me(token_value: str, proxy_value: str | None) -> dict:
captured.update(token=token_value, proxy=proxy_value)
return {"ok": True, "result": {"id": 42, "username": "new_bot"}}
monkeypatch.setattr(telegram_validation, "_get_me", fake_get_me)
result = validate_channel_config(
"telegram",
{
"channels.telegram.token": token,
"channels.telegram.proxy": proxy,
},
)
assert result["status"] == "connected"
assert captured == {"token": token, "proxy": proxy}
@pytest.mark.parametrize("proxy", ["127.0.0.1:7890", "http://[", "http://localhost:not-a-port"])
def test_validate_telegram_rejects_invalid_proxy_without_trying_token(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
proxy: str,
) -> None:
token = "123456:abcdefghijklmnopqrstuvwxyz"
config_path = tmp_path / "config.json"
save_config(Config(), config_path)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
def fail_get_me(*_args) -> dict:
pytest.fail("invalid proxy must stop before getMe")
monkeypatch.setattr(telegram_validation, "_get_me", fail_get_me)
result = validate_channel_config(
"telegram",
{
"channels.telegram.token": token,
"channels.telegram.proxy": proxy,
},
)
assert result["status"] == "invalid"
assert result["can_enable"] is False
assert proxy not in str(result)
assert any(check["id"] == "proxy_format" for check in result["checks"])
+82 -5
View File
@@ -2,24 +2,82 @@
import re
from typing import Any
from urllib.parse import urlparse
import httpx
from nanobot.channels.contracts import ChannelValidationContext
from nanobot.channels.validation import (
check,
http_get,
message_from_response,
payload,
required_checks,
status_from_checks,
string_value,
)
from nanobot.config.loader import resolve_env_refs
_TIMEOUT_SECONDS = 4.0
_SUPPORTED_PROXY_SCHEMES = {"http", "https", "socks5", "socks5h"}
def _proxy_url_is_valid(proxy: str) -> bool:
try:
parsed = urlparse(proxy)
hostname = parsed.hostname
parsed.port
except ValueError:
return False
return parsed.scheme.lower() in _SUPPORTED_PROXY_SCHEMES and bool(hostname)
def _get_me(token: str, proxy: str | None) -> dict[str, Any]:
client_kwargs: dict[str, Any] = {"timeout": _TIMEOUT_SECONDS}
if proxy:
client_kwargs.update(proxy=proxy, trust_env=False)
with httpx.Client(**client_kwargs) as client:
response = client.get(f"https://api.telegram.org/bot{token}/getMe")
response.raise_for_status()
data = response.json()
return data if isinstance(data, dict) else {}
def validate(values: dict[str, Any], _context: ChannelValidationContext) -> dict[str, Any]:
checks, missing = required_checks("telegram", values)
token = string_value(values.get("token"))
raw_token = string_value(values.get("token"))
raw_proxy = string_value(values.get("proxy"))
token = string_value(resolve_env_refs(raw_token))
proxy = string_value(resolve_env_refs(raw_proxy))
if raw_token and not token:
checks.append(
check(
"token_env",
"Token environment variable",
"fail",
"Set every environment variable referenced by the bot token.",
)
)
if raw_proxy and not proxy:
checks.append(
check(
"proxy_env",
"Proxy environment variable",
"fail",
"Set every environment variable referenced by the network proxy.",
)
)
if (raw_token and not token) or (raw_proxy and not proxy):
return status_from_checks("telegram", checks, missing)
if proxy and not _proxy_url_is_valid(proxy):
checks.append(
check(
"proxy_format",
"Network proxy",
"fail",
"Enter a full HTTP or SOCKS proxy URL.",
)
)
return status_from_checks("telegram", checks, missing)
if token:
if not re.match(r"^\d+:[A-Za-z0-9_-]{20,}$", token):
checks.append(
@@ -35,7 +93,7 @@ def validate(values: dict[str, Any], _context: ChannelValidationContext) -> dict
check("token_format", "Token format", "pass", "Looks like a BotFather token.")
)
try:
data = http_get(f"https://api.telegram.org/bot{token}/getMe")
data = _get_me(token, proxy or None)
if data.get("ok") and isinstance(data.get("result"), dict):
bot = data["result"]
identity = {
@@ -61,12 +119,31 @@ def validate(values: dict[str, Any], _context: ChannelValidationContext) -> dict
)
)
except httpx.HTTPStatusError as exc:
status_code = exc.response.status_code
rejected = status_code in {400, 401, 403, 404}
checks.append(
check(
"get_me",
"Bot identity",
"fail" if rejected else "warn",
(
f"Telegram rejected the token: HTTP {status_code}."
if rejected
else f"Telegram could not verify the token: HTTP {status_code}."
),
)
)
except httpx.TransportError:
checks.append(
check(
"proxy_connection" if proxy else "get_me",
"Network proxy" if proxy else "Bot identity",
"warn",
f"Telegram could not verify the token: HTTP {exc.response.status_code}.",
(
"Could not reach Telegram through the network proxy."
if proxy
else "Could not reach Telegram now. Try again later."
),
)
)
except Exception:
@@ -75,7 +152,7 @@ def validate(values: dict[str, Any], _context: ChannelValidationContext) -> dict
"get_me",
"Bot identity",
"warn",
"Could not reach Telegram now. Try again later.",
"Could not verify Telegram now. Try again later.",
)
)
return status_from_checks("telegram", checks, missing)
+1
View File
@@ -12,6 +12,7 @@ export default {
docsUrl: chatAppGuideUrl("telegram"),
fields: [
{ key: "channels.telegram.token" },
{ key: "channels.telegram.proxy" },
{ key: "channels.telegram.allowFrom" },
{ key: "channels.telegram.groupPolicy" },
],
@@ -17,6 +17,10 @@
"placeholder": "123456:ABC...",
"help": "Create it with BotFather."
},
"proxy": {
"label": "Network proxy",
"placeholder": "http://127.0.0.1:7890"
},
"allowFrom": {
"label": "Allowed users",
"placeholder": "* or Telegram user IDs",
@@ -17,6 +17,10 @@
"placeholder": "123456:ABC...",
"help": "Créalo con BotFather."
},
"proxy": {
"label": "Proxy de red",
"placeholder": "http://127.0.0.1:7890"
},
"allowFrom": {
"label": "Usuarios permitidos",
"placeholder": "* o ID de usuario de Telegram",
@@ -17,6 +17,10 @@
"placeholder": "123456:ABC...",
"help": "Créez-le avec BotFather."
},
"proxy": {
"label": "Proxy réseau",
"placeholder": "http://127.0.0.1:7890"
},
"allowFrom": {
"label": "Utilisateurs autorisés",
"placeholder": "* ou ID utilisateur Telegram",
@@ -17,6 +17,10 @@
"placeholder": "123456:ABC...",
"help": "Buat dengan BotFather."
},
"proxy": {
"label": "Proxy jaringan",
"placeholder": "http://127.0.0.1:7890"
},
"allowFrom": {
"label": "Pengguna yang diizinkan",
"placeholder": "* atau ID pengguna Telegram",
@@ -17,6 +17,10 @@
"placeholder": "123456:ABC...",
"help": "BotFather で作成します。"
},
"proxy": {
"label": "ネットワークプロキシ",
"placeholder": "http://127.0.0.1:7890"
},
"allowFrom": {
"label": "許可するユーザー",
"placeholder": "* または Telegram ユーザー ID",
@@ -17,6 +17,10 @@
"placeholder": "123456:ABC...",
"help": "BotFather에서 생성하세요."
},
"proxy": {
"label": "네트워크 프록시",
"placeholder": "http://127.0.0.1:7890"
},
"allowFrom": {
"label": "허용된 사용자",
"placeholder": "* 또는 Telegram 사용자 ID",
@@ -17,6 +17,10 @@
"placeholder": "123456:ABC...",
"help": "Crie-o com o BotFather."
},
"proxy": {
"label": "Proxy de rede",
"placeholder": "http://127.0.0.1:7890"
},
"allowFrom": {
"label": "Usuários permitidos",
"placeholder": "* ou IDs de usuário do Telegram",
@@ -17,6 +17,10 @@
"placeholder": "123456:ABC...",
"help": "Tạo bằng BotFather."
},
"proxy": {
"label": "Proxy mạng",
"placeholder": "http://127.0.0.1:7890"
},
"allowFrom": {
"label": "Người dùng được phép",
"placeholder": "* hoặc ID người dùng Telegram",
@@ -17,6 +17,10 @@
"placeholder": "123456:ABC...",
"help": "使用 BotFather 创建。"
},
"proxy": {
"label": "网络代理",
"placeholder": "http://127.0.0.1:7890"
},
"allowFrom": {
"label": "允许的用户",
"placeholder": "* 或 Telegram 用户 ID",
@@ -17,6 +17,10 @@
"placeholder": "123456:ABC...",
"help": "使用 BotFather 建立。"
},
"proxy": {
"label": "網路代理",
"placeholder": "http://127.0.0.1:7890"
},
"allowFrom": {
"label": "允許的使用者",
"placeholder": "* 或 Telegram 使用者 ID",