fix(security): prevent SSRF redirect bypass in Slack adapter

This commit is contained in:
Dusk1e 2026-04-10 13:40:12 +03:00 committed by Teknium
parent f4c7086035
commit 714809634f
2 changed files with 69 additions and 2 deletions

View file

@ -39,6 +39,7 @@ from gateway.platforms.base import (
MessageType,
SendResult,
SUPPORTED_DOCUMENT_TYPES,
_safe_url_for_log,
cache_document_from_bytes,
)
@ -656,8 +657,19 @@ class SlackAdapter(BasePlatformAdapter):
try:
import httpx
async def _ssrf_redirect_guard(response):
"""Re-check redirect targets so public URLs cannot bounce into private IPs."""
if response.is_redirect and response.next_request:
redirect_url = str(response.next_request.url)
if not is_safe_url(redirect_url):
raise ValueError("Blocked redirect to private/internal address")
# Download the image first
async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
async with httpx.AsyncClient(
timeout=30.0,
follow_redirects=True,
event_hooks={"response": [_ssrf_redirect_guard]},
) as client:
response = await client.get(image_url)
response.raise_for_status()
@ -674,7 +686,7 @@ class SlackAdapter(BasePlatformAdapter):
except Exception as e: # pragma: no cover - defensive logging
logger.warning(
"[Slack] Failed to upload image from URL %s, falling back to text: %s",
image_url,
_safe_url_for_log(image_url),
e,
exc_info=True,
)

View file

@ -1586,6 +1586,61 @@ class TestFallbackPreservesThreadContext:
assert "important screenshot" in call_kwargs["text"]
# ---------------------------------------------------------------------------
# TestSendImageSSRFGuards
# ---------------------------------------------------------------------------
class TestSendImageSSRFGuards:
"""send_image should reject redirects that land on private/internal hosts."""
@pytest.mark.asyncio
async def test_send_image_blocks_private_redirect_target(self, adapter):
redirect_response = MagicMock()
redirect_response.is_redirect = True
redirect_response.next_request = MagicMock(
url="http://169.254.169.254/latest/meta-data"
)
client_kwargs = {}
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
async def fake_get(_url):
for hook in client_kwargs["event_hooks"]["response"]:
await hook(redirect_response)
mock_client.get = AsyncMock(side_effect=fake_get)
adapter._app.client.files_upload_v2 = AsyncMock(return_value={"ok": True})
adapter._app.client.chat_postMessage = AsyncMock(return_value={"ts": "reply_ts"})
def fake_async_client(*args, **kwargs):
client_kwargs.update(kwargs)
return mock_client
def fake_is_safe_url(url):
return url == "https://public.example/image.png"
with (
patch("tools.url_safety.is_safe_url", side_effect=fake_is_safe_url),
patch("httpx.AsyncClient", side_effect=fake_async_client),
):
result = await adapter.send_image(
chat_id="C123",
image_url="https://public.example/image.png",
caption="see this",
)
assert result.success
assert client_kwargs["follow_redirects"] is True
assert client_kwargs["event_hooks"]["response"]
adapter._app.client.files_upload_v2.assert_not_awaited()
adapter._app.client.chat_postMessage.assert_awaited_once()
call_kwargs = adapter._app.client.chat_postMessage.call_args.kwargs
assert "see this" in call_kwargs["text"]
assert "https://public.example/image.png" in call_kwargs["text"]
# ---------------------------------------------------------------------------
# TestProgressMessageThread
# ---------------------------------------------------------------------------