/
/
/
1"""Tests for the dedicated SECURE_STRING encryption key and legacy secret migration."""
2
3from __future__ import annotations
4
5import base64
6from pathlib import Path
7from types import SimpleNamespace
8from uuid import uuid4
9
10import pytest
11from cryptography.fernet import Fernet, InvalidToken
12
13from music_assistant.constants import (
14 CONF_ENCRYPTION_KEY,
15 CONF_ENCRYPTION_KEY_MIGRATED,
16 CONF_SERVER_ID,
17 ENCRYPT_SUFFIX,
18)
19from music_assistant.controllers.config.controller import ConfigController
20
21
22def _make_controller(tmp_path: Path) -> ConfigController:
23 mass = SimpleNamespace(storage_path=str(tmp_path))
24 controller = ConfigController(mass) # type: ignore[arg-type]
25 controller.initialized = True
26 controller.save = lambda **_kwargs: None # type: ignore[method-assign]
27 return controller
28
29
30def _legacy_fernet(server_id: str) -> Fernet:
31 return Fernet(base64.urlsafe_b64encode(server_id.encode()[:32]))
32
33
34def test_encryption_key_not_derived_from_server_id(tmp_path: Path) -> None:
35 """The active key must not be the public, low-entropy server_id-derived key."""
36 controller = _make_controller(tmp_path)
37 server_id = uuid4().hex
38 controller.set(CONF_SERVER_ID, server_id)
39 controller._init_encryption()
40
41 stored_key = controller.get(CONF_ENCRYPTION_KEY)
42 assert stored_key
43 assert stored_key != base64.urlsafe_b64encode(server_id.encode()[:32]).decode()
44
45
46def test_encrypt_decrypt_round_trip(tmp_path: Path) -> None:
47 """A value encrypted with the new key decrypts back to the original."""
48 controller = _make_controller(tmp_path)
49 controller.set(CONF_SERVER_ID, uuid4().hex)
50 controller._init_encryption()
51
52 encrypted = controller.encrypt_string("s3cr3t-token")
53 assert encrypted.startswith(ENCRYPT_SUFFIX)
54 assert controller.decrypt_string(encrypted) == "s3cr3t-token"
55
56
57def test_legacy_encrypted_value_still_decrypts(tmp_path: Path) -> None:
58 """Secrets encrypted with the old server_id-derived key must still decrypt after setup."""
59 controller = _make_controller(tmp_path)
60 server_id = uuid4().hex
61 controller.set(CONF_SERVER_ID, server_id)
62 legacy_value = ENCRYPT_SUFFIX + _legacy_fernet(server_id).encrypt(b"old-secret").decode()
63 controller.set("providers/foo/token", legacy_value)
64
65 controller._init_encryption()
66 assert controller.decrypt_string(controller.get("providers/foo/token")) == "old-secret"
67
68
69def test_existing_secrets_migrated_off_legacy_key(tmp_path: Path) -> None:
70 """Secrets stored with the legacy key are proactively re-encrypted on setup."""
71 controller = _make_controller(tmp_path)
72 server_id = uuid4().hex
73 controller.set(CONF_SERVER_ID, server_id)
74 legacy_value = ENCRYPT_SUFFIX + _legacy_fernet(server_id).encrypt(b"old-secret").decode()
75 controller.set("providers/foo/token", legacy_value)
76
77 controller._init_encryption()
78
79 migrated = controller.get("providers/foo/token")
80 assert migrated != legacy_value
81 # the migrated token must decrypt with the new key alone, not just the legacy one
82 new_key: str = controller.get(CONF_ENCRYPTION_KEY)
83 token = migrated[len(ENCRYPT_SUFFIX) :].encode()
84 assert Fernet(new_key.encode()).decrypt(token) == b"old-secret"
85 with pytest.raises(InvalidToken):
86 _legacy_fernet(server_id).decrypt(token)
87
88
89def test_invalid_stored_key_regenerates_without_crash(tmp_path: Path) -> None:
90 """A corrupted/invalid stored encryption_key must not crash startup; a new key is generated."""
91 controller = _make_controller(tmp_path)
92 controller.set(CONF_SERVER_ID, uuid4().hex)
93 controller.set(CONF_ENCRYPTION_KEY, "not-a-valid-fernet-key")
94
95 controller._init_encryption()
96
97 new_key = controller.get(CONF_ENCRYPTION_KEY)
98 assert new_key != "not-a-valid-fernet-key"
99 round_tripped = controller.decrypt_string(controller.encrypt_string("s3cr3t"))
100 assert round_tripped == "s3cr3t"
101
102
103def test_non_string_stored_key_regenerates_without_crash(tmp_path: Path) -> None:
104 """A non-string encryption_key (e.g. settings hand-edited to a number) must not crash startup."""
105 controller = _make_controller(tmp_path)
106 controller.set(CONF_SERVER_ID, uuid4().hex)
107 controller.set(CONF_ENCRYPTION_KEY, 12345)
108
109 controller._init_encryption()
110
111 new_key = controller.get(CONF_ENCRYPTION_KEY)
112 assert isinstance(new_key, str)
113 assert controller.decrypt_string(controller.encrypt_string("s3cr3t")) == "s3cr3t"
114
115
116def test_legacy_secret_survives_invalid_stored_key(tmp_path: Path) -> None:
117 """Even with an invalid stored key, legacy server_id-encrypted secrets must still decrypt."""
118 controller = _make_controller(tmp_path)
119 server_id = uuid4().hex
120 controller.set(CONF_SERVER_ID, server_id)
121 controller.set(CONF_ENCRYPTION_KEY, "")
122 legacy_value = ENCRYPT_SUFFIX + _legacy_fernet(server_id).encrypt(b"old-secret").decode()
123 controller.set("providers/foo/token", legacy_value)
124
125 controller._init_encryption()
126
127 assert controller.decrypt_string(controller.get("providers/foo/token")) == "old-secret"
128
129
130def test_secrets_migrated_only_once(tmp_path: Path) -> None:
131 """After the one-time migration, later inits must not re-scan or re-migrate stored values."""
132 controller = _make_controller(tmp_path)
133 server_id = uuid4().hex
134 controller.set(CONF_SERVER_ID, server_id)
135 controller._init_encryption()
136 assert controller.get(CONF_ENCRYPTION_KEY_MIGRATED) is True
137
138 # a legacy-encrypted value planted after migration completed is left untouched next init
139 legacy_value = ENCRYPT_SUFFIX + _legacy_fernet(server_id).encrypt(b"late").decode()
140 controller.set("providers/foo/token", legacy_value)
141 controller._init_encryption()
142 assert controller.get("providers/foo/token") == legacy_value
143
144
145def test_invalid_key_regen_remigrates_legacy_secrets(tmp_path: Path) -> None:
146 """Regenerating an invalid key must re-run migration so legacy secrets move onto the new key."""
147 controller = _make_controller(tmp_path)
148 server_id = uuid4().hex
149 controller.set(CONF_SERVER_ID, server_id)
150 controller.set(CONF_ENCRYPTION_KEY, "not-a-valid-fernet-key")
151 controller.set(CONF_ENCRYPTION_KEY_MIGRATED, True)
152 legacy_value = ENCRYPT_SUFFIX + _legacy_fernet(server_id).encrypt(b"old-secret").decode()
153 controller.set("providers/foo/token", legacy_value)
154
155 controller._init_encryption()
156
157 migrated = controller.get("providers/foo/token")
158 assert migrated != legacy_value
159 assert controller.decrypt_string(migrated) == "old-secret"
160
161
162def test_new_encryption_not_readable_with_legacy_key(tmp_path: Path) -> None:
163 """New values must be encrypted with the new key, not the public legacy one."""
164 controller = _make_controller(tmp_path)
165 server_id = uuid4().hex
166 controller.set(CONF_SERVER_ID, server_id)
167 controller._init_encryption()
168
169 encrypted = controller.encrypt_string("new-secret")
170 token = encrypted.replace(ENCRYPT_SUFFIX, "").encode()
171
172 with pytest.raises(InvalidToken):
173 _legacy_fernet(server_id).decrypt(token)
174