music-assistant-server

1"""Tests for remote access feature."""
2
3import asyncio
4import base64
5import hashlib
6import json
7import logging
8from collections.abc import AsyncIterator, Callable, Coroutine
9from types import SimpleNamespace
10from typing import Any, cast
11from unittest.mock import AsyncMock, MagicMock, Mock, patch
12from urllib.parse import urlparse
13
14import aiohttp
15import pytest
16from aiolibdatachannel import (
17    DataChannel,
18    IceServer,
19    LogLevel,
20    PeerConnection,
21    RTCConfiguration,
22    RTCError,
23)
24from cryptography.hazmat.primitives import serialization
25
26from music_assistant.constants import VERBOSE_LOG_LEVEL
27from music_assistant.controllers.webserver.remote_access import (
28    STARTUP_DELAY,
29    TASK_ID_START_GATEWAY,
30    RemoteAccessInfo,
31    RemoteAccessManager,
32)
33from music_assistant.controllers.webserver.remote_access.gateway import (
34    DATA_CHANNEL_CHUNK_SIZE,
35    WebRTCGateway,
36    WebRTCSession,
37    _is_usable_ice_url,
38)
39from music_assistant.helpers.webrtc_certificate import (
40    _generate_certificate,
41    _remote_id_from_certificate,
42)
43
44
45async def test_remote_id_from_certificate() -> None:
46    """Test deterministic remote ID generation from a certificate."""
47    _, cert = _generate_certificate()
48    remote_id = _remote_id_from_certificate(cert)
49
50    # Should be base32 encoded, uppercase, no padding
51    assert remote_id.isalnum()
52    assert remote_id == remote_id.upper()
53    # 128 bits = 16 bytes -> 26 base32 chars (without padding)
54    assert len(remote_id) == 26
55    # deterministic: the same certificate always yields the same id
56    assert _remote_id_from_certificate(cert) == remote_id
57
58
59async def test_remote_access_info_dataclass() -> None:
60    """Test RemoteAccessInfo dataclass."""
61    info = RemoteAccessInfo(
62        enabled=True,
63        running=True,
64        connected=False,
65        remote_id="VVPN3TLP34YMGIZDINCEKQKSIR",
66        using_ha_cloud=False,
67        signaling_url="wss://signaling.music-assistant.io/ws",
68    )
69
70    assert info.enabled is True
71    assert info.running is True
72    assert info.connected is False
73    assert info.remote_id == "VVPN3TLP34YMGIZDINCEKQKSIR"
74    assert info.using_ha_cloud is False
75    assert info.signaling_url == "wss://signaling.music-assistant.io/ws"
76
77
78def _create_remote_access_manager() -> RemoteAccessManager:
79    """Create an enabled remote access manager with mocked dependencies."""
80    webserver = Mock()
81    webserver.mass = Mock()
82    webserver.logger = Mock()
83    manager = RemoteAccessManager(webserver)
84    manager._enabled = True
85    return manager
86
87
88def test_remote_access_debounces_restart_without_dropping_gateway() -> None:
89    """Keep the active gateway connected while a replacement is being debounced."""
90    manager = _create_remote_access_manager()
91    gateway = Mock()
92    gateway.stop = AsyncMock()
93    manager.gateway = gateway
94
95    manager._schedule_start()
96
97    gateway.stop.assert_not_awaited()
98    cast("Mock", manager.mass.cancel_timer).assert_called_once_with(TASK_ID_START_GATEWAY)
99    cast("Mock", manager.mass.call_later).assert_called_once_with(
100        STARTUP_DELAY,
101        manager._restart_gateway,
102        task_id=TASK_ID_START_GATEWAY,
103    )
104
105
106async def test_remote_access_stop_cancels_pending_restart() -> None:
107    """Cancel a restart after its debounce timer has promoted it to a task."""
108    manager = _create_remote_access_manager()
109    gateway = Mock()
110    gateway.stop = AsyncMock()
111    manager.gateway = gateway
112
113    await manager.stop()
114
115    cast("Mock", manager.mass.cancel_timer).assert_called_once_with(TASK_ID_START_GATEWAY)
116    cast("Mock", manager.mass.cancel_task).assert_called_once_with(TASK_ID_START_GATEWAY)
117    gateway.stop.assert_awaited_once_with()
118    assert manager.gateway is None
119
120
121async def test_remote_access_serializes_concurrent_starts() -> None:
122    """Create only one gateway when immediate start requests overlap."""
123    manager = _create_remote_access_manager()
124    start_entered = asyncio.Event()
125    allow_start = asyncio.Event()
126    gateway = Mock()
127    gateway.is_running = True
128
129    async def start_gateway() -> None:
130        start_entered.set()
131        await allow_start.wait()
132        manager.gateway = gateway
133
134    with patch.object(
135        manager,
136        "_start_gateway_locked",
137        new=AsyncMock(side_effect=start_gateway),
138    ) as start_gateway_locked:
139        first_start = asyncio.create_task(manager._start_gateway())
140        await start_entered.wait()
141        second_start = asyncio.create_task(manager._start_gateway())
142        await asyncio.sleep(0)
143        allow_start.set()
144        await asyncio.gather(first_start, second_start)
145
146    start_gateway_locked.assert_awaited_once()
147
148
149async def test_remote_access_provider_update_storm_schedules_one_restart() -> None:
150    """Schedule one restart when concurrent provider updates detect the same mode change."""
151    manager = _create_remote_access_manager()
152    ice_servers = [{"urls": "turn:example.com"}]
153
154    with (
155        patch.object(
156            manager,
157            "_get_ha_cloud_status",
158            new=AsyncMock(return_value=(True, ice_servers)),
159        ),
160        patch.object(manager, "_schedule_start") as schedule_start,
161    ):
162        await asyncio.gather(*(manager._on_providers_updated(Mock()) for _ in range(10)))
163
164    schedule_start.assert_called_once_with()
165    assert manager._target_using_ha_cloud is True
166
167
168async def test_remote_access_skips_restart_after_mode_flap_settles() -> None:
169    """Keep the current gateway when a fresh status check matches its active mode."""
170    manager = _create_remote_access_manager()
171    gateway = Mock()
172    gateway.is_running = True
173    manager.gateway = gateway
174    manager._target_using_ha_cloud = True
175
176    with (
177        patch.object(
178            manager,
179            "_get_ha_cloud_status",
180            new=AsyncMock(return_value=(False, None)),
181        ),
182        patch.object(
183            manager,
184            "_stop_gateway_locked",
185            new=AsyncMock(),
186        ) as stop_gateway,
187        patch.object(
188            manager,
189            "_start_gateway_locked",
190            new=AsyncMock(),
191        ) as start_gateway,
192    ):
193        await manager._restart_gateway()
194
195    stop_gateway.assert_not_awaited()
196    start_gateway.assert_not_awaited()
197    assert manager._target_using_ha_cloud is False
198
199
200async def test_remote_access_gateway_uses_internal_sendspin_url(
201    cert_pems: tuple[str, str],
202) -> None:
203    """Bridge the Sendspin data channel to the locally reachable Sendspin server."""
204    internal_url = "ws://127.0.0.1:8927/sendspin"
205    manager = _create_remote_access_manager()
206    manager._remote_id = "TEST-REMOTE-ID"
207    cast("Mock", manager.webserver).internal_base_url = "http://127.0.0.1:8095"
208    cast("Mock", manager.webserver).internal_sendspin_url = internal_url
209
210    with (
211        patch(
212            "music_assistant.controllers.webserver.remote_access"
213            ".get_or_create_webrtc_certificate_pems",
214            return_value=cert_pems,
215        ),
216        patch.object(manager, "_get_ha_cloud_status", new=AsyncMock(return_value=(False, None))),
217        patch.object(WebRTCGateway, "start", new=AsyncMock()),
218    ):
219        await manager._start_gateway_locked()
220
221    assert manager.gateway is not None
222    assert manager.gateway.sendspin_url == internal_url
223
224
225@pytest.mark.parametrize(
226    ("internal_base_url", "expected"),
227    [
228        ("http://127.0.0.1:8095", "ws://127.0.0.1:8095/ws"),
229        ("https://127.0.0.1:8095", "wss://127.0.0.1:8095/ws"),
230    ],
231)
232async def test_remote_access_gateway_uses_internal_base_url(
233    cert_pems: tuple[str, str],
234    internal_base_url: str,
235    expected: str,
236) -> None:
237    """Bridge the API data channel to the locally reachable webserver."""
238    manager = _create_remote_access_manager()
239    manager._remote_id = "TEST-REMOTE-ID"
240    # an external base URL must never be dialed back into this host
241    cast("Mock", manager.mass).webserver.base_url = "https://ma.example.com"
242    cast("Mock", manager.webserver).internal_base_url = internal_base_url
243
244    with (
245        patch(
246            "music_assistant.controllers.webserver.remote_access"
247            ".get_or_create_webrtc_certificate_pems",
248            return_value=cert_pems,
249        ),
250        patch.object(manager, "_get_ha_cloud_status", new=AsyncMock(return_value=(False, None))),
251        patch.object(WebRTCGateway, "start", new=AsyncMock()),
252    ):
253        await manager._start_gateway_locked()
254
255    assert manager.gateway is not None
256    assert manager.gateway.local_ws_url == expected
257
258
259async def test_webrtc_gateway_initialization(cert_pems: tuple[str, str]) -> None:
260    """Test WebRTCGateway initializes correctly."""
261    cert_pem, key_pem = cert_pems
262    mock_session = Mock()
263    gateway = WebRTCGateway(
264        http_session=mock_session,
265        remote_id="TEST-REMOTE-ID",
266        cert_pem=cert_pem,
267        key_pem=key_pem,
268        signaling_url="wss://test.example.com/ws",
269        local_ws_url="ws://localhost:8095/ws",
270    )
271
272    assert gateway._remote_id == "TEST-REMOTE-ID"
273    assert gateway.signaling_url == "wss://test.example.com/ws"
274    assert gateway.local_ws_url == "ws://localhost:8095/ws"
275    assert gateway.is_running is False
276    assert gateway.is_connected is False
277    assert len(gateway.ice_servers) > 0
278
279
280async def test_webrtc_gateway_custom_ice_servers(cert_pems: tuple[str, str]) -> None:
281    """Test WebRTCGateway accepts custom ICE servers."""
282    cert_pem, key_pem = cert_pems
283    mock_session = Mock()
284    custom_ice_servers = [
285        {"urls": "stun:custom.stun.server:3478"},
286        {"urls": "turn:custom.turn.server:3478", "username": "user", "credential": "pass"},
287    ]
288
289    gateway = WebRTCGateway(
290        http_session=mock_session,
291        remote_id="TEST-REMOTE-ID",
292        cert_pem=cert_pem,
293        key_pem=key_pem,
294        ice_servers=custom_ice_servers,
295    )
296
297    assert gateway.ice_servers == custom_ice_servers
298
299
300async def test_webrtc_gateway_local_ice_servers_skip_non_udp_turn(
301    cert_pems: tuple[str, str],
302) -> None:
303    """Test the local peer connection only gets ICE servers libjuice can use."""
304    cert_pem, key_pem = cert_pems
305    cred = {"username": "u", "credential": "p"}
306    # shape of what HA Cloud hands us: one UDP TURN url plus TCP/TLS variants
307    ha_cloud_ice_servers = [
308        {"urls": "stun:stun.cloudflare.com:3478"},
309        {"urls": "turn:turn.cloudflare.com:3478?transport=udp", **cred},
310        {"urls": "turn:turn.cloudflare.com:53", **cred},
311        {"urls": "turn:turn.cloudflare.com:3478?transport=tcp", **cred},
312        {"urls": "turns:turn.cloudflare.com:5349?transport=tcp", **cred},
313        {"urls": "turns:turn.cloudflare.com:443?transport=tcp", **cred},
314    ]
315
316    gateway = WebRTCGateway(
317        http_session=Mock(),
318        remote_id="TEST-REMOTE-ID",
319        cert_pem=cert_pem,
320        key_pem=key_pem,
321        ice_servers=ha_cloud_ice_servers,
322    )
323
324    assert [server.url for server in gateway._build_ice_servers(ha_cloud_ice_servers)] == [
325        "stun:stun.cloudflare.com:3478",
326        "turn:turn.cloudflare.com:3478?transport=udp",
327        "turn:turn.cloudflare.com:53",
328    ]
329    # remote clients still get the unfiltered list, since browsers do support TCP/TLS TURN
330    assert gateway.ice_servers == ha_cloud_ice_servers
331
332
333@pytest.mark.parametrize(
334    ("url", "usable"),
335    [
336        ("stun:stun.example.com:3478", True),
337        ("turn:turn.example.com:3478", True),
338        ("turn:turn.example.com:3478?transport=udp", True),
339        # the transport parameter wins over the scheme, matching rtc::IceServer
340        ("turns:turn.example.com:5349?transport=udp", True),
341        ("turn:turn.example.com:3478?transport=tcp", False),
342        ("turns:turn.example.com:5349", False),
343        ("turn:turn.example.com:3478?transport=tls", False),
344        ("https://turn.example.com", False),
345    ],
346)
347def test_is_usable_ice_url(url: str, usable: bool) -> None:
348    """Test only ICE server urls libjuice can actually use are kept."""
349    assert _is_usable_ice_url(url) is usable
350
351
352async def test_webrtc_gateway_start_stop(cert_pems: tuple[str, str]) -> None:
353    """Test WebRTCGateway start and stop."""
354    cert_pem, key_pem = cert_pems
355    mock_session = Mock()
356    gateway = WebRTCGateway(
357        http_session=mock_session,
358        remote_id="TEST-REMOTE-ID",
359        cert_pem=cert_pem,
360        key_pem=key_pem,
361    )
362
363    # Mock the _run method to avoid actual connection
364    with patch.object(gateway, "_run", new_callable=AsyncMock):
365        await gateway.start()
366        assert gateway.is_running is True
367        assert gateway._run_task is not None
368
369        await gateway.stop()
370        assert gateway.is_running is False
371
372
373@pytest.mark.parametrize(
374    ("logger_level", "expected_rtc_level"),
375    [
376        (VERBOSE_LOG_LEVEL, LogLevel.VERBOSE),
377        (logging.DEBUG, LogLevel.WARNING),
378        (logging.INFO, LogLevel.ERROR),
379        (logging.WARNING, LogLevel.ERROR),
380    ],
381)
382async def test_webrtc_gateway_native_log_level(
383    cert_pems: tuple[str, str], logger_level: int, expected_rtc_level: LogLevel
384) -> None:
385    """Test native libdatachannel logging is capped at ERROR unless debugging."""
386    cert_pem, key_pem = cert_pems
387    gateway = WebRTCGateway(
388        http_session=Mock(),
389        remote_id="TEST-REMOTE-ID",
390        cert_pem=cert_pem,
391        key_pem=key_pem,
392    )
393    gateway.logger = logging.getLogger("test_webrtc_native_log_level")
394    gateway.logger.setLevel(logger_level)
395
396    with (
397        patch.object(gateway, "_run", new_callable=AsyncMock),
398        patch(
399            "music_assistant.controllers.webserver.remote_access.gateway.install_python_logger"
400        ) as install_logger,
401    ):
402        await gateway.start()
403        await gateway.stop()
404
405    install_logger.assert_called_once_with(gateway.logger, level=expected_rtc_level)
406    assert not gateway.logger.filters
407
408
409async def test_webrtc_gateway_drops_benign_turn_warning_at_debug(
410    cert_pems: tuple[str, str], caplog: pytest.LogCaptureFixture
411) -> None:
412    """Test the benign Cloudflare CreatePermission warning is dropped at DEBUG level."""
413    cert_pem, key_pem = cert_pems
414    gateway = WebRTCGateway(
415        http_session=Mock(),
416        remote_id="TEST-REMOTE-ID",
417        cert_pem=cert_pem,
418        key_pem=key_pem,
419    )
420    gateway.logger = logging.getLogger("test_webrtc_benign_turn_warning")
421    gateway.logger.setLevel(logging.DEBUG)
422
423    with (
424        patch.object(gateway, "_run", new_callable=AsyncMock),
425        patch("music_assistant.controllers.webserver.remote_access.gateway.install_python_logger"),
426    ):
427        await gateway.start()
428        with caplog.at_level(logging.DEBUG, logger="test_webrtc_benign_turn_warning"):
429            gateway.logger.warning(
430                "rtc::impl::IceTransport::LogCallback@390: "
431                "juice: Got TURN CreatePermission error response, code=0"
432            )
433            gateway.logger.warning("juice: Lost connectivity")
434        await gateway.stop()
435
436    messages = [record.getMessage() for record in caplog.records if "juice" in record.getMessage()]
437    assert messages == ["juice: Lost connectivity"]
438    # stop() must remove the filter again
439    assert not gateway.logger.filters
440
441
442async def test_webrtc_gateway_handle_registration_message(cert_pems: tuple[str, str]) -> None:
443    """Test WebRTCGateway handles registration confirmation."""
444    cert_pem, key_pem = cert_pems
445    mock_session = Mock()
446    gateway = WebRTCGateway(
447        http_session=mock_session,
448        remote_id="TEST-REMOTE-ID",
449        cert_pem=cert_pem,
450        key_pem=key_pem,
451    )
452
453    # Mock signaling WebSocket
454    gateway._signaling_ws = Mock()
455
456    message = {"type": "registered", "remoteId": "TEST-REMOTE-ID"}
457    await gateway._handle_signaling_message(message)
458
459    # Should log but not crash
460
461
462async def test_webrtc_gateway_handle_error_message(cert_pems: tuple[str, str]) -> None:
463    """Test WebRTCGateway handles error messages."""
464    cert_pem, key_pem = cert_pems
465    mock_session = Mock()
466    gateway = WebRTCGateway(
467        http_session=mock_session,
468        remote_id="TEST-REMOTE-ID",
469        cert_pem=cert_pem,
470        key_pem=key_pem,
471    )
472
473    message = {"type": "error", "message": "Test error"}
474    # Should log error but not crash
475    await gateway._handle_signaling_message(message)
476
477
478async def test_webrtc_gateway_create_session(cert_pems: tuple[str, str]) -> None:
479    """Test WebRTCGateway creates sessions for clients."""
480    cert_pem, key_pem = cert_pems
481    mock_session = Mock()
482    gateway = WebRTCGateway(
483        http_session=mock_session,
484        remote_id="TEST-REMOTE-ID",
485        cert_pem=cert_pem,
486        key_pem=key_pem,
487    )
488
489    session_id = "test-session-123"
490    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
491        await gateway._create_session(session_id)
492
493        assert session_id in gateway.sessions
494        assert gateway.sessions[session_id].session_id == session_id
495        assert gateway.sessions[session_id].pc is not None
496
497        # Cleanup
498        await gateway._close_session(session_id)
499
500
501async def test_webrtc_gateway_close_session(cert_pems: tuple[str, str]) -> None:
502    """Test WebRTCGateway closes sessions properly."""
503    cert_pem, key_pem = cert_pems
504    mock_session = Mock()
505    gateway = WebRTCGateway(
506        http_session=mock_session,
507        remote_id="TEST-REMOTE-ID",
508        cert_pem=cert_pem,
509        key_pem=key_pem,
510    )
511
512    session_id = "test-session-456"
513    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
514        await gateway._create_session(session_id)
515        assert session_id in gateway.sessions
516
517        await gateway._close_session(session_id)
518        assert session_id not in gateway.sessions
519
520
521async def test_webrtc_gateway_close_nonexistent_session(cert_pems: tuple[str, str]) -> None:
522    """Test WebRTCGateway handles closing non-existent session gracefully."""
523    cert_pem, key_pem = cert_pems
524    mock_session = Mock()
525    gateway = WebRTCGateway(
526        http_session=mock_session,
527        remote_id="TEST-REMOTE-ID",
528        cert_pem=cert_pem,
529        key_pem=key_pem,
530    )
531
532    # Should not raise an error
533    await gateway._close_session("nonexistent-session")
534
535
536async def test_webrtc_gateway_default_ice_servers(cert_pems: tuple[str, str]) -> None:
537    """Test WebRTCGateway uses default ICE servers."""
538    cert_pem, key_pem = cert_pems
539    mock_session = Mock()
540    gateway = WebRTCGateway(
541        http_session=mock_session,
542        remote_id="TEST-REMOTE-ID",
543        cert_pem=cert_pem,
544        key_pem=key_pem,
545    )
546
547    assert len(gateway.ice_servers) > 0
548    # Should have at least one STUN server
549    assert any("stun:" in server["urls"] for server in gateway.ice_servers)
550
551
552async def test_webrtc_gateway_handle_client_connected(cert_pems: tuple[str, str]) -> None:
553    """Test WebRTCGateway handles client-connected message."""
554    cert_pem, key_pem = cert_pems
555    mock_session = Mock()
556    gateway = WebRTCGateway(
557        http_session=mock_session,
558        remote_id="TEST-REMOTE-ID",
559        cert_pem=cert_pem,
560        key_pem=key_pem,
561    )
562
563    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
564        message = {"type": "client-connected", "sessionId": "test-session"}
565        await gateway._handle_signaling_message(message)
566
567        # Session should be created
568        assert "test-session" in gateway.sessions
569
570        # Cleanup
571        await gateway._close_session("test-session")
572
573
574async def test_webrtc_gateway_handle_client_disconnected(cert_pems: tuple[str, str]) -> None:
575    """Test WebRTCGateway handles client-disconnected message."""
576    cert_pem, key_pem = cert_pems
577    mock_session = Mock()
578    gateway = WebRTCGateway(
579        http_session=mock_session,
580        remote_id="TEST-REMOTE-ID",
581        cert_pem=cert_pem,
582        key_pem=key_pem,
583    )
584
585    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
586        # Create a session first
587        session_id = "test-disconnect-session"
588        await gateway._create_session(session_id)
589        assert session_id in gateway.sessions
590
591        # Handle disconnect
592        message = {"type": "client-disconnected", "sessionId": session_id}
593        await gateway._handle_signaling_message(message)
594
595        # Session should be removed
596        assert session_id not in gateway.sessions
597
598
599async def test_webrtc_gateway_reconnection_logic(cert_pems: tuple[str, str]) -> None:
600    """Test WebRTCGateway has proper reconnection backoff."""
601    cert_pem, key_pem = cert_pems
602    mock_session = Mock()
603    gateway = WebRTCGateway(
604        http_session=mock_session,
605        remote_id="TEST-REMOTE-ID",
606        cert_pem=cert_pem,
607        key_pem=key_pem,
608    )
609
610    # Check initial reconnect delay
611    assert gateway._current_reconnect_delay == 10
612
613    # Simulate multiple failed connections (without actually connecting)
614    initial_delay = gateway._current_reconnect_delay
615    gateway._current_reconnect_delay = min(
616        gateway._current_reconnect_delay * 2, gateway._max_reconnect_delay
617    )
618
619    assert gateway._current_reconnect_delay == initial_delay * 2
620
621    # Should not exceed max
622    for _ in range(10):
623        gateway._current_reconnect_delay = min(
624            gateway._current_reconnect_delay * 2, gateway._max_reconnect_delay
625        )
626
627    assert gateway._current_reconnect_delay <= gateway._max_reconnect_delay
628
629
630async def test_webrtc_gateway_handle_offer_without_session(cert_pems: tuple[str, str]) -> None:
631    """Test WebRTCGateway handles offer for non-existent session gracefully."""
632    cert_pem, key_pem = cert_pems
633    mock_session = Mock()
634    gateway = WebRTCGateway(
635        http_session=mock_session,
636        remote_id="TEST-REMOTE-ID",
637        cert_pem=cert_pem,
638        key_pem=key_pem,
639    )
640
641    # Try to handle offer for non-existent session
642    offer_data = {"sdp": "test-sdp", "type": "offer"}
643    await gateway._handle_offer("nonexistent-session", offer_data)
644
645    # Should not crash
646
647
648async def test_webrtc_gateway_handle_ice_candidate_without_session(
649    cert_pems: tuple[str, str],
650) -> None:
651    """Test WebRTCGateway handles ICE candidate for non-existent session gracefully."""
652    cert_pem, key_pem = cert_pems
653    mock_session = Mock()
654    gateway = WebRTCGateway(
655        http_session=mock_session,
656        remote_id="TEST-REMOTE-ID",
657        cert_pem=cert_pem,
658        key_pem=key_pem,
659    )
660
661    # Try to handle ICE candidate for non-existent session
662    candidate_data = {
663        "candidate": "candidate:1 1 UDP 1234 192.168.1.1 12345 typ host",
664        "sdpMid": "0",
665        "sdpMLineIndex": 0,
666    }
667    await gateway._handle_ice_candidate("nonexistent-session", candidate_data)
668
669    # Should not crash
670
671
672@pytest.mark.parametrize(
673    "malicious_path",
674    [
675        "@evil.com",  # netloc becomes basic-auth creds, evil.com becomes the host
676        "//evil.com",  # protocol-relative URL pointing at another host
677        "@evil.com/foo",
678        "//evil.com/foo",
679    ],
680)
681async def test_http_proxy_request_cannot_change_host(
682    cert_pems: tuple[str, str], malicious_path: str
683) -> None:
684    """An attacker-controlled proxy path must never change the target host (SSRF guard)."""
685    cert_pem, key_pem = cert_pems
686    mock_session = Mock()
687    captured_url: dict[str, str] = {}
688
689    def fake_request(_method: str, url: str, **_kwargs: object) -> AsyncMock:
690        captured_url["url"] = url
691        response = AsyncMock()
692        response.status = 200
693        response.headers = {}
694        response.read = AsyncMock(return_value=b"")
695        ctx = AsyncMock()
696        ctx.__aenter__ = AsyncMock(return_value=response)
697        ctx.__aexit__ = AsyncMock(return_value=False)
698        return ctx
699
700    mock_session.request = fake_request
701
702    gateway = WebRTCGateway(
703        http_session=mock_session,
704        remote_id="TEST-REMOTE-ID",
705        cert_pem=cert_pem,
706        key_pem=key_pem,
707        local_ws_url="ws://localhost:8095/ws",
708    )
709
710    await gateway._handle_http_proxy_request(
711        None, {"id": "1", "method": "GET", "path": malicious_path}
712    )
713
714    parsed = urlparse(captured_url["url"])
715    assert parsed.hostname == "localhost"
716    assert parsed.port == 8095
717    assert parsed.username is None
718    assert "evil.com" not in (parsed.netloc or "")
719
720
721async def test_http_proxy_request_keeps_the_unverified_dial_on_this_host(
722    cert_pems: tuple[str, str],
723) -> None:
724    """The proxy must not carry its unverified TLS dial off this host."""
725    cert_pem, key_pem = cert_pems
726    mock_session = Mock()
727    captured_kwargs: dict[str, object] = {}
728
729    def fake_request(_method: str, _url: str, **kwargs: object) -> AsyncMock:
730        captured_kwargs.update(kwargs)
731        response = AsyncMock()
732        response.status = 200
733        response.headers = {}
734        response.read = AsyncMock(return_value=b"")
735        ctx = AsyncMock()
736        ctx.__aenter__ = AsyncMock(return_value=response)
737        ctx.__aexit__ = AsyncMock(return_value=False)
738        return ctx
739
740    mock_session.request = fake_request
741
742    gateway = WebRTCGateway(
743        http_session=mock_session,
744        remote_id="TEST-REMOTE-ID",
745        cert_pem=cert_pem,
746        key_pem=key_pem,
747        local_ws_url="wss://127.0.0.1:8095/ws",
748    )
749
750    await gateway._handle_http_proxy_request(None, {"id": "1", "method": "GET", "path": "/info"})
751
752    assert captured_kwargs["ssl"] is False
753    assert captured_kwargs["allow_redirects"] is False
754
755
756async def test_local_websocket_dial_skips_certificate_verification(
757    cert_pems: tuple[str, str],
758) -> None:
759    """Reach the local API on the bind address, which no certificate is issued for."""
760    cert_pem, key_pem = cert_pems
761    mock_session = Mock()
762    captured_kwargs: dict[str, object] = {}
763
764    async def fake_ws_connect(_url: str, **kwargs: object) -> AsyncMock:
765        captured_kwargs.update(kwargs)
766        return AsyncMock()
767
768    mock_session.ws_connect = fake_ws_connect
769
770    gateway = WebRTCGateway(
771        http_session=mock_session,
772        remote_id="TEST-REMOTE-ID",
773        cert_pem=cert_pem,
774        key_pem=key_pem,
775        local_ws_url="wss://127.0.0.1:8095/ws",
776    )
777    session = WebRTCSession(session_id="s1", pc=Mock())
778    channel = MagicMock()
779    channel.wait_open = AsyncMock()
780
781    with patch.object(gateway, "_schedule_close"):
782        await gateway._bridge_ma_api(session, channel)
783
784    assert captured_kwargs["ssl"] is False
785
786
787# ---- aiolibdatachannel loopback tests --------------------------------------
788#
789# These exercise the migrated gateway against real loopback PeerConnections
790# (offerer = browser role, answerer = gateway role) rather than mocking the
791# WebRTC layer. ICE servers are stubbed to [] so gathering completes on host
792# candidates only (fast, offline).
793
794
795@pytest.fixture(scope="session")
796def cert_pems() -> tuple[str, str]:
797    """Generate a throwaway DTLS cert/key as PEM strings for the gateway."""
798    private_key, cert = _generate_certificate()
799    cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode()
800    key_pem = private_key.private_bytes(
801        encoding=serialization.Encoding.PEM,
802        format=serialization.PrivateFormat.PKCS8,
803        encryption_algorithm=serialization.NoEncryption(),
804    ).decode()
805    return cert_pem, key_pem
806
807
808def _sha256_fingerprint(cert_pem: str) -> str:
809    """Compute the uppercase colon-separated SHA-256 fingerprint of a PEM certificate."""
810    body = "".join(line for line in cert_pem.splitlines() if line and not line.startswith("-----"))
811    der = base64.b64decode(body)
812    digest = hashlib.sha256(der).hexdigest().upper()
813    return ":".join(digest[i : i + 2] for i in range(0, len(digest), 2))
814
815
816class _FakeSignaling:
817    """Signaling WebSocket stand-in that captures outbound JSON messages."""
818
819    def __init__(self) -> None:
820        self.messages: list[dict[str, Any]] = []
821
822    async def send_json(self, data: dict[str, Any]) -> None:
823        self.messages.append(data)
824
825    @property
826    def answers(self) -> list[dict[str, Any]]:
827        return [m for m in self.messages if m.get("type") == "answer"]
828
829
830class _FakeLocalWS:
831    """Minimal aiohttp WebSocket stand-in for channel-bridging tests."""
832
833    def __init__(self) -> None:
834        self.closed = False
835        self.sent: list[str | bytes] = []
836        self._incoming: asyncio.Queue[SimpleNamespace | None] = asyncio.Queue()
837
838    async def send_str(self, data: str) -> None:
839        self.sent.append(data)
840
841    async def send_bytes(self, data: bytes) -> None:
842        self.sent.append(data)
843
844    async def close(self) -> None:
845        self.closed = True
846        self._incoming.put_nowait(None)
847
848    def feed_text(self, data: str) -> None:
849        """Queue a text message as if the local server sent it."""
850        self._incoming.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=data))
851
852    def feed_bytes(self, data: bytes) -> None:
853        """Queue a binary message as if the local server sent it."""
854        self._incoming.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.BINARY, data=data))
855
856    def __aiter__(self) -> AsyncIterator[SimpleNamespace]:
857        return self
858
859    async def __anext__(self) -> SimpleNamespace:
860        msg = await self._incoming.get()
861        if msg is None:
862            raise StopAsyncIteration
863        return msg
864
865
866class _FakeBidiChannel:
867    """
868    Async-iterable data-channel stand-in for bridging tests.
869
870    Drives the gateway's channel->local pump without a real WebRTC handshake: feed
871    inbound messages with :meth:`feed`, end the stream with :meth:`close`, and read
872    what the gateway sent back on ``sent``.
873    """
874
875    def __init__(self, label: str = "ma-api", max_message_size: int = 256 * 1024) -> None:
876        self.label = label
877        self.is_open = True
878        self.is_closed = False
879        self.closed = False
880        self.max_message_size = max_message_size
881        self.sent: list[str | bytes] = []
882        self._inbound: asyncio.Queue[str | bytes | None] = asyncio.Queue()
883
884    async def wait_open(self) -> None:
885        return
886
887    async def send(self, data: str | bytes) -> None:
888        # a real send always suspends, which is what lets concurrent senders interleave
889        await asyncio.sleep(0)
890        self.sent.append(data)
891
892    def feed(self, message: str | bytes) -> None:
893        """Queue an inbound message as if the browser sent it over the channel."""
894        self._inbound.put_nowait(message)
895
896    def close(self) -> None:
897        self.closed = True
898        self.is_closed = True
899        self.is_open = False
900        self._inbound.put_nowait(None)
901
902    async def aclose(self) -> None:
903        self.close()
904
905    def __aiter__(self) -> AsyncIterator[str | bytes]:
906        return self
907
908    async def __anext__(self) -> str | bytes:
909        message = await self._inbound.get()
910        if message is None:
911            raise StopAsyncIteration
912        return message
913
914
915class _FakeHttpSession:
916    """ClientSession stand-in handing out one fake WebSocket per dialed url."""
917
918    def __init__(self) -> None:
919        self.dialed: list[str] = []
920        self.dial_kwargs: list[dict[str, Any]] = []
921        self.websockets: dict[str, _FakeLocalWS] = {}
922        self.requested: list[str] = []
923        self.response_body = b""
924        self.bodies: dict[str, bytes] = {}
925
926    async def ws_connect(self, url: str, **kwargs: Any) -> _FakeLocalWS:
927        self.dialed.append(url)
928        self.dial_kwargs.append(kwargs)
929        local_ws = _FakeLocalWS()
930        self.websockets[url] = local_ws
931        return local_ws
932
933    def request(self, _method: str, url: str, **_kwargs: Any) -> AsyncMock:
934        """Serve this url's entry in ``bodies``, falling back to ``response_body``."""
935        self.requested.append(url)
936        response = AsyncMock()
937        response.status = 200
938        response.headers = {"Content-Type": "image/jpeg"}
939        response.read = AsyncMock(return_value=self.bodies.get(url, self.response_body))
940        ctx = AsyncMock()
941        ctx.__aenter__ = AsyncMock(return_value=response)
942        ctx.__aexit__ = AsyncMock(return_value=False)
943        return ctx
944
945
946class _FakePeerConnection:
947    """PeerConnection stand-in that runs gateway-spawned pumps as asyncio tasks."""
948
949    def __init__(self) -> None:
950        self._tasks: list[asyncio.Task[None]] = []
951        self._incoming: asyncio.Queue[_FakeBidiChannel] = asyncio.Queue()
952
953    def spawn_task(self, coro: Coroutine[Any, Any, None]) -> None:
954        self._tasks.append(asyncio.ensure_future(coro))
955
956    def offer_channel(self, channel: _FakeBidiChannel) -> None:
957        """Offer a data channel as if the browser had opened it."""
958        self._incoming.put_nowait(channel)
959
960    async def incoming_data_channels(self) -> AsyncIterator[DataChannel]:
961        while True:
962            yield cast("DataChannel", await self._incoming.get())
963
964    async def aclose(self) -> None:
965        for task in self._tasks:
966            task.cancel()
967        # await cancellation so no pump task lingers past teardown
968        await asyncio.gather(*self._tasks, return_exceptions=True)
969
970
971def _register_bridge_session(
972    gateway: WebRTCGateway, session_id: str, channel: _FakeBidiChannel
973) -> WebRTCSession:
974    """Register a session backed by a fake PeerConnection and ma-api channel."""
975    session = WebRTCSession(
976        session_id=session_id,
977        pc=cast("PeerConnection", _FakePeerConnection()),
978        data_channel=cast("DataChannel", channel),
979    )
980    gateway.sessions[session_id] = session
981    return session
982
983
984def _register_routed_session(
985    gateway: WebRTCGateway, session_id: str
986) -> tuple[WebRTCSession, _FakePeerConnection]:
987    """Register a session that routes the data channels offered to its PeerConnection."""
988    pc = _FakePeerConnection()
989    session = WebRTCSession(session_id=session_id, pc=cast("PeerConnection", pc))
990    gateway.sessions[session_id] = session
991    pc.spawn_task(gateway._accept_channels(session))
992    return session, pc
993
994
995async def _wait_for(predicate: Callable[[], bool], timeout: float = 15.0) -> None:
996    """Poll ``predicate`` until it is true or the timeout elapses."""
997    loop = asyncio.get_running_loop()
998    deadline = loop.time() + timeout
999    while loop.time() < deadline:
1000        if predicate():
1001            return
1002        await asyncio.sleep(0.02)
1003    raise AssertionError("condition not met within timeout")
1004
1005
1006async def test_handle_offer_answers_with_pinned_fingerprint(cert_pems: tuple[str, str]) -> None:
1007    """The answer SDP carries the DTLS fingerprint of the configured certificate (pinning)."""
1008    cert_pem, key_pem = cert_pems
1009    gateway = WebRTCGateway(
1010        http_session=Mock(),
1011        remote_id="TEST-REMOTE-ID",
1012        cert_pem=cert_pem,
1013        key_pem=key_pem,
1014    )
1015    signaling = _FakeSignaling()
1016    gateway._signaling_ws = cast("aiohttp.ClientWebSocketResponse", signaling)
1017    offerer = PeerConnection(RTCConfiguration())
1018    session_id = "fingerprint-session"
1019    try:
1020        await offerer.create_data_channel("ma-api")
1021        offer = await offerer.create_offer()
1022        with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
1023            await gateway._create_session(session_id)
1024            await asyncio.wait_for(
1025                gateway._handle_offer(session_id, {"sdp": offer.sdp, "type": "offer"}),
1026                timeout=15,
1027            )
1028
1029        assert len(signaling.answers) == 1
1030        answer = signaling.answers[0]
1031        assert answer["sessionId"] == session_id
1032        assert answer["data"]["type"] == "answer"
1033        fingerprint_line = next(
1034            line for line in answer["data"]["sdp"].splitlines() if line.startswith("a=fingerprint:")
1035        )
1036        assert _sha256_fingerprint(cert_pem) in fingerprint_line
1037    finally:
1038        await gateway._close_session(session_id)
1039        await offerer.aclose()
1040
1041
1042async def test_ma_api_channel_bridges_to_local_ws(cert_pems: tuple[str, str]) -> None:
1043    """Messages flow both ways across the ma-api data channel and the local WebSocket."""
1044    cert_pem, key_pem = cert_pems
1045    fake_ws = _FakeLocalWS()
1046    http_session = Mock()
1047    http_session.ws_connect = AsyncMock(
1048        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1049    )
1050    gateway = WebRTCGateway(
1051        http_session=http_session,
1052        remote_id="TEST-REMOTE-ID",
1053        cert_pem=cert_pem,
1054        key_pem=key_pem,
1055    )
1056    channel = _FakeBidiChannel()
1057    session = _register_bridge_session(gateway, "bridge-session", channel)
1058    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1059    try:
1060        await _wait_for(lambda: session.local_ws is not None)
1061
1062        # browser -> local WebSocket
1063        channel.feed("from browser")
1064        await _wait_for(lambda: fake_ws.sent == ["from browser"])
1065
1066        # local WebSocket -> browser
1067        fake_ws.feed_text("from local")
1068        await _wait_for(lambda: channel.sent == ["from local"])
1069    finally:
1070        channel.close()
1071        await asyncio.wait_for(bridge, timeout=5)
1072        await _wait_for(lambda: "bridge-session" not in gateway.sessions)
1073
1074
1075def test_build_ice_servers_maps_dicts() -> None:
1076    """ICE server dicts map to one IceServer per url, preserving TURN credentials."""
1077    gateway = WebRTCGateway(
1078        http_session=Mock(),
1079        remote_id="TEST-REMOTE-ID",
1080        cert_pem="cert",
1081        key_pem="key",
1082    )
1083    servers: list[dict[str, Any]] = [
1084        {"urls": "stun:stun.example.com:3478"},
1085        {"urls": "turn:turn.example.com:3478", "username": "user", "credential": "pass"},
1086        {
1087            "urls": ["stun:a.example.com:3478", "turn:b.example.com:3478"],
1088            "username": "u2",
1089            "credential": "c2",
1090        },
1091    ]
1092
1093    result = gateway._build_ice_servers(servers)
1094
1095    assert all(isinstance(server, IceServer) for server in result)
1096    # the two-url entry fans out, so 1 + 1 + 2 = 4 IceServers
1097    assert len(result) == 4
1098    assert result[0] == IceServer(url="stun:stun.example.com:3478")
1099    assert result[1] == IceServer(
1100        url="turn:turn.example.com:3478", username="user", credential="pass"
1101    )
1102    # to_url() inlines the TURN credentials for libdatachannel
1103    assert result[1].to_url() == "turn:user:[email protected]:3478"
1104    # list urls share the entry's credentials
1105    assert result[2] == IceServer(url="stun:a.example.com:3478", username="u2", credential="c2")
1106    assert result[3] == IceServer(url="turn:b.example.com:3478", username="u2", credential="c2")
1107
1108
1109async def test_session_closes_when_ma_api_channel_closes(cert_pems: tuple[str, str]) -> None:
1110    """Closing the browser ma-api channel tears down the whole gateway session."""
1111    cert_pem, key_pem = cert_pems
1112    fake_ws = _FakeLocalWS()
1113    http_session = Mock()
1114    http_session.ws_connect = AsyncMock(
1115        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1116    )
1117    gateway = WebRTCGateway(
1118        http_session=http_session,
1119        remote_id="TEST-REMOTE-ID",
1120        cert_pem=cert_pem,
1121        key_pem=key_pem,
1122    )
1123    channel = _FakeBidiChannel()
1124    session = _register_bridge_session(gateway, "channel-close-session", channel)
1125    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1126    await _wait_for(lambda: session.local_ws is not None)
1127
1128    # the browser closes the ma-api channel -> the whole session is torn down
1129    channel.close()
1130
1131    await _wait_for(lambda: "channel-close-session" not in gateway.sessions)
1132    assert "channel-close-session" not in gateway.sessions
1133    await asyncio.wait_for(bridge, timeout=5)
1134
1135
1136# ---- channel routing -------------------------------------------------------
1137
1138LOCAL_WS_URL = "ws://127.0.0.1:8095/ws"
1139SENDSPIN_URL = "ws://127.0.0.1:8927/sendspin"
1140
1141
1142def _routing_gateway(
1143    cert_pems: tuple[str, str],
1144    http_session: _FakeHttpSession,
1145    local_ws_url: str = LOCAL_WS_URL,
1146    set_sendspin_player_callback: Callable[[str, str], None] | None = None,
1147) -> WebRTCGateway:
1148    """Create a gateway whose local WebSockets are all served by the fake HTTP session."""
1149    cert_pem, key_pem = cert_pems
1150    return WebRTCGateway(
1151        http_session=cast("aiohttp.ClientSession", http_session),
1152        remote_id="TEST-REMOTE-ID",
1153        cert_pem=cert_pem,
1154        key_pem=key_pem,
1155        local_ws_url=local_ws_url,
1156        sendspin_url=SENDSPIN_URL,
1157        set_sendspin_player_callback=set_sendspin_player_callback,
1158    )
1159
1160
1161async def test_sendspin_channel_bridges_to_the_sendspin_server(cert_pems: tuple[str, str]) -> None:
1162    """A sendspin channel reaches the internal sendspin server, web player id and all."""
1163    http_session = _FakeHttpSession()
1164    announced_players: list[tuple[str, str]] = []
1165    gateway = _routing_gateway(
1166        cert_pems,
1167        http_session,
1168        set_sendspin_player_callback=lambda session_id, player_id: announced_players.append(
1169            (session_id, player_id)
1170        ),
1171    )
1172    session, pc = _register_routed_session(gateway, "sendspin-session")
1173    channel = _FakeBidiChannel(label="sendspin")
1174    pc.offer_channel(channel)
1175    try:
1176        await _wait_for(lambda: SENDSPIN_URL in http_session.websockets)
1177        local_ws = http_session.websockets[SENDSPIN_URL]
1178
1179        # the first message announces the web player, and is forwarded verbatim
1180        auth = json.dumps({"type": "auth", "token": "t", "client_id": "web-player-1"})
1181        channel.feed(auth)
1182        await _wait_for(lambda: local_ws.sent == [auth])
1183        assert announced_players == [("sendspin-session", "web-player-1")]
1184        assert session.sendspin_player_id == "web-player-1"
1185
1186        # audio keeps flowing in both directions, text and binary alike
1187        channel.feed(b"\x01\x02")
1188        await _wait_for(lambda: local_ws.sent == [auth, b"\x01\x02"])
1189        local_ws.feed_text('{"type":"hello"}')
1190        local_ws.feed_bytes(b"\x03\x04")
1191        await _wait_for(lambda: channel.sent == ['{"type":"hello"}', b"\x03\x04"])
1192    finally:
1193        await gateway._close_session("sendspin-session")
1194
1195
1196@pytest.mark.parametrize(
1197    ("local_ws_url", "expected_url"),
1198    [
1199        ("ws://127.0.0.1:8095/ws", "ws://127.0.0.1:8095/live_announcement"),
1200        # an https webserver is still dialed on its bind address, which no cert covers
1201        ("wss://127.0.0.1:8095/ws", "wss://127.0.0.1:8095/live_announcement"),
1202    ],
1203)
1204async def test_live_announcement_channel_bridges_to_the_webserver(
1205    cert_pems: tuple[str, str], local_ws_url: str, expected_url: str
1206) -> None:
1207    """A live announcement channel reaches the webserver route that takes the audio."""
1208    http_session = _FakeHttpSession()
1209    gateway = _routing_gateway(cert_pems, http_session, local_ws_url=local_ws_url)
1210    session, pc = _register_routed_session(gateway, "announce-session")
1211    channel = _FakeBidiChannel(label="live_announcement")
1212    pc.offer_channel(channel)
1213    try:
1214        await _wait_for(lambda: expected_url in http_session.websockets)
1215        local_ws = http_session.websockets[expected_url]
1216        assert http_session.dial_kwargs == [{"ssl": False}]
1217
1218        # the client authenticates on the route itself, so its handshake passes through
1219        handshake = [
1220            json.dumps({"type": "auth", "token": "t"}),
1221            json.dumps({"type": "start", "player_id": "player1", "sample_rate": 16000}),
1222        ]
1223        for message in handshake:
1224            channel.feed(message)
1225        await _wait_for(lambda: local_ws.sent == handshake)
1226        # the sendspin snoop belongs to the sendspin bridge only
1227        assert session.sendspin_player_id is None
1228
1229        # spoken audio goes up, the route's replies come back
1230        channel.feed(b"\x00\x01")
1231        await _wait_for(lambda: local_ws.sent == [*handshake, b"\x00\x01"])
1232        local_ws.feed_text('{"type":"started"}')
1233        await _wait_for(lambda: channel.sent == ['{"type":"started"}'])
1234    finally:
1235        await gateway._close_session("announce-session")
1236
1237
1238@pytest.mark.parametrize("closed_by", ["browser", "local"])
1239async def test_closing_a_bridged_channel_leaves_the_api_session_up(
1240    cert_pems: tuple[str, str], closed_by: str
1241) -> None:
1242    """Losing a bridged WebSocket tears down that bridge only, never the API session."""
1243    http_session = _FakeHttpSession()
1244    gateway = _routing_gateway(cert_pems, http_session)
1245    session, pc = _register_routed_session(gateway, "mixed-session")
1246    api_channel = _FakeBidiChannel()
1247    sendspin_channel = _FakeBidiChannel(label="sendspin")
1248    pc.offer_channel(api_channel)
1249    pc.offer_channel(sendspin_channel)
1250    try:
1251        await _wait_for(
1252            lambda: session.local_ws is not None and SENDSPIN_URL in http_session.dialed
1253        )
1254        sendspin_ws = http_session.websockets[SENDSPIN_URL]
1255
1256        if closed_by == "browser":
1257            sendspin_channel.close()
1258        else:
1259            await sendspin_ws.close()
1260        await _wait_for(lambda: not session.channels)
1261
1262        assert sendspin_ws.closed is True
1263        assert sendspin_channel.closed is True
1264        # the API session is untouched: still registered, still bridged, channel still open
1265        assert "mixed-session" in gateway.sessions
1266        assert session.local_ws is not None
1267        assert cast("_FakeLocalWS", session.local_ws).closed is False
1268        assert api_channel.closed is False
1269    finally:
1270        await gateway._close_session("mixed-session")
1271
1272
1273async def test_the_first_channel_is_the_api_channel_whatever_its_label(
1274    cert_pems: tuple[str, str],
1275) -> None:
1276    """Clients may label their API channel freely, so the first channel bridges to the API."""
1277    http_session = _FakeHttpSession()
1278    gateway = _routing_gateway(cert_pems, http_session)
1279    session, pc = _register_routed_session(gateway, "labelled-session")
1280    channel = _FakeBidiChannel(label="ma-api-v2")
1281    pc.offer_channel(channel)
1282    try:
1283        await _wait_for(lambda: session.local_ws is not None)
1284        assert session.data_channel is cast("DataChannel", channel)
1285        assert http_session.dialed == [f"{LOCAL_WS_URL}?webrtc_session_id=labelled-session"]
1286    finally:
1287        await gateway._close_session("labelled-session")
1288
1289
1290async def test_unknown_channel_label_cannot_replace_the_api_channel(
1291    cert_pems: tuple[str, str],
1292) -> None:
1293    """A channel this server has no route for is refused instead of hijacking the session."""
1294    http_session = _FakeHttpSession()
1295    gateway = _routing_gateway(cert_pems, http_session)
1296    session, pc = _register_routed_session(gateway, "unknown-session")
1297    api_channel = _FakeBidiChannel()
1298    unknown_channel = _FakeBidiChannel(label="channel-from-the-future")
1299    pc.offer_channel(api_channel)
1300    try:
1301        await _wait_for(lambda: session.local_ws is not None)
1302        pc.offer_channel(unknown_channel)
1303        await _wait_for(lambda: unknown_channel.closed)
1304
1305        assert session.data_channel is cast("DataChannel", api_channel)
1306        assert "unknown-session" in gateway.sessions
1307        # only the API channel was ever bridged
1308        assert http_session.dialed == [f"{LOCAL_WS_URL}?webrtc_session_id=unknown-session"]
1309    finally:
1310        await gateway._close_session("unknown-session")
1311
1312
1313def _proxy_request(request_id: str, path: str) -> str:
1314    """Build the http-proxy-request message a client sends for a proxied path."""
1315    return json.dumps(
1316        {"type": "http-proxy-request", "id": request_id, "method": "GET", "path": path}
1317    )
1318
1319
1320def _body_delivered(sent: list[str | bytes], size: int) -> bool:
1321    """Return whether the binary frames sent so far add up to a whole body of ``size``."""
1322    return sum(len(m) for m in sent if isinstance(m, bytes)) >= size
1323
1324
1325def _read_proxy_response(sent: list[str | bytes]) -> tuple[dict[str, Any], bytes]:
1326    """
1327    Read the binary-framed response a client would reassemble from the proxy channel.
1328
1329    :param sent: Messages the gateway sent, starting at the response's JSON header.
1330    """
1331    header = json.loads(cast("str", sent[0]))
1332    body = b""
1333    for message in sent[1:]:
1334        assert isinstance(message, bytes), f"expected a binary body frame, got {message!r}"
1335        body += message
1336        if len(body) >= header["size"]:
1337            break
1338    assert len(body) == header["size"]
1339    return header, body
1340
1341
1342async def test_http_proxy_channel_answers_on_its_own_channel(cert_pems: tuple[str, str]) -> None:
1343    """A proxied request on the http proxy channel is answered there, not on the API channel."""
1344    http_session = _FakeHttpSession()
1345    http_session.response_body = b"\xff\xd8jpeg-bytes"
1346    gateway = _routing_gateway(cert_pems, http_session)
1347    session, pc = _register_routed_session(gateway, "proxy-session")
1348    api_channel = _FakeBidiChannel()
1349    proxy_channel = _FakeBidiChannel(label="http_proxy")
1350    pc.offer_channel(api_channel)
1351    pc.offer_channel(proxy_channel)
1352    try:
1353        await _wait_for(lambda: session.local_ws is not None)
1354
1355        proxy_channel.feed(_proxy_request("img-1", "/imageproxy/abc"))
1356        await _wait_for(lambda: len(proxy_channel.sent) >= 2)
1357
1358        assert http_session.requested == ["http://127.0.0.1:8095/imageproxy/abc"]
1359        response, body = _read_proxy_response(proxy_channel.sent)
1360        assert response["type"] == "http-proxy-response"
1361        assert response["id"] == "img-1"
1362        assert response["status"] == 200
1363        # the body rides as raw binary, so it costs its own size on the wire and no more
1364        assert body == b"\xff\xd8jpeg-bytes"
1365        assert "body" not in response
1366        # the image never touches the API channel, nor the local API WebSocket
1367        assert api_channel.sent == []
1368        assert cast("_FakeLocalWS", session.local_ws).sent == []
1369    finally:
1370        await gateway._close_session("proxy-session")
1371
1372
1373async def test_http_proxy_request_on_the_api_channel_is_still_answered(
1374    cert_pems: tuple[str, str],
1375) -> None:
1376    """Clients that predate the http proxy channel keep proxying over the API channel."""
1377    http_session = _FakeHttpSession()
1378    http_session.response_body = b"legacy-body"
1379    gateway = _routing_gateway(cert_pems, http_session)
1380    session, pc = _register_routed_session(gateway, "legacy-session")
1381    api_channel = _FakeBidiChannel()
1382    pc.offer_channel(api_channel)
1383    try:
1384        await _wait_for(lambda: session.local_ws is not None)
1385
1386        api_channel.feed(_proxy_request("img-2", "/imageproxy/def"))
1387        await _wait_for(lambda: bool(api_channel.sent))
1388
1389        assert http_session.requested == ["http://127.0.0.1:8095/imageproxy/def"]
1390        response = json.loads(cast("str", api_channel.sent[0]))
1391        assert response["type"] == "http-proxy-response"
1392        assert response["id"] == "img-2"
1393        assert bytes.fromhex(response["body"]) == b"legacy-body"
1394        # the proxy request is served here, never forwarded to the local API WebSocket
1395        assert cast("_FakeLocalWS", session.local_ws).sent == []
1396    finally:
1397        await gateway._close_session("legacy-session")
1398
1399
1400async def test_http_proxy_channel_reports_a_failed_fetch_on_its_own_channel(
1401    cert_pems: tuple[str, str],
1402) -> None:
1403    """A fetch that raises still answers the client, on the channel it asked over."""
1404    http_session = _FakeHttpSession()
1405    http_session.request = Mock(side_effect=RuntimeError("boom"))  # type: ignore[method-assign]
1406    gateway = _routing_gateway(cert_pems, http_session)
1407    session, pc = _register_routed_session(gateway, "proxy-error-session")
1408    api_channel = _FakeBidiChannel()
1409    proxy_channel = _FakeBidiChannel(label="http_proxy")
1410    pc.offer_channel(api_channel)
1411    pc.offer_channel(proxy_channel)
1412    try:
1413        await _wait_for(lambda: session.local_ws is not None)
1414
1415        proxy_channel.feed(_proxy_request("img-3", "/imageproxy/boom"))
1416        await _wait_for(lambda: len(proxy_channel.sent) >= 2)
1417
1418        response, body = _read_proxy_response(proxy_channel.sent)
1419        assert response["id"] == "img-3"
1420        assert response["status"] == 500
1421        assert b"boom" in body
1422        assert api_channel.sent == []
1423    finally:
1424        await gateway._close_session("proxy-error-session")
1425
1426
1427async def test_http_proxy_channel_splits_a_large_body_into_binary_frames(
1428    cert_pems: tuple[str, str],
1429) -> None:
1430    """A body past the channel's message limit arrives as raw frames that concatenate back."""
1431    http_session = _FakeHttpSession()
1432    http_session.response_body = bytes(range(256)) * 2048  # 512 KiB
1433    gateway = _routing_gateway(cert_pems, http_session)
1434    session, pc = _register_routed_session(gateway, "proxy-large-session")
1435    proxy_channel = _FakeBidiChannel(label="http_proxy")
1436    pc.offer_channel(proxy_channel)
1437    try:
1438        await _wait_for(lambda: "http_proxy" in session.channels)
1439
1440        proxy_channel.feed(_proxy_request("img-5", "/imageproxy/large"))
1441        await _wait_for(
1442            lambda: _body_delivered(proxy_channel.sent, len(http_session.response_body))
1443        )
1444
1445        response, body = _read_proxy_response(proxy_channel.sent)
1446        assert response["size"] == len(http_session.response_body)
1447        assert body == http_session.response_body
1448        frames = proxy_channel.sent[1:]
1449        assert all(len(frame) <= proxy_channel.max_message_size for frame in frames)
1450        # the wire cost is the body itself, not the ~2.7x a hex-in-base64 response took
1451        assert sum(len(frame) for frame in frames) == len(http_session.response_body)
1452    finally:
1453        await gateway._close_session("proxy-large-session")
1454
1455
1456async def test_http_proxy_channel_honours_the_negotiated_message_limit(
1457    cert_pems: tuple[str, str],
1458) -> None:
1459    """A peer that advertises a small limit gets frames it can actually accept."""
1460    http_session = _FakeHttpSession()
1461    http_session.response_body = b"x" * (200 * 1024)
1462    gateway = _routing_gateway(cert_pems, http_session)
1463    session, pc = _register_routed_session(gateway, "proxy-small-frames-session")
1464    # what a peer that advertises no a=max-message-size in its SDP is assumed to accept
1465    proxy_channel = _FakeBidiChannel(label="http_proxy", max_message_size=64 * 1024)
1466    pc.offer_channel(proxy_channel)
1467    try:
1468        await _wait_for(lambda: "http_proxy" in session.channels)
1469
1470        proxy_channel.feed(_proxy_request("img-6", "/imageproxy/small-frames"))
1471        await _wait_for(
1472            lambda: _body_delivered(proxy_channel.sent, len(http_session.response_body))
1473        )
1474
1475        _, body = _read_proxy_response(proxy_channel.sent)
1476        assert body == http_session.response_body
1477        assert all(len(frame) <= 64 * 1024 for frame in proxy_channel.sent[1:])
1478    finally:
1479        await gateway._close_session("proxy-small-frames-session")
1480
1481
1482async def test_http_proxy_channel_never_interleaves_two_responses(
1483    cert_pems: tuple[str, str],
1484) -> None:
1485    """Body frames carry no request id, so each response must reach the client in one run."""
1486    http_session = _FakeHttpSession()
1487    http_session.bodies = {
1488        "http://127.0.0.1:8095/imageproxy/one": b"1" * (300 * 1024),
1489        "http://127.0.0.1:8095/imageproxy/two": b"2" * (300 * 1024),
1490    }
1491    gateway = _routing_gateway(cert_pems, http_session)
1492    session, pc = _register_routed_session(gateway, "proxy-concurrent-session")
1493    proxy_channel = _FakeBidiChannel(label="http_proxy")
1494    pc.offer_channel(proxy_channel)
1495    try:
1496        await _wait_for(lambda: "http_proxy" in session.channels)
1497
1498        proxy_channel.feed(_proxy_request("img-one", "/imageproxy/one"))
1499        proxy_channel.feed(_proxy_request("img-two", "/imageproxy/two"))
1500        await _wait_for(lambda: sum(isinstance(m, str) for m in proxy_channel.sent) == 2)
1501        await _wait_for(lambda: len(proxy_channel.sent) >= 6)
1502
1503        # split at the second header: each response owns an unbroken run of body frames
1504        second = next(i for i, m in enumerate(proxy_channel.sent) if i and isinstance(m, str))
1505        first_header, first_body = _read_proxy_response(proxy_channel.sent[:second])
1506        second_header, second_body = _read_proxy_response(proxy_channel.sent[second:])
1507        assert {first_header["id"], second_header["id"]} == {"img-one", "img-two"}
1508        # each body is one repeated byte, so any interleaving shows up as a mixed run
1509        filler = {"img-one": ord("1"), "img-two": ord("2")}
1510        assert set(first_body) == {filler[first_header["id"]]}
1511        assert set(second_body) == {filler[second_header["id"]]}
1512        assert len(first_body) == len(second_body) == 300 * 1024
1513    finally:
1514        await gateway._close_session("proxy-concurrent-session")
1515
1516
1517@pytest.mark.parametrize(
1518    "junk",
1519    [
1520        "not json at all",
1521        json.dumps(["not", "a", "dict"]),
1522        json.dumps({"type": "something-else"}),
1523        b"\x00\x01binary",
1524    ],
1525)
1526async def test_http_proxy_channel_survives_junk(
1527    cert_pems: tuple[str, str], junk: str | bytes
1528) -> None:
1529    """Anything that is not a proxy request is ignored without killing the channel."""
1530    http_session = _FakeHttpSession()
1531    http_session.response_body = b"still-here"
1532    gateway = _routing_gateway(cert_pems, http_session)
1533    session, pc = _register_routed_session(gateway, "proxy-junk-session")
1534    proxy_channel = _FakeBidiChannel(label="http_proxy")
1535    pc.offer_channel(proxy_channel)
1536    try:
1537        await _wait_for(lambda: "http_proxy" in session.channels)
1538
1539        proxy_channel.feed(junk)
1540        proxy_channel.feed(_proxy_request("img-4", "/imageproxy/ok"))
1541        await _wait_for(lambda: len(proxy_channel.sent) >= 2)
1542
1543        response, body = _read_proxy_response(proxy_channel.sent)
1544        assert response["id"] == "img-4"
1545        assert body == b"still-here"
1546    finally:
1547        await gateway._close_session("proxy-junk-session")
1548
1549
1550async def test_closing_the_http_proxy_channel_leaves_the_api_session_up(
1551    cert_pems: tuple[str, str],
1552) -> None:
1553    """Losing the http proxy channel tears down that channel only, never the API session."""
1554    http_session = _FakeHttpSession()
1555    gateway = _routing_gateway(cert_pems, http_session)
1556    session, pc = _register_routed_session(gateway, "proxy-close-session")
1557    api_channel = _FakeBidiChannel()
1558    proxy_channel = _FakeBidiChannel(label="http_proxy")
1559    pc.offer_channel(api_channel)
1560    pc.offer_channel(proxy_channel)
1561    try:
1562        await _wait_for(lambda: "http_proxy" in session.channels)
1563
1564        proxy_channel.close()
1565        await _wait_for(lambda: not session.channels)
1566
1567        assert "proxy-close-session" in gateway.sessions
1568        assert session.local_ws is not None
1569        assert api_channel.closed is False
1570    finally:
1571        await gateway._close_session("proxy-close-session")
1572
1573
1574async def test_a_second_http_proxy_channel_is_refused(cert_pems: tuple[str, str]) -> None:
1575    """A duplicate label is refused so the running handler is never left untracked."""
1576    http_session = _FakeHttpSession()
1577    gateway = _routing_gateway(cert_pems, http_session)
1578    session, pc = _register_routed_session(gateway, "duplicate-session")
1579    first = _FakeBidiChannel(label="http_proxy")
1580    second = _FakeBidiChannel(label="http_proxy")
1581    pc.offer_channel(first)
1582    try:
1583        await _wait_for(lambda: "http_proxy" in session.channels)
1584        pc.offer_channel(second)
1585        await _wait_for(lambda: second.closed)
1586
1587        assert first.closed is False
1588        assert session.channels["http_proxy"].channel is cast("DataChannel", first)
1589    finally:
1590        await gateway._close_session("duplicate-session")
1591
1592
1593class _FakeDataChannel:
1594    """
1595    Data channel stand-in that captures outbound messages for proxy tests.
1596
1597    :param close_after: Close the channel once this many messages have been sent.
1598    """
1599
1600    def __init__(self, max_message_size: int = 256 * 1024, close_after: int | None = None) -> None:
1601        self.is_closed = False
1602        self.max_message_size = max_message_size
1603        self.sent: list[str] = []
1604        # the gateway consults the channel once per outbound message, so this counts how far
1605        # a send loop got even when the messages themselves are discarded
1606        self.open_checks = 0
1607        self._is_open = True
1608        self._close_after = close_after
1609
1610    @property
1611    def is_open(self) -> bool:
1612        self.open_checks += 1
1613        return self._is_open
1614
1615    async def send(self, data: str) -> None:
1616        self.sent.append(data)
1617        if self._close_after is not None and len(self.sent) >= self._close_after:
1618            self._is_open = False
1619            self.is_closed = True
1620
1621
1622def _proxy_gateway(cert_pems: tuple[str, str]) -> WebRTCGateway:
1623    cert_pem, key_pem = cert_pems
1624    return WebRTCGateway(
1625        http_session=Mock(),
1626        remote_id="TEST-REMOTE-ID",
1627        cert_pem=cert_pem,
1628        key_pem=key_pem,
1629    )
1630
1631
1632def _reassemble_chunks(frames: list[str]) -> str:
1633    """Reassemble __chunk__ frames back into the original text (base64 -> bytes -> utf-8)."""
1634    parsed = [json.loads(f) for f in frames]
1635    assert all(f["type"] == "__chunk__" for f in parsed)
1636    assert len({f["id"] for f in parsed}) == 1
1637    count = parsed[0]["count"]
1638    parts: list[bytes] = [b""] * count
1639    for frame in parsed:
1640        parts[frame["seq"]] = base64.b64decode(frame["b64"])
1641    return b"".join(parts).decode()
1642
1643
1644async def test_http_proxy_response_small_body_single_message(
1645    cert_pems: tuple[str, str],
1646) -> None:
1647    """A body within the chunk size is sent as one http-proxy-response message."""
1648    gateway = _proxy_gateway(cert_pems)
1649    channel = _FakeDataChannel()
1650    body = b"\x00\x01\x02small-body"
1651
1652    await gateway._send_http_proxy_response(
1653        cast("DataChannel", channel), "req-small", 200, {"X-Test": "y"}, body
1654    )
1655
1656    assert len(channel.sent) == 1
1657    msg = json.loads(channel.sent[0])
1658    assert msg["type"] == "http-proxy-response"
1659    assert msg["id"] == "req-small"
1660    assert msg["status"] == 200
1661    assert msg["headers"] == {"X-Test": "y"}
1662    assert bytes.fromhex(msg["body"]) == body
1663
1664
1665async def test_http_proxy_response_large_body_chunked(cert_pems: tuple[str, str]) -> None:
1666    """A large HTTP-proxy response is split into base64 chunk frames the client reassembles."""
1667    gateway = _proxy_gateway(cert_pems)
1668    channel = _FakeDataChannel()
1669    # big body -> big JSON message
1670    body = bytes(range(256)) * ((DATA_CHANNEL_CHUNK_SIZE * 5) // 512)
1671
1672    await gateway._send_http_proxy_response(cast("DataChannel", channel), "req-big", 200, {}, body)
1673
1674    assert len(channel.sent) > 1
1675    assert all(json.loads(m)["type"] == "__chunk__" for m in channel.sent)
1676    # every serialized frame must stay under the negotiated 256 KiB data-channel limit
1677    assert all(len(m.encode()) < 256 * 1024 for m in channel.sent)
1678
1679    reassembled = json.loads(_reassemble_chunks(channel.sent))
1680    assert reassembled["type"] == "http-proxy-response"
1681    assert reassembled["id"] == "req-big"
1682    assert reassembled["status"] == 200
1683    assert bytes.fromhex(reassembled["body"]) == body
1684
1685
1686async def test_send_chunked_small_message_passthrough(cert_pems: tuple[str, str]) -> None:
1687    """A message within the limit is sent verbatim, not chunked."""
1688    gateway = _proxy_gateway(cert_pems)
1689    channel = _FakeDataChannel()
1690    await gateway._send_chunked(cast("DataChannel", channel), '{"event":"player_updated"}')
1691    assert channel.sent == ['{"event":"player_updated"}']
1692
1693
1694async def test_send_chunked_large_message_chunked(cert_pems: tuple[str, str]) -> None:
1695    """A large message is chunked and reassembles byte-identically (multibyte-safe)."""
1696    gateway = _proxy_gateway(cert_pems)
1697    channel = _FakeDataChannel()
1698    # multibyte payload so chunk boundaries fall mid-character, exercising the byte-level split
1699    text = '{"data":"' + "音楽" * DATA_CHANNEL_CHUNK_SIZE + '"}'
1700
1701    await gateway._send_chunked(cast("DataChannel", channel), text)
1702
1703    assert len(channel.sent) > 1
1704    assert all(len(m.encode()) < 256 * 1024 for m in channel.sent)
1705    assert _reassemble_chunks(channel.sent) == text
1706
1707
1708async def test_send_chunked_keeps_the_framing_released_clients_expect(
1709    cert_pems: tuple[str, str],
1710) -> None:
1711    """Against the 256 KiB limit every browser advertises, the frames must not move."""
1712    gateway = _proxy_gateway(cert_pems)
1713    channel = _FakeDataChannel()
1714    text = '{"event":"' + "x" * (DATA_CHANNEL_CHUNK_SIZE * 2 + 5000) + '"}'
1715    data = text.encode()
1716
1717    await gateway._send_chunked(cast("DataChannel", channel), text)
1718
1719    # the exact wire format the bundled frontend and the mobile app reassemble: 64 KiB pieces,
1720    # numbered from zero within a group id that counts up per message
1721    assert channel.sent == [
1722        json.dumps(
1723            {
1724                "type": "__chunk__",
1725                "id": 1,
1726                "seq": seq,
1727                "count": 3,
1728                "b64": base64.b64encode(
1729                    data[seq * DATA_CHANNEL_CHUNK_SIZE : (seq + 1) * DATA_CHANNEL_CHUNK_SIZE]
1730                ).decode(),
1731            }
1732        )
1733        for seq in range(3)
1734    ]
1735    # a full piece has always serialised well past 64 KiB, which only a peer advertising no
1736    # limit of its own would reject
1737    assert len(channel.sent[0].encode()) == 87447
1738
1739
1740async def test_send_chunked_honours_the_negotiated_message_limit(
1741    cert_pems: tuple[str, str],
1742) -> None:
1743    """A peer that advertises no limit in its SDP gets frames it can actually accept."""
1744    gateway = _proxy_gateway(cert_pems)
1745    # what libdatachannel assumes when the peer advertises no a=max-message-size
1746    channel = _FakeDataChannel(max_message_size=64 * 1024)
1747    text = '{"data":"' + "音楽" * DATA_CHANNEL_CHUNK_SIZE + '"}'
1748
1749    await gateway._send_chunked(cast("DataChannel", channel), text)
1750
1751    assert len(channel.sent) > 1
1752    assert all(len(m.encode()) <= channel.max_message_size for m in channel.sent)
1753    assert _reassemble_chunks(channel.sent) == text
1754
1755
1756async def test_send_chunked_passthrough_stops_at_the_negotiated_limit(
1757    cert_pems: tuple[str, str],
1758) -> None:
1759    """A message past a small peer's limit is chunked rather than sent whole."""
1760    gateway = _proxy_gateway(cert_pems)
1761    channel = _FakeDataChannel(max_message_size=16 * 1024)
1762    text = '{"data":"' + "x" * (32 * 1024) + '"}'
1763
1764    await gateway._send_chunked(cast("DataChannel", channel), text)
1765
1766    assert len(channel.sent) > 1
1767    assert all(len(m.encode()) <= channel.max_message_size for m in channel.sent)
1768    assert _reassemble_chunks(channel.sent) == text
1769
1770
1771async def test_send_chunked_stops_framing_once_the_channel_closes(
1772    cert_pems: tuple[str, str],
1773) -> None:
1774    """A channel that goes away mid-message stops the loop instead of encoding the rest."""
1775    gateway = _proxy_gateway(cert_pems)
1776    channel = _FakeDataChannel(close_after=1)
1777    text = '{"data":"' + "x" * (DATA_CHANNEL_CHUNK_SIZE * 10) + '"}'
1778
1779    await gateway._send_chunked(cast("DataChannel", channel), text)
1780
1781    assert len(channel.sent) == 1
1782    # the guard and the send of the first piece, then the guard again: the other ten pieces
1783    # are never framed
1784    assert channel.open_checks == 3
1785
1786
1787async def test_dropped_message_is_logged_with_its_size_on_the_wire(
1788    cert_pems: tuple[str, str], caplog: pytest.LogCaptureFixture
1789) -> None:
1790    """The size in the drop warning counts bytes, so it can be read against the channel limit."""
1791    gateway = _proxy_gateway(cert_pems)
1792    gateway.logger = logging.getLogger("test_webrtc_dropped_message_size")
1793    channel = Mock()
1794    channel.is_open = True
1795    channel.send = AsyncMock(side_effect=RTCError("message too large"))
1796    # 100 characters, 300 bytes once encoded
1797    text = "音" * 100
1798
1799    with caplog.at_level(logging.WARNING, logger="test_webrtc_dropped_message_size"):
1800        await gateway._send_on_channel(cast("DataChannel", channel), text)
1801
1802    assert "Dropping 300-byte data channel message" in caplog.text
1803
1804
1805async def test_ma_api_channel_chunks_within_the_negotiated_limit(
1806    cert_pems: tuple[str, str],
1807) -> None:
1808    """A large API event reaches a peer that advertises no limit instead of being dropped."""
1809    cert_pem, key_pem = cert_pems
1810    fake_ws = _FakeLocalWS()
1811    http_session = Mock()
1812    http_session.ws_connect = AsyncMock(
1813        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1814    )
1815    gateway = WebRTCGateway(
1816        http_session=http_session,
1817        remote_id="TEST-REMOTE-ID",
1818        cert_pem=cert_pem,
1819        key_pem=key_pem,
1820    )
1821    # what libdatachannel assumes when the peer advertises no a=max-message-size
1822    channel = _FakeBidiChannel(max_message_size=64 * 1024)
1823    session = _register_bridge_session(gateway, "small-limit-session", channel)
1824    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1825    try:
1826        await _wait_for(lambda: session.local_ws is not None)
1827
1828        event = json.dumps({"event": "queue_updated", "data": "x" * (200 * 1024)})
1829        fake_ws.feed_text(event)
1830        await _wait_for(
1831            lambda: bool(channel.sent) and len(channel.sent) == json.loads(channel.sent[0])["count"]
1832        )
1833
1834        frames = cast("list[str]", channel.sent)
1835        assert all(len(frame.encode()) <= channel.max_message_size for frame in frames)
1836        assert _reassemble_chunks(frames) == event
1837    finally:
1838        channel.close()
1839        await asyncio.wait_for(bridge, timeout=5)
1840        await _wait_for(lambda: "small-limit-session" not in gateway.sessions)
1841