From f0eda345064a05feb463353c1c813d17d791a64b Mon Sep 17 00:00:00 2001 From: Marvin Schenkel Date: Sun, 23 Nov 2025 19:58:30 +0100 Subject: [PATCH] Disallow malicious playlist updates (#2661) --- music_assistant/controllers/media/base.py | 5 ++- .../controllers/media/playlists.py | 40 ++++++++++++++++++- 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/music_assistant/controllers/media/base.py b/music_assistant/controllers/media/base.py index d1a4b582..bd3a88ef 100644 --- 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 c9ee2d10..d08fa504 100644 --- 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 -- 2.34.1