/
/
/
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, ConfigValueType
11from music_assistant_models.enums import ConfigEntryType
12from music_assistant_models.errors import LoginFailed
13
14from music_assistant.constants import CONF_PASSWORD, CONF_USERNAME, VERBOSE_LOG_LEVEL
15from music_assistant.helpers.process import check_output
16from music_assistant.helpers.util import get_ip_from_host
17from music_assistant.providers.filesystem_local import LocalFileSystemProvider, exists, makedirs
18from music_assistant.providers.filesystem_local.constants import (
19 CONF_ENTRY_CONTENT_TYPE,
20 CONF_ENTRY_CONTENT_TYPE_READ_ONLY,
21 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
22 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
23 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
24 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
25 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
26 CONF_ENTRY_MISSING_ALBUM_ARTIST,
27)
28
29if TYPE_CHECKING:
30 from music_assistant_models.config_entries import ProviderConfig
31 from music_assistant_models.provider import ProviderManifest
32
33 from music_assistant.mass import MusicAssistant
34 from music_assistant.models import ProviderInstanceType
35
36CONF_HOST = "host"
37CONF_SHARE = "share"
38CONF_SUBFOLDER = "subfolder"
39CONF_SMB_VERSION = "smb_version"
40CONF_CACHE_MODE = "cache_mode"
41
42
43async def setup(
44 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
45) -> ProviderInstanceType:
46 """Initialize provider(instance) with given configuration."""
47 # check if valid dns name is given for the host
48 server = str(config.get_value(CONF_HOST))
49 if not await get_ip_from_host(server):
50 msg = f"Unable to resolve {server}, make sure the address is resolveable."
51 raise LoginFailed(msg)
52 # check if share is valid
53 share = str(config.get_value(CONF_SHARE))
54 if not share or "/" in share or "\\" in share:
55 msg = "Invalid share name"
56 raise LoginFailed(msg)
57 # base_path will be the path where we're going to mount the remote share
58 base_path = f"/tmp/{config.instance_id}" # noqa: S108
59 return SMBFileSystemProvider(mass, manifest, config, base_path)
60
61
62async def get_config_entries(
63 mass: MusicAssistant,
64 instance_id: str | None = None,
65 action: str | None = None,
66 values: dict[str, ConfigValueType] | None = None,
67) -> tuple[ConfigEntry, ...]:
68 """
69 Return Config entries to setup this provider.
70
71 instance_id: id of an existing provider instance (None if new instance setup).
72 action: [optional] action key called from config entries UI.
73 values: the (intermediate) raw values for config entries sent with the action.
74 """
75 # ruff: noqa: ARG001
76 base_entries = (
77 ConfigEntry(
78 key=CONF_HOST,
79 type=ConfigEntryType.STRING,
80 label="Server",
81 required=True,
82 description="The (fqdn) hostname of the SMB/CIFS/DFS server to connect to."
83 "For example mynas.local.",
84 ),
85 ConfigEntry(
86 key=CONF_SHARE,
87 type=ConfigEntryType.STRING,
88 label="Share",
89 required=True,
90 description="The name of the share/service you'd like to connect to on "
91 "the remote host, For example 'media'.",
92 ),
93 ConfigEntry(
94 key=CONF_USERNAME,
95 type=ConfigEntryType.STRING,
96 label="Username",
97 required=False,
98 default_value="guest",
99 description="The username to authenticate to the remote server. "
100 "Leave as 'guest' or empty for anonymous access.",
101 ),
102 ConfigEntry(
103 key=CONF_PASSWORD,
104 type=ConfigEntryType.SECURE_STRING,
105 label="Password",
106 required=False,
107 default_value=None,
108 description="The password to authenticate to the remote server. "
109 "Leave empty for anonymous/guest access.",
110 ),
111 ConfigEntry(
112 key=CONF_SUBFOLDER,
113 type=ConfigEntryType.STRING,
114 label="Subfolder",
115 required=False,
116 default_value="",
117 description="[optional] Use if your music is stored in a sublevel of the share. "
118 "E.g. 'collections' or 'albums/A-K'.",
119 ),
120 ConfigEntry(
121 key=CONF_SMB_VERSION,
122 type=ConfigEntryType.STRING,
123 label="SMB Version",
124 required=False,
125 advanced=True,
126 default_value="3.0",
127 options=[
128 ConfigValueOption("Auto", ""),
129 ConfigValueOption("SMB 1.0", "1.0"),
130 ConfigValueOption("SMB 2.0", "2.0"),
131 ConfigValueOption("SMB 2.1", "2.1"),
132 ConfigValueOption("SMB 3.0", "3.0"),
133 ConfigValueOption("SMB 3.1.1", "3.1.1"),
134 ],
135 description="The SMB protocol version to use. SMB 3.0 or higher is recommended for "
136 "better performance and security. Use Auto to let the system negotiate.",
137 ),
138 ConfigEntry(
139 key=CONF_CACHE_MODE,
140 type=ConfigEntryType.STRING,
141 label="Cache Mode",
142 required=False,
143 advanced=True,
144 default_value="loose",
145 options=[
146 ConfigValueOption("Strict", "strict"),
147 ConfigValueOption("Loose (Recommended)", "loose"),
148 ConfigValueOption("None", "none"),
149 ],
150 description="Cache mode affects performance and consistency. "
151 "'Loose' provides better performance for read-heavy workloads "
152 "and is recommended for music libraries.",
153 ),
154 CONF_ENTRY_MISSING_ALBUM_ARTIST,
155 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
156 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
157 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
158 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
159 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
160 )
161
162 if instance_id is None or values is None:
163 return (
164 CONF_ENTRY_CONTENT_TYPE,
165 *base_entries,
166 )
167 return (
168 *base_entries,
169 CONF_ENTRY_CONTENT_TYPE_READ_ONLY,
170 )
171
172
173class SMBFileSystemProvider(LocalFileSystemProvider):
174 """
175 Implementation of an SMB File System Provider.
176
177 Basically this is just a wrapper around the regular local files provider,
178 except for the fact that it will mount a remote folder to a temporary location.
179 We went for this OS-depdendent approach because there is no solid async-compatible
180 smb library for Python (and we tried both pysmb and smbprotocol).
181 """
182
183 @property
184 def instance_name_postfix(self) -> str | None:
185 """Return a (default) instance name postfix for this provider instance."""
186 share = str(self.config.get_value(CONF_SHARE))
187 subfolder = str(self.config.get_value(CONF_SUBFOLDER))
188 if subfolder:
189 return subfolder
190 if share:
191 return share
192 return None
193
194 async def handle_async_init(self) -> None:
195 """Handle async initialization of the provider."""
196 if not await exists(self.base_path):
197 await makedirs(self.base_path)
198 try:
199 # do unmount first to cleanup any unexpected state
200 await self.unmount(ignore_error=True)
201 await self.mount()
202 except Exception as err:
203 msg = f"Connection failed for the given details: {err}"
204 raise LoginFailed(msg) from err
205 await self.check_write_access()
206
207 async def unload(self, is_removed: bool = False) -> None:
208 """
209 Handle unload/close of the provider.
210
211 Called when provider is deregistered (e.g. MA exiting or config reloading).
212 """
213 await self.unmount()
214
215 async def mount(self) -> None:
216 """Mount the SMB location to a temporary folder."""
217 server = str(self.config.get_value(CONF_HOST))
218 username = str(self.config.get_value(CONF_USERNAME) or "guest")
219 password = self.config.get_value(CONF_PASSWORD)
220 # Type narrowing: password can be str or None
221 password_str: str | None = str(password) if password is not None else None
222 share = str(self.config.get_value(CONF_SHARE))
223
224 # handle optional subfolder
225 subfolder = str(self.config.get_value(CONF_SUBFOLDER) or "")
226 if subfolder:
227 subfolder = subfolder.replace("\\", "/")
228 if not subfolder.startswith("/"):
229 subfolder = "/" + subfolder
230 subfolder = subfolder.removesuffix("/")
231
232 env_vars = os.environ.copy()
233
234 if platform.system() == "Darwin":
235 mount_cmd = self._build_macos_mount_cmd(
236 server, username, password_str, share, subfolder
237 )
238 elif platform.system() == "Linux":
239 mount_cmd = self._build_linux_mount_cmd(
240 server, username, password_str, share, subfolder
241 )
242 else:
243 msg = f"SMB provider is not supported on {platform.system()}"
244 raise LoginFailed(msg)
245
246 self.logger.debug("Mounting //%s/%s%s to %s", server, share, subfolder, self.base_path)
247 self.logger.log(VERBOSE_LOG_LEVEL, "Using mount command: %s", " ".join(mount_cmd))
248 returncode, output = await check_output(*mount_cmd, env=env_vars)
249 if returncode != 0:
250 msg = f"SMB mount failed with error: {output.decode()}"
251 raise LoginFailed(msg)
252
253 def _build_macos_mount_cmd(
254 self, server: str, username: str, password: str | None, share: str, subfolder: str
255 ) -> list[str]:
256 """Build mount command for macOS."""
257 mount_options = []
258
259 # Add SMB version if specified
260 smb_version = str(self.config.get_value(CONF_SMB_VERSION) or "")
261 if smb_version:
262 # macOS uses different version format (e.g., smb2, smb3)
263 if smb_version.startswith("3"):
264 mount_options.extend(["-o", "protocol_vers_map=6"]) # SMB3
265 elif smb_version.startswith("2"):
266 mount_options.extend(["-o", "protocol_vers_map=4"]) # SMB2
267
268 # Construct credentials in URL format
269 # macOS mount_smbfs supports special characters in password when URL-encoded
270 encoded_password = f":{quote(str(password), safe='')}" if password else ""
271
272 return [
273 "mount",
274 "-t",
275 "smbfs",
276 *mount_options,
277 f"//{username}{encoded_password}@{server}/{share}{subfolder}",
278 self.base_path,
279 ]
280
281 def _build_linux_mount_cmd(
282 self, server: str, username: str, password: str | None, share: str, subfolder: str
283 ) -> list[str]:
284 """Build mount command for Linux."""
285 options = ["rw"] # read-write access
286
287 # Handle username and password
288 if username and username.lower() != "guest":
289 options.append(f"username={username}")
290 if password:
291 options.append(f"password={password}")
292 else:
293 # Guest/anonymous access
294 options.append("guest")
295
296 # SMB version for better compatibility and performance
297 smb_version = str(self.config.get_value(CONF_SMB_VERSION) or "")
298 if smb_version:
299 options.append(f"vers={smb_version}")
300
301 # Cache mode for better performance
302 cache_mode = str(self.config.get_value(CONF_CACHE_MODE) or "loose")
303 options.append(f"cache={cache_mode}")
304
305 # Case insensitive by default (standard for SMB) and other performance options
306 # Note: iocharset is omitted to allow CIFS native Unicode handling for emoji
307 # and other 4-byte UTF-8 characters.
308 options.extend(
309 [
310 "nocase",
311 "file_mode=0755",
312 "dir_mode=0755",
313 "uid=0",
314 "gid=0",
315 "noperm",
316 "nobrl",
317 "mfsymlinks",
318 "noserverino",
319 "actimeo=30",
320 ]
321 )
322
323 return [
324 "mount",
325 "-t",
326 "cifs",
327 "-o",
328 ",".join(options),
329 f"//{server}/{share}{subfolder}",
330 self.base_path,
331 ]
332
333 async def unmount(self, ignore_error: bool = False) -> None:
334 """Unmount the remote share."""
335 returncode, output = await check_output("umount", self.base_path)
336 if returncode != 0 and not ignore_error:
337 self.logger.warning("SMB unmount failed with error: %s", output.decode())
338