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    DATA_CHANNEL_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
703    await gateway._handle_http_proxy_request(
704        None, {"id": "1", "method": "GET", "path": malicious_path}
705    )
706
707    parsed = urlparse(captured_url["url"])
708    assert parsed.hostname == "localhost"
709    assert parsed.port == 8095
710    assert parsed.username is None
711    assert "evil.com" not in (parsed.netloc or "")
712
713
714async def test_http_proxy_request_keeps_the_unverified_dial_on_this_host(
715    cert_pems: tuple[str, str],
716) -> None:
717    """The proxy must not carry its unverified TLS dial off this host."""
718    cert_pem, key_pem = cert_pems
719    mock_session = Mock()
720    captured_kwargs: dict[str, object] = {}
721
722    def fake_request(_method: str, _url: str, **kwargs: object) -> AsyncMock:
723        captured_kwargs.update(kwargs)
724        response = AsyncMock()
725        response.status = 200
726        response.headers = {}
727        response.read = AsyncMock(return_value=b"")
728        ctx = AsyncMock()
729        ctx.__aenter__ = AsyncMock(return_value=response)
730        ctx.__aexit__ = AsyncMock(return_value=False)
731        return ctx
732
733    mock_session.request = fake_request
734
735    gateway = WebRTCGateway(
736        http_session=mock_session,
737        remote_id="TEST-REMOTE-ID",
738        cert_pem=cert_pem,
739        key_pem=key_pem,
740        local_ws_url="wss://127.0.0.1:8095/ws",
741    )
742
743    await gateway._handle_http_proxy_request(None, {"id": "1", "method": "GET", "path": "/info"})
744
745    assert captured_kwargs["ssl"] is False
746    assert captured_kwargs["allow_redirects"] is False
747
748
749async def test_local_websocket_dial_skips_certificate_verification(
750    cert_pems: tuple[str, str],
751) -> None:
752    """Reach the local API on the bind address, which no certificate is issued for."""
753    cert_pem, key_pem = cert_pems
754    mock_session = Mock()
755    captured_kwargs: dict[str, object] = {}
756
757    async def fake_ws_connect(_url: str, **kwargs: object) -> AsyncMock:
758        captured_kwargs.update(kwargs)
759        return AsyncMock()
760
761    mock_session.ws_connect = fake_ws_connect
762
763    gateway = WebRTCGateway(
764        http_session=mock_session,
765        remote_id="TEST-REMOTE-ID",
766        cert_pem=cert_pem,
767        key_pem=key_pem,
768        local_ws_url="wss://127.0.0.1:8095/ws",
769    )
770    session = WebRTCSession(session_id="s1", pc=Mock())
771    channel = MagicMock()
772    channel.wait_open = AsyncMock()
773
774    with patch.object(gateway, "_schedule_close"):
775        await gateway._bridge_ma_api(session, channel)
776
777    assert captured_kwargs["ssl"] is False
778
779
780# ---- aiolibdatachannel loopback tests --------------------------------------
781#
782# These exercise the migrated gateway against real loopback PeerConnections
783# (offerer = browser role, answerer = gateway role) rather than mocking the
784# WebRTC layer. ICE servers are stubbed to [] so gathering completes on host
785# candidates only (fast, offline).
786
787
788@pytest.fixture(scope="session")
789def cert_pems() -> tuple[str, str]:
790    """Generate a throwaway DTLS cert/key as PEM strings for the gateway."""
791    private_key, cert = _generate_certificate()
792    cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode()
793    key_pem = private_key.private_bytes(
794        encoding=serialization.Encoding.PEM,
795        format=serialization.PrivateFormat.PKCS8,
796        encryption_algorithm=serialization.NoEncryption(),
797    ).decode()
798    return cert_pem, key_pem
799
800
801def _sha256_fingerprint(cert_pem: str) -> str:
802    """Compute the uppercase colon-separated SHA-256 fingerprint of a PEM certificate."""
803    body = "".join(line for line in cert_pem.splitlines() if line and not line.startswith("-----"))
804    der = base64.b64decode(body)
805    digest = hashlib.sha256(der).hexdigest().upper()
806    return ":".join(digest[i : i + 2] for i in range(0, len(digest), 2))
807
808
809class _FakeSignaling:
810    """Signaling WebSocket stand-in that captures outbound JSON messages."""
811
812    def __init__(self) -> None:
813        self.messages: list[dict[str, Any]] = []
814
815    async def send_json(self, data: dict[str, Any]) -> None:
816        self.messages.append(data)
817
818    @property
819    def answers(self) -> list[dict[str, Any]]:
820        return [m for m in self.messages if m.get("type") == "answer"]
821
822
823class _FakeLocalWS:
824    """Minimal aiohttp WebSocket stand-in for channel-bridging tests."""
825
826    def __init__(self) -> None:
827        self.closed = False
828        self.sent: list[str | bytes] = []
829        self._incoming: asyncio.Queue[SimpleNamespace | None] = asyncio.Queue()
830
831    async def send_str(self, data: str) -> None:
832        self.sent.append(data)
833
834    async def send_bytes(self, data: bytes) -> None:
835        self.sent.append(data)
836
837    async def close(self) -> None:
838        self.closed = True
839        self._incoming.put_nowait(None)
840
841    def feed_text(self, data: str) -> None:
842        """Queue a text message as if the local server sent it."""
843        self._incoming.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=data))
844
845    def feed_bytes(self, data: bytes) -> None:
846        """Queue a binary message as if the local server sent it."""
847        self._incoming.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.BINARY, data=data))
848
849    def __aiter__(self) -> AsyncIterator[SimpleNamespace]:
850        return self
851
852    async def __anext__(self) -> SimpleNamespace:
853        msg = await self._incoming.get()
854        if msg is None:
855            raise StopAsyncIteration
856        return msg
857
858
859class _FakeBidiChannel:
860    """
861    Async-iterable data-channel stand-in for bridging tests.
862
863    Drives the gateway's channel->local pump without a real WebRTC handshake: feed
864    inbound messages with :meth:`feed`, end the stream with :meth:`close`, and read
865    what the gateway sent back on ``sent``.
866    """
867
868    def __init__(self, label: str = "ma-api", max_message_size: int = 256 * 1024) -> None:
869        self.label = label
870        self.is_open = True
871        self.closed = False
872        self.max_message_size = max_message_size
873        self.sent: list[str | bytes] = []
874        self._inbound: asyncio.Queue[str | bytes | None] = asyncio.Queue()
875
876    async def wait_open(self) -> None:
877        return
878
879    async def send(self, data: str | bytes) -> None:
880        # a real send always suspends, which is what lets concurrent senders interleave
881        await asyncio.sleep(0)
882        self.sent.append(data)
883
884    def feed(self, message: str | bytes) -> None:
885        """Queue an inbound message as if the browser sent it over the channel."""
886        self._inbound.put_nowait(message)
887
888    def close(self) -> None:
889        self.closed = True
890        self.is_open = False
891        self._inbound.put_nowait(None)
892
893    async def aclose(self) -> None:
894        self.close()
895
896    def __aiter__(self) -> AsyncIterator[str | bytes]:
897        return self
898
899    async def __anext__(self) -> str | bytes:
900        message = await self._inbound.get()
901        if message is None:
902            raise StopAsyncIteration
903        return message
904
905
906class _FakeHttpSession:
907    """ClientSession stand-in handing out one fake WebSocket per dialed url."""
908
909    def __init__(self) -> None:
910        self.dialed: list[str] = []
911        self.dial_kwargs: list[dict[str, Any]] = []
912        self.websockets: dict[str, _FakeLocalWS] = {}
913        self.requested: list[str] = []
914        self.response_body = b""
915        self.bodies: dict[str, bytes] = {}
916
917    async def ws_connect(self, url: str, **kwargs: Any) -> _FakeLocalWS:
918        self.dialed.append(url)
919        self.dial_kwargs.append(kwargs)
920        local_ws = _FakeLocalWS()
921        self.websockets[url] = local_ws
922        return local_ws
923
924    def request(self, _method: str, url: str, **_kwargs: Any) -> AsyncMock:
925        """Serve this url's entry in ``bodies``, falling back to ``response_body``."""
926        self.requested.append(url)
927        response = AsyncMock()
928        response.status = 200
929        response.headers = {"Content-Type": "image/jpeg"}
930        response.read = AsyncMock(return_value=self.bodies.get(url, self.response_body))
931        ctx = AsyncMock()
932        ctx.__aenter__ = AsyncMock(return_value=response)
933        ctx.__aexit__ = AsyncMock(return_value=False)
934        return ctx
935
936
937class _FakePeerConnection:
938    """PeerConnection stand-in that runs gateway-spawned pumps as asyncio tasks."""
939
940    def __init__(self) -> None:
941        self._tasks: list[asyncio.Task[None]] = []
942        self._incoming: asyncio.Queue[_FakeBidiChannel] = asyncio.Queue()
943
944    def spawn_task(self, coro: Coroutine[Any, Any, None]) -> None:
945        self._tasks.append(asyncio.ensure_future(coro))
946
947    def offer_channel(self, channel: _FakeBidiChannel) -> None:
948        """Offer a data channel as if the browser had opened it."""
949        self._incoming.put_nowait(channel)
950
951    async def incoming_data_channels(self) -> AsyncIterator[DataChannel]:
952        while True:
953            yield cast("DataChannel", await self._incoming.get())
954
955    async def aclose(self) -> None:
956        for task in self._tasks:
957            task.cancel()
958        # await cancellation so no pump task lingers past teardown
959        await asyncio.gather(*self._tasks, return_exceptions=True)
960
961
962def _register_bridge_session(
963    gateway: WebRTCGateway, session_id: str, channel: _FakeBidiChannel
964) -> WebRTCSession:
965    """Register a session backed by a fake PeerConnection and ma-api channel."""
966    session = WebRTCSession(
967        session_id=session_id,
968        pc=cast("PeerConnection", _FakePeerConnection()),
969        data_channel=cast("DataChannel", channel),
970    )
971    gateway.sessions[session_id] = session
972    return session
973
974
975def _register_routed_session(
976    gateway: WebRTCGateway, session_id: str
977) -> tuple[WebRTCSession, _FakePeerConnection]:
978    """Register a session that routes the data channels offered to its PeerConnection."""
979    pc = _FakePeerConnection()
980    session = WebRTCSession(session_id=session_id, pc=cast("PeerConnection", pc))
981    gateway.sessions[session_id] = session
982    pc.spawn_task(gateway._accept_channels(session))
983    return session, pc
984
985
986async def _wait_for(predicate: Callable[[], bool], timeout: float = 15.0) -> None:
987    """Poll ``predicate`` until it is true or the timeout elapses."""
988    loop = asyncio.get_running_loop()
989    deadline = loop.time() + timeout
990    while loop.time() < deadline:
991        if predicate():
992            return
993        await asyncio.sleep(0.02)
994    raise AssertionError("condition not met within timeout")
995
996
997async def test_handle_offer_answers_with_pinned_fingerprint(cert_pems: tuple[str, str]) -> None:
998    """The answer SDP carries the DTLS fingerprint of the configured certificate (pinning)."""
999    cert_pem, key_pem = cert_pems
1000    gateway = WebRTCGateway(
1001        http_session=Mock(),
1002        remote_id="TEST-REMOTE-ID",
1003        cert_pem=cert_pem,
1004        key_pem=key_pem,
1005    )
1006    signaling = _FakeSignaling()
1007    gateway._signaling_ws = cast("aiohttp.ClientWebSocketResponse", signaling)
1008    offerer = PeerConnection(RTCConfiguration())
1009    session_id = "fingerprint-session"
1010    try:
1011        await offerer.create_data_channel("ma-api")
1012        offer = await offerer.create_offer()
1013        with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
1014            await gateway._create_session(session_id)
1015            await asyncio.wait_for(
1016                gateway._handle_offer(session_id, {"sdp": offer.sdp, "type": "offer"}),
1017                timeout=15,
1018            )
1019
1020        assert len(signaling.answers) == 1
1021        answer = signaling.answers[0]
1022        assert answer["sessionId"] == session_id
1023        assert answer["data"]["type"] == "answer"
1024        fingerprint_line = next(
1025            line for line in answer["data"]["sdp"].splitlines() if line.startswith("a=fingerprint:")
1026        )
1027        assert _sha256_fingerprint(cert_pem) in fingerprint_line
1028    finally:
1029        await gateway._close_session(session_id)
1030        await offerer.aclose()
1031
1032
1033async def test_ma_api_channel_bridges_to_local_ws(cert_pems: tuple[str, str]) -> None:
1034    """Messages flow both ways across the ma-api data channel and the local WebSocket."""
1035    cert_pem, key_pem = cert_pems
1036    fake_ws = _FakeLocalWS()
1037    http_session = Mock()
1038    http_session.ws_connect = AsyncMock(
1039        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1040    )
1041    gateway = WebRTCGateway(
1042        http_session=http_session,
1043        remote_id="TEST-REMOTE-ID",
1044        cert_pem=cert_pem,
1045        key_pem=key_pem,
1046    )
1047    channel = _FakeBidiChannel()
1048    session = _register_bridge_session(gateway, "bridge-session", channel)
1049    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1050    try:
1051        await _wait_for(lambda: session.local_ws is not None)
1052
1053        # browser -> local WebSocket
1054        channel.feed("from browser")
1055        await _wait_for(lambda: fake_ws.sent == ["from browser"])
1056
1057        # local WebSocket -> browser
1058        fake_ws.feed_text("from local")
1059        await _wait_for(lambda: channel.sent == ["from local"])
1060    finally:
1061        channel.close()
1062        await asyncio.wait_for(bridge, timeout=5)
1063        await _wait_for(lambda: "bridge-session" not in gateway.sessions)
1064
1065
1066def test_build_ice_servers_maps_dicts() -> None:
1067    """ICE server dicts map to one IceServer per url, preserving TURN credentials."""
1068    gateway = WebRTCGateway(
1069        http_session=Mock(),
1070        remote_id="TEST-REMOTE-ID",
1071        cert_pem="cert",
1072        key_pem="key",
1073    )
1074    servers: list[dict[str, Any]] = [
1075        {"urls": "stun:stun.example.com:3478"},
1076        {"urls": "turn:turn.example.com:3478", "username": "user", "credential": "pass"},
1077        {
1078            "urls": ["stun:a.example.com:3478", "turn:b.example.com:3478"],
1079            "username": "u2",
1080            "credential": "c2",
1081        },
1082    ]
1083
1084    result = gateway._build_ice_servers(servers)
1085
1086    assert all(isinstance(server, IceServer) for server in result)
1087    # the two-url entry fans out, so 1 + 1 + 2 = 4 IceServers
1088    assert len(result) == 4
1089    assert result[0] == IceServer(url="stun:stun.example.com:3478")
1090    assert result[1] == IceServer(
1091        url="turn:turn.example.com:3478", username="user", credential="pass"
1092    )
1093    # to_url() inlines the TURN credentials for libdatachannel
1094    assert result[1].to_url() == "turn:user:[email protected]:3478"
1095    # list urls share the entry's credentials
1096    assert result[2] == IceServer(url="stun:a.example.com:3478", username="u2", credential="c2")
1097    assert result[3] == IceServer(url="turn:b.example.com:3478", username="u2", credential="c2")
1098
1099
1100async def test_session_closes_when_ma_api_channel_closes(cert_pems: tuple[str, str]) -> None:
1101    """Closing the browser ma-api channel tears down the whole gateway session."""
1102    cert_pem, key_pem = cert_pems
1103    fake_ws = _FakeLocalWS()
1104    http_session = Mock()
1105    http_session.ws_connect = AsyncMock(
1106        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1107    )
1108    gateway = WebRTCGateway(
1109        http_session=http_session,
1110        remote_id="TEST-REMOTE-ID",
1111        cert_pem=cert_pem,
1112        key_pem=key_pem,
1113    )
1114    channel = _FakeBidiChannel()
1115    session = _register_bridge_session(gateway, "channel-close-session", channel)
1116    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1117    await _wait_for(lambda: session.local_ws is not None)
1118
1119    # the browser closes the ma-api channel -> the whole session is torn down
1120    channel.close()
1121
1122    await _wait_for(lambda: "channel-close-session" not in gateway.sessions)
1123    assert "channel-close-session" not in gateway.sessions
1124    await asyncio.wait_for(bridge, timeout=5)
1125
1126
1127# ---- channel routing -------------------------------------------------------
1128
1129LOCAL_WS_URL = "ws://127.0.0.1:8095/ws"
1130SENDSPIN_URL = "ws://127.0.0.1:8927/sendspin"
1131
1132
1133def _routing_gateway(
1134    cert_pems: tuple[str, str],
1135    http_session: _FakeHttpSession,
1136    local_ws_url: str = LOCAL_WS_URL,
1137    set_sendspin_player_callback: Callable[[str, str], None] | None = None,
1138) -> WebRTCGateway:
1139    """Create a gateway whose local WebSockets are all served by the fake HTTP session."""
1140    cert_pem, key_pem = cert_pems
1141    return WebRTCGateway(
1142        http_session=cast("aiohttp.ClientSession", http_session),
1143        remote_id="TEST-REMOTE-ID",
1144        cert_pem=cert_pem,
1145        key_pem=key_pem,
1146        local_ws_url=local_ws_url,
1147        sendspin_url=SENDSPIN_URL,
1148        set_sendspin_player_callback=set_sendspin_player_callback,
1149    )
1150
1151
1152async def test_sendspin_channel_bridges_to_the_sendspin_server(cert_pems: tuple[str, str]) -> None:
1153    """A sendspin channel reaches the internal sendspin server, web player id and all."""
1154    http_session = _FakeHttpSession()
1155    announced_players: list[tuple[str, str]] = []
1156    gateway = _routing_gateway(
1157        cert_pems,
1158        http_session,
1159        set_sendspin_player_callback=lambda session_id, player_id: announced_players.append(
1160            (session_id, player_id)
1161        ),
1162    )
1163    session, pc = _register_routed_session(gateway, "sendspin-session")
1164    channel = _FakeBidiChannel(label="sendspin")
1165    pc.offer_channel(channel)
1166    try:
1167        await _wait_for(lambda: SENDSPIN_URL in http_session.websockets)
1168        local_ws = http_session.websockets[SENDSPIN_URL]
1169
1170        # the first message announces the web player, and is forwarded verbatim
1171        auth = json.dumps({"type": "auth", "token": "t", "client_id": "web-player-1"})
1172        channel.feed(auth)
1173        await _wait_for(lambda: local_ws.sent == [auth])
1174        assert announced_players == [("sendspin-session", "web-player-1")]
1175        assert session.sendspin_player_id == "web-player-1"
1176
1177        # audio keeps flowing in both directions, text and binary alike
1178        channel.feed(b"\x01\x02")
1179        await _wait_for(lambda: local_ws.sent == [auth, b"\x01\x02"])
1180        local_ws.feed_text('{"type":"hello"}')
1181        local_ws.feed_bytes(b"\x03\x04")
1182        await _wait_for(lambda: channel.sent == ['{"type":"hello"}', b"\x03\x04"])
1183    finally:
1184        await gateway._close_session("sendspin-session")
1185
1186
1187@pytest.mark.parametrize(
1188    ("local_ws_url", "expected_url"),
1189    [
1190        ("ws://127.0.0.1:8095/ws", "ws://127.0.0.1:8095/live_announcement"),
1191        # an https webserver is still dialed on its bind address, which no cert covers
1192        ("wss://127.0.0.1:8095/ws", "wss://127.0.0.1:8095/live_announcement"),
1193    ],
1194)
1195async def test_live_announcement_channel_bridges_to_the_webserver(
1196    cert_pems: tuple[str, str], local_ws_url: str, expected_url: str
1197) -> None:
1198    """A live announcement channel reaches the webserver route that takes the audio."""
1199    http_session = _FakeHttpSession()
1200    gateway = _routing_gateway(cert_pems, http_session, local_ws_url=local_ws_url)
1201    session, pc = _register_routed_session(gateway, "announce-session")
1202    channel = _FakeBidiChannel(label="live_announcement")
1203    pc.offer_channel(channel)
1204    try:
1205        await _wait_for(lambda: expected_url in http_session.websockets)
1206        local_ws = http_session.websockets[expected_url]
1207        assert http_session.dial_kwargs == [{"ssl": False}]
1208
1209        # the client authenticates on the route itself, so its handshake passes through
1210        handshake = [
1211            json.dumps({"type": "auth", "token": "t"}),
1212            json.dumps({"type": "start", "player_id": "player1", "sample_rate": 16000}),
1213        ]
1214        for message in handshake:
1215            channel.feed(message)
1216        await _wait_for(lambda: local_ws.sent == handshake)
1217        # the sendspin snoop belongs to the sendspin bridge only
1218        assert session.sendspin_player_id is None
1219
1220        # spoken audio goes up, the route's replies come back
1221        channel.feed(b"\x00\x01")
1222        await _wait_for(lambda: local_ws.sent == [*handshake, b"\x00\x01"])
1223        local_ws.feed_text('{"type":"started"}')
1224        await _wait_for(lambda: channel.sent == ['{"type":"started"}'])
1225    finally:
1226        await gateway._close_session("announce-session")
1227
1228
1229@pytest.mark.parametrize("closed_by", ["browser", "local"])
1230async def test_closing_a_bridged_channel_leaves_the_api_session_up(
1231    cert_pems: tuple[str, str], closed_by: str
1232) -> None:
1233    """Losing a bridged WebSocket tears down that bridge only, never the API session."""
1234    http_session = _FakeHttpSession()
1235    gateway = _routing_gateway(cert_pems, http_session)
1236    session, pc = _register_routed_session(gateway, "mixed-session")
1237    api_channel = _FakeBidiChannel()
1238    sendspin_channel = _FakeBidiChannel(label="sendspin")
1239    pc.offer_channel(api_channel)
1240    pc.offer_channel(sendspin_channel)
1241    try:
1242        await _wait_for(
1243            lambda: session.local_ws is not None and SENDSPIN_URL in http_session.dialed
1244        )
1245        sendspin_ws = http_session.websockets[SENDSPIN_URL]
1246
1247        if closed_by == "browser":
1248            sendspin_channel.close()
1249        else:
1250            await sendspin_ws.close()
1251        await _wait_for(lambda: not session.channels)
1252
1253        assert sendspin_ws.closed is True
1254        assert sendspin_channel.closed is True
1255        # the API session is untouched: still registered, still bridged, channel still open
1256        assert "mixed-session" in gateway.sessions
1257        assert session.local_ws is not None
1258        assert cast("_FakeLocalWS", session.local_ws).closed is False
1259        assert api_channel.closed is False
1260    finally:
1261        await gateway._close_session("mixed-session")
1262
1263
1264async def test_the_first_channel_is_the_api_channel_whatever_its_label(
1265    cert_pems: tuple[str, str],
1266) -> None:
1267    """Clients may label their API channel freely, so the first channel bridges to the API."""
1268    http_session = _FakeHttpSession()
1269    gateway = _routing_gateway(cert_pems, http_session)
1270    session, pc = _register_routed_session(gateway, "labelled-session")
1271    channel = _FakeBidiChannel(label="ma-api-v2")
1272    pc.offer_channel(channel)
1273    try:
1274        await _wait_for(lambda: session.local_ws is not None)
1275        assert session.data_channel is cast("DataChannel", channel)
1276        assert http_session.dialed == [f"{LOCAL_WS_URL}?webrtc_session_id=labelled-session"]
1277    finally:
1278        await gateway._close_session("labelled-session")
1279
1280
1281async def test_unknown_channel_label_cannot_replace_the_api_channel(
1282    cert_pems: tuple[str, str],
1283) -> None:
1284    """A channel this server has no route for is refused instead of hijacking the session."""
1285    http_session = _FakeHttpSession()
1286    gateway = _routing_gateway(cert_pems, http_session)
1287    session, pc = _register_routed_session(gateway, "unknown-session")
1288    api_channel = _FakeBidiChannel()
1289    unknown_channel = _FakeBidiChannel(label="channel-from-the-future")
1290    pc.offer_channel(api_channel)
1291    try:
1292        await _wait_for(lambda: session.local_ws is not None)
1293        pc.offer_channel(unknown_channel)
1294        await _wait_for(lambda: unknown_channel.closed)
1295
1296        assert session.data_channel is cast("DataChannel", api_channel)
1297        assert "unknown-session" in gateway.sessions
1298        # only the API channel was ever bridged
1299        assert http_session.dialed == [f"{LOCAL_WS_URL}?webrtc_session_id=unknown-session"]
1300    finally:
1301        await gateway._close_session("unknown-session")
1302
1303
1304def _proxy_request(request_id: str, path: str) -> str:
1305    """Build the http-proxy-request message a client sends for a proxied path."""
1306    return json.dumps(
1307        {"type": "http-proxy-request", "id": request_id, "method": "GET", "path": path}
1308    )
1309
1310
1311def _body_delivered(sent: list[str | bytes], size: int) -> bool:
1312    """Return whether the binary frames sent so far add up to a whole body of ``size``."""
1313    return sum(len(m) for m in sent if isinstance(m, bytes)) >= size
1314
1315
1316def _read_proxy_response(sent: list[str | bytes]) -> tuple[dict[str, Any], bytes]:
1317    """
1318    Read the binary-framed response a client would reassemble from the proxy channel.
1319
1320    :param sent: Messages the gateway sent, starting at the response's JSON header.
1321    """
1322    header = json.loads(cast("str", sent[0]))
1323    body = b""
1324    for message in sent[1:]:
1325        assert isinstance(message, bytes), f"expected a binary body frame, got {message!r}"
1326        body += message
1327        if len(body) >= header["size"]:
1328            break
1329    assert len(body) == header["size"]
1330    return header, body
1331
1332
1333async def test_http_proxy_channel_answers_on_its_own_channel(cert_pems: tuple[str, str]) -> None:
1334    """A proxied request on the http proxy channel is answered there, not on the API channel."""
1335    http_session = _FakeHttpSession()
1336    http_session.response_body = b"\xff\xd8jpeg-bytes"
1337    gateway = _routing_gateway(cert_pems, http_session)
1338    session, pc = _register_routed_session(gateway, "proxy-session")
1339    api_channel = _FakeBidiChannel()
1340    proxy_channel = _FakeBidiChannel(label="http_proxy")
1341    pc.offer_channel(api_channel)
1342    pc.offer_channel(proxy_channel)
1343    try:
1344        await _wait_for(lambda: session.local_ws is not None)
1345
1346        proxy_channel.feed(_proxy_request("img-1", "/imageproxy/abc"))
1347        await _wait_for(lambda: len(proxy_channel.sent) >= 2)
1348
1349        assert http_session.requested == ["http://127.0.0.1:8095/imageproxy/abc"]
1350        response, body = _read_proxy_response(proxy_channel.sent)
1351        assert response["type"] == "http-proxy-response"
1352        assert response["id"] == "img-1"
1353        assert response["status"] == 200
1354        # the body rides as raw binary, so it costs its own size on the wire and no more
1355        assert body == b"\xff\xd8jpeg-bytes"
1356        assert "body" not in response
1357        # the image never touches the API channel, nor the local API WebSocket
1358        assert api_channel.sent == []
1359        assert cast("_FakeLocalWS", session.local_ws).sent == []
1360    finally:
1361        await gateway._close_session("proxy-session")
1362
1363
1364async def test_http_proxy_request_on_the_api_channel_is_still_answered(
1365    cert_pems: tuple[str, str],
1366) -> None:
1367    """Clients that predate the http proxy channel keep proxying over the API channel."""
1368    http_session = _FakeHttpSession()
1369    http_session.response_body = b"legacy-body"
1370    gateway = _routing_gateway(cert_pems, http_session)
1371    session, pc = _register_routed_session(gateway, "legacy-session")
1372    api_channel = _FakeBidiChannel()
1373    pc.offer_channel(api_channel)
1374    try:
1375        await _wait_for(lambda: session.local_ws is not None)
1376
1377        api_channel.feed(_proxy_request("img-2", "/imageproxy/def"))
1378        await _wait_for(lambda: bool(api_channel.sent))
1379
1380        assert http_session.requested == ["http://127.0.0.1:8095/imageproxy/def"]
1381        response = json.loads(cast("str", api_channel.sent[0]))
1382        assert response["type"] == "http-proxy-response"
1383        assert response["id"] == "img-2"
1384        assert bytes.fromhex(response["body"]) == b"legacy-body"
1385        # the proxy request is served here, never forwarded to the local API WebSocket
1386        assert cast("_FakeLocalWS", session.local_ws).sent == []
1387    finally:
1388        await gateway._close_session("legacy-session")
1389
1390
1391async def test_http_proxy_channel_reports_a_failed_fetch_on_its_own_channel(
1392    cert_pems: tuple[str, str],
1393) -> None:
1394    """A fetch that raises still answers the client, on the channel it asked over."""
1395    http_session = _FakeHttpSession()
1396    http_session.request = Mock(side_effect=RuntimeError("boom"))  # type: ignore[method-assign]
1397    gateway = _routing_gateway(cert_pems, http_session)
1398    session, pc = _register_routed_session(gateway, "proxy-error-session")
1399    api_channel = _FakeBidiChannel()
1400    proxy_channel = _FakeBidiChannel(label="http_proxy")
1401    pc.offer_channel(api_channel)
1402    pc.offer_channel(proxy_channel)
1403    try:
1404        await _wait_for(lambda: session.local_ws is not None)
1405
1406        proxy_channel.feed(_proxy_request("img-3", "/imageproxy/boom"))
1407        await _wait_for(lambda: len(proxy_channel.sent) >= 2)
1408
1409        response, body = _read_proxy_response(proxy_channel.sent)
1410        assert response["id"] == "img-3"
1411        assert response["status"] == 500
1412        assert b"boom" in body
1413        assert api_channel.sent == []
1414    finally:
1415        await gateway._close_session("proxy-error-session")
1416
1417
1418async def test_http_proxy_channel_splits_a_large_body_into_binary_frames(
1419    cert_pems: tuple[str, str],
1420) -> None:
1421    """A body past the channel's message limit arrives as raw frames that concatenate back."""
1422    http_session = _FakeHttpSession()
1423    http_session.response_body = bytes(range(256)) * 2048  # 512 KiB
1424    gateway = _routing_gateway(cert_pems, http_session)
1425    session, pc = _register_routed_session(gateway, "proxy-large-session")
1426    proxy_channel = _FakeBidiChannel(label="http_proxy")
1427    pc.offer_channel(proxy_channel)
1428    try:
1429        await _wait_for(lambda: "http_proxy" in session.channels)
1430
1431        proxy_channel.feed(_proxy_request("img-5", "/imageproxy/large"))
1432        await _wait_for(
1433            lambda: _body_delivered(proxy_channel.sent, len(http_session.response_body))
1434        )
1435
1436        response, body = _read_proxy_response(proxy_channel.sent)
1437        assert response["size"] == len(http_session.response_body)
1438        assert body == http_session.response_body
1439        frames = proxy_channel.sent[1:]
1440        assert all(len(frame) <= proxy_channel.max_message_size for frame in frames)
1441        # the wire cost is the body itself, not the ~2.7x a hex-in-base64 response took
1442        assert sum(len(frame) for frame in frames) == len(http_session.response_body)
1443    finally:
1444        await gateway._close_session("proxy-large-session")
1445
1446
1447async def test_http_proxy_channel_honours_the_negotiated_message_limit(
1448    cert_pems: tuple[str, str],
1449) -> None:
1450    """A peer that advertises a small limit gets frames it can actually accept."""
1451    http_session = _FakeHttpSession()
1452    http_session.response_body = b"x" * (200 * 1024)
1453    gateway = _routing_gateway(cert_pems, http_session)
1454    session, pc = _register_routed_session(gateway, "proxy-small-frames-session")
1455    # what a peer that advertises no a=max-message-size in its SDP is assumed to accept
1456    proxy_channel = _FakeBidiChannel(label="http_proxy", max_message_size=64 * 1024)
1457    pc.offer_channel(proxy_channel)
1458    try:
1459        await _wait_for(lambda: "http_proxy" in session.channels)
1460
1461        proxy_channel.feed(_proxy_request("img-6", "/imageproxy/small-frames"))
1462        await _wait_for(
1463            lambda: _body_delivered(proxy_channel.sent, len(http_session.response_body))
1464        )
1465
1466        _, body = _read_proxy_response(proxy_channel.sent)
1467        assert body == http_session.response_body
1468        assert all(len(frame) <= 64 * 1024 for frame in proxy_channel.sent[1:])
1469    finally:
1470        await gateway._close_session("proxy-small-frames-session")
1471
1472
1473async def test_http_proxy_channel_never_interleaves_two_responses(
1474    cert_pems: tuple[str, str],
1475) -> None:
1476    """Body frames carry no request id, so each response must reach the client in one run."""
1477    http_session = _FakeHttpSession()
1478    http_session.bodies = {
1479        "http://127.0.0.1:8095/imageproxy/one": b"1" * (300 * 1024),
1480        "http://127.0.0.1:8095/imageproxy/two": b"2" * (300 * 1024),
1481    }
1482    gateway = _routing_gateway(cert_pems, http_session)
1483    session, pc = _register_routed_session(gateway, "proxy-concurrent-session")
1484    proxy_channel = _FakeBidiChannel(label="http_proxy")
1485    pc.offer_channel(proxy_channel)
1486    try:
1487        await _wait_for(lambda: "http_proxy" in session.channels)
1488
1489        proxy_channel.feed(_proxy_request("img-one", "/imageproxy/one"))
1490        proxy_channel.feed(_proxy_request("img-two", "/imageproxy/two"))
1491        await _wait_for(lambda: sum(isinstance(m, str) for m in proxy_channel.sent) == 2)
1492        await _wait_for(lambda: len(proxy_channel.sent) >= 6)
1493
1494        # split at the second header: each response owns an unbroken run of body frames
1495        second = next(i for i, m in enumerate(proxy_channel.sent) if i and isinstance(m, str))
1496        first_header, first_body = _read_proxy_response(proxy_channel.sent[:second])
1497        second_header, second_body = _read_proxy_response(proxy_channel.sent[second:])
1498        assert {first_header["id"], second_header["id"]} == {"img-one", "img-two"}
1499        # each body is one repeated byte, so any interleaving shows up as a mixed run
1500        filler = {"img-one": ord("1"), "img-two": ord("2")}
1501        assert set(first_body) == {filler[first_header["id"]]}
1502        assert set(second_body) == {filler[second_header["id"]]}
1503        assert len(first_body) == len(second_body) == 300 * 1024
1504    finally:
1505        await gateway._close_session("proxy-concurrent-session")
1506
1507
1508@pytest.mark.parametrize(
1509    "junk",
1510    [
1511        "not json at all",
1512        json.dumps(["not", "a", "dict"]),
1513        json.dumps({"type": "something-else"}),
1514        b"\x00\x01binary",
1515    ],
1516)
1517async def test_http_proxy_channel_survives_junk(
1518    cert_pems: tuple[str, str], junk: str | bytes
1519) -> None:
1520    """Anything that is not a proxy request is ignored without killing the channel."""
1521    http_session = _FakeHttpSession()
1522    http_session.response_body = b"still-here"
1523    gateway = _routing_gateway(cert_pems, http_session)
1524    session, pc = _register_routed_session(gateway, "proxy-junk-session")
1525    proxy_channel = _FakeBidiChannel(label="http_proxy")
1526    pc.offer_channel(proxy_channel)
1527    try:
1528        await _wait_for(lambda: "http_proxy" in session.channels)
1529
1530        proxy_channel.feed(junk)
1531        proxy_channel.feed(_proxy_request("img-4", "/imageproxy/ok"))
1532        await _wait_for(lambda: len(proxy_channel.sent) >= 2)
1533
1534        response, body = _read_proxy_response(proxy_channel.sent)
1535        assert response["id"] == "img-4"
1536        assert body == b"still-here"
1537    finally:
1538        await gateway._close_session("proxy-junk-session")
1539
1540
1541async def test_closing_the_http_proxy_channel_leaves_the_api_session_up(
1542    cert_pems: tuple[str, str],
1543) -> None:
1544    """Losing the http proxy channel tears down that channel only, never the API session."""
1545    http_session = _FakeHttpSession()
1546    gateway = _routing_gateway(cert_pems, http_session)
1547    session, pc = _register_routed_session(gateway, "proxy-close-session")
1548    api_channel = _FakeBidiChannel()
1549    proxy_channel = _FakeBidiChannel(label="http_proxy")
1550    pc.offer_channel(api_channel)
1551    pc.offer_channel(proxy_channel)
1552    try:
1553        await _wait_for(lambda: "http_proxy" in session.channels)
1554
1555        proxy_channel.close()
1556        await _wait_for(lambda: not session.channels)
1557
1558        assert "proxy-close-session" in gateway.sessions
1559        assert session.local_ws is not None
1560        assert api_channel.closed is False
1561    finally:
1562        await gateway._close_session("proxy-close-session")
1563
1564
1565async def test_a_second_http_proxy_channel_is_refused(cert_pems: tuple[str, str]) -> None:
1566    """A duplicate label is refused so the running handler is never left untracked."""
1567    http_session = _FakeHttpSession()
1568    gateway = _routing_gateway(cert_pems, http_session)
1569    session, pc = _register_routed_session(gateway, "duplicate-session")
1570    first = _FakeBidiChannel(label="http_proxy")
1571    second = _FakeBidiChannel(label="http_proxy")
1572    pc.offer_channel(first)
1573    try:
1574        await _wait_for(lambda: "http_proxy" in session.channels)
1575        pc.offer_channel(second)
1576        await _wait_for(lambda: second.closed)
1577
1578        assert first.closed is False
1579        assert session.channels["http_proxy"].channel is cast("DataChannel", first)
1580    finally:
1581        await gateway._close_session("duplicate-session")
1582
1583
1584class _FakeDataChannel:
1585    """Data channel stand-in that captures outbound messages for proxy tests."""
1586
1587    def __init__(self) -> None:
1588        self.is_open = True
1589        self.sent: list[str] = []
1590
1591    async def send(self, data: str) -> None:
1592        self.sent.append(data)
1593
1594
1595def _proxy_gateway(cert_pems: tuple[str, str]) -> WebRTCGateway:
1596    cert_pem, key_pem = cert_pems
1597    return WebRTCGateway(
1598        http_session=Mock(),
1599        remote_id="TEST-REMOTE-ID",
1600        cert_pem=cert_pem,
1601        key_pem=key_pem,
1602    )
1603
1604
1605def _reassemble_chunks(frames: list[str]) -> str:
1606    """Reassemble __chunk__ frames back into the original text (base64 -> bytes -> utf-8)."""
1607    parsed = [json.loads(f) for f in frames]
1608    assert all(f["type"] == "__chunk__" for f in parsed)
1609    assert len({f["id"] for f in parsed}) == 1
1610    count = parsed[0]["count"]
1611    parts: list[bytes] = [b""] * count
1612    for frame in parsed:
1613        parts[frame["seq"]] = base64.b64decode(frame["b64"])
1614    return b"".join(parts).decode()
1615
1616
1617async def test_http_proxy_response_small_body_single_message(
1618    cert_pems: tuple[str, str],
1619) -> None:
1620    """A body within the chunk size is sent as one http-proxy-response message."""
1621    gateway = _proxy_gateway(cert_pems)
1622    channel = _FakeDataChannel()
1623    body = b"\x00\x01\x02small-body"
1624
1625    await gateway._send_http_proxy_response(
1626        cast("DataChannel", channel), "req-small", 200, {"X-Test": "y"}, body
1627    )
1628
1629    assert len(channel.sent) == 1
1630    msg = json.loads(channel.sent[0])
1631    assert msg["type"] == "http-proxy-response"
1632    assert msg["id"] == "req-small"
1633    assert msg["status"] == 200
1634    assert msg["headers"] == {"X-Test": "y"}
1635    assert bytes.fromhex(msg["body"]) == body
1636
1637
1638async def test_http_proxy_response_large_body_chunked(cert_pems: tuple[str, str]) -> None:
1639    """A large HTTP-proxy response is split into base64 chunk frames the client reassembles."""
1640    gateway = _proxy_gateway(cert_pems)
1641    channel = _FakeDataChannel()
1642    # big body -> big JSON message
1643    body = bytes(range(256)) * ((DATA_CHANNEL_CHUNK_SIZE * 5) // 512)
1644
1645    await gateway._send_http_proxy_response(cast("DataChannel", channel), "req-big", 200, {}, body)
1646
1647    assert len(channel.sent) > 1
1648    assert all(json.loads(m)["type"] == "__chunk__" for m in channel.sent)
1649    # every serialized frame must stay under the negotiated 256 KiB data-channel limit
1650    assert all(len(m.encode()) < 256 * 1024 for m in channel.sent)
1651
1652    reassembled = json.loads(_reassemble_chunks(channel.sent))
1653    assert reassembled["type"] == "http-proxy-response"
1654    assert reassembled["id"] == "req-big"
1655    assert reassembled["status"] == 200
1656    assert bytes.fromhex(reassembled["body"]) == body
1657
1658
1659async def test_send_chunked_small_message_passthrough(cert_pems: tuple[str, str]) -> None:
1660    """A message within the limit is sent verbatim, not chunked."""
1661    gateway = _proxy_gateway(cert_pems)
1662    channel = _FakeDataChannel()
1663    await gateway._send_chunked(cast("DataChannel", channel), '{"event":"player_updated"}')
1664    assert channel.sent == ['{"event":"player_updated"}']
1665
1666
1667async def test_send_chunked_large_message_chunked(cert_pems: tuple[str, str]) -> None:
1668    """A large message is chunked and reassembles byte-identically (multibyte-safe)."""
1669    gateway = _proxy_gateway(cert_pems)
1670    channel = _FakeDataChannel()
1671    # multibyte payload so chunk boundaries fall mid-character, exercising the byte-level split
1672    text = '{"data":"' + "音楽" * DATA_CHANNEL_CHUNK_SIZE + '"}'
1673
1674    await gateway._send_chunked(cast("DataChannel", channel), text)
1675
1676    assert len(channel.sent) > 1
1677    assert all(len(m.encode()) < 256 * 1024 for m in channel.sent)
1678    assert _reassemble_chunks(channel.sent) == text
1679