Feature/new events

.test.env

@@ -32,6 +32,8 @@ CHANNEL_SPOILER=2769521890099371011
 CHANNEL_BOT_LOGS=1105517088266788925
 
 # Roles
+ROLE_VERIFIED=1333333333333333337
+
 ROLE_BIZCTF2022=7629466241011276950
 ROLE_NOAH_GANG=6706800691011276950
 ROLE_BUDDY_GANG=6706800681011276950

src/bot.py

@@ -12,6 +12,7 @@
     MissingRequiredArgument, NoPrivateMessage, UserInputError,
 )
 from sqlalchemy.exc import NoResultFound
+from typing import TypeVar
 
 from src import trace_config
 from src.core import constants, settings
@@ -20,6 +21,9 @@
 logger = logging.getLogger(__name__)
 
 
+BOT_TYPE = TypeVar("BOT_TYPE", "Bot", DiscordBot)
+
+
 class Bot(DiscordBot):
     """Base bot class."""
 

src/cmds/automation/auto_verify.py

@@ -2,12 +2,8 @@
 
 from discord import Member, Message, User
 from discord.ext import commands
-from sqlalchemy import select
 
 from src.bot import Bot
-from src.database.models import HtbDiscordLink
-from src.database.session import AsyncSessionLocal
-from src.helpers.verification import get_user_details, process_identification
 
 logger = logging.getLogger(__name__)
 
@@ -19,31 +15,11 @@ def __init__(self, bot: Bot):
         self.bot = bot
 
     async def process_reverification(self, member: Member | User) -> None:
-        """Re-verifation process for a member."""
-        async with AsyncSessionLocal() as session:
-            stmt = (
-                select(HtbDiscordLink)
-                .where(HtbDiscordLink.discord_user_id == member.id)
-                .order_by(HtbDiscordLink.id)
-                .limit(1)
-            )
-            result = await session.scalars(stmt)
-            htb_discord_link: HtbDiscordLink = result.first()
-
-        if not htb_discord_link:
-            raise VerificationError(f"HTB Discord link for user {member.name} with ID {member}")
-
-        member_token: str = htb_discord_link.account_identifier
-
-        if member_token is None:
-            raise VerificationError(f"HTB account identifier for user {member.name} with ID {member.id} not found")
-
-        logger.debug(f"Processing re-verify of member {member.name} ({member.id}).")
-        htb_details = await get_user_details(member_token)
-        if htb_details is None:
-            raise VerificationError(f"Retrieving user details for user {member.name} with ID {member.id} failed")
-
-        await process_identification(htb_details, user=member, bot=self.bot)
+        """Re-verifation process for a member.
+
+        TODO: Reimplement once it's possible to fetch link state from the HTB Account.
+        """
+        raise VerificationError("Not implemented")
 
     @commands.Cog.listener()
     @commands.cooldown(1, 60, commands.BucketType.user)
@@ -74,4 +50,5 @@ class VerificationError(Exception):
 
 def setup(bot: Bot) -> None:
     """Load the `MessageHandler` cog."""
-    bot.add_cog(MessageHandler(bot))
+    # bot.add_cog(MessageHandler(bot))
+    pass

src/cmds/core/identify.py

@@ -1,17 +1,12 @@
 import logging
-from typing import Sequence
 
-import discord
 from discord import ApplicationContext, Interaction, WebhookMessage, slash_command
 from discord.ext import commands
 from discord.ext.commands import cooldown
-from sqlalchemy import select
 
 from src.bot import Bot
 from src.core import settings
-from src.database.models import HtbDiscordLink
-from src.database.session import AsyncSessionLocal
-from src.helpers.verification import get_user_details, process_identification
+from src.helpers.verification import send_verification_instructions
 
 logger = logging.getLogger(__name__)
 
@@ -25,121 +20,15 @@ def __init__(self, bot: Bot):
     @slash_command(
         guild_ids=settings.guild_ids,
         description="Identify yourself on the HTB Discord server by linking your HTB account ID to your Discord user "
-                    "ID.", guild_only=False
+        "ID.",
+        guild_only=False,
     )
     @cooldown(1, 60, commands.BucketType.user)
-    async def identify(self, ctx: ApplicationContext, account_identifier: str) -> Interaction | WebhookMessage:
-        """Identify yourself on the HTB Discord server by linking your HTB account ID to your Discord user ID."""
-        if len(account_identifier) != 60:
-            return await ctx.respond(
-                "This Account Identifier does not appear to be the right length (must be 60 characters long).",
-                ephemeral=True
-            )
-
-        await ctx.respond("Identification initiated, please wait...", ephemeral=True)
-        htb_user_details = await get_user_details(account_identifier)
-        if htb_user_details is None:
-            embed = discord.Embed(title="Error: Invalid account identifier.", color=0xFF0000)
-            return await ctx.respond(embed=embed, ephemeral=True)
-
-        json_htb_user_id = htb_user_details["user_id"]
-
-        author = ctx.user
-        member = await self.bot.get_or_fetch_user(author.id)
-        if not member:
-            return await ctx.respond(f"Error getting guild member with id: {author.id}.")
-
-        # Step 1: Check if the Account Identifier has already been recorded and if they are the previous owner.
-        # Scenario:
-        #   - I create a new Discord account.
-        #   - I reuse my previous Account Identifier.
-        #   - I now have an "alt account" with the same roles.
-        async with AsyncSessionLocal() as session:
-            stmt = (
-                select(HtbDiscordLink)
-                .filter(HtbDiscordLink.account_identifier == account_identifier)
-                .order_by(HtbDiscordLink.id.desc())
-                .limit(1)
-            )
-            result = await session.scalars(stmt)
-            most_recent_rec: HtbDiscordLink = result.first()
-
-        if most_recent_rec and most_recent_rec.discord_user_id_as_int != member.id:
-            error_desc = (
-                f"Verified user {member.mention} tried to identify as another identified user.\n"
-                f"Current Discord UID: {member.id}\n"
-                f"Other Discord UID: {most_recent_rec.discord_user_id}\n"
-                f"Related HTB UID: {most_recent_rec.htb_user_id}"
-            )
-            embed = discord.Embed(title="Identification error", description=error_desc, color=0xFF2429)
-            await self.bot.get_channel(settings.channels.VERIFY_LOGS).send(embed=embed)
-
-            return await ctx.respond(
-                "Identification error: please contact an online Moderator or Administrator for help.", ephemeral=True
-            )
-
-        # Step 2: Given the htb_user_id from JSON, check if each discord_user_id are different from member.id.
-        # Scenario:
-        #   - I have a Discord account that is linked already to a "Hacker" role.
-        #   - I create a new HTB account.
-        #   - I identify with the new account.
-        #   - `SELECT * FROM htb_discord_link WHERE htb_user_id = %s` will be empty,
-        #         because the new account has not been verified before. All is good.
-        #   - I am now "Noob" rank.
-        async with AsyncSessionLocal() as session:
-            stmt = select(HtbDiscordLink).filter(HtbDiscordLink.htb_user_id == json_htb_user_id)
-            result = await session.scalars(stmt)
-            user_links: Sequence[HtbDiscordLink] = result.all()
-
-        discord_user_ids = {u_link.discord_user_id_as_int for u_link in user_links}
-        if discord_user_ids and member.id not in discord_user_ids:
-            orig_discord_ids = ", ".join([f"<@{id_}>" for id_ in discord_user_ids])
-            error_desc = (f"The HTB account {json_htb_user_id} attempted to be identified by user <@{member.id}>, "
-                          f"but is tied to another Discord account.\n"
-                          f"Originally linked to Discord UID {orig_discord_ids}.")
-            embed = discord.Embed(title="Identification error", description=error_desc, color=0xFF2429)
-            await self.bot.get_channel(settings.channels.VERIFY_LOGS).send(embed=embed)
-
-            return await ctx.respond(
-                "Identification error: please contact an online Moderator or Administrator for help.", ephemeral=True
-            )
-
-        # Step 3: Check if discord_user_id already linked to an htb_user_id, and if JSON/db HTB IDs are the same.
-        # Scenario:
-        #   - I have a new, unlinked Discord account.
-        #   - Clubby generates a new token and gives it to me.
-        #   - `SELECT * FROM htb_discord_link WHERE discord_user_id = %s`
-        #         will be empty because I have not identified before.
-        #   - I am now Clubby.
-        async with AsyncSessionLocal() as session:
-            stmt = select(HtbDiscordLink).filter(HtbDiscordLink.discord_user_id == member.id)
-            result = await session.scalars(stmt)
-            user_links: Sequence[HtbDiscordLink] = result.all()
-
-        user_htb_ids = {u_link.htb_user_id_as_int for u_link in user_links}
-        if user_htb_ids and json_htb_user_id not in user_htb_ids:
-            error_desc = (f"User {member.mention} ({member.id}) tried to identify with a new HTB account.\n"
-                          f"Original HTB UIDs: {', '.join([str(i) for i in user_htb_ids])}, new HTB UID: "
-                          f"{json_htb_user_id}.")
-            embed = discord.Embed(title="Identification error", description=error_desc, color=0xFF2429)
-            await self.bot.get_channel(settings.channels.VERIFY_LOGS).send(embed=embed)
-
-            return await ctx.respond(
-                "Identification error: please contact an online Moderator or Administrator for help.", ephemeral=True
-            )
-
-        htb_discord_link = HtbDiscordLink(
-            account_identifier=account_identifier, discord_user_id=member.id, htb_user_id=json_htb_user_id
-        )
-        async with AsyncSessionLocal() as session:
-            session.add(htb_discord_link)
-            await session.commit()
-
-        await process_identification(htb_user_details, user=member, bot=self.bot)
-
-        return await ctx.respond(
-            f"Your Discord user has been successfully identified as HTB user {json_htb_user_id}.", ephemeral=True
-        )
+    async def identify(
+        self, ctx: ApplicationContext, account_identifier: str
+    ) -> Interaction | WebhookMessage:
+        """Legacy command. Now sends instructions to identify with HTB account."""
+        await send_verification_instructions(ctx, ctx.author)
 
 
 def setup(bot: Bot) -> None:

src/cmds/core/verify.py

@@ -1,14 +1,12 @@
 import logging
 
-import discord
 from discord import ApplicationContext, Interaction, WebhookMessage, slash_command
-from discord.errors import Forbidden, HTTPException
 from discord.ext import commands
 from discord.ext.commands import cooldown
 
 from src.bot import Bot
 from src.core import settings
-from src.helpers.verification import process_certification
+from src.helpers.verification import process_certification, send_verification_instructions
 
 logger = logging.getLogger(__name__)
 
@@ -47,57 +45,7 @@ async def verifycertification(self, ctx: ApplicationContext, certid: str, fullna
     @cooldown(1, 60, commands.BucketType.user)
     async def verify(self, ctx: ApplicationContext) -> Interaction | WebhookMessage:
         """Receive instructions in a DM on how to identify yourself with your HTB account."""
-        member = ctx.user
-
-        # Step one
-        embed_step1 = discord.Embed(color=0x9ACC14)
-        embed_step1.add_field(
-            name="Step 1: Log in at Hack The Box",
-            value="Go to the Hack The Box website at <https://www.hackthebox.com/>"
-                  " and navigate to **Login > HTB Labs**. Log in to your HTB Account."
-                  , inline=False, )
-        embed_step1.set_image(
-            url="https://media.discordapp.net/attachments/724587782755844098/839871275627315250/unknown.png"
-        )
-
-        # Step two
-        embed_step2 = discord.Embed(color=0x9ACC14)
-        embed_step2.add_field(
-            name="Step 2: Locate the Account Identifier",
-            value='Click on your profile name, then select **My Profile**. '
-                  'In the Profile Settings tab, find the field labeled **Account Identifier**. (<https://app.hackthebox.com/profile/settings>) '
-                  "Click the green button to copy your secret identifier.", inline=False, )
-        embed_step2.set_image(
-            url="https://media.discordapp.net/attachments/724587782755844098/839871332963188766/unknown.png"
-        )
-
-        # Step three
-        embed_step3 = discord.Embed(color=0x9ACC14)
-        embed_step3.add_field(
-            name="Step 3: Identification",
-            value="Now type `/identify IDENTIFIER_HERE` in the bot-commands channel.\n\nYour roles will be "
-                  "applied automatically.", inline=False
-        )
-        embed_step3.set_image(
-            url="https://media.discordapp.net/attachments/709907130102317093/904744444539076618/unknown.png"
-        )
-
-        try:
-            await member.send(embed=embed_step1)
-            await member.send(embed=embed_step2)
-            await member.send(embed=embed_step3)
-        except Forbidden as ex:
-            logger.error("Exception during verify call", exc_info=ex)
-            return await ctx.respond(
-                "Whoops! I cannot DM you after all due to your privacy settings. Please allow DMs from other server "
-                "members and try again in 1 minute."
-            )
-        except HTTPException as ex:
-            logger.error("Exception during verify call.", exc_info=ex)
-            return await ctx.respond(
-                "An unexpected error happened (HTTP 400, bad request). Please contact an Administrator."
-            )
-        return await ctx.respond("Please check your DM for instructions.", ephemeral=True)
+        await send_verification_instructions(ctx, ctx.author)
 
 
 def setup(bot: Bot) -> None:

src/core/config.py

@@ -91,6 +91,7 @@ class AcademyCertificates(BaseSettings):
 
 class Roles(BaseSettings):
     """The roles settings."""
+    VERIFIED: int
 
     # Moderation
     COMMUNITY_MANAGER: int
@@ -330,6 +331,8 @@ def load_settings(env_file: str | None = None):
             global_settings.roles.NOOB,
             global_settings.roles.VIP,
             global_settings.roles.VIP_PLUS,
+        ],
+        "ALL_SEASON_RANKS": [
             global_settings.roles.SEASON_HOLO,
             global_settings.roles.SEASON_PLATINUM,
             global_settings.roles.SEASON_RUBY,