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 DataChannel, IceServer, LogLevel, PeerConnection, RTCConfiguration
17from cryptography.hazmat.primitives import serialization
18
19from music_assistant.constants import VERBOSE_LOG_LEVEL
20from music_assistant.controllers.webserver.remote_access import (
21    STARTUP_DELAY,
22    TASK_ID_START_GATEWAY,
23    RemoteAccessInfo,
24    RemoteAccessManager,
25)
26from music_assistant.controllers.webserver.remote_access.gateway import (
27    MA_API_CHUNK_SIZE,
28    WebRTCGateway,
29    WebRTCSession,
30    _is_usable_ice_url,
31)
32from music_assistant.helpers.webrtc_certificate import (
33    _generate_certificate,
34    _remote_id_from_certificate,
35)
36
37
38async def test_remote_id_from_certificate() -> None:
39    """Test deterministic remote ID generation from a certificate."""
40    _, cert = _generate_certificate()
41    remote_id = _remote_id_from_certificate(cert)
42
43    # Should be base32 encoded, uppercase, no padding
44    assert remote_id.isalnum()
45    assert remote_id == remote_id.upper()
46    # 128 bits = 16 bytes -> 26 base32 chars (without padding)
47    assert len(remote_id) == 26
48    # deterministic: the same certificate always yields the same id
49    assert _remote_id_from_certificate(cert) == remote_id
50
51
52async def test_remote_access_info_dataclass() -> None:
53    """Test RemoteAccessInfo dataclass."""
54    info = RemoteAccessInfo(
55        enabled=True,
56        running=True,
57        connected=False,
58        remote_id="VVPN3TLP34YMGIZDINCEKQKSIR",
59        using_ha_cloud=False,
60        signaling_url="wss://signaling.music-assistant.io/ws",
61    )
62
63    assert info.enabled is True
64    assert info.running is True
65    assert info.connected is False
66    assert info.remote_id == "VVPN3TLP34YMGIZDINCEKQKSIR"
67    assert info.using_ha_cloud is False
68    assert info.signaling_url == "wss://signaling.music-assistant.io/ws"
69
70
71def _create_remote_access_manager() -> RemoteAccessManager:
72    """Create an enabled remote access manager with mocked dependencies."""
73    webserver = Mock()
74    webserver.mass = Mock()
75    webserver.logger = Mock()
76    manager = RemoteAccessManager(webserver)
77    manager._enabled = True
78    return manager
79
80
81def test_remote_access_debounces_restart_without_dropping_gateway() -> None:
82    """Keep the active gateway connected while a replacement is being debounced."""
83    manager = _create_remote_access_manager()
84    gateway = Mock()
85    gateway.stop = AsyncMock()
86    manager.gateway = gateway
87
88    manager._schedule_start()
89
90    gateway.stop.assert_not_awaited()
91    cast("Mock", manager.mass.cancel_timer).assert_called_once_with(TASK_ID_START_GATEWAY)
92    cast("Mock", manager.mass.call_later).assert_called_once_with(
93        STARTUP_DELAY,
94        manager._restart_gateway,
95        task_id=TASK_ID_START_GATEWAY,
96    )
97
98
99async def test_remote_access_stop_cancels_pending_restart() -> None:
100    """Cancel a restart after its debounce timer has promoted it to a task."""
101    manager = _create_remote_access_manager()
102    gateway = Mock()
103    gateway.stop = AsyncMock()
104    manager.gateway = gateway
105
106    await manager.stop()
107
108    cast("Mock", manager.mass.cancel_timer).assert_called_once_with(TASK_ID_START_GATEWAY)
109    cast("Mock", manager.mass.cancel_task).assert_called_once_with(TASK_ID_START_GATEWAY)
110    gateway.stop.assert_awaited_once_with()
111    assert manager.gateway is None
112
113
114async def test_remote_access_serializes_concurrent_starts() -> None:
115    """Create only one gateway when immediate start requests overlap."""
116    manager = _create_remote_access_manager()
117    start_entered = asyncio.Event()
118    allow_start = asyncio.Event()
119    gateway = Mock()
120    gateway.is_running = True
121
122    async def start_gateway() -> None:
123        start_entered.set()
124        await allow_start.wait()
125        manager.gateway = gateway
126
127    with patch.object(
128        manager,
129        "_start_gateway_locked",
130        new=AsyncMock(side_effect=start_gateway),
131    ) as start_gateway_locked:
132        first_start = asyncio.create_task(manager._start_gateway())
133        await start_entered.wait()
134        second_start = asyncio.create_task(manager._start_gateway())
135        await asyncio.sleep(0)
136        allow_start.set()
137        await asyncio.gather(first_start, second_start)
138
139    start_gateway_locked.assert_awaited_once()
140
141
142async def test_remote_access_provider_update_storm_schedules_one_restart() -> None:
143    """Schedule one restart when concurrent provider updates detect the same mode change."""
144    manager = _create_remote_access_manager()
145    ice_servers = [{"urls": "turn:example.com"}]
146
147    with (
148        patch.object(
149            manager,
150            "_get_ha_cloud_status",
151            new=AsyncMock(return_value=(True, ice_servers)),
152        ),
153        patch.object(manager, "_schedule_start") as schedule_start,
154    ):
155        await asyncio.gather(*(manager._on_providers_updated(Mock()) for _ in range(10)))
156
157    schedule_start.assert_called_once_with()
158    assert manager._target_using_ha_cloud is True
159
160
161async def test_remote_access_skips_restart_after_mode_flap_settles() -> None:
162    """Keep the current gateway when a fresh status check matches its active mode."""
163    manager = _create_remote_access_manager()
164    gateway = Mock()
165    gateway.is_running = True
166    manager.gateway = gateway
167    manager._target_using_ha_cloud = True
168
169    with (
170        patch.object(
171            manager,
172            "_get_ha_cloud_status",
173            new=AsyncMock(return_value=(False, None)),
174        ),
175        patch.object(
176            manager,
177            "_stop_gateway_locked",
178            new=AsyncMock(),
179        ) as stop_gateway,
180        patch.object(
181            manager,
182            "_start_gateway_locked",
183            new=AsyncMock(),
184        ) as start_gateway,
185    ):
186        await manager._restart_gateway()
187
188    stop_gateway.assert_not_awaited()
189    start_gateway.assert_not_awaited()
190    assert manager._target_using_ha_cloud is False
191
192
193async def test_remote_access_gateway_uses_internal_sendspin_url(
194    cert_pems: tuple[str, str],
195) -> None:
196    """Bridge the Sendspin data channel to the locally reachable Sendspin server."""
197    internal_url = "ws://127.0.0.1:8927/sendspin"
198    manager = _create_remote_access_manager()
199    manager._remote_id = "TEST-REMOTE-ID"
200    cast("Mock", manager.webserver).internal_base_url = "http://127.0.0.1:8095"
201    cast("Mock", manager.webserver).internal_sendspin_url = internal_url
202
203    with (
204        patch(
205            "music_assistant.controllers.webserver.remote_access"
206            ".get_or_create_webrtc_certificate_pems",
207            return_value=cert_pems,
208        ),
209        patch.object(manager, "_get_ha_cloud_status", new=AsyncMock(return_value=(False, None))),
210        patch.object(WebRTCGateway, "start", new=AsyncMock()),
211    ):
212        await manager._start_gateway_locked()
213
214    assert manager.gateway is not None
215    assert manager.gateway.sendspin_url == internal_url
216
217
218@pytest.mark.parametrize(
219    ("internal_base_url", "expected"),
220    [
221        ("http://127.0.0.1:8095", "ws://127.0.0.1:8095/ws"),
222        ("https://127.0.0.1:8095", "wss://127.0.0.1:8095/ws"),
223    ],
224)
225async def test_remote_access_gateway_uses_internal_base_url(
226    cert_pems: tuple[str, str],
227    internal_base_url: str,
228    expected: str,
229) -> None:
230    """Bridge the API data channel to the locally reachable webserver."""
231    manager = _create_remote_access_manager()
232    manager._remote_id = "TEST-REMOTE-ID"
233    # an external base URL must never be dialed back into this host
234    cast("Mock", manager.mass).webserver.base_url = "https://ma.example.com"
235    cast("Mock", manager.webserver).internal_base_url = internal_base_url
236
237    with (
238        patch(
239            "music_assistant.controllers.webserver.remote_access"
240            ".get_or_create_webrtc_certificate_pems",
241            return_value=cert_pems,
242        ),
243        patch.object(manager, "_get_ha_cloud_status", new=AsyncMock(return_value=(False, None))),
244        patch.object(WebRTCGateway, "start", new=AsyncMock()),
245    ):
246        await manager._start_gateway_locked()
247
248    assert manager.gateway is not None
249    assert manager.gateway.local_ws_url == expected
250
251
252async def test_webrtc_gateway_initialization(cert_pems: tuple[str, str]) -> None:
253    """Test WebRTCGateway initializes correctly."""
254    cert_pem, key_pem = cert_pems
255    mock_session = Mock()
256    gateway = WebRTCGateway(
257        http_session=mock_session,
258        remote_id="TEST-REMOTE-ID",
259        cert_pem=cert_pem,
260        key_pem=key_pem,
261        signaling_url="wss://test.example.com/ws",
262        local_ws_url="ws://localhost:8095/ws",
263    )
264
265    assert gateway._remote_id == "TEST-REMOTE-ID"
266    assert gateway.signaling_url == "wss://test.example.com/ws"
267    assert gateway.local_ws_url == "ws://localhost:8095/ws"
268    assert gateway.is_running is False
269    assert gateway.is_connected is False
270    assert len(gateway.ice_servers) > 0
271
272
273async def test_webrtc_gateway_custom_ice_servers(cert_pems: tuple[str, str]) -> None:
274    """Test WebRTCGateway accepts custom ICE servers."""
275    cert_pem, key_pem = cert_pems
276    mock_session = Mock()
277    custom_ice_servers = [
278        {"urls": "stun:custom.stun.server:3478"},
279        {"urls": "turn:custom.turn.server:3478", "username": "user", "credential": "pass"},
280    ]
281
282    gateway = WebRTCGateway(
283        http_session=mock_session,
284        remote_id="TEST-REMOTE-ID",
285        cert_pem=cert_pem,
286        key_pem=key_pem,
287        ice_servers=custom_ice_servers,
288    )
289
290    assert gateway.ice_servers == custom_ice_servers
291
292
293async def test_webrtc_gateway_local_ice_servers_skip_non_udp_turn(
294    cert_pems: tuple[str, str],
295) -> None:
296    """Test the local peer connection only gets ICE servers libjuice can use."""
297    cert_pem, key_pem = cert_pems
298    cred = {"username": "u", "credential": "p"}
299    # shape of what HA Cloud hands us: one UDP TURN url plus TCP/TLS variants
300    ha_cloud_ice_servers = [
301        {"urls": "stun:stun.cloudflare.com:3478"},
302        {"urls": "turn:turn.cloudflare.com:3478?transport=udp", **cred},
303        {"urls": "turn:turn.cloudflare.com:53", **cred},
304        {"urls": "turn:turn.cloudflare.com:3478?transport=tcp", **cred},
305        {"urls": "turns:turn.cloudflare.com:5349?transport=tcp", **cred},
306        {"urls": "turns:turn.cloudflare.com:443?transport=tcp", **cred},
307    ]
308
309    gateway = WebRTCGateway(
310        http_session=Mock(),
311        remote_id="TEST-REMOTE-ID",
312        cert_pem=cert_pem,
313        key_pem=key_pem,
314        ice_servers=ha_cloud_ice_servers,
315    )
316
317    assert [server.url for server in gateway._build_ice_servers(ha_cloud_ice_servers)] == [
318        "stun:stun.cloudflare.com:3478",
319        "turn:turn.cloudflare.com:3478?transport=udp",
320        "turn:turn.cloudflare.com:53",
321    ]
322    # remote clients still get the unfiltered list, since browsers do support TCP/TLS TURN
323    assert gateway.ice_servers == ha_cloud_ice_servers
324
325
326@pytest.mark.parametrize(
327    ("url", "usable"),
328    [
329        ("stun:stun.example.com:3478", True),
330        ("turn:turn.example.com:3478", True),
331        ("turn:turn.example.com:3478?transport=udp", True),
332        # the transport parameter wins over the scheme, matching rtc::IceServer
333        ("turns:turn.example.com:5349?transport=udp", True),
334        ("turn:turn.example.com:3478?transport=tcp", False),
335        ("turns:turn.example.com:5349", False),
336        ("turn:turn.example.com:3478?transport=tls", False),
337        ("https://turn.example.com", False),
338    ],
339)
340def test_is_usable_ice_url(url: str, usable: bool) -> None:
341    """Test only ICE server urls libjuice can actually use are kept."""
342    assert _is_usable_ice_url(url) is usable
343
344
345async def test_webrtc_gateway_start_stop(cert_pems: tuple[str, str]) -> None:
346    """Test WebRTCGateway start and stop."""
347    cert_pem, key_pem = cert_pems
348    mock_session = Mock()
349    gateway = WebRTCGateway(
350        http_session=mock_session,
351        remote_id="TEST-REMOTE-ID",
352        cert_pem=cert_pem,
353        key_pem=key_pem,
354    )
355
356    # Mock the _run method to avoid actual connection
357    with patch.object(gateway, "_run", new_callable=AsyncMock):
358        await gateway.start()
359        assert gateway.is_running is True
360        assert gateway._run_task is not None
361
362        await gateway.stop()
363        assert gateway.is_running is False
364
365
366@pytest.mark.parametrize(
367    ("logger_level", "expected_rtc_level"),
368    [
369        (VERBOSE_LOG_LEVEL, LogLevel.VERBOSE),
370        (logging.DEBUG, LogLevel.WARNING),
371        (logging.INFO, LogLevel.ERROR),
372        (logging.WARNING, LogLevel.ERROR),
373    ],
374)
375async def test_webrtc_gateway_native_log_level(
376    cert_pems: tuple[str, str], logger_level: int, expected_rtc_level: LogLevel
377) -> None:
378    """Test native libdatachannel logging is capped at ERROR unless debugging."""
379    cert_pem, key_pem = cert_pems
380    gateway = WebRTCGateway(
381        http_session=Mock(),
382        remote_id="TEST-REMOTE-ID",
383        cert_pem=cert_pem,
384        key_pem=key_pem,
385    )
386    gateway.logger = logging.getLogger("test_webrtc_native_log_level")
387    gateway.logger.setLevel(logger_level)
388
389    with (
390        patch.object(gateway, "_run", new_callable=AsyncMock),
391        patch(
392            "music_assistant.controllers.webserver.remote_access.gateway.install_python_logger"
393        ) as install_logger,
394    ):
395        await gateway.start()
396        await gateway.stop()
397
398    install_logger.assert_called_once_with(gateway.logger, level=expected_rtc_level)
399    assert not gateway.logger.filters
400
401
402async def test_webrtc_gateway_drops_benign_turn_warning_at_debug(
403    cert_pems: tuple[str, str], caplog: pytest.LogCaptureFixture
404) -> None:
405    """Test the benign Cloudflare CreatePermission warning is dropped at DEBUG level."""
406    cert_pem, key_pem = cert_pems
407    gateway = WebRTCGateway(
408        http_session=Mock(),
409        remote_id="TEST-REMOTE-ID",
410        cert_pem=cert_pem,
411        key_pem=key_pem,
412    )
413    gateway.logger = logging.getLogger("test_webrtc_benign_turn_warning")
414    gateway.logger.setLevel(logging.DEBUG)
415
416    with (
417        patch.object(gateway, "_run", new_callable=AsyncMock),
418        patch("music_assistant.controllers.webserver.remote_access.gateway.install_python_logger"),
419    ):
420        await gateway.start()
421        with caplog.at_level(logging.DEBUG, logger="test_webrtc_benign_turn_warning"):
422            gateway.logger.warning(
423                "rtc::impl::IceTransport::LogCallback@390: "
424                "juice: Got TURN CreatePermission error response, code=0"
425            )
426            gateway.logger.warning("juice: Lost connectivity")
427        await gateway.stop()
428
429    messages = [record.getMessage() for record in caplog.records if "juice" in record.getMessage()]
430    assert messages == ["juice: Lost connectivity"]
431    # stop() must remove the filter again
432    assert not gateway.logger.filters
433
434
435async def test_webrtc_gateway_handle_registration_message(cert_pems: tuple[str, str]) -> None:
436    """Test WebRTCGateway handles registration confirmation."""
437    cert_pem, key_pem = cert_pems
438    mock_session = Mock()
439    gateway = WebRTCGateway(
440        http_session=mock_session,
441        remote_id="TEST-REMOTE-ID",
442        cert_pem=cert_pem,
443        key_pem=key_pem,
444    )
445
446    # Mock signaling WebSocket
447    gateway._signaling_ws = Mock()
448
449    message = {"type": "registered", "remoteId": "TEST-REMOTE-ID"}
450    await gateway._handle_signaling_message(message)
451
452    # Should log but not crash
453
454
455async def test_webrtc_gateway_handle_error_message(cert_pems: tuple[str, str]) -> None:
456    """Test WebRTCGateway handles error messages."""
457    cert_pem, key_pem = cert_pems
458    mock_session = Mock()
459    gateway = WebRTCGateway(
460        http_session=mock_session,
461        remote_id="TEST-REMOTE-ID",
462        cert_pem=cert_pem,
463        key_pem=key_pem,
464    )
465
466    message = {"type": "error", "message": "Test error"}
467    # Should log error but not crash
468    await gateway._handle_signaling_message(message)
469
470
471async def test_webrtc_gateway_create_session(cert_pems: tuple[str, str]) -> None:
472    """Test WebRTCGateway creates sessions for clients."""
473    cert_pem, key_pem = cert_pems
474    mock_session = Mock()
475    gateway = WebRTCGateway(
476        http_session=mock_session,
477        remote_id="TEST-REMOTE-ID",
478        cert_pem=cert_pem,
479        key_pem=key_pem,
480    )
481
482    session_id = "test-session-123"
483    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
484        await gateway._create_session(session_id)
485
486        assert session_id in gateway.sessions
487        assert gateway.sessions[session_id].session_id == session_id
488        assert gateway.sessions[session_id].pc is not None
489
490        # Cleanup
491        await gateway._close_session(session_id)
492
493
494async def test_webrtc_gateway_close_session(cert_pems: tuple[str, str]) -> None:
495    """Test WebRTCGateway closes sessions properly."""
496    cert_pem, key_pem = cert_pems
497    mock_session = Mock()
498    gateway = WebRTCGateway(
499        http_session=mock_session,
500        remote_id="TEST-REMOTE-ID",
501        cert_pem=cert_pem,
502        key_pem=key_pem,
503    )
504
505    session_id = "test-session-456"
506    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
507        await gateway._create_session(session_id)
508        assert session_id in gateway.sessions
509
510        await gateway._close_session(session_id)
511        assert session_id not in gateway.sessions
512
513
514async def test_webrtc_gateway_close_nonexistent_session(cert_pems: tuple[str, str]) -> None:
515    """Test WebRTCGateway handles closing non-existent session gracefully."""
516    cert_pem, key_pem = cert_pems
517    mock_session = Mock()
518    gateway = WebRTCGateway(
519        http_session=mock_session,
520        remote_id="TEST-REMOTE-ID",
521        cert_pem=cert_pem,
522        key_pem=key_pem,
523    )
524
525    # Should not raise an error
526    await gateway._close_session("nonexistent-session")
527
528
529async def test_webrtc_gateway_default_ice_servers(cert_pems: tuple[str, str]) -> None:
530    """Test WebRTCGateway uses default ICE servers."""
531    cert_pem, key_pem = cert_pems
532    mock_session = Mock()
533    gateway = WebRTCGateway(
534        http_session=mock_session,
535        remote_id="TEST-REMOTE-ID",
536        cert_pem=cert_pem,
537        key_pem=key_pem,
538    )
539
540    assert len(gateway.ice_servers) > 0
541    # Should have at least one STUN server
542    assert any("stun:" in server["urls"] for server in gateway.ice_servers)
543
544
545async def test_webrtc_gateway_handle_client_connected(cert_pems: tuple[str, str]) -> None:
546    """Test WebRTCGateway handles client-connected message."""
547    cert_pem, key_pem = cert_pems
548    mock_session = Mock()
549    gateway = WebRTCGateway(
550        http_session=mock_session,
551        remote_id="TEST-REMOTE-ID",
552        cert_pem=cert_pem,
553        key_pem=key_pem,
554    )
555
556    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
557        message = {"type": "client-connected", "sessionId": "test-session"}
558        await gateway._handle_signaling_message(message)
559
560        # Session should be created
561        assert "test-session" in gateway.sessions
562
563        # Cleanup
564        await gateway._close_session("test-session")
565
566
567async def test_webrtc_gateway_handle_client_disconnected(cert_pems: tuple[str, str]) -> None:
568    """Test WebRTCGateway handles client-disconnected message."""
569    cert_pem, key_pem = cert_pems
570    mock_session = Mock()
571    gateway = WebRTCGateway(
572        http_session=mock_session,
573        remote_id="TEST-REMOTE-ID",
574        cert_pem=cert_pem,
575        key_pem=key_pem,
576    )
577
578    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
579        # Create a session first
580        session_id = "test-disconnect-session"
581        await gateway._create_session(session_id)
582        assert session_id in gateway.sessions
583
584        # Handle disconnect
585        message = {"type": "client-disconnected", "sessionId": session_id}
586        await gateway._handle_signaling_message(message)
587
588        # Session should be removed
589        assert session_id not in gateway.sessions
590
591
592async def test_webrtc_gateway_reconnection_logic(cert_pems: tuple[str, str]) -> None:
593    """Test WebRTCGateway has proper reconnection backoff."""
594    cert_pem, key_pem = cert_pems
595    mock_session = Mock()
596    gateway = WebRTCGateway(
597        http_session=mock_session,
598        remote_id="TEST-REMOTE-ID",
599        cert_pem=cert_pem,
600        key_pem=key_pem,
601    )
602
603    # Check initial reconnect delay
604    assert gateway._current_reconnect_delay == 10
605
606    # Simulate multiple failed connections (without actually connecting)
607    initial_delay = gateway._current_reconnect_delay
608    gateway._current_reconnect_delay = min(
609        gateway._current_reconnect_delay * 2, gateway._max_reconnect_delay
610    )
611
612    assert gateway._current_reconnect_delay == initial_delay * 2
613
614    # Should not exceed max
615    for _ in range(10):
616        gateway._current_reconnect_delay = min(
617            gateway._current_reconnect_delay * 2, gateway._max_reconnect_delay
618        )
619
620    assert gateway._current_reconnect_delay <= gateway._max_reconnect_delay
621
622
623async def test_webrtc_gateway_handle_offer_without_session(cert_pems: tuple[str, str]) -> None:
624    """Test WebRTCGateway handles offer for non-existent session gracefully."""
625    cert_pem, key_pem = cert_pems
626    mock_session = Mock()
627    gateway = WebRTCGateway(
628        http_session=mock_session,
629        remote_id="TEST-REMOTE-ID",
630        cert_pem=cert_pem,
631        key_pem=key_pem,
632    )
633
634    # Try to handle offer for non-existent session
635    offer_data = {"sdp": "test-sdp", "type": "offer"}
636    await gateway._handle_offer("nonexistent-session", offer_data)
637
638    # Should not crash
639
640
641async def test_webrtc_gateway_handle_ice_candidate_without_session(
642    cert_pems: tuple[str, str],
643) -> None:
644    """Test WebRTCGateway handles ICE candidate for non-existent session gracefully."""
645    cert_pem, key_pem = cert_pems
646    mock_session = Mock()
647    gateway = WebRTCGateway(
648        http_session=mock_session,
649        remote_id="TEST-REMOTE-ID",
650        cert_pem=cert_pem,
651        key_pem=key_pem,
652    )
653
654    # Try to handle ICE candidate for non-existent session
655    candidate_data = {
656        "candidate": "candidate:1 1 UDP 1234 192.168.1.1 12345 typ host",
657        "sdpMid": "0",
658        "sdpMLineIndex": 0,
659    }
660    await gateway._handle_ice_candidate("nonexistent-session", candidate_data)
661
662    # Should not crash
663
664
665@pytest.mark.parametrize(
666    "malicious_path",
667    [
668        "@evil.com",  # netloc becomes basic-auth creds, evil.com becomes the host
669        "//evil.com",  # protocol-relative URL pointing at another host
670        "@evil.com/foo",
671        "//evil.com/foo",
672    ],
673)
674async def test_http_proxy_request_cannot_change_host(
675    cert_pems: tuple[str, str], malicious_path: str
676) -> None:
677    """An attacker-controlled proxy path must never change the target host (SSRF guard)."""
678    cert_pem, key_pem = cert_pems
679    mock_session = Mock()
680    captured_url: dict[str, str] = {}
681
682    def fake_request(_method: str, url: str, **_kwargs: object) -> AsyncMock:
683        captured_url["url"] = url
684        response = AsyncMock()
685        response.status = 200
686        response.headers = {}
687        response.read = AsyncMock(return_value=b"")
688        ctx = AsyncMock()
689        ctx.__aenter__ = AsyncMock(return_value=response)
690        ctx.__aexit__ = AsyncMock(return_value=False)
691        return ctx
692
693    mock_session.request = fake_request
694
695    gateway = WebRTCGateway(
696        http_session=mock_session,
697        remote_id="TEST-REMOTE-ID",
698        cert_pem=cert_pem,
699        key_pem=key_pem,
700        local_ws_url="ws://localhost:8095/ws",
701    )
702    session = WebRTCSession(session_id="s1", pc=Mock())
703
704    await gateway._handle_http_proxy_request(
705        session, {"id": "1", "method": "GET", "path": malicious_path}
706    )
707
708    parsed = urlparse(captured_url["url"])
709    assert parsed.hostname == "localhost"
710    assert parsed.port == 8095
711    assert parsed.username is None
712    assert "evil.com" not in (parsed.netloc or "")
713
714
715async def test_http_proxy_request_keeps_the_unverified_dial_on_this_host(
716    cert_pems: tuple[str, str],
717) -> None:
718    """The proxy must not carry its unverified TLS dial off this host."""
719    cert_pem, key_pem = cert_pems
720    mock_session = Mock()
721    captured_kwargs: dict[str, object] = {}
722
723    def fake_request(_method: str, _url: str, **kwargs: object) -> AsyncMock:
724        captured_kwargs.update(kwargs)
725        response = AsyncMock()
726        response.status = 200
727        response.headers = {}
728        response.read = AsyncMock(return_value=b"")
729        ctx = AsyncMock()
730        ctx.__aenter__ = AsyncMock(return_value=response)
731        ctx.__aexit__ = AsyncMock(return_value=False)
732        return ctx
733
734    mock_session.request = fake_request
735
736    gateway = WebRTCGateway(
737        http_session=mock_session,
738        remote_id="TEST-REMOTE-ID",
739        cert_pem=cert_pem,
740        key_pem=key_pem,
741        local_ws_url="wss://127.0.0.1:8095/ws",
742    )
743    session = WebRTCSession(session_id="s1", pc=Mock())
744
745    await gateway._handle_http_proxy_request(session, {"id": "1", "method": "GET", "path": "/info"})
746
747    assert captured_kwargs["ssl"] is False
748    assert captured_kwargs["allow_redirects"] is False
749
750
751async def test_local_websocket_dial_skips_certificate_verification(
752    cert_pems: tuple[str, str],
753) -> None:
754    """Reach the local API on the bind address, which no certificate is issued for."""
755    cert_pem, key_pem = cert_pems
756    mock_session = Mock()
757    captured_kwargs: dict[str, object] = {}
758
759    async def fake_ws_connect(_url: str, **kwargs: object) -> AsyncMock:
760        captured_kwargs.update(kwargs)
761        return AsyncMock()
762
763    mock_session.ws_connect = fake_ws_connect
764
765    gateway = WebRTCGateway(
766        http_session=mock_session,
767        remote_id="TEST-REMOTE-ID",
768        cert_pem=cert_pem,
769        key_pem=key_pem,
770        local_ws_url="wss://127.0.0.1:8095/ws",
771    )
772    session = WebRTCSession(session_id="s1", pc=Mock())
773    channel = MagicMock()
774    channel.wait_open = AsyncMock()
775
776    with patch.object(gateway, "_schedule_close"):
777        await gateway._bridge_ma_api(session, channel)
778
779    assert captured_kwargs["ssl"] is False
780
781
782# ---- aiolibdatachannel loopback tests --------------------------------------
783#
784# These exercise the migrated gateway against real loopback PeerConnections
785# (offerer = browser role, answerer = gateway role) rather than mocking the
786# WebRTC layer. ICE servers are stubbed to [] so gathering completes on host
787# candidates only (fast, offline).
788
789
790@pytest.fixture(scope="session")
791def cert_pems() -> tuple[str, str]:
792    """Generate a throwaway DTLS cert/key as PEM strings for the gateway."""
793    private_key, cert = _generate_certificate()
794    cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode()
795    key_pem = private_key.private_bytes(
796        encoding=serialization.Encoding.PEM,
797        format=serialization.PrivateFormat.PKCS8,
798        encryption_algorithm=serialization.NoEncryption(),
799    ).decode()
800    return cert_pem, key_pem
801
802
803def _sha256_fingerprint(cert_pem: str) -> str:
804    """Compute the uppercase colon-separated SHA-256 fingerprint of a PEM certificate."""
805    body = "".join(line for line in cert_pem.splitlines() if line and not line.startswith("-----"))
806    der = base64.b64decode(body)
807    digest = hashlib.sha256(der).hexdigest().upper()
808    return ":".join(digest[i : i + 2] for i in range(0, len(digest), 2))
809
810
811class _FakeSignaling:
812    """Signaling WebSocket stand-in that captures outbound JSON messages."""
813
814    def __init__(self) -> None:
815        self.messages: list[dict[str, Any]] = []
816
817    async def send_json(self, data: dict[str, Any]) -> None:
818        self.messages.append(data)
819
820    @property
821    def answers(self) -> list[dict[str, Any]]:
822        return [m for m in self.messages if m.get("type") == "answer"]
823
824
825class _FakeLocalWS:
826    """Minimal aiohttp WebSocket stand-in for channel-bridging tests."""
827
828    def __init__(self) -> None:
829        self.closed = False
830        self.sent: list[str | bytes] = []
831        self._incoming: asyncio.Queue[SimpleNamespace | None] = asyncio.Queue()
832
833    async def send_str(self, data: str) -> None:
834        self.sent.append(data)
835
836    async def send_bytes(self, data: bytes) -> None:
837        self.sent.append(data)
838
839    async def close(self) -> None:
840        self.closed = True
841        self._incoming.put_nowait(None)
842
843    def feed_text(self, data: str) -> None:
844        """Queue a text message as if the local server sent it."""
845        self._incoming.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=data))
846
847    def feed_bytes(self, data: bytes) -> None:
848        """Queue a binary message as if the local server sent it."""
849        self._incoming.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.BINARY, data=data))
850
851    def __aiter__(self) -> AsyncIterator[SimpleNamespace]:
852        return self
853
854    async def __anext__(self) -> SimpleNamespace:
855        msg = await self._incoming.get()
856        if msg is None:
857            raise StopAsyncIteration
858        return msg
859
860
861class _FakeBidiChannel:
862    """
863    Async-iterable data-channel stand-in for bridging tests.
864
865    Drives the gateway's channel->local pump without a real WebRTC handshake: feed
866    inbound messages with :meth:`feed`, end the stream with :meth:`close`, and read
867    what the gateway sent back on ``sent``.
868    """
869
870    def __init__(self, label: str = "ma-api") -> None:
871        self.label = label
872        self.is_open = True
873        self.closed = False
874        self.sent: list[str | bytes] = []
875        self._inbound: asyncio.Queue[str | bytes | None] = asyncio.Queue()
876
877    async def wait_open(self) -> None:
878        return
879
880    async def send(self, data: str | bytes) -> None:
881        self.sent.append(data)
882
883    def feed(self, message: str | bytes) -> None:
884        """Queue an inbound message as if the browser sent it over the channel."""
885        self._inbound.put_nowait(message)
886
887    def close(self) -> None:
888        self.closed = True
889        self.is_open = False
890        self._inbound.put_nowait(None)
891
892    async def aclose(self) -> None:
893        self.close()
894
895    def __aiter__(self) -> AsyncIterator[str | bytes]:
896        return self
897
898    async def __anext__(self) -> str | bytes:
899        message = await self._inbound.get()
900        if message is None:
901            raise StopAsyncIteration
902        return message
903
904
905class _FakeHttpSession:
906    """ClientSession stand-in handing out one fake WebSocket per dialed url."""
907
908    def __init__(self) -> None:
909        self.dialed: list[str] = []
910        self.dial_kwargs: list[dict[str, Any]] = []
911        self.websockets: dict[str, _FakeLocalWS] = {}
912
913    async def ws_connect(self, url: str, **kwargs: Any) -> _FakeLocalWS:
914        self.dialed.append(url)
915        self.dial_kwargs.append(kwargs)
916        local_ws = _FakeLocalWS()
917        self.websockets[url] = local_ws
918        return local_ws
919
920
921class _FakePeerConnection:
922    """PeerConnection stand-in that runs gateway-spawned pumps as asyncio tasks."""
923
924    def __init__(self) -> None:
925        self._tasks: list[asyncio.Task[None]] = []
926        self._incoming: asyncio.Queue[_FakeBidiChannel] = asyncio.Queue()
927
928    def spawn_task(self, coro: Coroutine[Any, Any, None]) -> None:
929        self._tasks.append(asyncio.ensure_future(coro))
930
931    def offer_channel(self, channel: _FakeBidiChannel) -> None:
932        """Offer a data channel as if the browser had opened it."""
933        self._incoming.put_nowait(channel)
934
935    async def incoming_data_channels(self) -> AsyncIterator[DataChannel]:
936        while True:
937            yield cast("DataChannel", await self._incoming.get())
938
939    async def aclose(self) -> None:
940        for task in self._tasks:
941            task.cancel()
942        # await cancellation so no pump task lingers past teardown
943        await asyncio.gather(*self._tasks, return_exceptions=True)
944
945
946def _register_bridge_session(
947    gateway: WebRTCGateway, session_id: str, channel: _FakeBidiChannel
948) -> WebRTCSession:
949    """Register a session backed by a fake PeerConnection and ma-api channel."""
950    session = WebRTCSession(
951        session_id=session_id,
952        pc=cast("PeerConnection", _FakePeerConnection()),
953        data_channel=cast("DataChannel", channel),
954    )
955    gateway.sessions[session_id] = session
956    return session
957
958
959def _register_routed_session(
960    gateway: WebRTCGateway, session_id: str
961) -> tuple[WebRTCSession, _FakePeerConnection]:
962    """Register a session that routes the data channels offered to its PeerConnection."""
963    pc = _FakePeerConnection()
964    session = WebRTCSession(session_id=session_id, pc=cast("PeerConnection", pc))
965    gateway.sessions[session_id] = session
966    pc.spawn_task(gateway._accept_channels(session))
967    return session, pc
968
969
970async def _wait_for(predicate: Callable[[], bool], timeout: float = 15.0) -> None:
971    """Poll ``predicate`` until it is true or the timeout elapses."""
972    loop = asyncio.get_running_loop()
973    deadline = loop.time() + timeout
974    while loop.time() < deadline:
975        if predicate():
976            return
977        await asyncio.sleep(0.02)
978    raise AssertionError("condition not met within timeout")
979
980
981async def test_handle_offer_answers_with_pinned_fingerprint(cert_pems: tuple[str, str]) -> None:
982    """The answer SDP carries the DTLS fingerprint of the configured certificate (pinning)."""
983    cert_pem, key_pem = cert_pems
984    gateway = WebRTCGateway(
985        http_session=Mock(),
986        remote_id="TEST-REMOTE-ID",
987        cert_pem=cert_pem,
988        key_pem=key_pem,
989    )
990    signaling = _FakeSignaling()
991    gateway._signaling_ws = cast("aiohttp.ClientWebSocketResponse", signaling)
992    offerer = PeerConnection(RTCConfiguration())
993    session_id = "fingerprint-session"
994    try:
995        await offerer.create_data_channel("ma-api")
996        offer = await offerer.create_offer()
997        with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
998            await gateway._create_session(session_id)
999            await asyncio.wait_for(
1000                gateway._handle_offer(session_id, {"sdp": offer.sdp, "type": "offer"}),
1001                timeout=15,
1002            )
1003
1004        assert len(signaling.answers) == 1
1005        answer = signaling.answers[0]
1006        assert answer["sessionId"] == session_id
1007        assert answer["data"]["type"] == "answer"
1008        fingerprint_line = next(
1009            line for line in answer["data"]["sdp"].splitlines() if line.startswith("a=fingerprint:")
1010        )
1011        assert _sha256_fingerprint(cert_pem) in fingerprint_line
1012    finally:
1013        await gateway._close_session(session_id)
1014        await offerer.aclose()
1015
1016
1017async def test_ma_api_channel_bridges_to_local_ws(cert_pems: tuple[str, str]) -> None:
1018    """Messages flow both ways across the ma-api data channel and the local WebSocket."""
1019    cert_pem, key_pem = cert_pems
1020    fake_ws = _FakeLocalWS()
1021    http_session = Mock()
1022    http_session.ws_connect = AsyncMock(
1023        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1024    )
1025    gateway = WebRTCGateway(
1026        http_session=http_session,
1027        remote_id="TEST-REMOTE-ID",
1028        cert_pem=cert_pem,
1029        key_pem=key_pem,
1030    )
1031    channel = _FakeBidiChannel()
1032    session = _register_bridge_session(gateway, "bridge-session", channel)
1033    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1034    try:
1035        await _wait_for(lambda: session.local_ws is not None)
1036
1037        # browser -> local WebSocket
1038        channel.feed("from browser")
1039        await _wait_for(lambda: fake_ws.sent == ["from browser"])
1040
1041        # local WebSocket -> browser
1042        fake_ws.feed_text("from local")
1043        await _wait_for(lambda: channel.sent == ["from local"])
1044    finally:
1045        channel.close()
1046        await asyncio.wait_for(bridge, timeout=5)
1047        await _wait_for(lambda: "bridge-session" not in gateway.sessions)
1048
1049
1050def test_build_ice_servers_maps_dicts() -> None:
1051    """ICE server dicts map to one IceServer per url, preserving TURN credentials."""
1052    gateway = WebRTCGateway(
1053        http_session=Mock(),
1054        remote_id="TEST-REMOTE-ID",
1055        cert_pem="cert",
1056        key_pem="key",
1057    )
1058    servers: list[dict[str, Any]] = [
1059        {"urls": "stun:stun.example.com:3478"},
1060        {"urls": "turn:turn.example.com:3478", "username": "user", "credential": "pass"},
1061        {
1062            "urls": ["stun:a.example.com:3478", "turn:b.example.com:3478"],
1063            "username": "u2",
1064            "credential": "c2",
1065        },
1066    ]
1067
1068    result = gateway._build_ice_servers(servers)
1069
1070    assert all(isinstance(server, IceServer) for server in result)
1071    # the two-url entry fans out, so 1 + 1 + 2 = 4 IceServers
1072    assert len(result) == 4
1073    assert result[0] == IceServer(url="stun:stun.example.com:3478")
1074    assert result[1] == IceServer(
1075        url="turn:turn.example.com:3478", username="user", credential="pass"
1076    )
1077    # to_url() inlines the TURN credentials for libdatachannel
1078    assert result[1].to_url() == "turn:user:[email protected]:3478"
1079    # list urls share the entry's credentials
1080    assert result[2] == IceServer(url="stun:a.example.com:3478", username="u2", credential="c2")
1081    assert result[3] == IceServer(url="turn:b.example.com:3478", username="u2", credential="c2")
1082
1083
1084async def test_session_closes_when_ma_api_channel_closes(cert_pems: tuple[str, str]) -> None:
1085    """Closing the browser ma-api channel tears down the whole gateway session."""
1086    cert_pem, key_pem = cert_pems
1087    fake_ws = _FakeLocalWS()
1088    http_session = Mock()
1089    http_session.ws_connect = AsyncMock(
1090        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1091    )
1092    gateway = WebRTCGateway(
1093        http_session=http_session,
1094        remote_id="TEST-REMOTE-ID",
1095        cert_pem=cert_pem,
1096        key_pem=key_pem,
1097    )
1098    channel = _FakeBidiChannel()
1099    session = _register_bridge_session(gateway, "channel-close-session", channel)
1100    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1101    await _wait_for(lambda: session.local_ws is not None)
1102
1103    # the browser closes the ma-api channel -> the whole session is torn down
1104    channel.close()
1105
1106    await _wait_for(lambda: "channel-close-session" not in gateway.sessions)
1107    assert "channel-close-session" not in gateway.sessions
1108    await asyncio.wait_for(bridge, timeout=5)
1109
1110
1111# ---- channel routing -------------------------------------------------------
1112
1113LOCAL_WS_URL = "ws://127.0.0.1:8095/ws"
1114SENDSPIN_URL = "ws://127.0.0.1:8927/sendspin"
1115
1116
1117def _routing_gateway(
1118    cert_pems: tuple[str, str],
1119    http_session: _FakeHttpSession,
1120    local_ws_url: str = LOCAL_WS_URL,
1121    set_sendspin_player_callback: Callable[[str, str], None] | None = None,
1122) -> WebRTCGateway:
1123    """Create a gateway whose local WebSockets are all served by the fake HTTP session."""
1124    cert_pem, key_pem = cert_pems
1125    return WebRTCGateway(
1126        http_session=cast("aiohttp.ClientSession", http_session),
1127        remote_id="TEST-REMOTE-ID",
1128        cert_pem=cert_pem,
1129        key_pem=key_pem,
1130        local_ws_url=local_ws_url,
1131        sendspin_url=SENDSPIN_URL,
1132        set_sendspin_player_callback=set_sendspin_player_callback,
1133    )
1134
1135
1136async def test_sendspin_channel_bridges_to_the_sendspin_server(cert_pems: tuple[str, str]) -> None:
1137    """A sendspin channel reaches the internal sendspin server, web player id and all."""
1138    http_session = _FakeHttpSession()
1139    announced_players: list[tuple[str, str]] = []
1140    gateway = _routing_gateway(
1141        cert_pems,
1142        http_session,
1143        set_sendspin_player_callback=lambda session_id, player_id: announced_players.append(
1144            (session_id, player_id)
1145        ),
1146    )
1147    session, pc = _register_routed_session(gateway, "sendspin-session")
1148    channel = _FakeBidiChannel(label="sendspin")
1149    pc.offer_channel(channel)
1150    try:
1151        await _wait_for(lambda: SENDSPIN_URL in http_session.websockets)
1152        local_ws = http_session.websockets[SENDSPIN_URL]
1153
1154        # the first message announces the web player, and is forwarded verbatim
1155        auth = json.dumps({"type": "auth", "token": "t", "client_id": "web-player-1"})
1156        channel.feed(auth)
1157        await _wait_for(lambda: local_ws.sent == [auth])
1158        assert announced_players == [("sendspin-session", "web-player-1")]
1159        assert session.sendspin_player_id == "web-player-1"
1160
1161        # audio keeps flowing in both directions, text and binary alike
1162        channel.feed(b"\x01\x02")
1163        await _wait_for(lambda: local_ws.sent == [auth, b"\x01\x02"])
1164        local_ws.feed_text('{"type":"hello"}')
1165        local_ws.feed_bytes(b"\x03\x04")
1166        await _wait_for(lambda: channel.sent == ['{"type":"hello"}', b"\x03\x04"])
1167    finally:
1168        await gateway._close_session("sendspin-session")
1169
1170
1171@pytest.mark.parametrize(
1172    ("local_ws_url", "expected_url"),
1173    [
1174        ("ws://127.0.0.1:8095/ws", "ws://127.0.0.1:8095/live_announcement"),
1175        # an https webserver is still dialed on its bind address, which no cert covers
1176        ("wss://127.0.0.1:8095/ws", "wss://127.0.0.1:8095/live_announcement"),
1177    ],
1178)
1179async def test_live_announcement_channel_bridges_to_the_webserver(
1180    cert_pems: tuple[str, str], local_ws_url: str, expected_url: str
1181) -> None:
1182    """A live announcement channel reaches the webserver route that takes the audio."""
1183    http_session = _FakeHttpSession()
1184    gateway = _routing_gateway(cert_pems, http_session, local_ws_url=local_ws_url)
1185    session, pc = _register_routed_session(gateway, "announce-session")
1186    channel = _FakeBidiChannel(label="live_announcement")
1187    pc.offer_channel(channel)
1188    try:
1189        await _wait_for(lambda: expected_url in http_session.websockets)
1190        local_ws = http_session.websockets[expected_url]
1191        assert http_session.dial_kwargs == [{"ssl": False}]
1192
1193        # the client authenticates on the route itself, so its handshake passes through
1194        handshake = [
1195            json.dumps({"type": "auth", "token": "t"}),
1196            json.dumps({"type": "start", "player_id": "player1", "sample_rate": 16000}),
1197        ]
1198        for message in handshake:
1199            channel.feed(message)
1200        await _wait_for(lambda: local_ws.sent == handshake)
1201        # the sendspin snoop belongs to the sendspin bridge only
1202        assert session.sendspin_player_id is None
1203
1204        # spoken audio goes up, the route's replies come back
1205        channel.feed(b"\x00\x01")
1206        await _wait_for(lambda: local_ws.sent == [*handshake, b"\x00\x01"])
1207        local_ws.feed_text('{"type":"started"}')
1208        await _wait_for(lambda: channel.sent == ['{"type":"started"}'])
1209    finally:
1210        await gateway._close_session("announce-session")
1211
1212
1213@pytest.mark.parametrize("closed_by", ["browser", "local"])
1214async def test_closing_a_bridged_channel_leaves_the_api_session_up(
1215    cert_pems: tuple[str, str], closed_by: str
1216) -> None:
1217    """Losing a bridged WebSocket tears down that bridge only, never the API session."""
1218    http_session = _FakeHttpSession()
1219    gateway = _routing_gateway(cert_pems, http_session)
1220    session, pc = _register_routed_session(gateway, "mixed-session")
1221    api_channel = _FakeBidiChannel()
1222    sendspin_channel = _FakeBidiChannel(label="sendspin")
1223    pc.offer_channel(api_channel)
1224    pc.offer_channel(sendspin_channel)
1225    try:
1226        await _wait_for(
1227            lambda: session.local_ws is not None and SENDSPIN_URL in http_session.dialed
1228        )
1229        sendspin_ws = http_session.websockets[SENDSPIN_URL]
1230
1231        if closed_by == "browser":
1232            sendspin_channel.close()
1233        else:
1234            await sendspin_ws.close()
1235        await _wait_for(lambda: not session.ws_bridges)
1236
1237        assert sendspin_ws.closed is True
1238        assert sendspin_channel.closed is True
1239        # the API session is untouched: still registered, still bridged, channel still open
1240        assert "mixed-session" in gateway.sessions
1241        assert session.local_ws is not None
1242        assert cast("_FakeLocalWS", session.local_ws).closed is False
1243        assert api_channel.closed is False
1244    finally:
1245        await gateway._close_session("mixed-session")
1246
1247
1248async def test_the_first_channel_is_the_api_channel_whatever_its_label(
1249    cert_pems: tuple[str, str],
1250) -> None:
1251    """Clients may label their API channel freely, so the first channel bridges to the API."""
1252    http_session = _FakeHttpSession()
1253    gateway = _routing_gateway(cert_pems, http_session)
1254    session, pc = _register_routed_session(gateway, "labelled-session")
1255    channel = _FakeBidiChannel(label="ma-api-v2")
1256    pc.offer_channel(channel)
1257    try:
1258        await _wait_for(lambda: session.local_ws is not None)
1259        assert session.data_channel is cast("DataChannel", channel)
1260        assert http_session.dialed == [f"{LOCAL_WS_URL}?webrtc_session_id=labelled-session"]
1261    finally:
1262        await gateway._close_session("labelled-session")
1263
1264
1265async def test_unknown_channel_label_cannot_replace_the_api_channel(
1266    cert_pems: tuple[str, str],
1267) -> None:
1268    """A channel this server has no route for is refused instead of hijacking the session."""
1269    http_session = _FakeHttpSession()
1270    gateway = _routing_gateway(cert_pems, http_session)
1271    session, pc = _register_routed_session(gateway, "unknown-session")
1272    api_channel = _FakeBidiChannel()
1273    unknown_channel = _FakeBidiChannel(label="channel-from-the-future")
1274    pc.offer_channel(api_channel)
1275    try:
1276        await _wait_for(lambda: session.local_ws is not None)
1277        pc.offer_channel(unknown_channel)
1278        await _wait_for(lambda: unknown_channel.closed)
1279
1280        assert session.data_channel is cast("DataChannel", api_channel)
1281        assert "unknown-session" in gateway.sessions
1282        # only the API channel was ever bridged
1283        assert http_session.dialed == [f"{LOCAL_WS_URL}?webrtc_session_id=unknown-session"]
1284    finally:
1285        await gateway._close_session("unknown-session")
1286
1287
1288class _FakeDataChannel:
1289    """Data channel stand-in that captures outbound messages for proxy tests."""
1290
1291    def __init__(self) -> None:
1292        self.is_open = True
1293        self.sent: list[str] = []
1294
1295    async def send(self, data: str) -> None:
1296        self.sent.append(data)
1297
1298
1299def _proxy_gateway(cert_pems: tuple[str, str]) -> WebRTCGateway:
1300    cert_pem, key_pem = cert_pems
1301    return WebRTCGateway(
1302        http_session=Mock(),
1303        remote_id="TEST-REMOTE-ID",
1304        cert_pem=cert_pem,
1305        key_pem=key_pem,
1306    )
1307
1308
1309def _reassemble_chunks(frames: list[str]) -> str:
1310    """Reassemble __chunk__ frames back into the original text (base64 -> bytes -> utf-8)."""
1311    parsed = [json.loads(f) for f in frames]
1312    assert all(f["type"] == "__chunk__" for f in parsed)
1313    assert len({f["id"] for f in parsed}) == 1
1314    count = parsed[0]["count"]
1315    parts: list[bytes] = [b""] * count
1316    for frame in parsed:
1317        parts[frame["seq"]] = base64.b64decode(frame["b64"])
1318    return b"".join(parts).decode()
1319
1320
1321async def test_http_proxy_response_small_body_single_message(
1322    cert_pems: tuple[str, str],
1323) -> None:
1324    """A body within the chunk size is sent as one legacy http-proxy-response message."""
1325    gateway = _proxy_gateway(cert_pems)
1326    channel = _FakeDataChannel()
1327    session = cast("WebRTCSession", SimpleNamespace(data_channel=channel))
1328    body = b"\x00\x01\x02small-body"
1329
1330    await gateway._send_http_proxy_response(session, "req-small", 200, {"X-Test": "y"}, body)
1331
1332    assert len(channel.sent) == 1
1333    msg = json.loads(channel.sent[0])
1334    assert msg["type"] == "http-proxy-response"
1335    assert msg["id"] == "req-small"
1336    assert msg["status"] == 200
1337    assert msg["headers"] == {"X-Test": "y"}
1338    assert bytes.fromhex(msg["body"]) == body
1339
1340
1341async def test_http_proxy_response_large_body_chunked(cert_pems: tuple[str, str]) -> None:
1342    """A large HTTP-proxy response is split into base64 chunk frames the client reassembles."""
1343    gateway = _proxy_gateway(cert_pems)
1344    channel = _FakeDataChannel()
1345    session = cast("WebRTCSession", SimpleNamespace(data_channel=channel))
1346    body = bytes(range(256)) * ((MA_API_CHUNK_SIZE * 5) // 512)  # big body -> big JSON message
1347
1348    await gateway._send_http_proxy_response(session, "req-big", 200, {}, body)
1349
1350    assert len(channel.sent) > 1
1351    assert all(json.loads(m)["type"] == "__chunk__" for m in channel.sent)
1352    # every serialized frame must stay under the negotiated 256 KiB data-channel limit
1353    assert all(len(m.encode()) < 256 * 1024 for m in channel.sent)
1354
1355    reassembled = json.loads(_reassemble_chunks(channel.sent))
1356    assert reassembled["type"] == "http-proxy-response"
1357    assert reassembled["id"] == "req-big"
1358    assert reassembled["status"] == 200
1359    assert bytes.fromhex(reassembled["body"]) == body
1360
1361
1362async def test_send_ma_api_small_message_passthrough(cert_pems: tuple[str, str]) -> None:
1363    """A ma-api message within the limit is sent verbatim, not chunked."""
1364    gateway = _proxy_gateway(cert_pems)
1365    channel = _FakeDataChannel()
1366    await gateway._send_ma_api(cast("DataChannel", channel), '{"event":"player_updated"}')
1367    assert channel.sent == ['{"event":"player_updated"}']
1368
1369
1370async def test_send_ma_api_large_message_chunked(cert_pems: tuple[str, str]) -> None:
1371    """A large ma-api message is chunked and reassembles byte-identically (multibyte-safe)."""
1372    gateway = _proxy_gateway(cert_pems)
1373    channel = _FakeDataChannel()
1374    # multibyte payload so chunk boundaries fall mid-character, exercising the byte-level split
1375    text = '{"data":"' + "音楽" * MA_API_CHUNK_SIZE + '"}'
1376
1377    await gateway._send_ma_api(cast("DataChannel", channel), text)
1378
1379    assert len(channel.sent) > 1
1380    assert all(len(m.encode()) < 256 * 1024 for m in channel.sent)
1381    assert _reassemble_chunks(channel.sent) == text
1382