Disallow malicious playlist updates (#2661)

diff --git a/music_assistant/controllers/media/base.py b/music_assistant/controllers/media/base.py
index d1a4b58212fa9a0df4a32ecd435ea9e86fc85699..bd3a88ef54a92b3763592427468be4132fe300ac 100644 (file)
--- a/music_assistant/controllers/media/base.py
+++ b/music_assistant/controllers/media/base.py
@@ -10,7 +10,10 @@ from contextlib import suppress
 from typing import TYPE_CHECKING, Any, TypeVar, cast
 
 from music_assistant_models.enums import EventType, ExternalID, MediaType, ProviderFeature
-from music_assistant_models.errors import MediaNotFoundError, ProviderUnavailableError
+from music_assistant_models.errors import (
+    MediaNotFoundError,
+    ProviderUnavailableError,
+)
 from music_assistant_models.media_items import ItemMapping, MediaItemType, ProviderMapping, Track
 
 from music_assistant.constants import DB_TABLE_PLAYLOG, DB_TABLE_PROVIDER_MAPPINGS, MASS_LOGGER_NAME

diff --git a/music_assistant/controllers/media/playlists.py b/music_assistant/controllers/media/playlists.py
index c9ee2d104ee0f93c185ecc32287d272cfccd5dd4..d08fa5044e4a3ac0fddb4e78ab6fd2080acac45b 100644 (file)
--- a/music_assistant/controllers/media/playlists.py
+++ b/music_assistant/controllers/media/playlists.py
@@ -8,6 +8,7 @@ from typing import cast
 from music_assistant_models.enums import MediaType, ProviderFeature
 from music_assistant_models.errors import (
     InvalidDataError,
+    InvalidProviderURI,
     MediaNotFoundError,
     ProviderUnavailableError,
 )
@@ -44,6 +45,33 @@ class PlaylistController(MediaControllerBase[Playlist]):
             "music/playlists/remove_playlist_tracks", self.remove_playlist_tracks
         )
 
+    def _verify_update_allowed(self, current_item: Playlist, update: Playlist) -> None:
+        """Verify that the update is allowed from a security perspective.
+
+        Prevents updating item_id for non-streaming providers to prevent path traversal attacks.
+        """
+        # Build lookup dict of current mappings: provider_instance -> item_id
+        current_mappings = {
+            mapping.provider_instance: mapping.item_id for mapping in current_item.provider_mappings
+        }
+
+        # Check if any existing mapping's item_id has been modified for non-streaming providers
+        for update_mapping in update.provider_mappings:
+            # Only check if this is an existing mapping being modified
+            if update_mapping.provider_instance in current_mappings:
+                current_item_id = current_mappings[update_mapping.provider_instance]
+
+                # Disallow item_id changes for filesystem-based providers (filesystem, builtin)
+                if (
+                    current_item_id != update_mapping.item_id
+                    and update_mapping.provider_instance.startswith(("filesystem", "builtin"))
+                ):
+                    msg = (
+                        f"Updating item_id is not allowed for filesystem-based providers: "
+                        f"attempted to change '{current_item_id}' to '{update_mapping.item_id}'"
+                    )
+                    raise InvalidDataError(msg)
+
     async def tracks(
         self,
         item_id: str,
@@ -85,6 +113,9 @@ class PlaylistController(MediaControllerBase[Playlist]):
         else:
             provider = self.mass.get_provider("builtin")
 
+        if "/" in name or "\\" in name or ".." in name:
+            msg = f"{name} is not a valid Playlist name"
+            raise InvalidDataError(msg)
         # create playlist on the provider
         playlist = await provider.create_playlist(name)
         # add the new playlist to the library
@@ -101,7 +132,13 @@ class PlaylistController(MediaControllerBase[Playlist]):
         if not playlist.is_editable:
             msg = f"Playlist {playlist.name} is not editable"
             raise InvalidDataError(msg)
-
+        # Validate uris to prevent code injection
+        for uri in uris:
+            # Prevent code injection via newlines in URIs
+            if "\n" in uri or "\r" in uri:
+                msg = "Invalid URI: newlines not allowed"
+                raise InvalidProviderURI(msg)
+            await parse_uri(uri)
         # grab all existing track ids in the playlist so we can check for duplicates
         playlist_prov_map = next(iter(playlist.provider_mappings))
         playlist_prov = self.mass.get_provider(playlist_prov_map.provider_instance)
@@ -330,6 +367,7 @@ class PlaylistController(MediaControllerBase[Playlist]):
         """Update existing record in the database."""
         db_id = int(item_id)  # ensure integer
         cur_item = await self.get_library_item(db_id)
+        self._verify_update_allowed(cur_item, update)
         metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
         cur_item.external_ids.update(update.external_ids)
         name = update.name if overwrite else cur_item.name