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