Enforce body-size limits on chunked webhook requests

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #58536 fix(gateway): enforce body-size limits on chunked requests (salvage #3955 + #3949)

  merged to main

  files: gateway/platforms/api_server.py, gateway/platforms/webhook.py, tests/gateway/test_webhook_adapter.py

• #58902 security(raft): enforce body-size limit on chunked requests

  merged to main

  files: plugins/platforms/raft/adapter.py, tests/gateway/test_raft_adapter.py

  Show the diff

  • plugins/platforms/raft/adapter.py

    @@ -473,7 +473,11 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
                 self._bridge_token = [redacted:secret-assignment]
                 logger.info("[raft] Auto-generated bridge token")
     
    -        app = web.Application()
    +        # client_max_size makes aiohttp enforce the cap on every read path,
    +        # including Transfer-Encoding: chunked bodies that carry no
    +        # Content-Length and would otherwise bypass the header checks below
    +        # (mirrors gateway/platforms/webhook.py's connect()).
    +        app = web.Application(client_max_size=self._max_body_bytes)
             app.router.add_get("/health", self._handle_health)
             app.router.add_post(self._path, self._handle_wake)
             app.router.add_post("/activity", self._handle_activity)
    @@ -600,8 +604,16 @@ async def _handle_wake(self, request: "web.Request") -> "web.Response":
     
             try:
                 raw_body = await request.read()
    +        except web.HTTPRequestEntityTooLarge:
    +            # aiohttp's client_max_size tripped — chunked or lying
    +            # Content-Length. Same 413 as the header check above.
    +            return web.json_response({"ok": False, "error": "payload_too_large"}, status=413)
             except Exception:
                 return web.json_response({"ok": False, "error": "bad_request"}, status=400)
    +        if len(raw_body) > self._max_body_bytes:
    +            # Defense in depth: enforce the cap on the actual bytes read even
    +            # if the server-level limit was bypassed or misconfigured.
    +            return web.json_response({"ok": False, "error": "payload_too_large"}, status=413)
     
             payload: Dict[str, Any] = {}
             if raw_body.strip():
    @@ -646,7 +658,20 @@ async def _handle_activity(self, request: "web.Request") -> "web.Response":
                 return web.json_response({"ok": False, "error": "payload_too_large"}, status=413)
     
             try:
    -            payload = json.loads(await request.text())
    +            raw_text = await request.text()
    +        except web.HTTPRequestEntityTooLarge:
    +            # aiohttp's client_max_size tripped — chunked or lying
    +            # Content-Length. Same 413 as the header check above.
    +            return web.json_response({"ok": False, "error": "payload_too_large"}, status=413)
    +        except Exception as exc:
    +            return web.json_response({"ok": False, "error": str(exc)}, status=400)
    +        if len(raw_text.encode("utf-8")) > self._max_body_bytes:
    +            # Defense in depth: enforce the cap on the actual bytes read even
    +            # if the server-level limit was bypassed or misconfigured.
    +            return web.json_response({"ok": False, "error": "payload_too_large"}, status=413)
    +
    +        try:
    +            payload = json.loads(raw_text)
                 self._activity_queue.push(payload)
             except json.JSONDecodeError:
                 return web.json_response({"ok": False, "error": "invalid_json"}, status=400)

  • tests/gateway/test_raft_adapter.py

    @@ -1,5 +1,7 @@
     """Tests for the Raft channel adapter."""
     
    +import asyncio
    +import json
     import os
     from unittest.mock import AsyncMock, patch
     
    @@ -56,7 +58,8 @@ def _make_adapter(**extra):
     
     
     def _create_app(adapter: RaftAdapter) -> web.Application:
    -    app = web.Application()
    +    # Mirror connect(): client_max_size enforces the cap on chunked bodies.
    +    app = web.Application(client_max_size=adapter._max_body_bytes)
         app.router.add_get("/health", adapter._handle_health)
         app.router.add_post(adapter._path, adapter._handle_wake)
         app.router.add_post("/activity", adapter._handle_activity)
    @@ -407,6 +410,68 @@ def test_interrupted_turn_reports_error_stop(self):
             assert drain["events"][1]["errorClass"] == "interrupted"
     
     
    +class TestBodySize:
    +    """The wake/activity endpoints enforced max_body_bytes only via the
    +    Content-Length header; a Transfer-Encoding: chunked request
    +    (content_length=None) bypassed the cap entirely and read the full body,
    +    bounded only by aiohttp's implicit 1 MiB default. Mirrors
    +    gateway/platforms/webhook.py's TestBodySize."""
    +
    +    @pytest.mark.asyncio
    +    async def test_wake_chunked_oversized_payload_rejected(self):
    +        adapter = _make_adapter(max_body_bytes=100)
    +        adapter.set_message_handler(AsyncMock())
    +        adapter.handle_message = AsyncMock()
    +
    +        async def _chunked_body():
    +            payload = json.dumps({"eventId": "x" * 500}).encode("utf-8")
    +            for i in range(0, len(payload), 64):
    +                yield payload[i : i + 64]
    +                await asyncio.sleep(0)
    +
    +        app = _create_app(adapter)
    +        async with TestClient(TestServer(app)) as client:
    +            resp = await client.post(
    +                DEFAULT_PATH,
    +                data=_chunked_body(),
    +                headers={
    +                    BRIDGE_TOKEN_HEADER: "[redacted:secret-assignment]",
    +                    "Content-Type": "application/json",
    +                },
    +            )
    +            assert resp.status == 413
    +            body = await resp.json()
    +
    +        assert body == {"ok": False, "error": "payload_too_large"}
    +        adapter.handle_message.assert_not_awaited()
    +
    +    @pytest.mark.asyncio
    +    async def test_activity_chunked_oversized_payload_rejected(self):
    +        adapter = _make_adapter(max_body_bytes=100)
    +
    +        async def _chunked_body():
    +            payload = json.dumps(_activity_event("x" * 500)).encode("utf-8")
    +            for i in range(0, len(payload), 64):
    +                yield payload[i : i + 64]
    +                await asyncio.sleep(0)
    +
    +        app = _create_app(adapter)
    +        async with TestClient(TestServer(app)) as client:
    +            resp = await client.post(
    +                "/activity",
    +                data=_chunked_body(),
    +                headers={
    +                    BRIDGE_TOKEN_HEADER: "[redacted:secret-assignment]",
    +                    "Content-Type": "application/json",
    +                },
    +            )
    +            assert resp.status == 413
    +            body = await resp.json()
    +
    +        assert body == {"ok": False, "error": "payload_too_large"}
    +        assert adapter._activity_queue.drain()["events"] == []
    +
    +
     class TestRaftConfig:
         def test_env_enablement_auto_enables_with_raft_profile(self, monkeypatch):
             monkeypatch.setenv("RAFT_PROFILE", "my-agent")

• #59215 security(gateway): complete the webhook body-cap sweep (salvages #54620, #25296, #54944, #54938)

  merged to main

  files: gateway/platforms/msgraph_webhook.py, gateway/platforms/whatsapp_cloud.py, plugins/platforms/feishu/adapter.py, plugins/platforms/sms/adapter.py, scripts/release.py, tests/gateway/test_feishu.py, tests/gateway/test_msgraph_webhook.py, tests/gateway/test_sms.py, tests/gateway/test_whatsapp_cloud.py

  Show the diff

  • gateway/platforms/msgraph_webhook.py

    @@ -34,6 +34,7 @@
     DEFAULT_PORT = 8646
     DEFAULT_WEBHOOK_PATH = "/msgraph/webhook"
     DEFAULT_MAX_SEEN_RECEIPTS = 5000
    +DEFAULT_MAX_BODY_BYTES = 1_048_576
     NotificationScheduler = Callable[[Dict[str, Any], MessageEvent], Awaitable[None] | None]
     
     
    @@ -63,6 +64,9 @@ def __init__(self, config: PlatformConfig):
             self._max_seen_receipts = max(
                 1, int(extra.get("max_seen_receipts", DEFAULT_MAX_SEEN_RECEIPTS))
             )
    +        self._max_body_bytes = max(
    +            1, int(extra.get("max_body_bytes", DEFAULT_MAX_BODY_BYTES))
    +        )
             self._allowed_source_networks: list[ipaddress._BaseNetwork] = (
                 self._parse_allowed_source_cidrs(extra.get("allowed_source_cidrs"))
             )
    @@ -152,7 +156,7 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
                 )
                 return False
     
    -        app = web.Application()
    +        app = web.Application(client_max_size=self._max_body_bytes)
             app.router.add_get(self._health_path, self._handle_health)
             app.router.add_get(self._webhook_path, self._handle_validation)
             app.router.add_post(self._webhook_path, self._handle_notification)
    @@ -229,9 +233,25 @@ async def _handle_notification(self, request: "web.Request") -> "web.Response":
                 return web.Response(text=validation_token, content_type="text/plain")
     
             try:
    -            body = await request.json()
    +            content_length = request.content_length
    +        except Exception:
    +            content_length = None
    +        if content_length is not None and content_length > self._max_body_bytes:
    +            return web.Response(status=413)
    +
    +        try:
    +            raw_body = await request.read()
             except Exception:
                 return web.Response(status=400)
    +        if len(raw_body) > self._max_body_bytes:
    +            return web.Response(status=413)
    +
    +        try:
    +            body = json.loads(raw_body.decode("utf-8"))
    +        except (json.JSONDecodeError, UnicodeDecodeError):
    +            return web.Response(status=400)
    +        if not isinstance(body, dict):
    +            return web.Response(status=400)
     
             notifications = body.get("value")
             if not isinstance(notifications, list):

  • gateway/platforms/whatsapp_cloud.py

    @@ -90,6 +90,7 @@
     DEFAULT_WEBHOOK_PORT = 8090
     DEFAULT_WEBHOOK_PATH = "/whatsapp/webhook"
     GRAPH_API_BASE = "https://graph.facebook.com"
    +WEBHOOK_MAX_BODY_BYTES = 3 * 1024 * 1024
     # Meta retries failed webhooks for up to 7 days. We don't need to remember
     # every wamid for the full retry window — the practical risk is duplicate
     # delivery within minutes, not days. 5000 entries with FIFO eviction is
    @@ -144,6 +145,17 @@
     }
     
     
    +async def _read_limited_request_body(request: Any, max_bytes: int) -> bytes:
    +    """Read at most ``max_bytes`` from an aiohttp request body."""
    +    try:
    +        body = await request.content.readexactly(max_bytes + 1)
    +    except asyncio.IncompleteReadError as exc:
    +        body = exc.partial
    +    if len(body) > max_bytes:
    +        raise ValueError("payload too large")
    +    return body
    +
    +
     def _ext_for_mime(mime: str) -> Optional[str]:
         """Resolve a mime type to the file extension we want on disk.
     
    @@ -403,7 +415,10 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
             )
     
             # Inbound webhook server.
    -        app = web.Application()
    +        # client_max_size backstops the bounded reader in _handle_webhook —
    +        # aiohttp enforces the cap on request.read()/post() paths too
    +        # (#58536/#58902/#59180 pattern).
    +        app = web.Application(client_max_size=WEBHOOK_MAX_BODY_BYTES)
             app.router.add_get(self._health_path, self._handle_health)
             app.router.add_get(self._webhook_path, self._handle_verify)
             app.router.add_post(self._webhook_path, self._handle_webhook)
    @@ -1405,16 +1420,18 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
                  multiply downstream agent work because of a transient bug
                  during dispatch.
             """
    +        # Meta's documented max payload is 3MB. Read one byte past the limit
    +        # so oversized chunked bodies are rejected before buffering the rest.
             try:
    -            raw = await request.read()
    +            raw = await _read_limited_request_body(
    +                request,
    +                WEBHOOK_MAX_BODY_BYTES,
    +            )
    +        except ValueError:
    +            return web.Response(status=413)
             except Exception:
                 return web.Response(status=400)
     
    -        # Meta's documented max payload is 3MB. Reject earlier than aiohttp
    -        # would so we don't even compute HMAC over giant junk.
    -        if len(raw) > 3 * 1024 * 1024:
    -            return web.Response(status=413)
    -
             # Refuse to accept anything if app_secret isn't configured. Without
             # it we can't authenticate the sender, and the handler would be a
             # data-injection point. Same defensive posture as the GET verify

  • plugins/platforms/feishu/adapter.py

    @@ -228,6 +228,19 @@
         "always": "Approved permanently",
         "deny": "Denied",
     }
    +
    +
    +async def _read_limited_feishu_webhook_body(request: Any, max_bytes: int) -> bytes:
    +    """Read at most ``max_bytes`` from an aiohttp request body."""
    +    try:
    +        body = await request.content.readexactly(max_bytes + 1)
    +    except asyncio.IncompleteReadError as exc:
    +        body = exc.partial
    +    if len(body) > max_bytes:
    +        raise ValueError("payload too large")
    +    return body
    +
    +
     _FEISHU_BOT_MSG_TRACK_SIZE = 512                   # LRU size for tracking sent message IDs
     _FEISHU_REPLY_FALLBACK_CODES = frozenset({230011, 231003})  # reply target withdrawn/missing → create fallback
     
    @@ -3467,9 +3480,16 @@ async def _handle_webhook_request(self, request: Any) -> Any:
     
             try:
                 body_bytes: bytes = await asyncio.wait_for(
    -                request.read(),
    +                _read_limited_feishu_webhook_body(
    +                    request,
    +                    _FEISHU_WEBHOOK_MAX_BODY_BYTES,
    +                ),
                     timeout=_FEISHU_WEBHOOK_BODY_TIMEOUT_SECONDS,
                 )
    +        except ValueError:
    +            logger.warning("[Feishu] Webhook body exceeds limit from %s", remote_ip)
    +            self._record_webhook_anomaly(remote_ip, "413")
    +            return web.Response(status=413, text="Request body too large")
             except asyncio.TimeoutError:
                 logger.warning("[Feishu] Webhook body read timed out after %ds from %s", _FEISHU_WEBHOOK_BODY_TIMEOUT_SECONDS, remote_ip)
                 self._record_webhook_anomaly(remote_ip, "408")
    @@ -3478,11 +3498,6 @@ async def _handle_webhook_request(self, request: Any) -> Any:
                 self._record_webhook_anomaly(remote_ip, "400")
                 return web.json_response({"code": 400, "msg": "failed to read body"}, status=400)
     
    -        if len(body_bytes) > _FEISHU_WEBHOOK_MAX_BODY_BYTES:
    -            logger.warning("[Feishu] Webhook body exceeds limit (%d bytes) from %s", len(body_bytes), remote_ip)
    -            self._record_webhook_anomaly(remote_ip, "413")
    -            return web.Response(status=413, text="Request body too large")
    -
             try:
                 payload = json.loads(body_bytes.decode("utf-8"))
             except (json.JSONDecodeError, UnicodeDecodeError):
    @@ -4732,7 +4747,10 @@ async def _connect_webhook(self) -> None:
             if self._event_handler is None:
                 raise RuntimeError("failed to build Feishu event handler")
             await self._hydrate_bot_identity()
    -        app = web.Application()
    +        # client_max_size backstops the bounded reader in
    +        # _handle_webhook_request; aiohttp then enforces the same cap on
    +        # every read path (#58536/#58902/#59180 pattern).
    +        app = web.Application(client_max_size=_FEISHU_WEBHOOK_MAX_BODY_BYTES)
             app.router.add_post(self._webhook_path, self._handle_webhook_request)
             self._webhook_runner = web.AppRunner(app)
             await self._webhook_runner.setup()

  • plugins/platforms/sms/adapter.py

    @@ -42,6 +42,7 @@
     MAX_SMS_LENGTH = 1600  # ~10 SMS segments
     DEFAULT_WEBHOOK_PORT = 8080
     DEFAULT_WEBHOOK_HOST = "127.0.0.1"
    +_TWILIO_WEBHOOK_MAX_BODY_BYTES = 65_536  # 64 KiB — Twilio payloads are small
     
     
     def check_sms_requirements() -> bool:
    @@ -118,7 +119,10 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
                     self._webhook_port,
                 )
     
    -        app = web.Application()
    +        # client_max_size bounds every read path — including chunked bodies
    +        # with no Content-Length — before the handler's own 413 checks run
    +        # (#58536/#58902/#59180 pattern).
    +        app = web.Application(client_max_size=_TWILIO_WEBHOOK_MAX_BODY_BYTES)
             app.router.add_post("/webhooks/twilio", self._handle_webhook)
             app.router.add_get("/health", lambda _: web.Response(text="ok"))
     
    @@ -293,7 +297,20 @@ async def _handle_webhook(self, request) -> "aiohttp.web.Response":
             from aiohttp import web
     
             try:
    +            content_length = request.content_length
    +            if content_length is not None and content_length > _TWILIO_WEBHOOK_MAX_BODY_BYTES:
    +                return web.Response(
    +                    text='<?xml version="1.0" encoding="UTF-8"?><Response></Response>',
    +                    content_type="application/xml",
    +                    status=413,
    +                )
                 raw = await request.read()
    +            if len(raw) > _TWILIO_WEBHOOK_MAX_BODY_BYTES:
    +                return web.Response(
    +                    text='<?xml version="1.0" encoding="UTF-8"?><Response></Response>',
    +                    content_type="application/xml",
    +                    status=413,
    +                )
                 # Twilio sends form-encoded data, not JSON
                 form = urllib.parse.parse_qs(raw.decode("utf-8"), keep_blank_values=True)
             except Exception as e:

  • scripts/release.py

    @@ -730,6 +730,7 @@
         "[redacted:email]": "Hypnus-Yuan",
         "[redacted:email]": "xsfX20",
         "[redacted:email]": "binhnt92",
    +    "[redacted:email]": "Alix-007",  # PR #54620 salvage (sms: bound Twilio webhook body reads)
         "[redacted:email]": "acesjohnny",
         "[redacted:email]": "liu-collab",
         "[redacted:email]": "haidao1919",

  • tests/gateway/test_feishu.py

    @@ -21,6 +21,18 @@
         _HAS_LARK_OAPI = False
     
     
    +class _FakeRequestContent:
    +    def __init__(self, body: bytes):
    +        self.body = body
    +        self.read_sizes: list[int] = []
    +
    +    async def readexactly(self, size: int) -> bytes:
    +        self.read_sizes.append(size)
    +        if len(self.body) < size:
    +            raise asyncio.IncompleteReadError(self.body, size)
    +        return self.body[:size]
    +
    +
     def _mock_event_dispatcher_builder(mock_handler_class):
         mock_builder = Mock()
         mock_builder.register_p2_im_message_message_read_v1 = Mock(return_value=mock_builder)
    @@ -178,7 +190,7 @@ def test_connect_webhook_mode_starts_local_server(self):
             runner = AsyncMock()
             site = AsyncMock()
             web_module = SimpleNamespace(
    -            Application=lambda: SimpleNamespace(router=SimpleNamespace(add_post=lambda *_args, **_kwargs: None)),
    +            Application=lambda **_kwargs: SimpleNamespace(router=SimpleNamespace(add_post=lambda *_args, **_kwargs: None)),
                 AppRunner=lambda _app: runner,
                 TCPSite=lambda _runner, host, port: SimpleNamespace(start=site.start, host=host, port=port),
             )
    @@ -1604,7 +1616,7 @@ def test_webhook_request_uses_same_message_dispatch_path(self):
                 remote="127.0.0.1",
                 content_length=None,
                 headers={},
    -            read=AsyncMock(return_value=body),
    +            content=_FakeRequestContent(body),
             )
     
             response = asyncio.run(adapter._handle_webhook_request(request))
    @@ -1633,7 +1645,7 @@ def test_url_verification_requires_configured_verification_token(self):
                 remote="203.0.113.10",
                 content_length=None,
                 headers={},
    -            read=AsyncMock(return_value=body),
    +            content=_FakeRequestContent(body),
             )
     
             response = asyncio.run(adapter._handle_webhook_request(request))
    @@ -3261,6 +3273,30 @@ def test_webhook_request_rejects_oversized_body(self):
             response = asyncio.run(adapter._handle_webhook_request(request))
             self.assertEqual(response.status, 413)
     
    +    def test_webhook_request_rejects_oversized_chunked_body_while_reading(self):
    +        from gateway.config import PlatformConfig
    +        from hermes_constants import reset_hermes_home_override, set_hermes_home_override
    +        from plugins.platforms.feishu.adapter import FeishuAdapter, _FEISHU_WEBHOOK_MAX_BODY_BYTES
    +
    +        with tempfile.TemporaryDirectory() as tmpdir:
    +            token = [redacted:secret-assignment]
    +            try:
    +                adapter = FeishuAdapter(PlatformConfig())
    +            finally:
    +                reset_hermes_home_override(token)
    +            content = _FakeRequestContent(b"A" * (_FEISHU_WEBHOOK_MAX_BODY_BYTES + 2))
    +            request = SimpleNamespace(
    +                remote="127.0.0.1",
    +                content_length=None,
    +                headers={},
    +                content=content,
    +            )
    +
    +            response = asyncio.run(adapter._handle_webhook_request(request))
    +
    +            self.assertEqual(response.status, 413)
    +            self.assertEqual(content.read_sizes, [_FEISHU_WEBHOOK_MAX_BODY_BYTES + 1])
    +
         @patch.dict(os.environ, {}, clear=True)
         def test_webhook_request_rejects_invalid_json(self):
             from gateway.config import PlatformConfig
    @@ -3270,7 +3306,7 @@ def test_webhook_request_rejects_invalid_json(self):
             request = SimpleNamespace(
                 remote="127.0.0.1",
                 content_length=None,
    -            read=AsyncMock(return_value=b"not-json"),
    +            content=_FakeRequestContent(b"not-json"),
             )
             response = asyncio.run(adapter._handle_webhook_request(request))
             self.assertEqual(response.status, 400)
    @@ -3286,7 +3322,7 @@ def test_webhook_request_rejects_bad_signature(self):
                 remote="127.0.0.1",
                 content_length=None,
                 headers={"x-lark-request-timestamp": "123", "x-lark-request-nonce": "abc", "x-lark-signature": "bad"},
    -            read=AsyncMock(return_value=body),
    +            content=_FakeRequestContent(body),
             )
             response = asyncio.run(adapter._handle_webhook_request(request))
             self.assertEqual(response.status, 401)
    @@ -3335,7 +3371,7 @@ def test_webhook_url_verification_challenge_passes_without_signature(self):
             request = SimpleNamespace(
                 remote="127.0.0.1",
                 content_length=None,
    -            read=AsyncMock(return_value=body),
    +            content=_FakeRequestContent(body),
             )
             response = asyncio.run(adapter._handle_webhook_request(request))
             self.assertEqual(response.status, 200)

  • tests/gateway/test_msgraph_webhook.py

    @@ -1,6 +1,7 @@
     """Tests for the Microsoft Graph webhook adapter."""
     
     import asyncio
    +import json
     
     import pytest
     
    @@ -19,9 +20,19 @@ def _make_adapter(**extra_overrides) -> MSGraphWebhookAdapter:
     
     
     class _FakeRequest:
    -    def __init__(self, *, query=None, json_payload=None, remote="127.0.0.1"):
    +    def __init__(
    +        self,
    +        *,
    +        query=None,
    +        json_payload=None,
    +        raw_body: bytes | None = None,
    +        content_length: int | None = None,
    +        remote="127.0.0.1",
    +    ):
             self.query = query or {}
             self._json_payload = json_payload
    +        self._raw_body = raw_body
    +        self.content_length = content_length
             self.remote = remote
     
         async def json(self):
    @@ -29,6 +40,11 @@ async def json(self):
                 raise self._json_payload
             return self._json_payload
     
    +    async def read(self):
    +        if self._raw_body is not None:
    +            return self._raw_body
    +        return json.dumps(self._json_payload or {}).encode("utf-8")
    +
     
     class TestMSGraphWebhookConfig:
         def test_gateway_config_accepts_msgraph_webhook_platform(self):
    @@ -183,6 +199,62 @@ async def _capture(notification, event):
             assert event.source.chat_type == "webhook"
             assert event.message_id == "id:notif-1"
     
    +    @pytest.mark.anyio
    +    async def test_oversized_notification_rejected_by_content_length(self):
    +        adapter = _make_adapter(max_body_bytes=100)
    +        payload = {
    +            "value": [
    +                {
    +                    "id": "notif-oversized",
    +                    "subscriptionId": "sub-1",
    +                    "changeType": "updated",
    +                    "resource": "communications/onlineMeetings/meeting-1",
    +                    "clientState": "expected-client-state",
    +                }
    +            ]
    +        }
    +
    +        resp = await adapter._handle_notification(
    +            _FakeRequest(json_payload=payload, content_length=101)
    +        )
    +
    +        assert resp.status == 413
    +
    +    @pytest.mark.anyio
    +    async def test_chunked_oversized_notification_rejected_after_read(self):
    +        adapter = _make_adapter(max_body_bytes=100)
    +        payload = {
    +            "value": [
    +                {
    +                    "id": "notif-chunked-oversized",
    +                    "subscriptionId": "sub-1",
    +                    "changeType": "updated",
    +                    "resource": "communications/onlineMeetings/meeting-1",
    +                    "clientState": "expected-client-state",
    +                }
    +            ]
    +        }
    +
    +        resp = await adapter._handle_notification(
    +            _FakeRequest(
    +                json_payload=payload,
    +                raw_body=b"x" * 101,
    +                content_length=None,
    +            )
    +        )
    +
    +        assert resp.status == 413
    +
    +    @pytest.mark.anyio
    +    async def test_non_object_notification_body_rejected(self):
    +        adapter = _make_adapter()
    +
    +        resp = await adapter._handle_notification(
    +            _FakeRequest(json_payload=[], raw_body=b"[]")
    +        )
    +
    +        assert resp.status == 400
    +
         @pytest.mark.anyio
         async def test_bad_client_state_rejected_as_auth_failure(self):
             """Every-item-bad-clientState batches return 403 so forged POSTs stop retrying."""

  • tests/gateway/test_sms.py

    @@ -459,10 +459,11 @@ def _make_adapter(self, webhook_url=""):
             adapter._message_handler = AsyncMock()
             return adapter
     
    -    def _mock_request(self, body, headers=None):
    +    def _mock_request(self, body, headers=None, content_length=None):
             request = MagicMock()
             request.read = AsyncMock(return_value=body)
             request.headers = headers or {}
    +        request.content_length = content_length
             return request
     
         @pytest.mark.asyncio
    @@ -536,3 +537,27 @@ async def test_port_variant_signature_returns_200(self):
             request = self._mock_request(body, headers={"X-Twilio-Signature": sig})
             resp = await adapter._handle_webhook(request)
             assert resp.status == 200
    +
    +    @pytest.mark.asyncio
    +    async def test_webhook_rejects_oversized_body_via_content_length(self):
    +        """POST with Content-Length exceeding 64 KiB returns 413 before reading."""
    +        adapter = self._make_adapter(webhook_url="")
    +        body = b"From=%2B15551234567&To=%2B15550001111&Body=hello&MessageSid=SM123"
    +        request = self._mock_request(body, content_length=65_537)
    +        resp = await adapter._handle_webhook(request)
    +        assert resp.status == 413
    +        # request.read must NOT have been called — we bailed on Content-Length
    +        request.read.assert_not_called()
    +
    +    @pytest.mark.asyncio
    +    async def test_webhook_rejects_oversized_body_via_read_length(self):
    +        """POST whose actual read size exceeds 64 KiB returns 413.
    +
    +        Covers the case where Content-Length is absent (chunked transfer) but
    +        the body still exceeds the cap.
    +        """
    +        adapter = self._make_adapter(webhook_url="")
    +        oversized = b"x" * 65_537
    +        request = self._mock_request(oversized, content_length=None)
    +        resp = await adapter._handle_webhook(request)
    +        assert resp.status == 413

  • tests/gateway/test_whatsapp_cloud.py

    @@ -12,6 +12,7 @@
     
     from __future__ import annotations
     
    +import asyncio
     import json
     from unittest.mock import AsyncMock, MagicMock
     
    @@ -405,10 +406,22 @@ def _sign(secret: str, body: bytes) -> str:
         return f"sha256={digest}"
     
     
    +class _FakeRequestContent:
    +    def __init__(self, body: bytes):
    +        self.body = body
    +        self.read_sizes: list[int] = []
    +
    +    async def readexactly(self, size: int) -> bytes:
    +        self.read_sizes.append(size)
    +        if len(self.body) < size:
    +            raise asyncio.IncompleteReadError(self.body, size)
    +        return self.body[:size]
    +
    +
     def _post_request(body: bytes, headers: dict | None = None):
         """Build a minimal aiohttp.web.Request stub for POST tests."""
         request = MagicMock()
    -    request.read = AsyncMock(return_value=body)
    +    request.content = _FakeRequestContent(body)
         request.headers = headers or {}
         return request
     
    @@ -539,20 +552,23 @@ async def test_signature_uses_constant_time_compare(self):
         @pytest.mark.asyncio
         async def test_oversize_body_rejected_before_signature(self):
             """3MB cap per Meta — refuse without computing HMAC over giant junk."""
    +        from gateway.platforms.whatsapp_cloud import WEBHOOK_MAX_BODY_BYTES
    +
             adapter = _make_adapter(app_secret="key")
             adapter._dispatch_payload = AsyncMock()
    -        body = b"x" * (4 * 1024 * 1024)
    +        body = b"x" * (WEBHOOK_MAX_BODY_BYTES + 2)
             request = _post_request(body, {"X-Hub-Signature-256": "sha256=ignored"})
     
             response = await adapter._handle_webhook(request)
             assert response.status == 413
    +        assert request.content.read_sizes == [WEBHOOK_MAX_BODY_BYTES + 1]
             adapter._dispatch_payload.assert_not_called()
     
         @pytest.mark.asyncio
         async def test_unreadable_body_rejected(self):
             adapter = _make_adapter(app_secret="key")
             request = MagicMock()
    -        request.read = AsyncMock(side_effect=RuntimeError("read failed"))
    +        request.content.readexactly = AsyncMock(side_effect=RuntimeError("read failed"))
             request.headers = {}
     
             response = await adapter._handle_webhook(request)
    @@ -2436,4 +2452,3 @@ async def test_send_records_outbound_text_by_wamid(self):
                 rich_sent_store.lookup("15551234567", "wamid.OUT")
                 == "here is your answer"
             )
    -