Add claims-based permission system

music_assistant/controllers/config.py

@@ -108,6 +108,7 @@
 )
 from music_assistant.helpers.api import api_command
 from music_assistant.helpers.json import JSON_DECODE_EXCEPTIONS, async_json_dumps, async_json_loads
+from music_assistant.helpers.permissions import Permission
 from music_assistant.helpers.util import load_provider_module, validate_announcement_chime_url
 from music_assistant.models import ProviderModuleType
 from music_assistant.models.music_provider import MusicProvider
@@ -475,7 +476,7 @@ async def get_provider_config_entries(  # noqa: PLR0915
                     entry.value = values.get(entry.key, entry.default_value)
         return all_entries
 
-    @api_command("config/providers/save", required_role="admin")
+    @api_command("config/providers/save", required_permissions=[Permission.PROVIDERS_MANAGE])
     async def save_provider_config(
         self,
         provider_domain: str,
@@ -496,7 +497,7 @@ async def save_provider_config(
         # return full config, just in case
         return await self.get_provider_config(config.instance_id)
 
-    @api_command("config/providers/remove", required_role="admin")
+    @api_command("config/providers/remove", required_permissions=[Permission.PROVIDERS_MANAGE])
     async def remove_provider_config(self, instance_id: str) -> None:
         """Remove ProviderConfig."""
         conf_key = f"{CONF_PROVIDERS}/{instance_id}"
@@ -796,7 +797,7 @@ def get_base_player_config(self, player_id: str, provider: str) -> PlayerConfig:
             }
         return cast("PlayerConfig", PlayerConfig.parse([], raw_conf))
 
-    @api_command("config/players/save", required_role="admin")
+    @api_command("config/players/save", required_permissions=[Permission.PLAYERS_CONFIGURE])
     async def save_player_config(
         self, player_id: str, values: dict[str, ConfigValueType]
     ) -> PlayerConfig:
@@ -838,7 +839,7 @@ async def save_player_config(
         # return full player config (just in case)
         return await self.get_player_config(player_id)
 
-    @api_command("config/players/remove", required_role="admin")
+    @api_command("config/players/remove", required_permissions=[Permission.PLAYERS_CONFIGURE])
     async def remove_player_config(self, player_id: str) -> None:
         """Remove PlayerConfig."""
         conf_key = f"{CONF_PLAYERS}/{player_id}"
@@ -930,7 +931,7 @@ def get_player_dsp_config(self, player_id: str) -> DSPConfig:
         dsp_config.enabled = False
         return dsp_config
 
-    @api_command("config/players/dsp/save", required_role="admin")
+    @api_command("config/players/dsp/save", required_permissions=[Permission.PLAYERS_CONFIGURE])
     async def save_dsp_config(self, player_id: str, config: DSPConfig) -> DSPConfig:
         """
         Save/update DSPConfig for a player.
@@ -957,7 +958,7 @@ async def get_dsp_presets(self) -> list[DSPConfigPreset]:
         raw_presets = self.get(CONF_PLAYER_DSP_PRESETS, {})
         return [DSPConfigPreset.from_dict(preset) for preset in raw_presets.values()]
 
-    @api_command("config/dsp_presets/save", required_role="admin")
+    @api_command("config/dsp_presets/save", required_permissions=[Permission.CONFIG_WRITE])
     async def save_dsp_presets(self, preset: DSPConfigPreset) -> DSPConfigPreset:
         """
         Save/update a user-defined DSP presets.
@@ -982,7 +983,7 @@ async def save_dsp_presets(self, preset: DSPConfigPreset) -> DSPConfigPreset:
 
         return preset
 
-    @api_command("config/dsp_presets/remove", required_role="admin")
+    @api_command("config/dsp_presets/remove", required_permissions=[Permission.CONFIG_WRITE])
     async def remove_dsp_preset(self, preset_id: str) -> None:
         """Remove a user-defined DSP preset."""
         self.mass.config.remove(f"{CONF_PLAYER_DSP_PRESETS}/preset_{preset_id}")
@@ -1152,7 +1153,7 @@ async def get_core_config_entries(
                     entry.value = values.get(entry.key, entry.default_value)
         return all_entries
 
-    @api_command("config/core/save", required_role="admin")
+    @api_command("config/core/save", required_permissions=[Permission.CONFIG_WRITE])
     async def save_core_config(
         self,
         domain: str,
@@ -1469,7 +1470,7 @@ async def _async_save(self) -> None:
             await _file.write(await async_json_dumps(self._data, indent=True))
         LOGGER.debug("Saved data to persistent storage")
 
-    @api_command("config/providers/reload", required_role="admin")
+    @api_command("config/providers/reload", required_permissions=[Permission.PROVIDERS_MANAGE])
     async def _reload_provider(self, instance_id: str) -> None:
         """Reload provider."""
         try:

music_assistant/controllers/media/base.py

@@ -34,6 +34,7 @@
 from music_assistant.helpers.compare import compare_media_item, create_safe_string
 from music_assistant.helpers.database import UNSET
 from music_assistant.helpers.json import json_loads, serialize_to_json
+from music_assistant.helpers.permissions import Permission
 from music_assistant.helpers.util import guard_single_request, parse_optional_bool
 
 if TYPE_CHECKING:
@@ -124,10 +125,14 @@ def __init__(self, mass: MusicAssistant) -> None:
             f"music/{api_base}/get_{self.media_type}", self.get, alias=True
         )
         self.mass.register_api_command(
-            f"music/{api_base}/update", self.update_item_in_library, required_role="admin"
+            f"music/{api_base}/update",
+            self.update_item_in_library,
+            required_permissions=[Permission.LIBRARY_WRITE],
         )
         self.mass.register_api_command(
-            f"music/{api_base}/remove", self.remove_item_from_library, required_role="admin"
+            f"music/{api_base}/remove",
+            self.remove_item_from_library,
+            required_permissions=[Permission.LIBRARY_DELETE],
         )
         self._db_add_lock = asyncio.Lock()
 

music_assistant/controllers/media/genres.py

@@ -37,6 +37,7 @@
 from music_assistant.helpers.compare import create_safe_string
 from music_assistant.helpers.database import UNSET
 from music_assistant.helpers.json import serialize_to_json
+from music_assistant.helpers.permissions import Permission
 
 from .base import MediaControllerBase
 
@@ -82,33 +83,39 @@ def __init__(self, mass: MusicAssistant) -> None:
 
         # register extra api handlers
         self.mass.register_api_command(
-            "music/genres/add_alias", self.add_alias, required_role="admin"
+            "music/genres/add_alias",
+            self.add_alias,
+            required_permissions=[Permission.LIBRARY_WRITE],
         )
         self.mass.register_api_command(
-            "music/genres/remove_alias", self.remove_alias, required_role="admin"
+            "music/genres/remove_alias",
+            self.remove_alias,
+            required_permissions=[Permission.LIBRARY_WRITE],
         )
         self.mass.register_api_command(
-            "music/genres/add_media_mapping", self.add_media_mapping, required_role="admin"
+            "music/genres/add_media_mapping",
+            self.add_media_mapping,
+            required_permissions=[Permission.LIBRARY_WRITE],
         )
         self.mass.register_api_command(
             "music/genres/remove_media_mapping",
             self.remove_media_mapping,
-            required_role="admin",
+            required_permissions=[Permission.LIBRARY_WRITE],
         )
         self.mass.register_api_command(
             "music/genres/promote_alias",
             self.promote_alias_to_genre,
-            required_role="admin",
+            required_permissions=[Permission.LIBRARY_WRITE],
         )
         self.mass.register_api_command(
             "music/genres/restore_defaults",
             self.restore_default_genres,
-            required_role="admin",
+            required_permissions=[Permission.LIBRARY_WRITE],
         )
         self.mass.register_api_command(
             "music/genres/add",
             self.add_item_to_library,
-            required_role="admin",
+            required_permissions=[Permission.LIBRARY_WRITE],
         )
         self.mass.register_api_command(
             "music/genres/overview",
@@ -121,7 +128,7 @@ def __init__(self, mass: MusicAssistant) -> None:
         self.mass.register_api_command(
             "music/genres/scan_mappings",
             self.scan_mappings,
-            required_role="admin",
+            required_permissions=[Permission.LIBRARY_WRITE],
         )
         self.mass.register_api_command(
             "music/genres/scanner_status",
@@ -134,7 +141,7 @@ def __init__(self, mass: MusicAssistant) -> None:
         self.mass.register_api_command(
             "music/genres/merge",
             self.merge_genres,
-            required_role="admin",
+            required_permissions=[Permission.LIBRARY_WRITE],
         )
 
         # Run genre mapping scanner after library sync completes

music_assistant/controllers/players/controller.py

@@ -83,6 +83,7 @@
     get_sendspin_player_id,
 )
 from music_assistant.helpers.api import api_command
+from music_assistant.helpers.permissions import Permission
 from music_assistant.helpers.tags import async_parse_tags
 from music_assistant.helpers.throttle_retry import Throttler
 from music_assistant.helpers.util import TaskManager, validate_announcement_chime_url
@@ -1114,7 +1115,7 @@ async def cmd_ungroup_many(self, player_ids: list[str]) -> None:
         for player_id in list(player_ids):
             await self.cmd_ungroup(player_id)
 
-    @api_command("players/create_group_player", required_role="admin")
+    @api_command("players/create_group_player", required_permissions=[Permission.PLAYERS_CONFIGURE])
     async def create_group_player(
         self, provider: str, name: str, members: list[str], dynamic: bool = True
     ) -> Player:
@@ -1135,7 +1136,7 @@ async def create_group_player(
             )
         return await provider_instance.create_group_player(name, members, dynamic)
 
-    @api_command("players/remove_group_player", required_role="admin")
+    @api_command("players/remove_group_player", required_permissions=[Permission.PLAYERS_CONFIGURE])
     async def remove_group_player(self, player_id: str) -> None:
         """Remove a group player."""
         if not (player := self.get_player(player_id)):
@@ -1360,7 +1361,7 @@ async def unregister(self, player_id: str, permanent: bool = False) -> None:
         # Schedule debounced update of all players since can_group_with values may change
         self._schedule_update_all_players()
 
-    @api_command("players/remove", required_role="admin")
+    @api_command("players/remove", required_permissions=[Permission.PLAYERS_CONFIGURE])
     async def remove(self, player_id: str) -> None:
         """
         Remove a player from a provider.

music_assistant/controllers/webserver/api_docs.py

@@ -1117,6 +1117,7 @@ def generate_commands_json(command_handlers: dict[str, APICommandHandler]) -> li
         "return_type": str,  # Return type
         "authenticated": bool,  # Whether authentication is required
         "required_role": str | None,  # Required user role (if any)
+        "required_permissions": list[str] | None,  # Required permission scopes (if any)
     }
     """
     commands_data = []
@@ -1171,6 +1172,11 @@ def generate_commands_json(command_handlers: dict[str, APICommandHandler]) -> li
                 "return_type": return_type_str,
                 "authenticated": handler.authenticated,
                 "required_role": handler.required_role,
+                "required_permissions": (
+                    [p.value for p in handler.required_permissions]
+                    if handler.required_permissions
+                    else None
+                ),
             }
         )
 

music_assistant/controllers/webserver/auth.py

@@ -42,6 +42,7 @@
 from music_assistant.helpers.datetime import utc
 from music_assistant.helpers.json import json_dumps, json_loads
 from music_assistant.helpers.jwt_auth import JWTHelper
+from music_assistant.helpers.permissions import Permission
 
 if TYPE_CHECKING:
     from music_assistant.controllers.webserver import WebserverController
@@ -443,7 +444,10 @@ async def authenticate_with_token(self, token: str) -> User | None:
                 updates,
             )
 
-            return await self.get_user(user_id)
+            user = await self.get_user(user_id)
+            if user:
+                self._attach_permissions(user, payload)
+            return user
 
         except pyjwt.ExpiredSignatureError:
             if token_id := self.jwt_helper.get_token_id(token):
@@ -485,8 +489,29 @@ async def authenticate_with_token(self, token: str) -> User | None:
             legacy_updates,
         )
 
-        # Get user
-        return await self.get_user(token_row["user_id"])
+        # Get user (legacy token — generate permissions from role)
+        user = await self.get_user(token_row["user_id"])
+        if user:
+            self._attach_permissions(user)
+        return user
+
+    def _attach_permissions(self, user: User, jwt_payload: dict[str, Any] | None = None) -> None:
+        """Attach permission and claims data to a User object.
+
+        When a JWT payload is available, permissions and claims are extracted
+        from it. Otherwise, permissions are generated from the user's role.
+
+        :param user: User object to attach permissions to.
+        :param jwt_payload: Decoded JWT payload (if available).
+        """
+        from music_assistant.helpers.permissions import get_permissions_for_role  # noqa: PLC0415
+
+        if jwt_payload and "permissions" in jwt_payload:
+            user.permissions = jwt_payload["permissions"]  # type: ignore[attr-defined]
+            user.claims = jwt_payload  # type: ignore[attr-defined]
+        else:
+            user.permissions = get_permissions_for_role(user.role)  # type: ignore[attr-defined]
+            user.claims = {}  # type: ignore[attr-defined]
 
     async def get_token_id_from_token(self, token: str) -> str | None:
         """
@@ -506,7 +531,7 @@ async def get_token_id_from_token(self, token: str) -> str | None:
             return None
         return str(token_row["token_id"])
 
-    @api_command("auth/user", required_role="admin")
+    @api_command("auth/user", required_permissions=[Permission.USERS_MANAGE])
     async def get_user(self, user_id: str) -> User | None:
         """
         Get user by ID (admin only).
@@ -976,7 +1001,7 @@ async def get_user_tokens(self, user_id: str | None = None) -> list[AuthToken]:
         )
         return [AuthToken.from_dict(dict(row)) for row in token_rows]
 
-    @api_command("auth/users", required_role="admin")
+    @api_command("auth/users", required_permissions=[Permission.USERS_MANAGE])
     async def list_users(self) -> list[User]:
         """
         Get all users (admin only).
@@ -1029,7 +1054,7 @@ async def update_user_role(self, user_id: str, new_role: UserRole, admin_user: U
         )
         return True
 
-    @api_command("auth/user/enable", required_role="admin")
+    @api_command("auth/user/enable", required_permissions=[Permission.USERS_MANAGE])
     async def enable_user(self, user_id: str) -> None:
         """
         Enable user account (admin only).
@@ -1042,7 +1067,7 @@ async def enable_user(self, user_id: str) -> None:
             {"enabled": 1},
         )
 
-    @api_command("auth/user/disable", required_role="admin")
+    @api_command("auth/user/disable", required_permissions=[Permission.USERS_MANAGE])
     async def disable_user(self, user_id: str) -> None:
         """
         Disable user account (admin only).
@@ -1249,7 +1274,7 @@ async def create_long_lived_token(self, name: str, user_id: str | None = None) -
         self.logger.info("Created long-lived token '%s' for user '%s'", name, target_user.username)
         return token
 
-    @api_command("auth/user/create", required_role="admin")
+    @api_command("auth/user/create", required_permissions=[Permission.USERS_MANAGE])
     async def create_user_with_api(
         self,
         username: str,
@@ -1310,7 +1335,7 @@ async def create_user_with_api(
         self.logger.info("User created by admin: %s (role: %s)", username, role)
         return user
 
-    @api_command("auth/user/delete", required_role="admin")
+    @api_command("auth/user/delete", required_permissions=[Permission.USERS_MANAGE])
     async def delete_user(self, user_id: str) -> None:
         """
         Delete user account (admin only).
@@ -1522,7 +1547,7 @@ async def get_my_providers(self) -> list[dict[str, Any]]:
         providers = [UserAuthProvider.from_dict(dict(row)) for row in rows]
         return [p.to_dict() for p in providers]
 
-    @api_command("auth/user/unlink_provider", required_role="admin")
+    @api_command("auth/user/unlink_provider", required_permissions=[Permission.USERS_MANAGE])
     async def unlink_provider(self, user_id: str, provider_type: str) -> bool:
         """
         Unlink authentication provider from user (admin only).
@@ -1748,7 +1773,7 @@ async def exchange_join_code(self, code: str) -> dict[str, Any]:
                 "error": "Failed to create access token",
             }
 
-    @api_command("auth/join_codes", required_role="admin")
+    @api_command("auth/join_codes", required_permissions=[Permission.USERS_MANAGE])
     async def list_join_codes(self, user_id: str | None = None) -> list[dict[str, Any]]:
         """List join codes, optionally filtered by user (admin only).
 
@@ -1759,7 +1784,7 @@ async def list_join_codes(self, user_id: str | None = None) -> list[dict[str, An
         rows = await self.database.get_rows("join_codes", filter_args, limit=100)
         return [dict(row) for row in rows]
 
-    @api_command("auth/join_code/revoke", required_role="admin")
+    @api_command("auth/join_code/revoke", required_permissions=[Permission.USERS_MANAGE])
     async def revoke_join_code(self, code_id: str) -> None:
         """Revoke a specific join code (admin only).
 

music_assistant/controllers/webserver/controller.py

@@ -565,9 +565,18 @@ async def _handle_jsonrpc_api_command(self, request: web.Request) -> web.Respons
                     headers={"WWW-Authenticate": 'Bearer realm="Music Assistant"'},
                 )
 
-            # Set user in context and check role
+            # Set user in context and check permissions/role
             set_current_user(user)
-            if handler.required_role == "admin" and user.role != UserRole.ADMIN:
+            if handler.required_permissions:
+                from music_assistant.helpers.permissions import has_permission  # noqa: PLC0415
+
+                if not has_permission(user, *handler.required_permissions):
+                    perm_names = [p.value for p in handler.required_permissions]
+                    return web.Response(
+                        status=403,
+                        text=f"Missing required permissions: {perm_names}",
+                    )
+            elif handler.required_role == "admin" and user.role != UserRole.ADMIN:
                 return web.Response(
                     status=403,
                     text="Admin access required",