feat(pterodactyl): added a regex blacklist command

This commit is contained in:
SeaswimmerTheFsh 2024-03-02 19:27:36 -05:00
parent 5c2bfea238
commit f6ebbae583
Signed by untrusted user: cswimr
GPG key ID: B8953EC01E5C4063
4 changed files with 58 additions and 3 deletions

View file

@ -21,4 +21,5 @@ def register_config(config_obj: Config) -> None:
join_msg='Welcome to the server! 👋',
leave_msg='Goodbye! 👋',
mask_ip=True,
regex_blacklist={},
)

View file

@ -8,6 +8,7 @@ from pydactyl import PterodactylClient
from redbot.core import commands
from redbot.core.bot import Red
from redbot.core.utils.chat_formatting import box
from redbot.core.utils.views import ConfirmView
from pterodactyl.config import config, register_config
from pterodactyl.logger import logger
@ -286,6 +287,45 @@ class Pterodactyl(commands.Cog):
await config.api_endpoint.set(endpoint)
await ctx.send(f"API endpoint set to {endpoint}")
@pterodactyl_config_regex.group(name = "blacklist", aliases = ['block', 'blocklist'])
async def pterodactyl_config_regex_blacklist(self, ctx: commands.Context):
"""Blacklist regex patterns."""
@pterodactyl_config_regex_blacklist.command(name = "add")
async def pterodactyl_config_regex_blacklist_add(self, ctx: commands.Context, name: str, *, regex: str) -> None:
"""Add a regex pattern to the blacklist."""
async with config.regex_blacklist() as blacklist:
blacklist: dict
if name not in blacklist:
blacklist.update({name: regex})
await ctx.send(f"Added `{name}` to the regex blacklist.\n{box(regex, 're')}")
else:
view = ConfirmView(ctx.author, disable_buttons=True)
msg = await ctx.send(f"Name `{name}` already exists in the blacklist. Would you like to update it? Current value:\n{box(blacklist[name], 're')}", view=view)
await view.wait()
if view.result is True:
blacklist.update({name: regex})
await msg.edit(f"Updated `{name}` in the regex blacklist.\n{box(regex, 're')}")
else:
await msg.edit(content="Cancelled.")
@pterodactyl_config_regex_blacklist.command(name = "remove")
async def pterodactyl_config_regex_blacklist_remove(self, ctx: commands.Context, name: str) -> None:
"""Remove a regex pattern from the blacklist."""
async with config.regex_blacklist() as blacklist:
blacklist: dict
if name in blacklist:
view = ConfirmView(ctx.author, disable_buttons=True)
msg = await ctx.send(f"Are you sure you want to remove `{name}` from the regex blacklist?\n{box(blacklist[name], 're')}", view=view)
await view.wait()
if view.result is True:
del blacklist[name]
await msg.edit(f"Removed `{name}` from the regex blacklist.")
else:
await msg.edit(content="Cancelled.")
else:
await ctx.send(f"Name `{name}` does not exist in the blacklist.")
@pterodactyl_config.command(name = 'view', aliases = ['show'])
async def pterodactyl_config_view(self, ctx: commands.Context) -> None:
"""View the current configuration."""
@ -305,6 +345,7 @@ class Pterodactyl(commands.Cog):
leave_msg = await config.leave_msg()
mask_ip = await config.mask_ip()
api_endpoint = await config.api_endpoint()
regex_blacklist = await config.regex_blacklist()
embed = discord.Embed(color = await ctx.embed_color(), title="Pterodactyl Configuration")
embed.description = f"""**Base URL:** {base_url}
**Server ID:** `{server_id}`
@ -323,7 +364,14 @@ class Pterodactyl(commands.Cog):
**Join Regex:** {box(join_regex, 're')}
**Leave Regex:** {box(leave_regex, 're')}
**Achievement Regex:** {box(achievement_regex, 're')}"""
await ctx.send(embed=embed)
await ctx.send(embes=embed)
if not len(regex_blacklist) == 0:
regex_blacklist_embed = discord.Embed(color = await ctx.embed_color(), title="Regex Blacklist")
regex_string = ''
for regex in regex_blacklist:
regex_string += f"{box(regex, 're')}\n"
regex_blacklist_embed.description = regex_string
await ctx.send(embed=regex_blacklist_embed)
def get_bool_str(self, inp: bool) -> str:
"""Return a string representation of a boolean."""

View file

@ -45,8 +45,10 @@ async def establish_websocket_connection(coginstance: Pterodactyl) -> None:
logger.info("WebSocket authentication successful")
if json.loads(message)['event'] == 'console output' and await config.console_channel() is not None:
if await config.current_status() in ('running', 'offline', ''):
content = remove_ansi_escape_codes(json.loads(message)['args'][0])
msg = json.loads(message)['args'][0]
regex_blacklist = await config.regex_blacklist()
if await config.current_status() in ('running', 'offline', '') and not any(re.match(regex, msg) for regex in regex_blacklist):
content = remove_ansi_escape_codes(msg)
if await config.mask_ip() is True:
content = mask_ip(content)