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