fix(webrtc): harden /sdp HTTP server against memory-amplification DoS (#1354)

libp2p/transport/webrtc/_aiortc_helpers.py

@@ -37,6 +37,12 @@
 # Timeout for HTTP SDP exchange (seconds).
 _SDP_HTTP_TIMEOUT = 15.0
 
+# Bounds for the HTTP /sdp dev harness — defend against memory-amplification
+# DoS while the harness exists (until the STUN-based listener lands, #1352).
+_MAX_SDP_BODY_SIZE = 32 * 1024  # 32 KiB; SDP offers are typically 1–4 KiB
+_MAX_HEADER_LINES = 64
+_MAX_HEADER_BYTES = 8 * 1024  # 8 KiB total across all header lines
+
 
 # ------------------------------------------------------------------
 # Peer-connection lifecycle
@@ -380,31 +386,62 @@ async def _handle(
         writer: asyncio.StreamWriter,
     ) -> None:
         try:
-            # Read HTTP request line (consumed but not used) + headers
+            # Request line (consumed but not validated)
             await asyncio.wait_for(reader.readline(), timeout=_SDP_HTTP_TIMEOUT)
+
+            # Headers — bounded by line count and accumulated byte size.
+            # The for/else fires when the loop exhausts _MAX_HEADER_LINES
+            # reads without finding the blank-line terminator.
             headers: dict[str, str] = {}
-            while True:
+            header_bytes = 0
+            for _ in range(_MAX_HEADER_LINES):
                 line = await asyncio.wait_for(
                     reader.readline(), timeout=_SDP_HTTP_TIMEOUT
                 )
                 if line in (b"\r\n", b"\n", b""):
                     break
+                header_bytes += len(line)
+                if header_bytes > _MAX_HEADER_BYTES:
+                    writer.write(
+                        b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\n\r\n"
+                    )
+                    await writer.drain()
+                    return
                 key, _, value = line.decode().partition(":")
                 headers[key.strip().lower()] = value.strip()
+            else:
+                writer.write(b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\n\r\n")
+                await writer.drain()
+                return
+
+            # Validate Content-Length before touching any body buffer.
+            raw_cl = headers.get("content-length", "0")
+            try:
+                content_length = int(raw_cl)
+            except ValueError:
+                writer.write(b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\n\r\n")
+                await writer.drain()
+                return
+            if content_length < 0:
+                writer.write(b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\n\r\n")
+                await writer.drain()
+                return
+            if content_length > _MAX_SDP_BODY_SIZE:
+                writer.write(
+                    b"HTTP/1.1 413 Payload Too Large\r\nContent-Length: 0\r\n\r\n"
+                )
+                await writer.drain()
+                return
 
-            content_length = int(headers.get("content-length", "0"))
             body = b""
             if content_length > 0:
                 body = await asyncio.wait_for(
                     reader.readexactly(content_length),
                     timeout=_SDP_HTTP_TIMEOUT,
                 )
 
-            # Process: call the offer handler, get answer SDP
             answer_sdp = await on_offer(body.decode())
             answer_bytes = answer_sdp.encode()
-
-            # Send HTTP response
             response = (
                 b"HTTP/1.1 200 OK\r\n"
                 b"Content-Type: application/sdp\r\n"

newsfragments/1354.bugfix.rst

@@ -0,0 +1 @@
+Hardened the WebRTC Direct ``POST /sdp`` dev-harness HTTP server against memory-amplification DoS: bounded ``Content-Length`` to 32 KiB (returns 413 on excess), rejected negative/non-integer values with 400, and capped header parsing at 64 lines / 8 KiB to prevent indefinite task hold via slow header flooding.

tests/core/pubsub/test_gossipsub_identify_aware_publish.py

@@ -216,6 +216,11 @@ async def test_publish_before_identify_completes():
 
         # Connect and publish in rapid succession – no sleep in between
         await connect(pubsubs[0].host, pubsubs[1].host)
+        # Wait for the peer pubsub stream to be registered so the pending
+        # message queue can target it.  The subscription from the hello
+        # packet may not have been processed yet, so this still exercises
+        # the identify-aware queuing path.
+        await pubsubs[0].wait_for_peer(pubsubs[1].host.get_id())
         await pubsubs[0].publish(topic, data)
 
         # Give enough time for:
@@ -291,6 +296,7 @@ async def test_multiple_rapid_publishes_before_identify():
         sub_1 = await pubsubs[1].subscribe(topic)
 
         await connect(pubsubs[0].host, pubsubs[1].host)
+        await pubsubs[0].wait_for_peer(pubsubs[1].host.get_id())
 
         # Rapid-fire publishes
         for data in messages:
@@ -334,7 +340,9 @@ async def test_three_nodes_publish_before_full_mesh():
 
         # Connect A-B and B-C
         await connect(pubsubs[0].host, pubsubs[1].host)
+        await pubsubs[0].wait_for_peer(pubsubs[1].host.get_id())
         await connect(pubsubs[1].host, pubsubs[2].host)
+        await pubsubs[1].wait_for_peer(pubsubs[2].host.get_id())
 
         # Publish immediately from A
         await pubsubs[0].publish(topic, data)

tests/core/transport/webrtc/test_aiortc_helpers.py

@@ -0,0 +1,228 @@
+"""
+Regression tests for _aiortc_helpers.run_signaling_server hardening.
+
+Covers the acceptance criteria from issue #1354:
+  - Content-Length > _MAX_SDP_BODY_SIZE  → 413, no body buffer allocated
+  - Non-integer / negative Content-Length → 400
+  - Header loop exceeds _MAX_HEADER_LINES or _MAX_HEADER_BYTES → 400
+  - Happy path still returns 200
+
+All tests are sync wrappers over asyncio coroutines so they run outside
+trio_mode and don't interfere with the project-wide trio backend.
+"""
+# pyrefly: ignore
+
+from __future__ import annotations
+
+import asyncio
+import sys
+
+import pytest
+
+try:
+    from libp2p.transport.webrtc._aiortc_helpers import (
+        _MAX_HEADER_BYTES,
+        _MAX_HEADER_LINES,
+        _MAX_SDP_BODY_SIZE,
+        run_signaling_server,
+    )
+
+    HAS_AIORTC = True
+except ImportError:
+    HAS_AIORTC = False
+
+pytestmark = pytest.mark.skipif(not HAS_AIORTC, reason="aiortc not installed")
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+async def _start(on_offer=None):
+    """Start the signaling server on a free port; return (server, port)."""
+    if on_offer is None:
+
+        async def on_offer(sdp: str) -> str:
+            return "v=0\r\n"
+
+    server = await run_signaling_server("127.0.0.1", 0, on_offer)
+    port = server.sockets[0].getsockname()[1]
+    return server, port
+
+
+async def _status(port: int, raw_request: bytes, timeout: float = 5.0) -> bytes:
+    """Send *raw_request* and return the HTTP status line."""
+    reader, writer = await asyncio.open_connection("127.0.0.1", port)
+    try:
+        writer.write(raw_request)
+        await writer.drain()
+        return await asyncio.wait_for(reader.readline(), timeout=timeout)
+    finally:
+        writer.close()
+
+
+# ---------------------------------------------------------------------------
+# Content-Length validation
+# ---------------------------------------------------------------------------
+
+
+class TestContentLength:
+    def test_too_large_returns_413_and_no_rss_spike(self) -> None:
+        asyncio.run(self._too_large())
+
+    async def _too_large(self) -> None:
+        # Snapshot RSS high-water mark before the request.
+        try:
+            import resource
+
+            rss_before = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+        except ImportError:
+            rss_before = None
+
+        server, port = await _start()
+        try:
+            line = await _status(
+                port,
+                b"POST /sdp HTTP/1.1\r\nContent-Length: 999999999\r\n\r\n",
+            )
+            assert b"413" in line
+        finally:
+            server.close()
+            await server.wait_closed()
+
+        # Soft RSS check: the body buffer (≈1 GiB) must NOT have been allocated.
+        # ru_maxrss is the high-water mark so a delta > 0 means new peak memory.
+        if rss_before is not None:
+            import resource
+
+            rss_after = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+            delta = rss_after - rss_before
+            # Linux: ru_maxrss in KiB; macOS: in bytes
+            one_mib = 1024 if sys.platform.startswith("linux") else 1024 * 1024
+            assert delta < one_mib, (
+                f"RSS high-water grew by {delta} units after 413 rejection; "
+                "body buffer may have been allocated before the check"
+            )
+
+    def test_non_integer_returns_400(self) -> None:
+        asyncio.run(self._non_integer())
+
+    async def _non_integer(self) -> None:
+        server, port = await _start()
+        try:
+            line = await _status(
+                port,
+                b"POST /sdp HTTP/1.1\r\nContent-Length: notanumber\r\n\r\n",
+            )
+            assert b"400" in line
+        finally:
+            server.close()
+            await server.wait_closed()
+
+    def test_negative_returns_400(self) -> None:
+        asyncio.run(self._negative())
+
+    async def _negative(self) -> None:
+        server, port = await _start()
+        try:
+            line = await _status(
+                port,
+                b"POST /sdp HTTP/1.1\r\nContent-Length: -1\r\n\r\n",
+            )
+            assert b"400" in line
+        finally:
+            server.close()
+            await server.wait_closed()
+
+    def test_exactly_at_limit_is_accepted(self) -> None:
+        asyncio.run(self._at_limit())
+
+    async def _at_limit(self) -> None:
+        body = b"x" * _MAX_SDP_BODY_SIZE
+
+        async def on_offer(sdp: str) -> str:
+            return "v=0\r\n"
+
+        server, port = await _start(on_offer)
+        try:
+            request = (
+                b"POST /sdp HTTP/1.1\r\n"
+                b"Content-Length: " + str(len(body)).encode() + b"\r\n"
+                b"\r\n" + body
+            )
+            line = await _status(port, request)
+            assert b"200" in line
+        finally:
+            server.close()
+            await server.wait_closed()
+
+
+# ---------------------------------------------------------------------------
+# Header loop bounds
+# ---------------------------------------------------------------------------
+
+
+class TestHeaderBounds:
+    def test_too_many_header_lines_returns_400(self) -> None:
+        asyncio.run(self._too_many_lines())
+
+    async def _too_many_lines(self) -> None:
+        # Send _MAX_HEADER_LINES + 1 non-blank headers — the loop must reject
+        # before ever reaching the blank-line terminator.
+        spam = b"".join(
+            f"X-Spam-{i}: v\r\n".encode() for i in range(_MAX_HEADER_LINES + 1)
+        )
+        server, port = await _start()
+        try:
+            line = await _status(port, b"POST /sdp HTTP/1.1\r\n" + spam + b"\r\n")
+            assert b"400" in line
+        finally:
+            server.close()
+            await server.wait_closed()
+
+    def test_header_bytes_over_limit_returns_400(self) -> None:
+        asyncio.run(self._too_many_bytes())
+
+    async def _too_many_bytes(self) -> None:
+        # A single header line that exceeds _MAX_HEADER_BYTES.
+        big = b"X-Big: " + b"x" * (_MAX_HEADER_BYTES + 1) + b"\r\n"
+        server, port = await _start()
+        try:
+            line = await _status(port, b"POST /sdp HTTP/1.1\r\n" + big + b"\r\n")
+            assert b"400" in line
+        finally:
+            server.close()
+            await server.wait_closed()
+
+
+# ---------------------------------------------------------------------------
+# Happy path
+# ---------------------------------------------------------------------------
+
+
+class TestHappyPath:
+    def test_valid_request_returns_200(self) -> None:
+        asyncio.run(self._valid())
+
+    async def _valid(self) -> None:
+        received: list[str] = []
+
+        async def on_offer(sdp: str) -> str:
+            received.append(sdp)
+            return "v=0\r\nanswer"
+
+        server, port = await _start(on_offer)
+        body = b"v=0\r\noffer"
+        try:
+            request = (
+                b"POST /sdp HTTP/1.1\r\n"
+                b"Content-Type: application/sdp\r\n"
+                b"Content-Length: " + str(len(body)).encode() + b"\r\n"
+                b"\r\n" + body
+            )
+            line = await _status(port, request)
+            assert b"200" in line
+            assert received == ["v=0\r\noffer"]
+        finally:
+            server.close()
+            await server.wait_closed()