diff --git a/libp2p/transport/webrtc/_aiortc_helpers.py b/libp2p/transport/webrtc/_aiortc_helpers.py index 9922babfe..faef723f0 100644 --- a/libp2p/transport/webrtc/_aiortc_helpers.py +++ b/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,19 +386,53 @@ 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( @@ -400,11 +440,8 @@ async def _handle( 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" diff --git a/newsfragments/1354.bugfix.rst b/newsfragments/1354.bugfix.rst new file mode 100644 index 000000000..1ba8f039c --- /dev/null +++ b/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. diff --git a/tests/core/pubsub/test_gossipsub_identify_aware_publish.py b/tests/core/pubsub/test_gossipsub_identify_aware_publish.py index b2a6300e2..c0e775942 100644 --- a/tests/core/pubsub/test_gossipsub_identify_aware_publish.py +++ b/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) diff --git a/tests/core/transport/webrtc/test_aiortc_helpers.py b/tests/core/transport/webrtc/test_aiortc_helpers.py new file mode 100644 index 000000000..4993f29cc --- /dev/null +++ b/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()