You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
224 lines
11 KiB
Python
224 lines
11 KiB
Python
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")) |