music-assistant-server

1"""Tests for ``provider.auth.MASTokenVerifier``."""
2
3from __future__ import annotations
4
5import base64
6import json
7import logging
8from unittest.mock import AsyncMock, MagicMock
9
10import pytest
11
12from music_assistant.providers.fastmcp_server.auth import MASTokenVerifier
13
14
15def _make_jwt(payload: dict[str, object]) -> str:
16    """
17    Forge an unsigned-but-structurally-valid JWT for audience-claim tests.
18
19    The signature isn't checked by ``MASTokenVerifier`` (verification is MA's
20    job); we only inspect the payload's ``aud`` claim.
21    """
22    header = base64.urlsafe_b64encode(b'{"alg":"none","typ":"JWT"}').rstrip(b"=").decode()
23    body = base64.urlsafe_b64encode(json.dumps(payload).encode()).rstrip(b"=").decode()
24    return f"{header}.{body}.signature"
25
26
27@pytest.mark.asyncio
28async def test_valid_token_returns_access_token(mock_mass: MagicMock, mock_user: MagicMock) -> None:
29    """A valid token yields an AccessToken bound to the canonical resource URI."""
30    mock_mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=mock_user)
31    verifier = MASTokenVerifier(
32        mock_mass,
33        base_url="http://localhost:8095",
34        public_resource_uri="http://localhost:8095/mcp/v1",
35    )
36
37    token = await verifier.verify_token("valid-token")
38
39    assert token is not None
40    assert token.client_id == "u1"
41    assert token.scopes == []
42    assert token.resource == "http://localhost:8095/mcp/v1"
43    assert token.token == "valid-token"
44
45
46@pytest.mark.asyncio
47async def test_invalid_token_returns_none(mock_mass: MagicMock) -> None:
48    """An invalid (rejected) token returns None."""
49    mock_mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=None)
50    verifier = MASTokenVerifier(mock_mass)
51    assert await verifier.verify_token("nope") is None
52
53
54@pytest.mark.asyncio
55async def test_disabled_user_returns_none(mock_mass: MagicMock, mock_user: MagicMock) -> None:
56    """A user marked disabled is rejected even if the token is valid."""
57    mock_user.enabled = False
58    mock_mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=mock_user)
59    verifier = MASTokenVerifier(mock_mass)
60    assert await verifier.verify_token("valid-but-disabled") is None
61
62
63@pytest.mark.asyncio
64async def test_authenticate_called_once(mock_mass: MagicMock, mock_user: MagicMock) -> None:
65    """We delegate exactly once per verify_token call (no retry storm)."""
66    mock_mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=mock_user)
67    verifier = MASTokenVerifier(mock_mass)
68    await verifier.verify_token("t")
69    mock_mass.webserver.auth.authenticate_with_token.assert_awaited_once_with("t")
70
71
72@pytest.mark.asyncio
73async def test_underlying_exception_swallowed(
74    mock_mass: MagicMock, caplog: pytest.LogCaptureFixture
75) -> None:
76    """
77    If MA's auth raises, we log AT ERROR LEVEL and return None — never propagate.
78
79    The log level matters: a refactor that demotes ``LOGGER.exception`` to
80    ``LOGGER.debug`` would silently strip operator visibility for auth-system
81    outages while keeping the test green. Pinning the captured record's
82    level prevents that.
83    """
84    mock_mass.webserver.auth.authenticate_with_token = AsyncMock(
85        side_effect=RuntimeError("db down")
86    )
87    verifier = MASTokenVerifier(mock_mass)
88
89    with caplog.at_level(logging.ERROR, logger="music_assistant.providers.fastmcp_server.auth"):
90        assert await verifier.verify_token("any") is None
91
92    assert any(
93        rec.levelno >= logging.ERROR and "MA token verification raised" in rec.message
94        for rec in caplog.records
95    ), f"expected an ERROR-level 'MA token verification raised' log; got {caplog.records!r}"
96
97
98# ── audience binding (C6) ────────────────────────────────────────────────────
99
100
101_RESOURCE = "http://localhost:8095/mcp/v1"
102
103
104@pytest.mark.asyncio
105async def test_legacy_token_passes_in_soft_mode(mock_mass: MagicMock, mock_user: MagicMock) -> None:
106    """Non-JWT (legacy hash) tokens have no aud; soft mode accepts them."""
107    mock_mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=mock_user)
108    verifier = MASTokenVerifier(mock_mass, public_resource_uri=_RESOURCE, enforce_audience=False)
109    assert await verifier.verify_token("legacy-hash-token") is not None
110
111
112@pytest.mark.asyncio
113async def test_legacy_token_rejected_in_strict_mode(
114    mock_mass: MagicMock, mock_user: MagicMock
115) -> None:
116    """Strict mode rejects tokens that have no audience claim at all."""
117    mock_mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=mock_user)
118    verifier = MASTokenVerifier(mock_mass, public_resource_uri=_RESOURCE, enforce_audience=True)
119    assert await verifier.verify_token("legacy-hash-token") is None
120
121
122@pytest.mark.asyncio
123async def test_jwt_with_matching_aud_accepted_in_strict_mode(
124    mock_mass: MagicMock, mock_user: MagicMock
125) -> None:
126    """A JWT carrying ``aud == public_resource_uri`` passes strict enforcement."""
127    mock_mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=mock_user)
128    verifier = MASTokenVerifier(mock_mass, public_resource_uri=_RESOURCE, enforce_audience=True)
129    token = _make_jwt({"sub": "u1", "aud": _RESOURCE})
130    assert await verifier.verify_token(token) is not None
131
132
133@pytest.mark.asyncio
134async def test_jwt_with_mismatched_aud_rejected_in_strict_mode(
135    mock_mass: MagicMock, mock_user: MagicMock
136) -> None:
137    """A JWT issued for a different audience is rejected in strict mode."""
138    mock_mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=mock_user)
139    verifier = MASTokenVerifier(mock_mass, public_resource_uri=_RESOURCE, enforce_audience=True)
140    token = _make_jwt({"sub": "u1", "aud": "http://other.example/api"})
141    assert await verifier.verify_token(token) is None
142
143
144@pytest.mark.asyncio
145async def test_jwt_with_aud_list_accepted(mock_mass: MagicMock, mock_user: MagicMock) -> None:
146    """RFC 8707 allows ``aud`` to be a list — match is membership."""
147    mock_mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=mock_user)
148    verifier = MASTokenVerifier(mock_mass, public_resource_uri=_RESOURCE, enforce_audience=True)
149    token = _make_jwt({"sub": "u1", "aud": ["http://other.example", _RESOURCE]})
150    assert await verifier.verify_token(token) is not None
151
152
153@pytest.mark.asyncio
154async def test_strict_mode_audience_mismatch_skips_authenticate(
155    mock_mass: MagicMock, mock_user: MagicMock
156) -> None:
157    """
158    A token rejected on audience must not reach ``authenticate_with_token``.
159
160    Calling MA's auth refreshes the sliding-window expiry for the token. If
161    audience was checked second (after authenticate), an attacker holding a
162    valid MA token issued for a different endpoint could keep it alive
163    indefinitely by repeatedly hitting the MCP endpoint — even though MCP
164    rejects the request. The check now runs first; MA is never consulted.
165    """
166    mock_mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=mock_user)
167    verifier = MASTokenVerifier(mock_mass, public_resource_uri=_RESOURCE, enforce_audience=True)
168    token = _make_jwt({"sub": "u1", "aud": "http://other.example/api"})
169
170    assert await verifier.verify_token(token) is None
171    mock_mass.webserver.auth.authenticate_with_token.assert_not_awaited()
172
173
174# ── _extract_jwt_audience: malformed payload paths ───────────────────────────
175
176
177def _malformed_payload_jwt(raw_payload: bytes) -> str:
178    """
179    Forge a JWT-shaped token with a non-standard payload segment.
180
181    The payload segment is base64url-encoded (no padding); skipping the JSON
182    encoding step lets us hand the decoder arbitrary bytes to exercise the
183    error paths (binascii decode failure, non-object JSON, etc.).
184    """
185    header = base64.urlsafe_b64encode(b'{"alg":"none","typ":"JWT"}').rstrip(b"=").decode()
186    body = base64.urlsafe_b64encode(raw_payload).rstrip(b"=").decode()
187    return f"{header}.{body}.sig"
188
189
190def _payload_jwt(payload: object) -> str:
191    """Forge a JWT with ``json.dumps(payload)`` as the payload (no claims object guard)."""
192    header = base64.urlsafe_b64encode(b'{"alg":"none","typ":"JWT"}').rstrip(b"=").decode()
193    body = base64.urlsafe_b64encode(json.dumps(payload).encode()).rstrip(b"=").decode()
194    return f"{header}.{body}.sig"
195
196
197@pytest.mark.parametrize(
198    ("token", "case"),
199    [
200        # Not three dot-separated segments → not a JWT at all → None.
201        ("no.dots", "fewer than 3 segments"),
202        ("a.b.c.d", "more than 3 segments"),
203        ("plain-token-with-no-dots", "no dots at all (legacy hash form)"),
204        # Three segments but the payload isn't valid base64.
205        ("hdr.!!not-base64!!.sig", "payload is not valid base64url"),
206        # Three segments, payload decodes to bytes that aren't valid UTF-8.
207        (_malformed_payload_jwt(b"\xff\xfe\xfd"), "payload is not valid UTF-8"),
208        # Three segments, payload IS JSON but not an object — must not crash.
209        (_payload_jwt([1, 2, 3]), "payload is a JSON array, not an object"),
210        (_payload_jwt("just a string"), "payload is a JSON string, not an object"),
211        # ``aud`` is present but the wrong type (int) — must not crash either.
212        (_payload_jwt({"sub": "u1", "aud": 42}), "aud claim is an int"),
213        (_payload_jwt({"sub": "u1", "aud": {"weird": "shape"}}), "aud claim is a dict"),
214    ],
215)
216def test_extract_jwt_audience_malformed_returns_none(token: str, case: str) -> None:
217    """
218    Every malformed-JWT path collapses to ``None`` — never an unhandled raise.
219
220    Verified against ``provider.auth._extract_jwt_audience`` so a refactor
221    that broadens the try/except boundary (or worse, lets the exception
222    propagate to ``verify_token``) is caught.
223    """
224    from music_assistant.providers.fastmcp_server.auth import _extract_jwt_audience  # noqa: PLC0415
225
226    assert _extract_jwt_audience(token) is None, case
227