/
/
/
1"""Cross-cutting config security tests."""
2
3from __future__ import annotations
4
5import contextlib
6import logging
7from typing import Any
8
9import pytest
10from fastmcp import Client, FastMCP
11
12from music_assistant.providers.fastmcp_server.middleware import TagFilterMiddleware
13from music_assistant.providers.fastmcp_server.server import build_tag_lookup
14from music_assistant.providers.fastmcp_server.tags import Tag
15from music_assistant.providers.fastmcp_server.tools.config import build_config_server
16
17
18async def test_off_by_default_hides_all_config_tools(mounted_config_off: Any) -> None:
19 """Tag.CONFIG_* all disabled -> zero config tools visible via list_tools()."""
20 async with Client(mounted_config_off) as client:
21 tools = await client.list_tools()
22 leaked = [t.name for t in tools if t.name.startswith("config_")]
23 assert leaked == [], f"config tools leaked: {leaked}"
24
25
26async def test_enabling_read_tag_exposes_only_read_tools(mock_mass: Any) -> None:
27 """With only Tag.CONFIG_READ allowed, only the 6 read tools are visible."""
28 mcp = FastMCP(name="t")
29 mcp.mount(build_config_server(mock_mass, require_confirmation=False), namespace="config")
30 mcp.add_middleware(TagFilterMiddleware(lambda: {Tag.CONFIG_READ.value}, build_tag_lookup(mcp)))
31 try:
32 async with Client(mcp) as client:
33 names = {t.name for t in await client.list_tools() if t.name.startswith("config_")}
34 finally:
35 close = getattr(mcp, "close", None) or getattr(mcp, "shutdown", None)
36 if callable(close):
37 with contextlib.suppress(Exception):
38 close()
39 assert names == {
40 "config_list_targets",
41 "config_get_provider",
42 "config_get_core",
43 "config_get_player",
44 "config_get_entries",
45 "config_get_dsp",
46 }
47
48
49async def test_audit_log_written_before_save(mock_config_targets: Any, caplog: Any) -> None:
50 """Audit log is written before _do_save, with key but never value."""
51 # mounted_config fixture uses the bare mock_mass, but we need config mocks
52 # so we build our own here with mock_config_targets (which has config mocks set up)
53 mcp = FastMCP(name="test")
54 mcp.mount(
55 build_config_server(mock_config_targets, require_confirmation=False), namespace="config"
56 )
57 with caplog.at_level(logging.INFO, logger="music_assistant.providers.fastmcp_server.config"):
58 try:
59 async with Client(mcp) as client:
60 await client.call_tool(
61 "config_set_provider_value",
62 {"instance_id": "yandex_music", "key": "log_level", "value": "DEBUG"},
63 )
64 finally:
65 close = getattr(mcp, "close", None) or getattr(mcp, "shutdown", None)
66 if callable(close):
67 with contextlib.suppress(Exception):
68 close()
69 audit = [r for r in caplog.records if "config_write" in r.message]
70 assert audit, "expected audit line"
71 assert "log_level" in audit[0].message
72 # The value "DEBUG" must NOT appear in the audit message (key only).
73 assert "DEBUG" not in audit[0].message
74
75
76async def test_dry_run_not_audited(mock_config_targets: Any, caplog: Any) -> None:
77 """dry_run=True skips audit logging."""
78 mcp = FastMCP(name="test")
79 mcp.mount(
80 build_config_server(mock_config_targets, require_confirmation=False), namespace="config"
81 )
82 with caplog.at_level(logging.INFO, logger="music_assistant.providers.fastmcp_server.config"):
83 try:
84 async with Client(mcp) as client:
85 await client.call_tool(
86 "config_set_provider_value",
87 {
88 "instance_id": "yandex_music",
89 "key": "log_level",
90 "value": "DEBUG",
91 "dry_run": True,
92 },
93 )
94 finally:
95 close = getattr(mcp, "close", None) or getattr(mcp, "shutdown", None)
96 if callable(close):
97 with contextlib.suppress(Exception):
98 close()
99 assert not [r for r in caplog.records if "config_write" in r.message]
100
101
102_EXPECTED_DESC = {
103 "config_get_provider": ["see also"],
104 "config_set_provider_value": ["dry_run", "config:write:secret"],
105 "config_set_core_value": ["restart"],
106 "config_get_dsp": ["dsp"],
107 "config_save_dsp": ["dspconfig"],
108 "config_trigger_provider_action": ["confirmation"],
109}
110
111
112@pytest.mark.parametrize(("name", "subs"), list(_EXPECTED_DESC.items()))
113async def test_tool_descriptions_carry_breadcrumbs(
114 mounted_config: Any, name: str, subs: list[str]
115) -> None:
116 """Each config tool's description must include the planned workflow cross-references."""
117 async with Client(mounted_config) as client:
118 tools = {t.name: t for t in await client.list_tools()}
119 assert name in tools, f"{name} not exposed"
120 desc = (tools[name].description or "").lower()
121 for s in subs:
122 assert s.lower() in desc, f"{name} desc missing {s!r}: {desc!r}"
123