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