kdb-bot/src/bot_data/model/user_role_enum.py

@@ -0,0 +1,9 @@
+from enum import Enum
+
+
+class UserRoleEnum(Enum):
+
+    member = 0
+    moderator = 1
+    admin = 2
+    technician = 3

kdb-bot/src/bot_graphql/abc/query_abc.py

@@ -5,6 +5,8 @@ from cpl_core.dependency_injection import ServiceProviderABC
 from cpl_discord.service import DiscordBotServiceABC
 from cpl_query.extension import List
 
+from bot_api.exception.service_error_code_enum import ServiceErrorCode
+from bot_api.exception.service_exception import ServiceException
 from bot_api.route.route import Route
 from bot_data.model.auth_role_enum import AuthRoleEnum
 from bot_data.model.auth_user import AuthUser
@@ -17,6 +19,7 @@ from bot_data.model.server import Server
 from bot_data.model.user import User
 from bot_data.model.user_joined_server import UserJoinedServer
 from bot_data.model.user_joined_voice_channel import UserJoinedVoiceChannel
+from bot_data.model.user_role_enum import UserRoleEnum
 from bot_graphql.abc.filter_abc import FilterABC
 from bot_graphql.filter.page import Page
 from bot_graphql.filter.sort import Sort
@@ -140,6 +143,32 @@ class QueryABC(ObjectType):
 
         return access
 
+    @ServiceProviderABC.inject
+    def _can_user_mutate_data(self, server: Server, permission: UserRoleEnum, services: ServiceProviderABC):
+        permissions: PermissionService = services.get_service(PermissionService)
+        bot: DiscordBotServiceABC = services.get_service(DiscordBotServiceABC)
+
+        auth_user = Route.get_user()
+        if auth_user == "system" or auth_user.auth_role == AuthRoleEnum.admin:
+            return
+
+        member = bot.get_guild(server.discord_server_id).get_member(
+            auth_user.users.where(lambda x: x.server.server_id == server.server_id).single().discord_id
+        )
+
+        check_perm = lambda x: True
+        match permission:
+            case UserRoleEnum.moderator:
+                check_perm = lambda x: permissions.is_member_moderator(x)
+            case UserRoleEnum.admin:
+                check_perm = lambda x: permissions.is_member_admin(x)
+            case UserRoleEnum.technician:
+                check_perm = lambda x: permissions.is_member_technician(x)
+
+        if not check_perm(member):
+            ex = ServiceException(ServiceErrorCode.Forbidden, f"User not allowed to mutate data")
+            raise ex
+
     # @FilterABC.resolve_filter_annotation
     def _resolve_collection(self, collection: List, *_, filter: FilterABC = None, page: Page = None, sort: Sort = None):
         if filter is not None:

kdb-bot/src/bot_graphql/mutations/auto_role_mutation.py

@@ -3,6 +3,7 @@ from cpl_core.database.context import DatabaseContextABC
 from bot_data.abc.auto_role_repository_abc import AutoRoleRepositoryABC
 from bot_data.abc.server_repository_abc import ServerRepositoryABC
 from bot_data.model.auto_role import AutoRole
+from bot_data.model.user_role_enum import UserRoleEnum
 from bot_graphql.abc.query_abc import QueryABC
 
 
@@ -25,6 +26,8 @@ class AutoRoleMutation(QueryABC):
 
     def resolve_create_auto_role(self, *_, input: dict):
         auto_role = AutoRole(self._servers.get_server_by_id(input["serverId"]), input["channelId"], input["messageId"])
+        self._can_user_mutate_data(auto_role.server, UserRoleEnum.moderator)
+
         self._auto_roles.add_auto_role(auto_role)
         self._db.save_changes()
 
@@ -39,6 +42,8 @@ class AutoRoleMutation(QueryABC):
 
     def resolve_update_auto_role(self, *_, input: dict):
         auto_role = self._auto_roles.get_auto_role_by_id(input["id"])
+        self._can_user_mutate_data(auto_role.server, UserRoleEnum.moderator)
+
         auto_role.discord_channel_id = input["channelId"] if "channelId" in input else auto_role.discord_channel_id
         auto_role.discord_message_id = input["messageId"] if "messageId" in input else auto_role.discord_message_id
 
@@ -50,6 +55,8 @@ class AutoRoleMutation(QueryABC):
 
     def resolve_delete_auto_role(self, *_, id: int):
         auto_role = self._auto_roles.get_auto_role_by_id(id)
+        self._can_user_mutate_data(auto_role.server, UserRoleEnum.moderator)
+
         self._auto_roles.delete_auto_role(auto_role)
         self._db.save_changes()
         return auto_role

kdb-bot/src/bot_graphql/mutations/auto_role_rule_mutation.py

@@ -3,6 +3,7 @@ from cpl_core.database.context import DatabaseContextABC
 from bot_data.abc.auto_role_repository_abc import AutoRoleRepositoryABC
 from bot_data.abc.server_repository_abc import ServerRepositoryABC
 from bot_data.model.auto_role_rule import AutoRoleRule
+from bot_data.model.user_role_enum import UserRoleEnum
 from bot_graphql.abc.query_abc import QueryABC
 
 
@@ -27,6 +28,8 @@ class AutoRoleRuleMutation(QueryABC):
         auto_role_rule = AutoRoleRule(
             self._auto_roles.get_auto_role_by_id(input["autoRoleId"]), input["emojiName"], input["roleId"]
         )
+        self._can_user_mutate_data(auto_role_rule.auto_role.server, UserRoleEnum.moderator)
+
         self._auto_roles.add_auto_role_rule(auto_role_rule)
         self._db.save_changes()
 
@@ -45,6 +48,8 @@ class AutoRoleRuleMutation(QueryABC):
 
     def resolve_update_auto_role_rule(self, *_, input: dict):
         auto_role_rule = self._auto_roles.get_auto_role_rule_by_id(input["id"])
+        self._can_user_mutate_data(auto_role_rule.auto_role.server, UserRoleEnum.moderator)
+
         auto_role_rule.emoji_name = input["emojiName"] if "emojiName" in input else auto_role_rule.emoji_name
         auto_role_rule.role_id = input["roleId"] if "roleId" in input else auto_role_rule.role_id
 
@@ -56,6 +61,8 @@ class AutoRoleRuleMutation(QueryABC):
 
     def resolve_delete_auto_role_rule(self, *_, id: int):
         auto_role_rule = self._auto_roles.get_auto_role_rule_by_id(id)
+        self._can_user_mutate_data(auto_role_rule.auto_role.server, UserRoleEnum.moderator)
+
         self._auto_roles.delete_auto_role_rule(auto_role_rule)
         self._db.save_changes()
         return auto_role_rule

kdb-bot/src/bot_graphql/mutations/level_mutation.py

@@ -3,6 +3,7 @@ from cpl_core.database.context import DatabaseContextABC
 from bot_data.abc.level_repository_abc import LevelRepositoryABC
 from bot_data.abc.server_repository_abc import ServerRepositoryABC
 from bot_data.model.level import Level
+from bot_data.model.user_role_enum import UserRoleEnum
 from bot_graphql.abc.query_abc import QueryABC
 
 
@@ -24,12 +25,15 @@ class LevelMutation(QueryABC):
         self.set_field("deleteLevel", self.resolve_delete_level)
 
     def resolve_create_level(self, *_, input: dict):
+        server = self._servers.get_server_by_id(input["serverId"])
+        self._can_user_mutate_data(server, UserRoleEnum.admin)
+
         level = Level(
             input["name"],
             input["color"],
             int(input["minXp"]),
             int(input["permissions"]),
-            self._servers.get_server_by_id(input["serverId"]),
+            server,
         )
         self._levels.add_level(level)
         self._db.save_changes()
@@ -46,6 +50,8 @@ class LevelMutation(QueryABC):
 
     def resolve_update_level(self, *_, input: dict):
         level = self._levels.get_level_by_id(input["id"])
+        self._can_user_mutate_data(level.server, UserRoleEnum.admin)
+
         level.name = input["name"] if "name" in input else level.name
         level.color = input["color"] if "color" in input else level.color
         level.min_xp = input["minXp"] if "minXp" in input else level.min_xp
@@ -59,6 +65,8 @@ class LevelMutation(QueryABC):
 
     def resolve_delete_level(self, *_, id: int):
         level = self._levels.get_level_by_id(id)
+        self._can_user_mutate_data(level.server, UserRoleEnum.admin)
+
         self._levels.delete_level(level)
         self._db.save_changes()
         return level

kdb-bot/src/bot_graphql/mutations/user_mutation.py

@@ -1,8 +1,11 @@
 from cpl_core.database.context import DatabaseContextABC
+from cpl_discord.service import DiscordBotServiceABC
 
 from bot_data.abc.server_repository_abc import ServerRepositoryABC
 from bot_data.abc.user_repository_abc import UserRepositoryABC
+from bot_data.model.user_role_enum import UserRoleEnum
 from bot_graphql.abc.query_abc import QueryABC
+from modules.permission.service.permission_service import PermissionService
 
 
 class UserMutation(QueryABC):
@@ -10,18 +13,24 @@ class UserMutation(QueryABC):
         self,
         servers: ServerRepositoryABC,
         users: UserRepositoryABC,
+        bot: DiscordBotServiceABC,
         db: DatabaseContextABC,
+        permissions: PermissionService,
     ):
         QueryABC.__init__(self, "UserMutation")
 
         self._servers = servers
         self._users = users
+        self._bot = bot
         self._db = db
+        self._permissions = permissions
 
         self.set_field("updateUser", self.resolve_update_user)
 
     def resolve_update_user(self, *_, input: dict):
         user = self._users.get_user_by_id(input["id"])
+        self._can_user_mutate_data(user.server, UserRoleEnum.moderator)
+
         user.xp = input["xp"] if "xp" in input else user.xp
 
         self._users.update_user(user)