sh-edraft.de/sh_discord_bot
5
1
Fork 1
Code Issues 26 Pull Requests Actions Packages Projects 3 Releases 27 Wiki Activity

Added logic to check if user is allowed to see requested data #89

Browse Source
This commit is contained in:
Sven Heidemann
2023-02-11 12:54:14 +01:00
parent 84937dde0a
commit dd64435c65
15 changed files with 199 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
kdb-bot/src/bot_graphql/abc/query_abc.py
View File

@@ -1,11 +1,26 @@
from typing import Callable
from ariadne import ObjectType
from cpl_core.dependency_injection import ServiceProviderABC
from cpl_discord.service import DiscordBotServiceABC
from cpl_query.extension import List
from bot_api.route.route import Route
from bot_data.model.auth_role_enum import AuthRoleEnum
from bot_data.model.auth_user import AuthUser
from bot_data.model.auto_role import AutoRole
from bot_data.model.auto_role_rule import AutoRoleRule
from bot_data.model.client import Client
from bot_data.model.known_user import KnownUser
from bot_data.model.level import Level
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_graphql.abc.filter_abc import FilterABC
from bot_graphql.filter.page import Page
from bot_graphql.filter.sort import Sort
from modules.permission.service.permission_service import PermissionService
class QueryABC(ObjectType):
@@ -31,11 +46,100 @@ class QueryABC(ObjectType):
sort.from_dict(kwargs["sort"])
kwargs["sort"] = sort
return self._resolve_collection(get_collection(*args), *args, **kwargs)
collection = get_collection(*args)
user = Route.get_user()
if user == "system" or user.auth_role == AuthRoleEnum.admin:
return self._resolve_collection(collection, *args, **kwargs)
for x in collection:
if not self._can_user_see_element(user, x):
return List()
return self._resolve_collection(collection, *args, **kwargs)
self.set_field(f"{name}s", wrapper)
self.set_field(f"{name}Count", lambda *args: get_collection(*args).count())
@ServiceProviderABC.inject
def _can_user_see_element(self, user: AuthUser, element, services: ServiceProviderABC) -> bool:
permissions: PermissionService = services.get_service(PermissionService)
bot: DiscordBotServiceABC = services.get_service(DiscordBotServiceABC)
access = False
if type(element) == AutoRole:
element: AutoRole = element
for u in user.users:
u: User = u
guild = bot.get_guild(u.server.discord_server_id)
member = guild.get_member(u.discord_id)
if permissions.is_member_moderator(member) and u.server.server_id == element.server.server_id:
access = True
break
elif type(element) == AutoRoleRule:
element: AutoRole = element.auto_role
for u in user.users:
u: User = u
guild = bot.get_guild(u.server.discord_server_id)
member = guild.get_member(u.discord_id)
if permissions.is_member_moderator(member) and u.server.server_id == element.server.server_id:
access = True
break
elif type(element) == Client:
for u in user.users:
u: User = u
if u.server.server_id == element.server.server_id:
access = True
break
elif type(element) == KnownUser:
for u in user.users:
u: User = u
guild = bot.get_guild(u.server.discord_server_id)
member = guild.get_member(u.discord_id)
if permissions.is_member_moderator(member):
access = True
break
elif type(element) == Level:
for u in user.users:
u: User = u
if u.server.server_id == element.server.server_id:
access = True
break
elif type(element) == Server:
for u in user.users:
u: User = u
if u.server.server_id == element.server_id:
access = True
break
elif type(element) == User:
for u in user.users:
u: User = u
if u.user_id == element.user_id:
access = True
break
elif type(element) == UserJoinedServer:
for u in user.users:
u: User = u
if u.user_id == element.user.user_id:
access = True
break
elif type(element) == UserJoinedVoiceChannel:
for u in user.users:
u: User = u
if u.user_id == element.user.user_id:
access = True
break
return access
# @FilterABC.resolve_filter_annotation
def _resolve_collection(self, collection: List, *_, filter: FilterABC = None, page: Page = None, sort: Sort = None):
if filter is not None:

5
kdb-bot/src/bot_graphql/queries/auto_role_query.py
View File

@@ -4,6 +4,7 @@ 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_graphql.abc.data_query_abc import DataQueryABC
from bot_graphql.filter.auto_role_filter import AutoRoleFilter
from bot_graphql.filter.server_filter import ServerFilter
@@ -26,7 +27,9 @@ class AutoRoleQuery(DataQueryABC):
self.set_field("messageId", self.resolve_message_id)
self.set_field("server", self.resolve_server)
self.add_collection(
"autoRoleRule", lambda x, *_: self._auto_role_rules.get_auto_role_rules_by_auto_role_id(x.auto_role_id)
"autoRoleRule",
lambda x, *_: self._auto_role_rules.get_auto_role_rules_by_auto_role_id(x.auto_role_id),
AutoRoleFilter,
)
@staticmethod

15
kdb-bot/src/bot_graphql/queries/server_query.py
View File

@@ -8,6 +8,9 @@ from bot_data.abc.user_joined_voice_channel_repository_abc import UserJoinedVoic
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.server import Server
from bot_graphql.abc.data_query_abc import DataQueryABC
from bot_graphql.filter.auto_role_filter import AutoRoleFilter
from bot_graphql.filter.client_filter import ClientFilter
from bot_graphql.filter.level_filter import LevelFilter
from bot_graphql.filter.user_filter import UserFilter
@@ -38,10 +41,16 @@ class ServerQuery(DataQueryABC):
self.set_field("iconURL", self.resolve_icon_url)
self.add_collection(
"autoRole", lambda server, *_: self._auto_roles.get_auto_roles_by_server_id(server.server_id)
"autoRole",
lambda server, *_: self._auto_roles.get_auto_roles_by_server_id(server.server_id),
AutoRoleFilter,
)
self.add_collection(
"client", lambda server, *_: self._clients.get_clients_by_server_id(server.server_id), ClientFilter
)
self.add_collection(
"level", lambda server, *_: self._levels.get_levels_by_server_id(server.server_id), LevelFilter
)
self.add_collection("client", lambda server, *_: self._clients.get_clients_by_server_id(server.server_id))
self.add_collection("level", lambda server, *_: self._levels.get_levels_by_server_id(server.server_id))
self.add_collection("user", lambda server, *_: self._users.get_users_by_server_id(server.server_id), UserFilter)
@staticmethod

12
kdb-bot/src/bot_graphql/queries/user_query.py
View File

@@ -5,6 +5,8 @@ from bot_data.abc.user_joined_server_repository_abc import UserJoinedServerRepos
from bot_data.abc.user_joined_voice_channel_repository_abc import UserJoinedVoiceChannelRepositoryABC
from bot_data.model.user import User
from bot_graphql.abc.data_query_abc import DataQueryABC
from bot_graphql.filter.user_joined_server_filter import UserJoinedServerFilter
from bot_graphql.filter.user_joined_voice_channel_filter import UserJoinedVoiceChannelFilter
from modules.level.service.level_service import LevelService
@@ -31,9 +33,15 @@ class UserQuery(DataQueryABC):
self.set_field("xp", self.resolve_xp)
self.set_field("ontime", self.resolve_ontime)
self.set_field("level", self.resolve_level)
self.add_collection("joinedServer", lambda user, *_: self._ujs.get_user_joined_servers_by_user_id(user.user_id))
self.add_collection(
"joinedVoiceChannel", lambda user, *_: self._ujvs.get_user_joined_voice_channels_by_user_id(user.user_id)
"joinedServer",
lambda user, *_: self._ujs.get_user_joined_servers_by_user_id(user.user_id),
UserJoinedServerFilter,
)
self.add_collection(
"joinedVoiceChannel",
lambda user, *_: self._ujvs.get_user_joined_voice_channels_by_user_id(user.user_id),
UserJoinedVoiceChannelFilter,
)
self.set_field("server", self.resolve_server)

13
kdb-bot/src/bot_graphql/query.py
View File

@@ -9,9 +9,12 @@ from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_graphql.abc.query_abc import QueryABC
from bot_graphql.filter.auto_role_filter import AutoRoleFilter
from bot_graphql.filter.auto_role_rule_filter import AutoRoleRuleFilter
from bot_graphql.filter.client_filter import ClientFilter
from bot_graphql.filter.level_filter import LevelFilter
from bot_graphql.filter.server_filter import ServerFilter
from bot_graphql.filter.user_filter import UserFilter
from bot_graphql.filter.user_joined_server_filter import UserJoinedServerFilter
from bot_graphql.filter.user_joined_voice_channel_filter import UserJoinedVoiceChannelFilter
class Query(QueryABC):
@@ -38,12 +41,16 @@ class Query(QueryABC):
self.add_collection("autoRole", lambda *_: self._auto_roles.get_auto_roles(), AutoRoleFilter)
self.add_collection("autoRoleRule", lambda *_: self._auto_roles.get_auto_role_rules(), AutoRoleRuleFilter)
self.add_collection("client", lambda *_: self._clients.get_clients())
self.add_collection("client", lambda *_: self._clients.get_clients(), ClientFilter)
self.add_collection("knownUser", lambda *_: self._known_users.get_users())
self.add_collection("level", lambda *_: self._levels.get_levels(), LevelFilter)
self.add_collection("server", lambda *_: self._servers.get_servers(), ServerFilter)
self.add_collection("userJoinedServer", lambda *_: self._user_joined_servers.get_user_joined_servers())
self.add_collection(
"userJoinedVoiceChannel", lambda *_: self._user_joined_voice_channels.get_user_joined_voice_channels()
"userJoinedServer", lambda *_: self._user_joined_servers.get_user_joined_servers(), UserJoinedServerFilter
)
self.add_collection(
"userJoinedVoiceChannel",
lambda *_: self._user_joined_voice_channels.get_user_joined_voice_channels(),
UserJoinedVoiceChannelFilter,
)
self.add_collection("user", lambda *_: self._users.get_users(), UserFilter)