music-assistant-server

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