music-assistant-server

1"""SMB filesystem provider for Music Assistant."""
2
3from __future__ import annotations
4
5import os
6import platform
7from typing import TYPE_CHECKING
8from urllib.parse import quote
9
10from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
11from music_assistant_models.enums import ConfigEntryType
12from music_assistant_models.errors import LoginFailed, SetupFailedError, UnsupportedSystemError
13
14from music_assistant.constants import CONF_PASSWORD, CONF_USERNAME, VERBOSE_LOG_LEVEL
15from music_assistant.helpers.json import SerializableType
16from music_assistant.helpers.mount import error_summary, unmount
17from music_assistant.helpers.process import check_output
18from music_assistant.helpers.util import get_ip_from_host
19from music_assistant.providers.filesystem_local import (
20    LocalFileSystemProvider,
21    ismount,
22    makedirs,
23)
24from music_assistant.providers.filesystem_local.constants import (
25    CONF_CONTENT_TYPE,
26    CONF_ENTRY_CONTENT_TYPE,
27    CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
28    CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
29    CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
30    CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
31    CONF_ENTRY_LIBRARY_SYNC_TRACKS,
32    CONF_ENTRY_MISSING_ALBUM_ARTIST,
33    CONF_ENTRY_PROPAGATE_GENRES,
34    content_type_config_entry,
35)
36
37if TYPE_CHECKING:
38    from music_assistant_models.config_entries import ProviderConfig
39    from music_assistant_models.provider import ProviderManifest
40
41    from music_assistant.mass import MusicAssistant
42    from music_assistant.models import ProviderInstanceType
43
44CONF_HOST = "host"
45CONF_SHARE = "share"
46CONF_SUBFOLDER = "subfolder"
47CONF_SMB_VERSION = "smb_version"
48CONF_CACHE_MODE = "cache_mode"
49
50# lowercase fragments that both mount tools (Linux mount.cifs and macOS mount_smbfs) emit when
51# the server rejected the credentials - only those must be reported back as an auth problem
52_AUTH_FAILURE_MARKERS = (
53    "permission denied",
54    "authentication error",
55    "nt_status_logon_failure",
56    "nt_status_access_denied",
57    "nt_status_account_disabled",
58    "nt_status_account_locked_out",
59    "nt_status_password_expired",
60    "nt_status_wrong_password",
61)
62
63
64async def setup(
65    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
66) -> ProviderInstanceType:
67    """Initialize provider(instance) with given configuration."""
68    # base_path will be the path where we're going to mount the remote share
69    base_path = f"/tmp/{config.instance_id}"  # noqa: S108
70    return SMBFileSystemProvider(mass, manifest, config, base_path)
71
72
73class SMBFileSystemProvider(LocalFileSystemProvider):
74    """
75    Implementation of an SMB File System Provider.
76
77    Basically this is just a wrapper around the regular local files provider,
78    except for the fact that it will mount a remote folder to a temporary location.
79    We went for this OS-depdendent approach because there is no solid async-compatible
80    smb library for Python (and we tried both pysmb and smbprotocol).
81    """
82
83    @property
84    def instance_name_postfix(self) -> str | None:
85        """Return a (default) instance name postfix for this provider instance."""
86        share = str(self.get_setup_value(CONF_SHARE))
87        subfolder = str(self.get_setup_value(CONF_SUBFOLDER))
88        if subfolder:
89            return subfolder
90        if share:
91            return share
92        return None
93
94    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
95        """Return Config entries to configure this provider."""
96        # connection details and content type are collected by the setup flow; surface the
97        # (immutable) content type read-only so the sync options' depends_on chains resolve
98        content_type = str(
99            self.get_setup_value(CONF_CONTENT_TYPE, CONF_ENTRY_CONTENT_TYPE.default_value)
100        )
101        return (
102            content_type_config_entry(content_type),
103            ConfigEntry(
104                key=CONF_CACHE_MODE,
105                type=ConfigEntryType.STRING,
106                required=False,
107                advanced=True,
108                default_value="loose",
109                options=[
110                    ConfigValueOption("strict"),
111                    ConfigValueOption("loose"),
112                    ConfigValueOption("none"),
113                ],
114            ),
115            CONF_ENTRY_MISSING_ALBUM_ARTIST,
116            CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
117            CONF_ENTRY_LIBRARY_SYNC_TRACKS,
118            CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
119            CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
120            CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
121            CONF_ENTRY_PROPAGATE_GENRES,
122        )
123
124    async def handle_async_init(self) -> None:
125        """Handle async initialization of the provider."""
126        # validate the connection details before attempting to mount
127        server = str(self.get_setup_value(CONF_HOST))
128        if not await get_ip_from_host(server):
129            msg = f"Unable to resolve {server}, make sure the address is resolvable."
130            raise SetupFailedError(
131                msg,
132                translation_key="host_unresolvable",
133                translation_args=[server],
134            )
135        share = str(self.get_setup_value(CONF_SHARE))
136        if not share or "/" in share or "\\" in share:
137            msg = "Invalid share name"
138            raise SetupFailedError(msg)
139        # the mount point may already exist; checking first is not reliable because
140        # reading the path fails while the server is unreachable
141        await makedirs(self.base_path, exist_ok=True)
142        try:
143            # do unmount first to cleanup any unexpected state
144            await unmount(self.base_path, self.logger)
145            await self.mount()
146        except OSError as err:
147            msg = f"Unable to run the mount command: {err}"
148            raise SetupFailedError(msg) from err
149        await self.check_write_access()
150
151    async def unload(self, is_removed: bool = False) -> None:
152        """
153        Handle unload/close of the provider.
154
155        Called when provider is deregistered (e.g. MA exiting or config reloading).
156        """
157        await super().unload(is_removed)
158        await unmount(self.base_path, self.logger)
159
160    async def get_diagnostics(self) -> dict[str, SerializableType]:
161        """Return diagnostics info for this provider to include in diagnostics reports."""
162        return {
163            **await super().get_diagnostics(),
164            "mounted": await ismount(self.base_path),
165        }
166
167    async def mount(self) -> None:
168        """Mount the SMB location to a temporary folder."""
169        server = str(self.get_setup_value(CONF_HOST))
170        username = str(self.get_setup_value(CONF_USERNAME) or "guest")
171        password = self.get_setup_value(CONF_PASSWORD)
172        # Type narrowing: password can be str or None
173        password_str: str | None = str(password) if password is not None else None
174        share = str(self.get_setup_value(CONF_SHARE))
175
176        # handle optional subfolder
177        subfolder = str(self.get_setup_value(CONF_SUBFOLDER) or "")
178        if subfolder:
179            subfolder = subfolder.replace("\\", "/")
180            if not subfolder.startswith("/"):
181                subfolder = "/" + subfolder
182            subfolder = subfolder.removesuffix("/")
183
184        env_vars = os.environ.copy()
185
186        if platform.system() == "Darwin":
187            mount_cmd = self._build_macos_mount_cmd(
188                server, username, password_str, share, subfolder
189            )
190        elif platform.system() == "Linux":
191            mount_cmd, env_vars = self._build_linux_mount_cmd(
192                server, username, password_str, share, subfolder, env_vars
193            )
194        else:
195            msg = f"SMB provider is not supported on {platform.system()}"
196            raise UnsupportedSystemError(msg)
197
198        self.logger.debug("Mounting //%s/%s%s to %s", server, share, subfolder, self.base_path)
199        self.logger.log(VERBOSE_LOG_LEVEL, "Using mount command: %s", " ".join(mount_cmd))
200        returncode, output = await check_output(*mount_cmd, env=env_vars)
201        if returncode != 0:
202            raise _mount_error(output.decode().strip())
203
204    def _build_macos_mount_cmd(
205        self, server: str, username: str, password: str | None, share: str, subfolder: str
206    ) -> list[str]:
207        """Build mount command for macOS."""
208        mount_options = []
209
210        # Add SMB version if specified
211        smb_version = str(self.get_setup_value(CONF_SMB_VERSION) or "")
212        if smb_version:
213            # macOS uses different version format (e.g., smb2, smb3)
214            if smb_version.startswith("3"):
215                mount_options.extend(["-o", "protocol_vers_map=6"])  # SMB3
216            elif smb_version.startswith("2"):
217                mount_options.extend(["-o", "protocol_vers_map=4"])  # SMB2
218
219        # Construct credentials in URL format
220        # macOS mount_smbfs supports special characters in password when URL-encoded
221        encoded_password = f":{quote(str(password), safe='')}" if password else ""
222
223        return [
224            "mount",
225            "-t",
226            "smbfs",
227            *mount_options,
228            f"//{username}{encoded_password}@{server}/{share}{subfolder}",
229            self.base_path,
230        ]
231
232    def _build_linux_mount_cmd(
233        self,
234        server: str,
235        username: str,
236        password: str | None,
237        share: str,
238        subfolder: str,
239        env_vars: dict[str, str],
240    ) -> tuple[list[str], dict[str, str]]:
241        """
242        Build mount command for Linux.
243
244        Uses the PASSWD environment variable to handle passwords with special characters
245        (commas, etc.) that cannot be escaped on the command line.
246
247        :param server: The SMB server hostname or IP.
248        :param username: The username for authentication.
249        :param password: The password for authentication (can contain special chars).
250        :param share: The share name on the server.
251        :param subfolder: Optional subfolder path within the share.
252        :param env_vars: Environment variables dict to modify with PASSWD if needed.
253        :returns: Tuple of (mount command args, modified env vars).
254        """
255        options = ["rw"]  # read-write access
256
257        # We pass the password via the PASSWD environment variable to avoid
258        # improperly escaped passwords with special characters.
259        if username and username.lower() != "guest":
260            options.append(f"username={username}")
261            if password:
262                env_vars["PASSWD"] = password
263        else:
264            # Guest/anonymous access
265            options.append("guest")
266
267        # SMB version for better compatibility and performance
268        smb_version = str(self.get_setup_value(CONF_SMB_VERSION) or "")
269        if smb_version:
270            options.append(f"vers={smb_version}")
271
272        # Cache mode for better performance
273        cache_mode = str(self.config.get_value(CONF_CACHE_MODE) or "loose")
274        options.append(f"cache={cache_mode}")
275
276        # Case insensitive by default (standard for SMB) and other performance options.
277        # Note: emoji and other 4-byte UTF-8 characters (U+10000+) in folder/file names
278        # are NOT supported due to a Linux kernel limitation in the CIFS client's NLS layer.
279        # Items with such characters will be skipped during library sync.
280        options.extend(
281            [
282                "iocharset=utf8",
283                "nocase",
284                "file_mode=0755",
285                "dir_mode=0755",
286                "uid=0",
287                "gid=0",
288                "noperm",
289                "nobrl",
290                "mfsymlinks",
291                "noserverino",
292                "actimeo=30",
293            ]
294        )
295
296        mount_cmd = [
297            "mount",
298            "-t",
299            "cifs",
300            "-o",
301            ",".join(options),
302            f"//{server}/{share}{subfolder}",
303            self.base_path,
304        ]
305        return mount_cmd, env_vars
306
307
308def _mount_error(output: str) -> SetupFailedError | LoginFailed:
309    """
310    Return the error to raise for a failed mount command.
311
312    :param output: The (combined) output of the mount command.
313    """
314    lowered = output.lower()
315    if any(marker in lowered for marker in _AUTH_FAILURE_MARKERS):
316        return LoginFailed(f"SMB mount failed with error: {output}")
317    return SetupFailedError(
318        f"SMB mount failed with error: {output}",
319        translation_key="mount_failed",
320        translation_args=[error_summary(output)],
321    )
322