music-assistant-server

1"""Logic to handle storage of persistent (configuration) settings."""
2
3from __future__ import annotations
4
5import asyncio
6import base64
7import contextlib
8import logging
9import os
10import threading
11from pathlib import Path
12from typing import TYPE_CHECKING, Any
13from uuid import uuid4
14
15import aiofiles
16from cryptography.fernet import Fernet, InvalidToken
17from music_assistant_models import config_entries
18from music_assistant_models.errors import InvalidDataError
19
20from music_assistant.constants import (
21    CONF_ENCRYPTION_KEY,
22    CONF_ENCRYPTION_KEY_MIGRATED,
23    CONF_ONBOARD_DONE,
24    CONF_SERVER_ID,
25    ENCRYPT_SUFFIX,
26)
27from music_assistant.controllers.config.constants import DEFAULT_SAVE_DELAY
28from music_assistant.controllers.config.core import CoreConfigMixin
29from music_assistant.controllers.config.dsp import DSPConfigMixin
30from music_assistant.controllers.config.flows import SetupFlowMixin
31from music_assistant.controllers.config.migrations import (
32    migrate,
33    migrate_connected_player_plugins,
34    migrate_hass_engine_selection,
35    migrate_nfs_subfolder_into_export_path,
36    migrate_provider_setup_data,
37)
38from music_assistant.controllers.config.players import PlayerConfigMixin
39from music_assistant.controllers.config.providers import ProviderConfigMixin
40from music_assistant.controllers.config.queues import PlayerQueueConfigMixin
41from music_assistant.helpers.json import (
42    JSON_DECODE_EXCEPTIONS,
43    async_json_dumps,
44    async_json_loads,
45    json_loads,
46)
47
48if TYPE_CHECKING:
49    from music_assistant import MusicAssistant
50
51LOGGER = logging.getLogger(__name__)
52
53
54class ConfigController(
55    ProviderConfigMixin,
56    PlayerConfigMixin,
57    PlayerQueueConfigMixin,
58    DSPConfigMixin,
59    CoreConfigMixin,
60    SetupFlowMixin,
61):
62    """Controller that handles storage of persistent configuration settings."""
63
64    _fernet: Fernet | None = None
65
66    def __init__(self, mass: MusicAssistant) -> None:
67        """Initialize storage controller."""
68        self.mass = mass
69        self.initialized = False
70        self._data: dict[str, Any] = {}
71        self.filename = os.path.join(self.mass.storage_path, "settings.json")
72        self._timer_handle: asyncio.TimerHandle | None = None
73        self._save_requested = 0
74        self._save_written = 0
75        self._save_lock = asyncio.Lock()
76        self._disk_lock = threading.Lock()
77
78    async def setup(self) -> None:
79        """Async initialize of controller."""
80        await self._load()
81        self.initialized = True
82        # create default server ID if needed
83        self.set_default(CONF_SERVER_ID, uuid4().hex)
84        self._init_encryption()
85        config_entries.ENCRYPT_CALLBACK = self.encrypt_string
86        config_entries.DECRYPT_CALLBACK = self.decrypt_string
87        # one-off: move pre-setup-flow provider values into (encrypted) setup_data.
88        # runs here, after encryption is initialized, so string values are encrypted
89        # at rest (the migrate() pass in _load() runs before encryption is available).
90        setup_data_migrated = migrate_provider_setup_data(self._data, self.encrypt_string)
91        # one-off: fold a stored NFS subfolder into its export path. Same phase and reason as
92        # above, and after it so a legacy install's keys have landed in setup_data by now.
93        # TODO: remove after 2.10 release
94        nfs_subfolder_migrated = migrate_nfs_subfolder_into_export_path(
95            self._data, self.encrypt_string, self.decrypt_string
96        )
97        # one-off: move the connected-player plugins to the player-bound model (collapse
98        # spotify_connect/airplay_receiver instances, enforce the mandatory player on
99        # ariacast_receiver/yandex_ynison). Runs after the setup-data move above so a
100        # legacy install's keys have landed in setup_data by now.
101        # TODO: remove after 2.12 release
102        connected_plugins_migrated = migrate_connected_player_plugins(
103            self._data, self.decrypt_string, self.mass.storage_path
104        )
105        if setup_data_migrated or nfs_subfolder_migrated or connected_plugins_migrated:
106            self.save(immediate=True)
107        # one-off: hand the Home Assistant plugin's former single TTS/AI entity choice over to
108        # the providers that select their own engine now. Runs here for the same reason: the
109        # ai_radio selection lands in its encrypted setup_data.
110        if migrate_hass_engine_selection(self._data, self.encrypt_string):
111            self.save(immediate=True)
112        if not self.onboard_done:
113            self.mass.register_api_command(
114                "config/onboard_complete",
115                self.set_onboard_complete,
116                authenticated=True,
117                alias=True,  # hide from public API docs
118            )
119        LOGGER.debug("Started.")
120
121    @property
122    def onboard_done(self) -> bool:
123        """Return True if onboarding is done."""
124        return bool(self.get(CONF_ONBOARD_DONE, False))
125
126    async def set_onboard_complete(self) -> None:
127        """
128        Mark onboarding as complete.
129
130        This is called by the frontend after the user has completed the onboarding wizard.
131        Only available when onboarding is not yet complete.
132        """
133        if self.onboard_done:
134            msg = "Onboarding already completed"
135            raise InvalidDataError(msg)
136
137        self.set(CONF_ONBOARD_DONE, True)
138        self.save(immediate=True)
139        LOGGER.info("Onboarding completed")
140
141    async def close(self) -> None:
142        """Handle logic on server stop."""
143        if self._timer_handle is not None:
144            self._timer_handle.cancel()
145            self._timer_handle = None
146        if self._save_written != self._save_requested:
147            # the latest change never made it to disk: its save is either still waiting
148            # out the debounce delay or was cancelled on stop, so write it here
149            await self._async_save()
150        LOGGER.debug("Stopped.")
151
152    def get(self, key: str, default: Any = None) -> Any:
153        """Get value(s) for a specific key/path in persistent storage."""
154        assert self.initialized, "Not yet (async) initialized"
155        # we support a multi level hierarchy by providing the key as path,
156        # with a slash (/) as splitter. Sort that out here.
157        parent = self._data
158        subkeys = key.split("/")
159        for index, subkey in enumerate(subkeys):
160            if index == (len(subkeys) - 1):
161                value = parent.get(subkey, default)
162                if value is None:
163                    # replace None with default
164                    return default
165                return value
166            if subkey not in parent:
167                # requesting subkey from a non existing parent
168                return default
169            parent = parent[subkey]
170        return default
171
172    def set(self, key: str, value: Any, immediate: bool = False) -> None:
173        """Set value(s) for a specific key/path in persistent storage."""
174        assert self.initialized, "Not yet (async) initialized"
175        # we support a multi level hierarchy by providing the key as path,
176        # with a slash (/) as splitter.
177        parent = self._data
178        subkeys = key.split("/")
179        for index, subkey in enumerate(subkeys):
180            if index == (len(subkeys) - 1):
181                parent[subkey] = value
182            else:
183                parent.setdefault(subkey, {})
184                parent = parent[subkey]
185        self.save(immediate=immediate)
186
187    def set_default(self, key: str, default_value: Any) -> None:
188        """Set default value(s) for a specific key/path in persistent storage."""
189        assert self.initialized, "Not yet (async) initialized"
190        cur_value = self.get(key, "__MISSING__")
191        if cur_value == "__MISSING__":
192            self.set(key, default_value)
193
194    def remove(
195        self,
196        key: str,
197    ) -> None:
198        """Remove value(s) for a specific key/path in persistent storage."""
199        assert self.initialized, "Not yet (async) initialized"
200        parent = self._data
201        subkeys = key.split("/")
202        for index, subkey in enumerate(subkeys):
203            if subkey not in parent:
204                return
205            if index == (len(subkeys) - 1):
206                parent.pop(subkey)
207            else:
208                parent.setdefault(subkey, {})
209                parent = parent[subkey]
210
211        self.save()
212
213    def save(self, immediate: bool = False) -> None:
214        """Schedule save of data to disk."""
215        if self._timer_handle is not None:
216            self._timer_handle.cancel()
217            self._timer_handle = None
218
219        self._save_requested += 1
220        if immediate:
221            self.mass.create_task(self._async_save)
222        else:
223            # schedule the save for later
224            self._timer_handle = self.mass.loop.call_later(DEFAULT_SAVE_DELAY, self._start_save)
225
226    def encrypt_string(self, str_value: str) -> str:
227        """Encrypt a (password)string with Fernet."""
228        if str_value.startswith(ENCRYPT_SUFFIX):
229            return str_value
230        assert self._fernet is not None
231        return ENCRYPT_SUFFIX + self._fernet.encrypt(str_value.encode()).decode()
232
233    def decrypt_string(self, encrypted_str: str) -> str:
234        """Decrypt a (password)string with Fernet."""
235        if not encrypted_str:
236            return encrypted_str
237        if not encrypted_str.startswith(ENCRYPT_SUFFIX):
238            return encrypted_str
239        assert self._fernet is not None
240        try:
241            return self._fernet.decrypt(encrypted_str.replace(ENCRYPT_SUFFIX, "").encode()).decode()
242        except InvalidToken as err:
243            msg = "Password decryption failed"
244            raise InvalidDataError(msg) from err
245
246    def _init_encryption(self) -> None:
247        """Set up encryption for SECURE_STRING config values."""
248        self._fernet = self._load_or_create_encryption_key()
249        if not self.get(CONF_ENCRYPTION_KEY_MIGRATED):
250            self._migrate_legacy_secrets()
251            self.set(CONF_ENCRYPTION_KEY_MIGRATED, True)
252
253    def _load_or_create_encryption_key(self) -> Fernet:
254        """Return the stored encryption key, generating a new one if it is absent or invalid."""
255        encryption_key: Any = self.get(CONF_ENCRYPTION_KEY, "")
256        if isinstance(encryption_key, str) and encryption_key:
257            try:
258                return Fernet(encryption_key.encode())
259            except ValueError:
260                LOGGER.warning("Stored encryption key is invalid; generating a new one")
261                self.set(CONF_ENCRYPTION_KEY_MIGRATED, False)
262        encryption_key = Fernet.generate_key().decode()
263        self.set(CONF_ENCRYPTION_KEY, encryption_key)
264        return Fernet(encryption_key.encode())
265
266    def _migrate_legacy_secrets(self) -> None:
267        """One-time re-encryption of secrets that were encrypted with the server_id-derived key."""
268        server_id: str = self.get(CONF_SERVER_ID)
269        assert server_id
270        legacy_fernet = Fernet(base64.urlsafe_b64encode(server_id.encode()[:32]))
271        migrated = self._rotate_encrypted_values(self._data, legacy_fernet)
272        if migrated:
273            LOGGER.info("Re-encrypted %s secret(s) with the dedicated encryption key", migrated)
274            self.save(immediate=True)
275
276    def _rotate_encrypted_values(self, node: Any, legacy_fernet: Fernet) -> int:
277        """Recursively re-encrypt legacy-encrypted values, returning the count."""
278        assert self._fernet is not None
279        count = 0
280        values = node.items() if isinstance(node, dict) else enumerate(node)
281        for key, value in values:
282            if isinstance(value, (dict, list)):
283                count += self._rotate_encrypted_values(value, legacy_fernet)
284            elif isinstance(value, str) and value.startswith(ENCRYPT_SUFFIX):
285                token = value[len(ENCRYPT_SUFFIX) :].encode()
286                try:
287                    decrypted = legacy_fernet.decrypt(token)
288                except InvalidToken:
289                    continue
290                node[key] = ENCRYPT_SUFFIX + self._fernet.encrypt(decrypted).decode()
291                count += 1
292        return count
293
294    async def _load(self) -> None:
295        """Load data from persistent storage."""
296        assert not self._data, "Already loaded"
297
298        for filename in (self.filename, f"{self.filename}.backup"):
299            try:
300                async with aiofiles.open(filename, encoding="utf-8") as _file:
301                    self._data = await async_json_loads(await _file.read())
302                    LOGGER.debug("Loaded persistent settings from %s", filename)
303                    if await migrate(self._data):
304                        await self._async_save()
305                    return
306            except FileNotFoundError:
307                pass
308            except JSON_DECODE_EXCEPTIONS:
309                LOGGER.exception("Error while reading persistent storage file %s", filename)
310        LOGGER.debug("Started with empty storage: No persistent storage file found.")
311
312    def _start_save(self) -> None:
313        """Start the save task, called by the save timer."""
314        self._timer_handle = None
315        self.mass.create_task(self._async_save)
316
317    async def _async_save(self) -> None:
318        """Save persistent data to disk."""
319        async with self._save_lock:
320            # remember which change we are about to write: anything requested after this
321            # point is not part of it, and must leave the settings marked as unsaved
322            requested = self._save_requested
323            json_data = await async_json_dumps(self._data, indent=True)
324            await asyncio.to_thread(self._save_to_disk, json_data)
325            self._save_written = requested
326        LOGGER.debug("Saved data to persistent storage")
327
328    def _save_to_disk(self, json_data: str) -> None:
329        """Atomically write the settings file to disk, rotating the previous one to backup."""
330        # cancelling a save does not stop the worker thread it already handed the write
331        # to, so _save_lock is released while this is still running. guard the file
332        # itself here, in the thread that actually writes it, or a second writer would
333        # race this one over the same temp file and leave no settings at all
334        with self._disk_lock:
335            filename = Path(self.filename)
336            filename_temp = Path(f"{self.filename}.tmp")
337            with filename_temp.open("w", encoding="utf-8") as _file:
338                _file.write(json_data)
339                _file.flush()
340                # fsync so a power failure can not leave a zero-length file behind (#5716)
341                os.fsync(_file.fileno())
342            with contextlib.suppress(FileNotFoundError, *JSON_DECODE_EXCEPTIONS):
343                # only rotate a parseable file to the backup, so a corrupt
344                # (crash leftover) file can never clobber a possibly good backup
345                json_loads(filename.read_bytes())
346                filename.replace(f"{self.filename}.backup")
347            filename_temp.replace(filename)
348            # best effort: fsync the directory as well so the renames themselves
349            # survive a power failure (not supported on all platforms/filesystems)
350            with contextlib.suppress(OSError):
351                dir_fd = os.open(os.path.dirname(self.filename), os.O_RDONLY)
352                try:
353                    os.fsync(dir_fd)
354                finally:
355                    os.close(dir_fd)
356