music-assistant-server

1"""Tests for Origin allowlist computation, matching, and bridge enforcement (C1+C2)."""
2# mypy: disable-error-code="arg-type, no-untyped-def, type-arg, assignment, operator, misc"
3
4from __future__ import annotations
5
6from types import SimpleNamespace
7from typing import Any
8
9import pytest
10from aiohttp.test_utils import TestClient, TestServer
11
12from music_assistant.providers.fastmcp_server.http_bridge import (
13    build_protected_resource_metadata,
14    mount_into_mass,
15    mount_well_known,
16)
17from music_assistant.providers.fastmcp_server.origins import (
18    _is_origin_allowed,
19    _normalize_origin,
20)
21from music_assistant.providers.fastmcp_server.origins import (
22    compute_origin_allowlist as _compute_origin_allowlist,
23)
24from music_assistant.providers.fastmcp_server.origins import (
25    is_origin_allowed_for_request as _is_origin_allowed_for_request,
26)
27
28from .conftest import FakeWebserver, build_aiohttp_app
29
30
31@pytest.mark.parametrize(
32    ("raw", "expected"),
33    [
34        ("http://localhost:8095", "http://localhost:8095"),
35        ("HTTP://Localhost:8095", "http://localhost:8095"),
36        ("http://localhost:80", "http://localhost"),
37        ("https://example.com:443", "https://example.com"),
38        ("https://example.com/path", "https://example.com"),
39        ("null", "null"),
40        ("", None),
41        ("not-a-url", None),
42        ("http://", None),
43        # IPv6 literals: brackets must round-trip so the normalized form
44        # matches the allowlist entry the bridge synthesises.
45        ("http://[::1]", "http://[::1]"),
46        ("http://[::1]:8095", "http://[::1]:8095"),
47        ("HTTP://[::1]:8095", "http://[::1]:8095"),
48        ("http://[2001:db8::1]:80", "http://[2001:db8::1]"),
49    ],
50)
51def test_normalize_origin(raw: str, expected: str | None) -> None:
52    """Origin strings collapse to ``scheme://host[:port]`` lowercased, default ports stripped."""
53    assert _normalize_origin(raw) == expected
54
55
56def test_compute_allowlist_ipv6_publish_ip() -> None:
57    """An IPv6 publish_ip is bracketed in the allowlist so browsers' Origin matches."""
58    mass = SimpleNamespace(
59        webserver=SimpleNamespace(base_url="http://localhost:8095", publish_ip="::1"),
60    )
61    allow = _compute_origin_allowlist(mass)
62    assert "http://[::1]:8095" in allow
63    assert _is_origin_allowed("http://[::1]:8095", allow) is True
64    # And without an explicit port (still works because we bracket consistently).
65    assert "http://[::1]" in allow
66
67
68def _fake_mass(base_url: str = "http://localhost:8095", publish_ip: str = "127.0.0.1"):
69    return SimpleNamespace(webserver=SimpleNamespace(base_url=base_url, publish_ip=publish_ip))
70
71
72def test_compute_allowlist_default() -> None:
73    """Default allowlist contains loopbacks, base_url host, and publish_ip."""
74    allow = _compute_origin_allowlist(_fake_mass())
75    # loopbacks always there
76    assert "http://localhost" in allow
77    assert "http://127.0.0.1" in allow
78    assert "http://[::1]" in allow
79    # base_url with port
80    assert "http://localhost:8095" in allow
81    # https-twin of base_url
82    assert "https://localhost:8095" in allow
83    # publish_ip with derived port
84    assert "http://127.0.0.1:8095" in allow
85    assert "https://127.0.0.1:8095" in allow
86
87
88def test_compute_allowlist_with_extras() -> None:
89    """CSV ``extra_origins`` get normalized; bogus / empty entries silently dropped."""
90    allow = _compute_origin_allowlist(
91        _fake_mass(),
92        extra_origins_csv="https://ha.example.com, http://reverse.lan:8443 ,bogus,",
93    )
94    assert "https://ha.example.com" in allow
95    assert "http://reverse.lan:8443" in allow
96    # bogus + empty silently dropped
97    assert all(o != "" for o in allow)
98
99
100def test_compute_allowlist_with_https_base_url() -> None:
101    """When base_url uses https on the default port, no port suffix is added."""
102    allow = _compute_origin_allowlist(_fake_mass(base_url="https://mcp.example.com"))
103    assert "https://mcp.example.com" in allow
104    # publish_ip with no explicit port (https is scheme-default)
105    assert "http://127.0.0.1" in allow
106    assert "https://127.0.0.1" in allow
107
108
109@pytest.mark.parametrize(
110    ("origin", "allowed"),
111    [
112        (None, True),  # CLI / stdio-style — no Origin
113        ("http://localhost:8095", True),
114        ("http://LOCALHOST:8095", True),  # case-insensitive
115        ("http://localhost:8095/", True),  # trailing slash tolerated
116        ("http://evil.example", False),
117        ("https://localhost:8095", True),  # https-twin allowed by default
118        ("null", False),  # not in default allowlist
119    ],
120)
121def test_is_origin_allowed(origin: str | None, allowed: bool) -> None:
122    """Match Origin against the default allowlist (case-insensitive, trailing-slash tolerant)."""
123    allow = _compute_origin_allowlist(_fake_mass())
124    assert _is_origin_allowed(origin, allow) is allowed
125
126
127def test_is_origin_allowed_with_explicit_null() -> None:
128    """``Origin: null`` is accepted only when the operator opts in via ``extra_origins``."""
129    allow = _compute_origin_allowlist(_fake_mass(), extra_origins_csv="null")
130    assert "null" in allow
131    assert _is_origin_allowed("null", allow) is True
132
133
134def test_garbage_origin_rejected() -> None:
135    """Malformed Origin values fail closed with 403."""
136    allow = _compute_origin_allowlist(_fake_mass())
137    assert _is_origin_allowed("not-a-url", allow) is False
138    assert _is_origin_allowed("http://", allow) is False
139
140
141# ── HA-ingress fallback in `_is_origin_allowed_for_request` ─────────────────
142
143
144def _fake_request(
145    headers: dict[str, str] | None = None,
146    *,
147    scheme: str = "http",
148) -> Any:
149    """Build a minimal stand-in for ``aiohttp.web.Request`` for origin checks."""
150    return SimpleNamespace(
151        headers=headers or {},
152        remote="172.30.32.1",
153        scheme=scheme,
154    )
155
156
157def _install_ingress_stub(monkeypatch: pytest.MonkeyPatch, *, is_ingress: bool) -> None:
158    """Inject ``is_request_from_ingress`` lazily into ``sys.modules``."""
159    import sys  # noqa: PLC0415
160    import types  # noqa: PLC0415
161
162    pkg = types.ModuleType("music_assistant")
163    pkg.__path__ = []
164    controllers = types.ModuleType("music_assistant.controllers")
165    controllers.__path__ = []
166    webserver_pkg = types.ModuleType("music_assistant.controllers.webserver")
167    webserver_pkg.__path__ = []
168    helpers_pkg = types.ModuleType("music_assistant.controllers.webserver.helpers")
169    helpers_pkg.__path__ = []
170    auth_mod = types.ModuleType("music_assistant.controllers.webserver.helpers.auth_middleware")
171    auth_mod.is_request_from_ingress = lambda _req: is_ingress  # type: ignore[attr-defined]
172
173    monkeypatch.setitem(sys.modules, "music_assistant", pkg)
174    monkeypatch.setitem(sys.modules, "music_assistant.controllers", controllers)
175    monkeypatch.setitem(sys.modules, "music_assistant.controllers.webserver", webserver_pkg)
176    monkeypatch.setitem(sys.modules, "music_assistant.controllers.webserver.helpers", helpers_pkg)
177    monkeypatch.setitem(
178        sys.modules,
179        "music_assistant.controllers.webserver.helpers.auth_middleware",
180        auth_mod,
181    )
182
183
184def test_request_origin_allowed_via_allowlist(monkeypatch: pytest.MonkeyPatch) -> None:
185    """Falls through to the legacy allowlist when the basic check accepts."""
186    _install_ingress_stub(monkeypatch, is_ingress=False)
187    allow = _compute_origin_allowlist(_fake_mass())
188    req = _fake_request({"Origin": "http://localhost:8095"})
189    assert _is_origin_allowed_for_request(req, allow) is True
190
191
192def test_request_origin_accepts_ingress_forwarded_host(
193    monkeypatch: pytest.MonkeyPatch,
194) -> None:
195    """
196    Ingress request whose Origin matches X-Forwarded-Host is accepted.
197
198    Reproduces the HA add-on case where the user opens the Connect Wizard at
199    ``https://<ha>/<slug>/…`` and the browser sends ``Origin: https://<ha>``,
200    which is never on the static allowlist.
201    """
202    _install_ingress_stub(monkeypatch, is_ingress=True)
203    allow = _compute_origin_allowlist(_fake_mass())
204    req = _fake_request(
205        {
206            "Origin": "https://ha.example.com",
207            "X-Forwarded-Host": "ha.example.com",
208            "X-Forwarded-Proto": "https",
209        }
210    )
211    assert _is_origin_allowed_for_request(req, allow) is True
212
213
214def test_request_origin_rejects_forwarded_host_without_ingress(
215    monkeypatch: pytest.MonkeyPatch,
216) -> None:
217    """Without the trusted-ingress signal, ``X-Forwarded-Host`` is not trusted."""
218    _install_ingress_stub(monkeypatch, is_ingress=False)
219    allow = _compute_origin_allowlist(_fake_mass())
220    req = _fake_request(
221        {
222            "Origin": "https://attacker.example",
223            "X-Forwarded-Host": "attacker.example",
224            "X-Forwarded-Proto": "https",
225        }
226    )
227    assert _is_origin_allowed_for_request(req, allow) is False
228
229
230def test_request_origin_rejects_mismatched_forwarded_host(
231    monkeypatch: pytest.MonkeyPatch,
232) -> None:
233    """Origin must equal the forwarded host — a mismatch is rejected even via ingress."""
234    _install_ingress_stub(monkeypatch, is_ingress=True)
235    allow = _compute_origin_allowlist(_fake_mass())
236    req = _fake_request(
237        {
238            "Origin": "https://attacker.example",
239            "X-Forwarded-Host": "ha.example.com",
240            "X-Forwarded-Proto": "https",
241        }
242    )
243    assert _is_origin_allowed_for_request(req, allow) is False
244
245
246def test_request_origin_no_forward_header_rejected(
247    monkeypatch: pytest.MonkeyPatch,
248) -> None:
249    """Ingress without ``X-Forwarded-Host`` falls through to the strict allowlist."""
250    _install_ingress_stub(monkeypatch, is_ingress=True)
251    allow = _compute_origin_allowlist(_fake_mass())
252    req = _fake_request({"Origin": "https://ha.example.com"})
253    assert _is_origin_allowed_for_request(req, allow) is False
254
255
256def test_request_origin_missing_proto_falls_back_to_transport_scheme(
257    monkeypatch: pytest.MonkeyPatch,
258) -> None:
259    """
260    When ``X-Forwarded-Proto`` is absent, the aiohttp ``scheme`` fills in.
261
262    Inside an HA add-on the transport scheme is plain ``http`` because the
263    container is reached over the docker network. A proxy that forwards the
264    host header but omits the proto header should still validate against the
265    actual transport's scheme rather than guess ``https``.
266    """
267    _install_ingress_stub(monkeypatch, is_ingress=True)
268    allow = _compute_origin_allowlist(_fake_mass())
269    req = _fake_request(
270        {
271            "Origin": "http://ha.local:8123",
272            "X-Forwarded-Host": "ha.local:8123",
273        },
274        scheme="http",
275    )
276    assert _is_origin_allowed_for_request(req, allow) is True
277
278
279def test_request_origin_accepts_case_insensitive_origin(
280    monkeypatch: pytest.MonkeyPatch,
281) -> None:
282    """``Origin`` matching is case-insensitive on host (RFC 3986)."""
283    _install_ingress_stub(monkeypatch, is_ingress=True)
284    allow = _compute_origin_allowlist(_fake_mass())
285    req = _fake_request(
286        {
287            "Origin": "HTTPS://Ha.Example.COM",
288            "X-Forwarded-Host": "ha.example.com",
289            "X-Forwarded-Proto": "https",
290        }
291    )
292    assert _is_origin_allowed_for_request(req, allow) is True
293
294
295def test_request_origin_handles_missing_ma_module() -> None:
296    """When ``music_assistant`` is unavailable, never auto-accept (fail closed)."""
297    allow = _compute_origin_allowlist(_fake_mass())
298    req = _fake_request(
299        {
300            "Origin": "https://ha.example.com",
301            "X-Forwarded-Host": "ha.example.com",
302        }
303    )
304    # No stub installed; the import inside the helper raises, caught → reject.
305    assert _is_origin_allowed_for_request(req, allow) is False
306
307
308# ── End-to-end bridge enforcement (C2) ──────────────────────────────────────
309
310
311class _FakeMcp:
312    """Stand-in for FastMCP exposing an ASGI app via ``http_app(...)``."""
313
314    def __init__(self, asgi_app: Any) -> None:
315        self._app = asgi_app
316
317    def http_app(self, transport: str = "streamable-http", path: str = "/mcp") -> Any:
318        return self._app
319
320
321async def _echo_asgi(scope: dict, receive: Any, send: Any) -> None:
322    """Minimal ASGI app: handles lifespan events + returns 200 'OK' on http."""
323    if scope.get("type") == "lifespan":
324        while True:
325            msg = await receive()
326            if msg["type"] == "lifespan.startup":
327                await send({"type": "lifespan.startup.complete"})
328            elif msg["type"] == "lifespan.shutdown":
329                await send({"type": "lifespan.shutdown.complete"})
330                return
331    await send({"type": "http.response.start", "status": 200, "headers": []})
332    await send({"type": "http.response.body", "body": b"OK"})
333
334
335@pytest.fixture
336async def bridge_client() -> Any:
337    """Build the bridge handler against a fake MA + fake ASGI, expose via TestClient."""
338    fake_ws = FakeWebserver()
339    mass = SimpleNamespace(webserver=fake_ws)
340    mcp = _FakeMcp(_echo_asgi)
341    await mount_into_mass(mass, mcp, mount_path="/mcp/v1", extra_origins_csv="")
342
343    async with TestClient(TestServer(build_aiohttp_app(fake_ws))) as client:
344        yield client
345
346
347async def test_bridge_rejects_evil_origin(bridge_client: TestClient) -> None:
348    """A non-allow-listed Origin is blocked with 403, ASGI never invoked."""
349    resp = await bridge_client.post("/mcp/v1/", headers={"Origin": "http://evil.example"})
350    assert resp.status == 403
351
352
353async def test_bridge_allows_localhost(bridge_client: TestClient) -> None:
354    """Origin that matches base_url is forwarded to ASGI (200 from echo app)."""
355    resp = await bridge_client.post("/mcp/v1/", headers={"Origin": "http://localhost:8095"})
356    assert resp.status == 200
357    assert (await resp.read()) == b"OK"
358
359
360async def test_bridge_allows_no_origin(bridge_client: TestClient) -> None:
361    """Requests without ``Origin`` header (curl/CLI) pass through unchanged."""
362    resp = await bridge_client.post("/mcp/v1/")
363    assert resp.status == 200
364
365
366def test_build_protected_resource_metadata_minimal() -> None:
367    """RFC 9728 metadata always carries resource + authorization_servers + bearer methods."""
368    meta = build_protected_resource_metadata(
369        resource_uri="http://localhost:8095/mcp/v1",
370        authorization_servers=["http://localhost:8095"],
371    )
372    assert meta == {
373        "resource": "http://localhost:8095/mcp/v1",
374        "authorization_servers": ["http://localhost:8095"],
375        "bearer_methods_supported": ["header"],
376    }
377
378
379def test_build_protected_resource_metadata_full() -> None:
380    """Optional fields scopes_supported / resource_name appear when provided."""
381    meta = build_protected_resource_metadata(
382        resource_uri="http://localhost:8095/mcp/v1",
383        authorization_servers=["http://localhost:8095"],
384        scopes_supported=["query:library", "control:playback"],
385        resource_name="Music Assistant MCP",
386    )
387    assert meta["scopes_supported"] == ["query:library", "control:playback"]
388    assert meta["resource_name"] == "Music Assistant MCP"
389
390
391async def test_mount_well_known_with_dynamic_scopes_refreshes() -> None:
392    """
393    When scopes_supported is a callable, the body refreshes on each request.
394
395    Permission hot-swap mutates the closed-over set in MCPServerRuntime;
396    the well-known endpoint must reflect the change without a runtime rebuild.
397    """
398    fake_ws = FakeWebserver()
399    mass = SimpleNamespace(webserver=fake_ws)
400    scopes: list[str] = ["query:library"]
401    await mount_well_known(
402        mass,
403        mount_path="/mcp/v1",
404        resource_uri="http://localhost:8095/mcp/v1",
405        authorization_servers=["http://localhost:8095"],
406        scopes_supported=lambda: list(scopes),
407        resource_name="MA MCP",
408    )
409
410    async with TestClient(TestServer(build_aiohttp_app(fake_ws))) as client:
411        before = await (await client.get("/.well-known/oauth-protected-resource")).json()
412        assert before["scopes_supported"] == ["query:library"]
413        # Mutate the underlying set (simulates hot-swap of permission flags).
414        scopes.append("control:playback")
415        after = await (await client.get("/.well-known/oauth-protected-resource")).json()
416        assert "control:playback" in after["scopes_supported"]
417
418
419async def test_mount_well_known_serves_metadata() -> None:
420    """The well-known route returns the RFC 9728 JSON document for both URI forms."""
421    fake_ws = FakeWebserver()
422    mass = SimpleNamespace(webserver=fake_ws)
423    unmount = await mount_well_known(
424        mass,
425        mount_path="/mcp/v1",
426        resource_uri="http://localhost:8095/mcp/v1",
427        authorization_servers=["http://localhost:8095"],
428        scopes_supported=["query:library"],
429        resource_name="Music Assistant MCP",
430    )
431
432    paths = [r[0] for r in fake_ws.routes]
433    assert "/.well-known/oauth-protected-resource/mcp/v1" in paths
434    assert "/.well-known/oauth-protected-resource" in paths
435
436    async with TestClient(TestServer(build_aiohttp_app(fake_ws))) as client:
437        for path in paths:
438            resp = await client.get(path)
439            assert resp.status == 200
440            assert resp.headers["content-type"].startswith("application/json")
441            doc = await resp.json()
442            assert doc["resource"] == "http://localhost:8095/mcp/v1"
443            assert doc["authorization_servers"] == ["http://localhost:8095"]
444            assert doc["bearer_methods_supported"] == ["header"]
445            assert doc["scopes_supported"] == ["query:library"]
446            assert doc["resource_name"] == "Music Assistant MCP"
447
448    unmount()
449
450
451async def test_bridge_with_extra_origins() -> None:
452    """``extra_origins_csv`` widens the allowlist for reverse-proxy / HA ingress."""
453    fake_ws = FakeWebserver()
454    mass = SimpleNamespace(webserver=fake_ws)
455    mcp = _FakeMcp(_echo_asgi)
456    await mount_into_mass(
457        mass, mcp, mount_path="/mcp/v1", extra_origins_csv="https://ha.example.com"
458    )
459
460    async with TestClient(TestServer(build_aiohttp_app(fake_ws))) as client:
461        resp = await client.post("/mcp/v1/", headers={"Origin": "https://ha.example.com"})
462        assert resp.status == 200
463        # An origin not in the extras stays rejected.
464        resp = await client.post("/mcp/v1/", headers={"Origin": "http://evil.example"})
465        assert resp.status == 403
466