diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 71c244976ae..a70cb33b10d 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -581,6 +581,11 @@ _POLLING_ERROR_TASK_STUCK_TIMEOUT = 300.0 # A generation is not healthy until the dedicated getUpdates request returns # successfully. This exceeds a normal long-poll cycle for healthy idle bots. _POLLING_PROGRESS_TIMEOUT = 60.0 +# Telegram transcodes an uploaded video before it answers sendVideo, so the +# wait for the response is unrelated to how fast the bytes went out and +# routinely exceeds the 20s read timeout the rest of the Bot API is tuned for. +# Only media sends take this longer budget. +_MEDIA_SEND_READ_TIMEOUT = 180.0 _POLLING_GENERATION_CONTEXT: ContextVar[Optional[int]] = ContextVar( "telegram_polling_generation", default=None ) @@ -3664,6 +3669,13 @@ class TelegramAdapter(BasePlatformAdapter): "connect_timeout": _env_float("HERMES_TELEGRAM_HTTP_CONNECT_TIMEOUT", 10.0), "read_timeout": _env_float("HERMES_TELEGRAM_HTTP_READ_TIMEOUT", 20.0), "write_timeout": _env_float("HERMES_TELEGRAM_HTTP_WRITE_TIMEOUT", 20.0), + # Not a duplicate of write_timeout: PTB routes any request + # carrying files to media_write_timeout instead, so the line + # above never applied to an upload and every upload was pinned + # to PTB's own 20s default. + "media_write_timeout": _env_float( + "HERMES_TELEGRAM_HTTP_MEDIA_WRITE_TIMEOUT", 300.0 + ), } # CLOSE_WAIT fd leak (#31599, same class as #18451): PTB's @@ -7196,6 +7208,7 @@ class TelegramAdapter(BasePlatformAdapter): "video": f, "caption": caption[:1024] if caption else None, "reply_to_message_id": reply_to_id, + "read_timeout": _MEDIA_SEND_READ_TIMEOUT, **thread_kwargs, **self._notification_kwargs(metadata), }, diff --git a/tests/tools/test_managed_tool_gateway.py b/tests/tools/test_managed_tool_gateway.py index 1757d5141fa..5a02ddff471 100644 --- a/tests/tools/test_managed_tool_gateway.py +++ b/tests/tools/test_managed_tool_gateway.py @@ -1,3 +1,4 @@ +import asyncio import os import json from datetime import datetime, timedelta, timezone @@ -178,6 +179,187 @@ def test_managed_gateway_auth_headers_empty_without_a_token(): ) == {} +class TestManagedMediaUploader: + """The presign -> PUT -> ``nous-upload:`` protocol. + + This is the only way a local image or video reaches a managed vendor, and + the pieces it gets right are not incidental: the presigned URL signs the + content type and byte length, so a PUT that disagrees with the presign is + rejected by storage rather than by us. + """ + + GATEWAY = "https://tool-gateway.example.com" + BASE_URL = f"{GATEWAY}/api/bfl" + UPLOAD_PATH = "/api/uploads/bfl" + + @staticmethod + def _builder(vendor): + return f"https://{vendor}-gateway.example.com" + + def _uploader(self, **kwargs): + return managed_tool_gateway.build_managed_media_uploader( + kwargs.pop("server_url", self.BASE_URL), + kwargs.pop("upload_path", self.UPLOAD_PATH), + gateway_builder=lambda vendor: self.GATEWAY, + token_reader=kwargs.pop("token_reader", lambda: "nous-token"), + ) + + @staticmethod + def _response(status_code=200, payload=None): + class _R: + def __init__(self): + self.status_code = status_code + + def json(self): + if payload is None: + raise ValueError("no json") + return payload + + return _R() + + def _run(self, uploader, data=b"bytes", mime="image/png", presign=None, put=None): + """Drive one upload with both HTTP legs stubbed; returns the calls made.""" + import httpx + + from tools import url_safety + + calls = {"presign": [], "put": []} + presign = presign if presign is not None else self._response( + 200, {"uploadUrl": "https://storage.example/put?sig=abc", "token": "tok-1"} + ) + put = put if put is not None else self._response(200) + + class _PresignClient: + def __init__(self, **_kw): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *_exc): + return False + + async def post(self, url, headers=None, json=None): + calls["presign"].append({"url": url, "headers": headers, "json": json}) + return presign + + class _PutClient: + async def __aenter__(self): + return self + + async def __aexit__(self, *_exc): + return False + + async def put(self, url, content=None, headers=None): + calls["put"].append({"url": url, "content": content, "headers": headers}) + return put + + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True), \ + patch.object(httpx, "AsyncClient", _PresignClient), \ + patch.object(url_safety, "create_ssrf_safe_async_client", lambda **_kw: _PutClient()): + calls["result"] = asyncio.run(uploader(data, mime)) + return calls + + def test_presign_declares_the_exact_type_and_length_the_put_then_sends(self): + # Storage validates the PUT against what was signed, so a mismatch + # between these two is a rejection with no useful error. + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + uploader = self._uploader() + data = b"\x89PNG\r\n\x1a\n" + b"payload" * 100 + + calls = self._run(uploader, data=data, mime="image/png") + + assert calls["presign"][0]["url"] == f"{self.GATEWAY}{self.UPLOAD_PATH}" + assert calls["presign"][0]["json"] == { + "contentType": "image/png", + "contentLength": len(data), + } + assert calls["presign"][0]["headers"]["Authorization"] == "Bearer nous-token" + assert calls["put"][0]["url"] == "https://storage.example/put?sig=abc" + assert calls["put"][0]["content"] == data + assert calls["put"][0]["headers"] == {"Content-Type": "image/png"} + assert calls["result"] == "nous-upload:tok-1" + + def test_the_bytes_go_to_storage_and_never_through_the_gateway(self): + # The whole point of presigning is that the gateway's request-size + # ceiling does not apply to a 50MB clip. + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + uploader = self._uploader() + + calls = self._run(uploader, data=b"v" * 4096, mime="video/mp4") + + assert len(calls["presign"]) == 1 and len(calls["put"]) == 1 + assert self.GATEWAY not in calls["put"][0]["url"] + assert calls["presign"][0]["json"]["contentType"] == "video/mp4" + + def test_no_uploader_when_the_url_is_not_a_managed_gateway(self): + # Refusing to build is what makes the caller say "pass a URL instead" + # rather than forwarding a raw local path to a third party. + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + assert self._uploader(server_url="https://attacker.example/api/bfl") is None + + @pytest.mark.parametrize("upload_path", [None, "", "api/uploads/bfl", 42]) + def test_no_uploader_without_a_rooted_upload_path(self, upload_path): + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + assert self._uploader(upload_path=upload_path) is None + + def test_a_missing_credential_fails_before_any_request(self): + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + uploader = self._uploader() + + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True), \ + patch.object(managed_tool_gateway, "managed_gateway_auth_headers", return_value={}): + with pytest.raises(RuntimeError, match="no Nous credential"): + asyncio.run(uploader(b"x", "image/png")) + + def test_a_gateway_refusal_surfaces_its_own_message(self): + # Quota and size refusals carry guidance written for the model; a bare + # status code would throw that away. + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + uploader = self._uploader() + refusal = self._response( + 413, {"error": {"message": "That file is 82MB; the limit for video is 50MB."}} + ) + + with pytest.raises(RuntimeError, match="the limit for video is 50MB"): + self._run(uploader, presign=refusal) + + def test_an_unreadable_refusal_still_reports_the_status(self): + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + uploader = self._uploader() + + with pytest.raises(RuntimeError, match="HTTP 502"): + self._run(uploader, presign=self._response(502, None)) + + @pytest.mark.parametrize( + "payload", + [ + {}, + {"uploadUrl": "https://storage.example/put"}, + {"token": "tok-1"}, + {"uploadUrl": "", "token": "tok-1"}, + {"uploadUrl": "https://storage.example/put", "token": ""}, + ], + ) + def test_a_malformed_presign_response_is_refused_rather_than_guessed(self, payload): + # Half a presign must not become a PUT to nowhere or an empty token + # that later reads as a valid reference. + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + uploader = self._uploader() + + with pytest.raises(RuntimeError, match="malformed"): + self._run(uploader, presign=self._response(200, payload)) + + def test_a_storage_rejection_is_not_reported_as_a_successful_upload(self): + # A signature mismatch answers non-200 with an XML body; returning a + # token here would hand the vendor a reference to nothing. + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + uploader = self._uploader() + + with pytest.raises(RuntimeError, match="storage refused the upload"): + self._run(uploader, put=self._response(403)) + + def test_is_managed_tool_gateway_ready_skips_refresh_for_expired_cached_token(tmp_path, monkeypatch): monkeypatch.delenv("TOOL_GATEWAY_USER_TOKEN", raising=False) monkeypatch.setenv("HERMES_HOME", str(tmp_path))