Implement permissions system (#50)

Author: No767

bot/cogs/config.py

@@ -7,7 +7,8 @@
 import msgspec
 from async_lru import alru_cache
 from discord.ext import commands
-from libs.utils import GuildContext, is_manager
+from libs.utils import GuildContext
+from libs.utils.checks import bot_check_permissions, check_permissions
 
 if TYPE_CHECKING:
     from rodhaj import Rodhaj
@@ -119,7 +120,8 @@ async def get_guild_config(self, guild_id: int) -> Optional[GuildConfig]:
         config = GuildConfig(bot=self.bot, **dict(rows))
         return config
 
-    @is_manager()
+    @check_permissions(manage_guild=True)
+    @bot_check_permissions(manage_channels=True, manage_webhooks=True)
     @commands.guild_only()
     @commands.hybrid_group(name="config")
     async def config(self, ctx: GuildContext) -> None:
@@ -253,8 +255,8 @@ async def setup(self, ctx: GuildContext, *, flags: SetupFlags) -> None:
             return
 
         query = """
-        INSERT INTO guild_config (id, category_id, ticket_channel_id, logging_channel_id, logging_broadcast_url, ticket_broadcast_url)
-        VALUES ($1, $2, $3, $4, $5, $6);
+        INSERT INTO guild_config (id, category_id, ticket_channel_id, logging_channel_id, logging_broadcast_url, ticket_broadcast_url, prefix)
+        VALUES ($1, $2, $3, $4, $5, $6, $7);
         """
         try:
             await self.pool.execute(
@@ -265,6 +267,7 @@ async def setup(self, ctx: GuildContext, *, flags: SetupFlags) -> None:
                 logging_channel.id,
                 lgc_webhook.url,
                 tc_webhook.url,
+                [],
             )
         except asyncpg.UniqueViolationError:
             await ticket_channel.delete(reason=delete_reason)

bot/cogs/tickets.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-from functools import lru_cache
 from typing import TYPE_CHECKING, Annotated, NamedTuple, Optional, Union
 
 import asyncpg
@@ -14,6 +13,7 @@
     get_partial_ticket,
     safe_content,
 )
+from libs.utils.checks import bot_check_permissions
 from libs.utils.embeds import Embed, LoggingEmbed
 
 from .config import GuildWebhookDispatcher
@@ -23,7 +23,6 @@
     from rodhaj import Rodhaj
 
 
-STAFF_ROLE = 1184257456419913798
 TICKET_EMOJI = "\U0001f3ab"  # U+1F3AB Ticket
 
 
@@ -101,15 +100,6 @@ def add_status_checklist(
     ) -> StatusChecklist:
         return self.in_progress_tickets.setdefault(author_id, status)
 
-    #### Determining staff
-
-    @lru_cache(maxsize=64)
-    def get_staff(self, guild: discord.Guild) -> Optional[list[discord.Member]]:
-        mod_role = guild.get_role(STAFF_ROLE)
-        if mod_role is None:
-            return None
-        return [member for member in mod_role.members]
-
     ### Conditions for closing tickets
 
     async def can_close_ticket(self, ctx: RoboContext):
@@ -128,22 +118,17 @@ async def can_close_ticket(self, ctx: RoboContext):
         return False
 
     async def can_admin_close_ticket(self, ctx: RoboContext) -> bool:
-        guild_id = self.bot.transprogrammer_guild_id
-        guild = self.bot.get_guild(guild_id) or (await self.bot.fetch_guild(guild_id))
-        staff_members = self.get_staff(guild)
-
-        if staff_members is None:
-            return False
-
-        # TODO: Add the hierarchy system here
-        staff_ids = [member.id for member in staff_members]
+        # More than likely it will be closed through the threads
+        # That means, it must be done in a guild. Thus, we know that
+        # it will always be discord.Member
+        perms = ctx.channel.permissions_for(ctx.author)  # type: ignore
         from_ticket_channel = (
             isinstance(ctx.channel, discord.Thread)
             and ctx.partial_config is not None
             and ctx.channel.parent_id == ctx.partial_config.ticket_channel_id
         )
 
-        if ctx.author.id in staff_ids and from_ticket_channel is True:
+        if perms.manage_threads and from_ticket_channel is True:
             return True
         return False
 
@@ -319,11 +304,18 @@ def get_solved_tag(
 
     ### Feature commands
 
+    # This command requires the manage_threads permissions for the bot
     @is_ticket_or_dm()
+    @bot_check_permissions(manage_threads=True)
     @commands.cooldown(1, 20, commands.BucketType.channel)
     @commands.hybrid_command(name="close", aliases=["solved", "closed", "resolved"])
     async def close(self, ctx: RoboContext) -> None:
-        """Closes the thread"""
+        """Closes a ticket
+
+        If someone requests to close the ticket
+        and has Manage Threads permissions, then they can
+        also close the ticket.
+        """
         query = """
         DELETE FROM tickets
         WHERE thread_id = $1 AND owner_id = $2;

bot/libs/utils/checks.py

@@ -1,18 +1,22 @@
 from __future__ import annotations
 
 import os
-from typing import Callable, TypeVar
+from typing import TYPE_CHECKING, Callable, TypeVar
 
+import discord
 from discord import app_commands
 from discord.ext import commands
 
-T = TypeVar("T")
+# Although commands.HybridCommand (and it's group version) can be bound here for Type T,
+# it doesn't make sense as they are just subclasses of commands.Command and co.
+T = TypeVar("T", commands.Command, commands.Group)
+
+if TYPE_CHECKING:
+    from libs.utils.context import RoboContext
 
 
-# For time's sake I might as well take these from RDanny
-# There is really no used of creating my own system when there is one out there already
 async def check_guild_permissions(
-    ctx: commands.Context, perms: dict[str, bool], *, check=all
+    ctx: RoboContext, perms: dict[str, bool], *, check=all
 ) -> bool:
     is_owner = await ctx.bot.is_owner(ctx.author)
     if is_owner:
@@ -27,9 +31,48 @@ async def check_guild_permissions(
     )
 
 
-def hybrid_permissions_check(**perms: bool) -> Callable[[T], T]:
-    async def pred(ctx: commands.Context):
-        return await check_guild_permissions(ctx, perms)
+async def check_bot_permissions(
+    ctx: RoboContext, perms: dict[str, bool], *, check=all
+) -> bool:
+    is_owner = await ctx.bot.is_owner(ctx.author)
+    if is_owner:
+        return True
+
+    if ctx.guild is None:
+        return False
+
+    bot_resolved_perms = ctx.me.guild_permissions  # type: ignore
+    return check(
+        getattr(bot_resolved_perms, name, None) == value
+        for name, value in perms.items()
+    )
+
+
+def check_permissions(**perms: bool) -> Callable[[T], T]:
+    async def pred(ctx: RoboContext):
+        # Usually means this is in the context of a DM
+        if (
+            isinstance(ctx.me, discord.ClientUser)
+            or isinstance(ctx.author, discord.User)
+            or ctx.guild is None
+        ):
+            return False
+        guild_perms = await check_guild_permissions(ctx, perms)
+        can_run = ctx.me.top_role > ctx.author.top_role
+        return guild_perms and can_run
+
+    def decorator(func: T) -> T:
+        func.extras["permissions"] = perms
+        commands.check(pred)(func)
+        app_commands.default_permissions(**perms)(func)
+        return func
+
+    return decorator
+
+
+def bot_check_permissions(**perms: bool) -> Callable[[T], T]:
+    async def pred(ctx: RoboContext):
+        return await check_bot_permissions(ctx, perms)
 
     def decorator(func: T) -> T:
         commands.check(pred)(func)
@@ -40,17 +83,17 @@ def decorator(func: T) -> T:
 
 
 def is_manager():
-    return hybrid_permissions_check(manage_guild=True)
+    return check_permissions(manage_guild=True)
 
 
 def is_mod():
-    return hybrid_permissions_check(
+    return check_permissions(
         ban_members=True, manage_messages=True, kick_members=True, moderate_members=True
     )
 
 
 def is_admin():
-    return hybrid_permissions_check(administrator=True)
+    return check_permissions(administrator=True)
 
 
 def is_docker() -> bool:

bot/libs/utils/help.py

@@ -13,6 +13,28 @@
 # Light Orange (255, 199, 184) - Used for command pages
 
 
+def process_perms_name(
+    command: Union[commands.Group, commands.Command]
+) -> Optional[str]:
+    merge_list = []
+    if (
+        all(isinstance(parent, commands.Group) for parent in command.parents)
+        and len(command.parents) > 0
+    ):
+        # See https://stackoverflow.com/a/27638751
+        merge_list = [
+            next(iter(parent.extras["permissions"])) for parent in command.parents
+        ]
+
+    if "permissions" in command.extras:
+        merge_list.extend([*command.extras["permissions"]])
+
+    perms_set = sorted(set(merge_list))
+    if len(perms_set) == 0:
+        return None
+    return ", ".join(name.replace("_", " ").title() for name in perms_set)
+
+
 class GroupHelpPageSource(menus.ListPageSource):
     def __init__(
         self,
@@ -27,10 +49,15 @@ def __init__(
         self.title: str = f"{self.group.qualified_name} Commands"
         self.description: str = self.group.description
 
+    def _process_description(self, group: Union[commands.Group, commands.Cog]):
+        if isinstance(group, commands.Group) and "permissions" in group.extras:
+            return f"{self.description}\n\n**Required Permissions**: {process_perms_name(group)}"
+        return self.description
+
     async def format_page(self, menu: RoboPages, commands: list[commands.Command]):
         embed = discord.Embed(
             title=self.title,
-            description=self.description,
+            description=self._process_description(self.group),
             colour=discord.Colour.from_rgb(197, 184, 255),
         )
 
@@ -271,8 +298,16 @@ async def send_cog_help(self, cog):
         )
         await menu.start()
 
-    def common_command_formatting(self, embed_like, command):
+    def common_command_formatting(
+        self,
+        embed_like: Union[discord.Embed, GroupHelpPageSource],
+        command: commands.Command,
+    ):
         embed_like.title = self.get_command_signature(command)
+        processed_perms = process_perms_name(command)
+        if isinstance(embed_like, discord.Embed) and processed_perms is not None:
+            embed_like.add_field(name="Required Permissions", value=processed_perms)
+
         if command.description:
             embed_like.description = f"{command.description}\n\n{command.help}"
         else:

bot/migrations/V3__perms_and_config_changes.sql

@@ -0,0 +1,11 @@
+-- Revision Version: V3
+-- Revises: V2
+-- Creation Date: 2024-01-23 08:41:08.795638 UTC
+-- Reason: perms and config changes
+
+-- Remove this column as it was never used
+ALTER TABLE IF EXISTS guild_config DROP COLUMN locked;
+
+-- Also in lieu with permissions based commands, 
+-- we don't need to store perms levels on users
+ALTER TABLE IF EXISTS user_config DROP COLUMN permission_level;

bot/migrations/V4__custom_prefix.sql

@@ -0,0 +1,9 @@
+-- Revision Version: V4
+-- Revises: V3
+-- Creation Date: 2024-01-24 02:54:39.500620 UTC
+-- Reason: custom prefix support
+
+-- Allow for custom prefixes to be stored. This is simply setup work
+-- for another feature
+ALTER TABLE IF EXISTS guild_config ADD COLUMN prefix TEXT[];
+

bot/rodhaj.py

@@ -48,6 +48,9 @@ def __init__(
             activity=discord.Activity(
                 type=discord.ActivityType.watching, name="a game"
             ),
+            allowed_mentions=discord.AllowedMentions(
+                everyone=False, replied_user=False
+            ),
             command_prefix=["r>", "?", "!"],
             help_command=RodhajHelp(),
             intents=intents,

permissions.md

@@ -0,0 +1,9 @@
+# Required permissions for Rodhaj
+
+This document serves to provide the necessary permissions
+that Rodhaj requires. Currently these are the required
+permissions:
+
+- Manage Threads
+- Manage Channels
+- Manage Webhooks