import json import logging import os from time import sleep from typing import List import discord from mcrcon import MCRcon from sqlalchemy import select, and_ from sqlalchemy.orm import Session from config.config import ConfigLoader from database.DB import get_engine, MinecraftUser, MinecraftServer, DiscordUser from log.Log import Log def add_user_name_to_whitelist(domain: str, user_name: str): """ Adds a minecraft user_name to the whitelist :param domain: :param user_name: :return: """ log = logging.getLogger("MC-API") with MCRcon(os.environ.get('MCRCON_HOST'), os.environ.get('MCRCON_PW')) as mcr: log.info(f"Adding minecraft user \"{user_name}\" to the whitelist of \"{domain}\"") resp = mcr.command(f"/whitelist add {user_name}") print(resp) def get_all_users_on_whitelist(domain: str) -> dict: """ Gets a list of all users on the whitelist by reading the server's whitelist.json :param domain: :return: """ c: ConfigLoader = ConfigLoader.get_config_loader() with open(c.whitelist_location_per_domain[domain], 'r') as whitelist_file: whitelist = json.load(whitelist_file) return whitelist def get_list_of_users_on_whitelist(domain: str) -> List[str]: on_whitelist = [name for name in map(lambda x: x['name'], get_all_users_on_whitelist(domain))] return on_whitelist def remove_user_name_from_whitelist(domain: str, user_name: str): """ Removes a minecraft user_name from the whitelist :param domain: :param user_name: :return: """ log = logging.getLogger("MC-API") with MCRcon(os.environ.get('MCRCON_HOST'), os.environ.get('MCRCON_PW')) as mcr: log.info(f"Removing minecraft user \"{user_name}\" from the whitelist of \"{domain}\"") resp = mcr.command(f"/whitelist remove {user_name}") print(resp) async def handle_minecraft_application(mc_name: str, mc_uuid: str, domain: str, discord_user: discord.User, interaction: discord.Interaction): """ Applies an application to the server whitelist or ignores it depending on if the minecraft user is already whitelisted. Will alter the database to reflecte the current state. :param mc_name: :param mc_uuid: :param domain: :param discord_user: :param interaction :return: """ log = logging.getLogger("MC-API") c: ConfigLoader = ConfigLoader.get_config_loader() engine = get_engine() with Session(engine) as session: # query if the user is already related to our server stmt = select(MinecraftUser).where( MinecraftUser.minecraft_servers.any(and_(MinecraftServer.domain == domain, MinecraftUser.user_name == mc_name))) mc_user_on_server: MinecraftUser = session.scalars(stmt).first() # query for the server obj stmt = select(MinecraftServer).where(MinecraftServer.domain == domain) dc_user: DiscordUser = session.scalars( select(DiscordUser).where(DiscordUser.snowflake_id == str(discord_user.id)) ).first() curr_server: MinecraftServer = session.scalars(stmt).first() on_whitelist = get_list_of_users_on_whitelist(domain) if mc_user_on_server is not None: # this user is already associated with the server (could be missing on whitelist?) log.info( f"Found minecraft username \"{mc_name}\" to be already associated with the server. Aborting ...") if mc_name not in on_whitelist: log.error(f"Minecraft username \"{mc_name}\" was found in the database but is " f"not in \"whitelist.json\"! Did someone manually remove them? Removing the relationship ...") mc_user_on_server.minecraft_servers.remove(curr_server) session.add(mc_user_on_server) session.commit() if dc_user.minecraft_user_uuid == mc_uuid: log.info(f"Discord user \"{dc_user.handle}\" ({dc_user.snowflake_id}) tried to add minecraft user " f"\"{mc_name}\" ({mc_uuid}) twice! Aborting...") #interaction.response.send_message(f"The minecraft user \"{mc_name}\" was already added by you!\n" # f"Not doing anything because this seems unnecessary.\n\n" # f"If this is an error inform an admin!", # ephemeral=True) return # do not inform user about this, otherwise leaks who is on the server by guessing their name with the bot # minecraft user is not associated with server or was just dis-associated (could still be on whitelist) log.info( f"Did not find minecraft user \"{mc_name}\" in relation to server \"{curr_server.domain}\"") if mc_name in on_whitelist: log.warning( f"Minecraft username \"{mc_name}\" was not found in the database but is " f"in \"whitelist.json\"! Were they added manually? Ignoring that.") return # query for the minecraft user mc_user = session.scalars( select(MinecraftUser).where(MinecraftUser.user_name == str(mc_name)) ).first() if mc_user is None: # create mc_user mc_user = MinecraftUser(user_name=str(mc_name), uuid=mc_uuid) session.add(mc_user) session.commit() if dc_user is None: # got to create discord user dc_user = DiscordUser(snowflake_id=str(discord_user.id), handle=str(discord_user.name), minecraft_user_uuid=mc_uuid) session.add(dc_user) else: # discord user is known if dc_user.minecraft_user_uuid is not None and dc_user.minecraft_user_uuid != mc_uuid: # discord user already had another minecraft account old_mc_user: MinecraftUser = session.scalars(select(MinecraftUser).where( MinecraftUser.uuid == dc_user.minecraft_user_uuid )).first() log.info(f"Discord user \"{dc_user.handle}\" ({dc_user.snowflake_id}) already had a minecraft uuid " f"associated: \"{old_mc_user.user_name}\" ({old_mc_user.uuid})\n" f"Updating it to \"{mc_name}\" ({mc_uuid})!") remove_user_name_from_whitelist(domain, str(old_mc_user.user_name)) dc_user.minecraft_user_uuid = mc_uuid session.add(dc_user) mc_user.minecraft_servers.append(curr_server) session.commit() add_user_name_to_whitelist(domain, mc_name) #await interaction.response.send_message("Thanks for your response. You've been added to the server's whitelist!\n" # "Enjoy your time playing!\n\n" # "Also don't forget to inform yourself about available mods or " # "which minecraft version to use! Some are: \n" # f"{c.post_application_text(domain)}", # ephemeral=True) async def handle_minecraft_retirement(domain: str, discord_user: discord.User, interaction: discord.Interaction): """ Only removes any minecraft user associated with a discord user. :param domain: :param discord_user: :param interaction: :return: """ log = logging.getLogger("MC-API") engine = get_engine() with Session(engine) as session: dc_user: DiscordUser = session.scalars(select(DiscordUser).where( DiscordUser.snowflake_id == str(discord_user.id) )).first() if dc_user is None: # unknown discord user called this command log.info(f"Can't find discord user \"{discord_user.name}\" (\"{discord_user.id}\") in the database!") await interaction.response.send_message(f"Hey, {discord_user.name}!\nWould love to help but don't think we have met before!\n" f"If you're still on the whitelist of \"{domain}\" " "please tell an admin or moderator instead. " "They'll figure this out for you!", ephemeral=True) return else: # discord user has interacted with this bot before if dc_user.minecraft_user_uuid is None: # no minecraft user known (used this command before or manual edit)? log.info(f"Found discord user \"{dc_user.handle}\" ({dc_user.snowflake_id}) in the database " f"but there was no minecraft UUID associated with it.") await interaction.response.send_message("Couldn't find any minecraft user associated with your account.\n" "But I've seen you before! Have you used this command already?", ephemeral=True) return mc_user: MinecraftUser = session.scalars(select(MinecraftUser).where( MinecraftUser.uuid == dc_user.minecraft_user_uuid )).first() if mc_user is None: log.critical(f"Could not find minecraft user with UUID \"{dc_user.minecraft_user_uuid}\" but it was " f"associated with discord account \"{dc_user.handle}\" ({dc_user.snowflake_id}). " f"Something is broken in the database schema! Did a delete not cascade?") exit(1) curr_server: MinecraftServer = session.scalars(select(MinecraftServer).where( MinecraftServer.domain == domain )).first() if curr_server is None: # domain is unknown to the database?! log.critical(f"Tried to look up minecraft server \"{domain}\" but the database query was empty! \n" f"Something is broken in the database!") exit(1) dc_user.minecraft_user_uuid = None session.add(dc_user) mc_user.minecraft_servers.remove(curr_server) session.add(mc_user) log.info(f"Discord user \"{dc_user.handle}\" ({dc_user.snowflake_id}) is leaving server \"{domain}\" with " f"minecraft user \"{mc_user.user_name}\" ({mc_user.uuid})!") remove_user_name_from_whitelist(curr_server.domain, str(mc_user.user_name)) session.commit() await interaction.response.send_message(f"Removed your minecraft account \"{mc_user.user_name}\" " f"from the server's whitelist.\n" f"Take care!", ephemeral=True) if __name__ == "__main__": Log.setup() print(get_all_users_on_whitelist("feather-mc.pandro.de")) sleep(2) add_user_name_to_whitelist("feather-mc.pandro.de", "MausKomant") sleep(2) print(get_all_users_on_whitelist("feather-mc.pandro.de")) remove_user_name_from_whitelist("feather-mc.pandro.de", "MausKomant") sleep(2) print(get_all_users_on_whitelist("feather-mc.pandro.de"))