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") -> None:
869        self.label = label
870        self.is_open = True
871        self.closed = False
872        self.sent: list[str | bytes] = []
873        self._inbound: asyncio.Queue[str | bytes | None] = asyncio.Queue()
874
875    async def wait_open(self) -> None:
876        return
877
878    async def send(self, data: str | bytes) -> None:
879        self.sent.append(data)
880
881    def feed(self, message: str | bytes) -> None:
882        """Queue an inbound message as if the browser sent it over the channel."""
883        self._inbound.put_nowait(message)
884
885    def close(self) -> None:
886        self.closed = True
887        self.is_open = False
888        self._inbound.put_nowait(None)
889
890    async def aclose(self) -> None:
891        self.close()
892
893    def __aiter__(self) -> AsyncIterator[str | bytes]:
894        return self
895
896    async def __anext__(self) -> str | bytes:
897        message = await self._inbound.get()
898        if message is None:
899            raise StopAsyncIteration
900        return message
901
902
903class _FakeHttpSession:
904    """ClientSession stand-in handing out one fake WebSocket per dialed url."""
905
906    def __init__(self) -> None:
907        self.dialed: list[str] = []
908        self.dial_kwargs: list[dict[str, Any]] = []
909        self.websockets: dict[str, _FakeLocalWS] = {}
910        self.requested: list[str] = []
911        self.response_body = b""
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    def request(self, _method: str, url: str, **_kwargs: Any) -> AsyncMock:
921        """Serve ``response_body`` for a proxied HTTP request."""
922        self.requested.append(url)
923        response = AsyncMock()
924        response.status = 200
925        response.headers = {"Content-Type": "image/jpeg"}
926        response.read = AsyncMock(return_value=self.response_body)
927        ctx = AsyncMock()
928        ctx.__aenter__ = AsyncMock(return_value=response)
929        ctx.__aexit__ = AsyncMock(return_value=False)
930        return ctx
931
932
933class _FakePeerConnection:
934    """PeerConnection stand-in that runs gateway-spawned pumps as asyncio tasks."""
935
936    def __init__(self) -> None:
937        self._tasks: list[asyncio.Task[None]] = []
938        self._incoming: asyncio.Queue[_FakeBidiChannel] = asyncio.Queue()
939
940    def spawn_task(self, coro: Coroutine[Any, Any, None]) -> None:
941        self._tasks.append(asyncio.ensure_future(coro))
942
943    def offer_channel(self, channel: _FakeBidiChannel) -> None:
944        """Offer a data channel as if the browser had opened it."""
945        self._incoming.put_nowait(channel)
946
947    async def incoming_data_channels(self) -> AsyncIterator[DataChannel]:
948        while True:
949            yield cast("DataChannel", await self._incoming.get())
950
951    async def aclose(self) -> None:
952        for task in self._tasks:
953            task.cancel()
954        # await cancellation so no pump task lingers past teardown
955        await asyncio.gather(*self._tasks, return_exceptions=True)
956
957
958def _register_bridge_session(
959    gateway: WebRTCGateway, session_id: str, channel: _FakeBidiChannel
960) -> WebRTCSession:
961    """Register a session backed by a fake PeerConnection and ma-api channel."""
962    session = WebRTCSession(
963        session_id=session_id,
964        pc=cast("PeerConnection", _FakePeerConnection()),
965        data_channel=cast("DataChannel", channel),
966    )
967    gateway.sessions[session_id] = session
968    return session
969
970
971def _register_routed_session(
972    gateway: WebRTCGateway, session_id: str
973) -> tuple[WebRTCSession, _FakePeerConnection]:
974    """Register a session that routes the data channels offered to its PeerConnection."""
975    pc = _FakePeerConnection()
976    session = WebRTCSession(session_id=session_id, pc=cast("PeerConnection", pc))
977    gateway.sessions[session_id] = session
978    pc.spawn_task(gateway._accept_channels(session))
979    return session, pc
980
981
982async def _wait_for(predicate: Callable[[], bool], timeout: float = 15.0) -> None:
983    """Poll ``predicate`` until it is true or the timeout elapses."""
984    loop = asyncio.get_running_loop()
985    deadline = loop.time() + timeout
986    while loop.time() < deadline:
987        if predicate():
988            return
989        await asyncio.sleep(0.02)
990    raise AssertionError("condition not met within timeout")
991
992
993async def test_handle_offer_answers_with_pinned_fingerprint(cert_pems: tuple[str, str]) -> None:
994    """The answer SDP carries the DTLS fingerprint of the configured certificate (pinning)."""
995    cert_pem, key_pem = cert_pems
996    gateway = WebRTCGateway(
997        http_session=Mock(),
998        remote_id="TEST-REMOTE-ID",
999        cert_pem=cert_pem,
1000        key_pem=key_pem,
1001    )
1002    signaling = _FakeSignaling()
1003    gateway._signaling_ws = cast("aiohttp.ClientWebSocketResponse", signaling)
1004    offerer = PeerConnection(RTCConfiguration())
1005    session_id = "fingerprint-session"
1006    try:
1007        await offerer.create_data_channel("ma-api")
1008        offer = await offerer.create_offer()
1009        with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
1010            await gateway._create_session(session_id)
1011            await asyncio.wait_for(
1012                gateway._handle_offer(session_id, {"sdp": offer.sdp, "type": "offer"}),
1013                timeout=15,
1014            )
1015
1016        assert len(signaling.answers) == 1
1017        answer = signaling.answers[0]
1018        assert answer["sessionId"] == session_id
1019        assert answer["data"]["type"] == "answer"
1020        fingerprint_line = next(
1021            line for line in answer["data"]["sdp"].splitlines() if line.startswith("a=fingerprint:")
1022        )
1023        assert _sha256_fingerprint(cert_pem) in fingerprint_line
1024    finally:
1025        await gateway._close_session(session_id)
1026        await offerer.aclose()
1027
1028
1029async def test_ma_api_channel_bridges_to_local_ws(cert_pems: tuple[str, str]) -> None:
1030    """Messages flow both ways across the ma-api data channel and the local WebSocket."""
1031    cert_pem, key_pem = cert_pems
1032    fake_ws = _FakeLocalWS()
1033    http_session = Mock()
1034    http_session.ws_connect = AsyncMock(
1035        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1036    )
1037    gateway = WebRTCGateway(
1038        http_session=http_session,
1039        remote_id="TEST-REMOTE-ID",
1040        cert_pem=cert_pem,
1041        key_pem=key_pem,
1042    )
1043    channel = _FakeBidiChannel()
1044    session = _register_bridge_session(gateway, "bridge-session", channel)
1045    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1046    try:
1047        await _wait_for(lambda: session.local_ws is not None)
1048
1049        # browser -> local WebSocket
1050        channel.feed("from browser")
1051        await _wait_for(lambda: fake_ws.sent == ["from browser"])
1052
1053        # local WebSocket -> browser
1054        fake_ws.feed_text("from local")
1055        await _wait_for(lambda: channel.sent == ["from local"])
1056    finally:
1057        channel.close()
1058        await asyncio.wait_for(bridge, timeout=5)
1059        await _wait_for(lambda: "bridge-session" not in gateway.sessions)
1060
1061
1062def test_build_ice_servers_maps_dicts() -> None:
1063    """ICE server dicts map to one IceServer per url, preserving TURN credentials."""
1064    gateway = WebRTCGateway(
1065        http_session=Mock(),
1066        remote_id="TEST-REMOTE-ID",
1067        cert_pem="cert",
1068        key_pem="key",
1069    )
1070    servers: list[dict[str, Any]] = [
1071        {"urls": "stun:stun.example.com:3478"},
1072        {"urls": "turn:turn.example.com:3478", "username": "user", "credential": "pass"},
1073        {
1074            "urls": ["stun:a.example.com:3478", "turn:b.example.com:3478"],
1075            "username": "u2",
1076            "credential": "c2",
1077        },
1078    ]
1079
1080    result = gateway._build_ice_servers(servers)
1081
1082    assert all(isinstance(server, IceServer) for server in result)
1083    # the two-url entry fans out, so 1 + 1 + 2 = 4 IceServers
1084    assert len(result) == 4
1085    assert result[0] == IceServer(url="stun:stun.example.com:3478")
1086    assert result[1] == IceServer(
1087        url="turn:turn.example.com:3478", username="user", credential="pass"
1088    )
1089    # to_url() inlines the TURN credentials for libdatachannel
1090    assert result[1].to_url() == "turn:user:[email protected]:3478"
1091    # list urls share the entry's credentials
1092    assert result[2] == IceServer(url="stun:a.example.com:3478", username="u2", credential="c2")
1093    assert result[3] == IceServer(url="turn:b.example.com:3478", username="u2", credential="c2")
1094
1095
1096async def test_session_closes_when_ma_api_channel_closes(cert_pems: tuple[str, str]) -> None:
1097    """Closing the browser ma-api channel tears down the whole gateway session."""
1098    cert_pem, key_pem = cert_pems
1099    fake_ws = _FakeLocalWS()
1100    http_session = Mock()
1101    http_session.ws_connect = AsyncMock(
1102        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1103    )
1104    gateway = WebRTCGateway(
1105        http_session=http_session,
1106        remote_id="TEST-REMOTE-ID",
1107        cert_pem=cert_pem,
1108        key_pem=key_pem,
1109    )
1110    channel = _FakeBidiChannel()
1111    session = _register_bridge_session(gateway, "channel-close-session", channel)
1112    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1113    await _wait_for(lambda: session.local_ws is not None)
1114
1115    # the browser closes the ma-api channel -> the whole session is torn down
1116    channel.close()
1117
1118    await _wait_for(lambda: "channel-close-session" not in gateway.sessions)
1119    assert "channel-close-session" not in gateway.sessions
1120    await asyncio.wait_for(bridge, timeout=5)
1121
1122
1123# ---- channel routing -------------------------------------------------------
1124
1125LOCAL_WS_URL = "ws://127.0.0.1:8095/ws"
1126SENDSPIN_URL = "ws://127.0.0.1:8927/sendspin"
1127
1128
1129def _routing_gateway(
1130    cert_pems: tuple[str, str],
1131    http_session: _FakeHttpSession,
1132    local_ws_url: str = LOCAL_WS_URL,
1133    set_sendspin_player_callback: Callable[[str, str], None] | None = None,
1134) -> WebRTCGateway:
1135    """Create a gateway whose local WebSockets are all served by the fake HTTP session."""
1136    cert_pem, key_pem = cert_pems
1137    return WebRTCGateway(
1138        http_session=cast("aiohttp.ClientSession", http_session),
1139        remote_id="TEST-REMOTE-ID",
1140        cert_pem=cert_pem,
1141        key_pem=key_pem,
1142        local_ws_url=local_ws_url,
1143        sendspin_url=SENDSPIN_URL,
1144        set_sendspin_player_callback=set_sendspin_player_callback,
1145    )
1146
1147
1148async def test_sendspin_channel_bridges_to_the_sendspin_server(cert_pems: tuple[str, str]) -> None:
1149    """A sendspin channel reaches the internal sendspin server, web player id and all."""
1150    http_session = _FakeHttpSession()
1151    announced_players: list[tuple[str, str]] = []
1152    gateway = _routing_gateway(
1153        cert_pems,
1154        http_session,
1155        set_sendspin_player_callback=lambda session_id, player_id: announced_players.append(
1156            (session_id, player_id)
1157        ),
1158    )
1159    session, pc = _register_routed_session(gateway, "sendspin-session")
1160    channel = _FakeBidiChannel(label="sendspin")
1161    pc.offer_channel(channel)
1162    try:
1163        await _wait_for(lambda: SENDSPIN_URL in http_session.websockets)
1164        local_ws = http_session.websockets[SENDSPIN_URL]
1165
1166        # the first message announces the web player, and is forwarded verbatim
1167        auth = json.dumps({"type": "auth", "token": "t", "client_id": "web-player-1"})
1168        channel.feed(auth)
1169        await _wait_for(lambda: local_ws.sent == [auth])
1170        assert announced_players == [("sendspin-session", "web-player-1")]
1171        assert session.sendspin_player_id == "web-player-1"
1172
1173        # audio keeps flowing in both directions, text and binary alike
1174        channel.feed(b"\x01\x02")
1175        await _wait_for(lambda: local_ws.sent == [auth, b"\x01\x02"])
1176        local_ws.feed_text('{"type":"hello"}')
1177        local_ws.feed_bytes(b"\x03\x04")
1178        await _wait_for(lambda: channel.sent == ['{"type":"hello"}', b"\x03\x04"])
1179    finally:
1180        await gateway._close_session("sendspin-session")
1181
1182
1183@pytest.mark.parametrize(
1184    ("local_ws_url", "expected_url"),
1185    [
1186        ("ws://127.0.0.1:8095/ws", "ws://127.0.0.1:8095/live_announcement"),
1187        # an https webserver is still dialed on its bind address, which no cert covers
1188        ("wss://127.0.0.1:8095/ws", "wss://127.0.0.1:8095/live_announcement"),
1189    ],
1190)
1191async def test_live_announcement_channel_bridges_to_the_webserver(
1192    cert_pems: tuple[str, str], local_ws_url: str, expected_url: str
1193) -> None:
1194    """A live announcement channel reaches the webserver route that takes the audio."""
1195    http_session = _FakeHttpSession()
1196    gateway = _routing_gateway(cert_pems, http_session, local_ws_url=local_ws_url)
1197    session, pc = _register_routed_session(gateway, "announce-session")
1198    channel = _FakeBidiChannel(label="live_announcement")
1199    pc.offer_channel(channel)
1200    try:
1201        await _wait_for(lambda: expected_url in http_session.websockets)
1202        local_ws = http_session.websockets[expected_url]
1203        assert http_session.dial_kwargs == [{"ssl": False}]
1204
1205        # the client authenticates on the route itself, so its handshake passes through
1206        handshake = [
1207            json.dumps({"type": "auth", "token": "t"}),
1208            json.dumps({"type": "start", "player_id": "player1", "sample_rate": 16000}),
1209        ]
1210        for message in handshake:
1211            channel.feed(message)
1212        await _wait_for(lambda: local_ws.sent == handshake)
1213        # the sendspin snoop belongs to the sendspin bridge only
1214        assert session.sendspin_player_id is None
1215
1216        # spoken audio goes up, the route's replies come back
1217        channel.feed(b"\x00\x01")
1218        await _wait_for(lambda: local_ws.sent == [*handshake, b"\x00\x01"])
1219        local_ws.feed_text('{"type":"started"}')
1220        await _wait_for(lambda: channel.sent == ['{"type":"started"}'])
1221    finally:
1222        await gateway._close_session("announce-session")
1223
1224
1225@pytest.mark.parametrize("closed_by", ["browser", "local"])
1226async def test_closing_a_bridged_channel_leaves_the_api_session_up(
1227    cert_pems: tuple[str, str], closed_by: str
1228) -> None:
1229    """Losing a bridged WebSocket tears down that bridge only, never the API session."""
1230    http_session = _FakeHttpSession()
1231    gateway = _routing_gateway(cert_pems, http_session)
1232    session, pc = _register_routed_session(gateway, "mixed-session")
1233    api_channel = _FakeBidiChannel()
1234    sendspin_channel = _FakeBidiChannel(label="sendspin")
1235    pc.offer_channel(api_channel)
1236    pc.offer_channel(sendspin_channel)
1237    try:
1238        await _wait_for(
1239            lambda: session.local_ws is not None and SENDSPIN_URL in http_session.dialed
1240        )
1241        sendspin_ws = http_session.websockets[SENDSPIN_URL]
1242
1243        if closed_by == "browser":
1244            sendspin_channel.close()
1245        else:
1246            await sendspin_ws.close()
1247        await _wait_for(lambda: not session.channels)
1248
1249        assert sendspin_ws.closed is True
1250        assert sendspin_channel.closed is True
1251        # the API session is untouched: still registered, still bridged, channel still open
1252        assert "mixed-session" in gateway.sessions
1253        assert session.local_ws is not None
1254        assert cast("_FakeLocalWS", session.local_ws).closed is False
1255        assert api_channel.closed is False
1256    finally:
1257        await gateway._close_session("mixed-session")
1258
1259
1260async def test_the_first_channel_is_the_api_channel_whatever_its_label(
1261    cert_pems: tuple[str, str],
1262) -> None:
1263    """Clients may label their API channel freely, so the first channel bridges to the API."""
1264    http_session = _FakeHttpSession()
1265    gateway = _routing_gateway(cert_pems, http_session)
1266    session, pc = _register_routed_session(gateway, "labelled-session")
1267    channel = _FakeBidiChannel(label="ma-api-v2")
1268    pc.offer_channel(channel)
1269    try:
1270        await _wait_for(lambda: session.local_ws is not None)
1271        assert session.data_channel is cast("DataChannel", channel)
1272        assert http_session.dialed == [f"{LOCAL_WS_URL}?webrtc_session_id=labelled-session"]
1273    finally:
1274        await gateway._close_session("labelled-session")
1275
1276
1277async def test_unknown_channel_label_cannot_replace_the_api_channel(
1278    cert_pems: tuple[str, str],
1279) -> None:
1280    """A channel this server has no route for is refused instead of hijacking the session."""
1281    http_session = _FakeHttpSession()
1282    gateway = _routing_gateway(cert_pems, http_session)
1283    session, pc = _register_routed_session(gateway, "unknown-session")
1284    api_channel = _FakeBidiChannel()
1285    unknown_channel = _FakeBidiChannel(label="channel-from-the-future")
1286    pc.offer_channel(api_channel)
1287    try:
1288        await _wait_for(lambda: session.local_ws is not None)
1289        pc.offer_channel(unknown_channel)
1290        await _wait_for(lambda: unknown_channel.closed)
1291
1292        assert session.data_channel is cast("DataChannel", api_channel)
1293        assert "unknown-session" in gateway.sessions
1294        # only the API channel was ever bridged
1295        assert http_session.dialed == [f"{LOCAL_WS_URL}?webrtc_session_id=unknown-session"]
1296    finally:
1297        await gateway._close_session("unknown-session")
1298
1299
1300def _proxy_request(request_id: str, path: str) -> str:
1301    """Build the http-proxy-request message a client sends for a proxied path."""
1302    return json.dumps(
1303        {"type": "http-proxy-request", "id": request_id, "method": "GET", "path": path}
1304    )
1305
1306
1307async def test_http_proxy_channel_answers_on_its_own_channel(cert_pems: tuple[str, str]) -> None:
1308    """A proxied request on the http proxy channel is answered there, not on the API channel."""
1309    http_session = _FakeHttpSession()
1310    http_session.response_body = b"\xff\xd8jpeg-bytes"
1311    gateway = _routing_gateway(cert_pems, http_session)
1312    session, pc = _register_routed_session(gateway, "proxy-session")
1313    api_channel = _FakeBidiChannel()
1314    proxy_channel = _FakeBidiChannel(label="http_proxy")
1315    pc.offer_channel(api_channel)
1316    pc.offer_channel(proxy_channel)
1317    try:
1318        await _wait_for(lambda: session.local_ws is not None)
1319
1320        proxy_channel.feed(_proxy_request("img-1", "/imageproxy/abc"))
1321        await _wait_for(lambda: bool(proxy_channel.sent))
1322
1323        assert http_session.requested == ["http://127.0.0.1:8095/imageproxy/abc"]
1324        response = json.loads(cast("str", proxy_channel.sent[0]))
1325        assert response["type"] == "http-proxy-response"
1326        assert response["id"] == "img-1"
1327        assert response["status"] == 200
1328        assert bytes.fromhex(response["body"]) == b"\xff\xd8jpeg-bytes"
1329        # the image never touches the API channel, nor the local API WebSocket
1330        assert api_channel.sent == []
1331        assert cast("_FakeLocalWS", session.local_ws).sent == []
1332    finally:
1333        await gateway._close_session("proxy-session")
1334
1335
1336async def test_http_proxy_request_on_the_api_channel_is_still_answered(
1337    cert_pems: tuple[str, str],
1338) -> None:
1339    """Clients that predate the http proxy channel keep proxying over the API channel."""
1340    http_session = _FakeHttpSession()
1341    http_session.response_body = b"legacy-body"
1342    gateway = _routing_gateway(cert_pems, http_session)
1343    session, pc = _register_routed_session(gateway, "legacy-session")
1344    api_channel = _FakeBidiChannel()
1345    pc.offer_channel(api_channel)
1346    try:
1347        await _wait_for(lambda: session.local_ws is not None)
1348
1349        api_channel.feed(_proxy_request("img-2", "/imageproxy/def"))
1350        await _wait_for(lambda: bool(api_channel.sent))
1351
1352        assert http_session.requested == ["http://127.0.0.1:8095/imageproxy/def"]
1353        response = json.loads(cast("str", api_channel.sent[0]))
1354        assert response["type"] == "http-proxy-response"
1355        assert response["id"] == "img-2"
1356        assert bytes.fromhex(response["body"]) == b"legacy-body"
1357        # the proxy request is served here, never forwarded to the local API WebSocket
1358        assert cast("_FakeLocalWS", session.local_ws).sent == []
1359    finally:
1360        await gateway._close_session("legacy-session")
1361
1362
1363async def test_http_proxy_channel_reports_a_failed_fetch_on_its_own_channel(
1364    cert_pems: tuple[str, str],
1365) -> None:
1366    """A fetch that raises still answers the client, on the channel it asked over."""
1367    http_session = _FakeHttpSession()
1368    http_session.request = Mock(side_effect=RuntimeError("boom"))  # type: ignore[method-assign]
1369    gateway = _routing_gateway(cert_pems, http_session)
1370    session, pc = _register_routed_session(gateway, "proxy-error-session")
1371    api_channel = _FakeBidiChannel()
1372    proxy_channel = _FakeBidiChannel(label="http_proxy")
1373    pc.offer_channel(api_channel)
1374    pc.offer_channel(proxy_channel)
1375    try:
1376        await _wait_for(lambda: session.local_ws is not None)
1377
1378        proxy_channel.feed(_proxy_request("img-3", "/imageproxy/boom"))
1379        await _wait_for(lambda: bool(proxy_channel.sent))
1380
1381        response = json.loads(cast("str", proxy_channel.sent[0]))
1382        assert response["id"] == "img-3"
1383        assert response["status"] == 500
1384        assert api_channel.sent == []
1385    finally:
1386        await gateway._close_session("proxy-error-session")
1387
1388
1389@pytest.mark.parametrize(
1390    "junk",
1391    [
1392        "not json at all",
1393        json.dumps(["not", "a", "dict"]),
1394        json.dumps({"type": "something-else"}),
1395        b"\x00\x01binary",
1396    ],
1397)
1398async def test_http_proxy_channel_survives_junk(
1399    cert_pems: tuple[str, str], junk: str | bytes
1400) -> None:
1401    """Anything that is not a proxy request is ignored without killing the channel."""
1402    http_session = _FakeHttpSession()
1403    http_session.response_body = b"still-here"
1404    gateway = _routing_gateway(cert_pems, http_session)
1405    session, pc = _register_routed_session(gateway, "proxy-junk-session")
1406    proxy_channel = _FakeBidiChannel(label="http_proxy")
1407    pc.offer_channel(proxy_channel)
1408    try:
1409        await _wait_for(lambda: "http_proxy" in session.channels)
1410
1411        proxy_channel.feed(junk)
1412        proxy_channel.feed(_proxy_request("img-4", "/imageproxy/ok"))
1413        await _wait_for(lambda: bool(proxy_channel.sent))
1414
1415        response = json.loads(cast("str", proxy_channel.sent[0]))
1416        assert response["id"] == "img-4"
1417        assert bytes.fromhex(response["body"]) == b"still-here"
1418    finally:
1419        await gateway._close_session("proxy-junk-session")
1420
1421
1422async def test_closing_the_http_proxy_channel_leaves_the_api_session_up(
1423    cert_pems: tuple[str, str],
1424) -> None:
1425    """Losing the http proxy channel tears down that channel only, never the API session."""
1426    http_session = _FakeHttpSession()
1427    gateway = _routing_gateway(cert_pems, http_session)
1428    session, pc = _register_routed_session(gateway, "proxy-close-session")
1429    api_channel = _FakeBidiChannel()
1430    proxy_channel = _FakeBidiChannel(label="http_proxy")
1431    pc.offer_channel(api_channel)
1432    pc.offer_channel(proxy_channel)
1433    try:
1434        await _wait_for(lambda: "http_proxy" in session.channels)
1435
1436        proxy_channel.close()
1437        await _wait_for(lambda: not session.channels)
1438
1439        assert "proxy-close-session" in gateway.sessions
1440        assert session.local_ws is not None
1441        assert api_channel.closed is False
1442    finally:
1443        await gateway._close_session("proxy-close-session")
1444
1445
1446async def test_a_second_http_proxy_channel_is_refused(cert_pems: tuple[str, str]) -> None:
1447    """A duplicate label is refused so the running handler is never left untracked."""
1448    http_session = _FakeHttpSession()
1449    gateway = _routing_gateway(cert_pems, http_session)
1450    session, pc = _register_routed_session(gateway, "duplicate-session")
1451    first = _FakeBidiChannel(label="http_proxy")
1452    second = _FakeBidiChannel(label="http_proxy")
1453    pc.offer_channel(first)
1454    try:
1455        await _wait_for(lambda: "http_proxy" in session.channels)
1456        pc.offer_channel(second)
1457        await _wait_for(lambda: second.closed)
1458
1459        assert first.closed is False
1460        assert session.channels["http_proxy"].channel is cast("DataChannel", first)
1461    finally:
1462        await gateway._close_session("duplicate-session")
1463
1464
1465class _FakeDataChannel:
1466    """Data channel stand-in that captures outbound messages for proxy tests."""
1467
1468    def __init__(self) -> None:
1469        self.is_open = True
1470        self.sent: list[str] = []
1471
1472    async def send(self, data: str) -> None:
1473        self.sent.append(data)
1474
1475
1476def _proxy_gateway(cert_pems: tuple[str, str]) -> WebRTCGateway:
1477    cert_pem, key_pem = cert_pems
1478    return WebRTCGateway(
1479        http_session=Mock(),
1480        remote_id="TEST-REMOTE-ID",
1481        cert_pem=cert_pem,
1482        key_pem=key_pem,
1483    )
1484
1485
1486def _reassemble_chunks(frames: list[str]) -> str:
1487    """Reassemble __chunk__ frames back into the original text (base64 -> bytes -> utf-8)."""
1488    parsed = [json.loads(f) for f in frames]
1489    assert all(f["type"] == "__chunk__" for f in parsed)
1490    assert len({f["id"] for f in parsed}) == 1
1491    count = parsed[0]["count"]
1492    parts: list[bytes] = [b""] * count
1493    for frame in parsed:
1494        parts[frame["seq"]] = base64.b64decode(frame["b64"])
1495    return b"".join(parts).decode()
1496
1497
1498async def test_http_proxy_response_small_body_single_message(
1499    cert_pems: tuple[str, str],
1500) -> None:
1501    """A body within the chunk size is sent as one http-proxy-response message."""
1502    gateway = _proxy_gateway(cert_pems)
1503    channel = _FakeDataChannel()
1504    body = b"\x00\x01\x02small-body"
1505
1506    await gateway._send_http_proxy_response(
1507        cast("DataChannel", channel), "req-small", 200, {"X-Test": "y"}, body
1508    )
1509
1510    assert len(channel.sent) == 1
1511    msg = json.loads(channel.sent[0])
1512    assert msg["type"] == "http-proxy-response"
1513    assert msg["id"] == "req-small"
1514    assert msg["status"] == 200
1515    assert msg["headers"] == {"X-Test": "y"}
1516    assert bytes.fromhex(msg["body"]) == body
1517
1518
1519async def test_http_proxy_response_large_body_chunked(cert_pems: tuple[str, str]) -> None:
1520    """A large HTTP-proxy response is split into base64 chunk frames the client reassembles."""
1521    gateway = _proxy_gateway(cert_pems)
1522    channel = _FakeDataChannel()
1523    # big body -> big JSON message
1524    body = bytes(range(256)) * ((DATA_CHANNEL_CHUNK_SIZE * 5) // 512)
1525
1526    await gateway._send_http_proxy_response(cast("DataChannel", channel), "req-big", 200, {}, body)
1527
1528    assert len(channel.sent) > 1
1529    assert all(json.loads(m)["type"] == "__chunk__" for m in channel.sent)
1530    # every serialized frame must stay under the negotiated 256 KiB data-channel limit
1531    assert all(len(m.encode()) < 256 * 1024 for m in channel.sent)
1532
1533    reassembled = json.loads(_reassemble_chunks(channel.sent))
1534    assert reassembled["type"] == "http-proxy-response"
1535    assert reassembled["id"] == "req-big"
1536    assert reassembled["status"] == 200
1537    assert bytes.fromhex(reassembled["body"]) == body
1538
1539
1540async def test_send_chunked_small_message_passthrough(cert_pems: tuple[str, str]) -> None:
1541    """A message within the limit is sent verbatim, not chunked."""
1542    gateway = _proxy_gateway(cert_pems)
1543    channel = _FakeDataChannel()
1544    await gateway._send_chunked(cast("DataChannel", channel), '{"event":"player_updated"}')
1545    assert channel.sent == ['{"event":"player_updated"}']
1546
1547
1548async def test_send_chunked_large_message_chunked(cert_pems: tuple[str, str]) -> None:
1549    """A large message is chunked and reassembles byte-identically (multibyte-safe)."""
1550    gateway = _proxy_gateway(cert_pems)
1551    channel = _FakeDataChannel()
1552    # multibyte payload so chunk boundaries fall mid-character, exercising the byte-level split
1553    text = '{"data":"' + "音楽" * DATA_CHANNEL_CHUNK_SIZE + '"}'
1554
1555    await gateway._send_chunked(cast("DataChannel", channel), text)
1556
1557    assert len(channel.sent) > 1
1558    assert all(len(m.encode()) < 256 * 1024 for m in channel.sent)
1559    assert _reassemble_chunks(channel.sent) == text
1560