diff --git a/bot.py b/bot.py index 780e94c7ab..48ad5982bb 100644 --- a/bot.py +++ b/bot.py @@ -24,6 +24,8 @@ from emoji import UNICODE_EMOJI from pkg_resources import parse_version +from core.blocklist import Blocklist, BlockReason + try: # noinspection PyUnresolvedReferences from colorama import init @@ -87,6 +89,9 @@ def __init__(self): self._configure_logging() self.plugin_db = PluginDatabaseClient(self) # Deprecated + + self.blocklist = Blocklist(bot=self) + self.startup() def get_guild_icon( @@ -168,6 +173,11 @@ def startup(self): logger.line() logger.info("discord.py: v%s", discord.__version__) logger.line() + if not self.config["blocked"] or not self.config["blocked_roles"]: + logger.warning( + "Un-migrated blocklists found. Please run the '[p]migrate blocklist' command after backing " + "up your config/database. Blocklist functionality will be disabled until this is done." + ) async def load_extensions(self): for cog in self.loaded_cogs: @@ -462,10 +472,14 @@ def main_category(self) -> typing.Optional[discord.CategoryChannel]: @property def blocked_users(self) -> typing.Dict[str, str]: + """DEPRECATED, used blocklist instead""" + logger.warning("blocked_users is deprecated and does not function, its usage is a bug") return self.config["blocked"] @property def blocked_roles(self) -> typing.Dict[str, str]: + """DEPRECATED, used blocklist instead""" + logger.warning("blocked_roles is deprecated and does not function, its usage is a bug") return self.config["blocked_roles"] @property @@ -524,6 +538,7 @@ async def on_connect(self): logger.debug("Connected to gateway.") await self.config.refresh() await self.api.setup_indexes() + await self.blocklist.setup() await self.load_extensions() self._connected.set() @@ -718,6 +733,8 @@ def check_guild_age(self, author: discord.Member) -> bool: return True def check_manual_blocked_roles(self, author: discord.Member) -> bool: + """DEPRECATED""" + logger.error("check_manual_blocked_roles is deprecated, usage is a bug.") for role in author.roles: if str(role.id) not in self.blocked_roles: continue @@ -734,6 +751,8 @@ def check_manual_blocked_roles(self, author: discord.Member) -> bool: return True def check_manual_blocked(self, author: discord.User) -> bool: + """DEPRECATED""" + logger.error("check_manual_blocked is deprecated, usage is a bug.") if str(author.id) not in self.blocked_users: return True @@ -758,6 +777,7 @@ async def _process_blocked(self, message): # This is to store blocked message cooldown in memory _block_msg_cooldown = dict() + # This has a bunch of side effects async def is_blocked( self, author: discord.User, @@ -765,6 +785,24 @@ async def is_blocked( channel: discord.TextChannel = None, send_message: bool = False, ) -> bool: + """ + Check if a user is blocked for any reason and send a message if they are (if send_message is true). + + If you are using this method with send_message set to false or not set, + You should be using blocklist.is_user_blocked() or blocklist.is_id_blocked() + if you only care whether a user is manually blocked then use blocklist.is_id_blocked(). + + Parameters + ---------- + author + channel + send_message + + Returns + ------- + bool + Whether the user is blocked or not. + """ member = self.guild.get_member(author.id) or await MemberConverter.convert(author) if member is None: # try to find in other guilds @@ -794,33 +832,36 @@ async def send_embed(title=None, desc=None): self._block_msg_cooldown[str(author)] = now + timedelta(minutes=5) return await channel.send(embed=emb) - if str(author.id) in self.blocked_whitelisted_users: - if str(author.id) in self.blocked_users: - self.blocked_users.pop(str(author.id)) - await self.config.update() - return False - - if not self.check_account_age(author): - blocked_reason = "Sorry, your account is too new to initiate a ticket." - await send_embed(title="Message not sent!", desc=blocked_reason) - return True - - if not self.check_manual_blocked(author): - blocked_reason = "You have been blocked from contacting Modmail." - await send_embed(title="Message not sent!", desc=blocked_reason) - return True - - if not self.check_guild_age(member): - blocked_reason = "Sorry, you joined the server too recently to initiate a ticket." - await send_embed(title="Message not sent!", desc=blocked_reason) - return True - - if not self.check_manual_blocked_roles(member): - blocked_reason = "Sorry, your role(s) has been blacklisted from contacting Modmail." - await send_embed(title="Message not sent!", desc=blocked_reason) - return True + if member is not None: + blocked, block_type = self.blocklist.is_user_blocked(member) + else: + blocked, block_type = self.blocklist.is_id_blocked(author.id) + if not self.blocklist.is_valid_account_age(author): + blocked = True + block_type = BlockReason.ACCOUNT_AGE - return False + if blocked: + if block_type == BlockReason.ACCOUNT_AGE: + blocked_reason = "Sorry, your account is too new to initiate a ticket." + await send_embed(title="Message not sent!", desc=blocked_reason) + return True + + if block_type == BlockReason.BLOCKED_USER: + blocked_reason = "You have been blocked from contacting Modmail." + await send_embed(title="Message not sent!", desc=blocked_reason) + return True + + if block_type == BlockReason.GUILD_AGE: + blocked_reason = "Sorry, you joined the server too recently to initiate a ticket." + await send_embed(title="Message not sent!", desc=blocked_reason) + return True + + if block_type == BlockReason.BLOCKED_ROLE: + blocked_reason = "Sorry, your role(s) has been blacklisted from contacting Modmail." + await send_embed(title="Message not sent!", desc=blocked_reason) + return True + + return blocked async def get_thread_cooldown(self, author: discord.Member): thread_cooldown = self.config.get("thread_cooldown") diff --git a/cogs/modmail.py b/cogs/modmail.py index e41ecbe321..061640e264 100644 --- a/cogs/modmail.py +++ b/cogs/modmail.py @@ -1,5 +1,4 @@ import asyncio -import datetime import re from datetime import timezone from itertools import zip_longest @@ -12,7 +11,8 @@ from discord.ext.commands.cooldowns import BucketType from discord.ext.commands.view import StringView -from core import checks +from core import blocklist, checks +from core.blocklist import BlockType from core.models import DMDisabled, PermissionLevel, SimilarCategoryConverter, getLogger from core.paginator import EmbedPaginatorSession from core.thread import Thread @@ -1562,7 +1562,7 @@ async def contact( elif u.bot: errors.append(f"{u} is a bot, cannot add to thread.") users.remove(u) - elif await self.bot.is_blocked(u): + elif await self.bot.blocklist.is_user_blocked(u): ref = f"{u.mention} is" if ctx.author != u else "You are" errors.append(f"{ref} currently blocked from contacting {self.bot.user.name}.") users.remove(u) @@ -1645,57 +1645,34 @@ async def contact( async def blocked(self, ctx): """Retrieve a list of blocked users.""" - roles, users, now = [], [], discord.utils.utcnow() + roles, users = [], [] - blocked_users = list(self.bot.blocked_users.items()) - for id_, data in blocked_users: - blocked_by_id = data["blocked_by"] - blocked_at = parser.parse(data["blocked_at"]) - human_blocked_at = discord.utils.format_dt(blocked_at, style="R") - if "until" in data: - blocked_until = parser.parse(data["until"]) - human_blocked_until = discord.utils.format_dt(blocked_until, style="R") - else: - blocked_until = human_blocked_until = "Permanent" - - if isinstance(blocked_until, datetime.datetime) and blocked_until < now: - self.bot.blocked_users.pop(str(id_)) - logger.debug("No longer blocked, user %s.", id_) - continue - - string = f"<@{id_}> ({human_blocked_until})" - string += f"\n- Issued {human_blocked_at} by <@{blocked_by_id}>" - - reason = data.get("reason") - if reason: - string += f"\n- Blocked for {reason}" + blocked: list[blocklist.BlocklistEntry] = await self.bot.blocklist.get_all_blocks() - users.append(string + "\n") + for item in blocked: + human_blocked_at = discord.utils.format_dt(item.timestamp, style="R") + if item.expires_at is not None: + human_blocked_until = discord.utils.format_dt(item.expires_at, style="R") + else: + human_blocked_until = "Permanent" - blocked_roles = list(self.bot.blocked_roles.items()) - for id_, data in blocked_roles: - blocked_by_id = data["blocked_by"] - blocked_at = parser.parse(data["blocked_at"]) - human_blocked_at = discord.utils.format_dt(blocked_at, style="R") - if "until" in data: - blocked_until = parser.parse(data["until"]) - human_blocked_until = discord.utils.format_dt(blocked_until, style="R") + if item.type == blocklist.BlockType.USER: + string = f"<@{item.id}>" else: - blocked_until = human_blocked_until = "Permanent" + string = f"<@&{item.id}>" - if isinstance(blocked_until, datetime.datetime) and blocked_until < now: - self.bot.blocked_users.pop(str(id_)) - logger.debug("No longer blocked, user %s.", id_) - continue + string += f" ({human_blocked_until})" - string = f"<@&{id_}> ({human_blocked_until})" - string += f"\n- Issued {human_blocked_at} by <@{blocked_by_id}>" + string += f"\n- Issued {human_blocked_at} by <@{item.blocking_user_id}>" - reason = data.get("reason") - if reason: - string += f"\n- Blocked for {reason}" + if item.reason is not None: + string += f"\n- Blocked for {item.reason}" + string += "\n" - roles.append(string + "\n") + if item.type == blocklist.BlockType.USER: + users.append(string) + elif item.type == blocklist.BlockType.ROLE: + roles.append(string) user_embeds = [discord.Embed(title="Blocked Users", color=self.bot.main_color, description="")] @@ -1713,7 +1690,7 @@ async def blocked(self, ctx): else: embed.description += line else: - user_embeds[0].description = "Currently there are no blocked users." + user_embeds[0].description = "No users are currently blocked." if len(user_embeds) > 1: for n, em in enumerate(user_embeds): @@ -1736,7 +1713,7 @@ async def blocked(self, ctx): else: embed.description += line else: - role_embeds[-1].description = "Currently there are no blocked roles." + role_embeds[-1].description = "No roles are currently blocked." if len(role_embeds) > 1: for n, em in enumerate(role_embeds): @@ -1763,7 +1740,6 @@ async def blocked_whitelist(self, ctx, *, user: User = None): return await ctx.send_help(ctx.command) mention = getattr(user, "mention", f"`{user.id}`") - msg = "" if str(user.id) in self.bot.blocked_whitelisted_users: embed = discord.Embed( @@ -1775,21 +1751,20 @@ async def blocked_whitelist(self, ctx, *, user: User = None): return await ctx.send(embed=embed) self.bot.blocked_whitelisted_users.append(str(user.id)) - - if str(user.id) in self.bot.blocked_users: - msg = self.bot.blocked_users.get(str(user.id)) or "" - self.bot.blocked_users.pop(str(user.id)) - await self.bot.config.update() - if msg.startswith("System Message: "): - # If the user is blocked internally (for example: below minimum account age) - # Show an extended message stating the original internal message - reason = msg[16:].strip().rstrip(".") + blocked: bool + blocklist_entry: blocklist.BlocklistEntry + + blocked, blocklist_entry = await self.bot.blocklist.is_id_blocked(user.id) + if blocked: + await self.bot.blocklist.unblock_id(user.id) embed = discord.Embed( title="Success", - description=f"{mention} was previously blocked internally for " - f'"{reason}". {mention} is now whitelisted.', + description=f""" + {mention} has been whitelisted. + They were previously blocked by <@{blocklist_entry.blocking_user_id}> {" for "+blocklist_entry.reason if blocklist_entry.reason is not None else ""}. + """, color=self.bot.main_color, ) else: @@ -1843,8 +1818,6 @@ async def send_embed(title: str, message: str): ): return await send_embed("Error", f"Cannot block {mention}, user is whitelisted.") - now, blocked = discord.utils.utcnow(), dict() - desc = f"{mention} is now blocked." if duration: desc += f"\n- Expires: {discord.utils.format_dt(duration.dt, style='R')}" @@ -1852,24 +1825,24 @@ async def send_embed(title: str, message: str): if reason: desc += f"\n- Reason: {reason}" - blocked["blocked_at"] = str(now) - blocked["blocked_by"] = ctx.author.id - if duration: - blocked["until"] = str(duration.dt) - if reason: - blocked["reason"] = reason + blocktype: BlockType if isinstance(user_or_role, discord.Role): - self.bot.blocked_roles[str(user_or_role.id)] = blocked + blocktype = BlockType.ROLE elif isinstance(user_or_role, discord.User): - blocked_users = self.bot.blocked_users - blocked_users[str(user_or_role.id)] = blocked + blocktype = BlockType.USER else: return logger.warning( f"{__name__}: cannot block user, user is neither an instance of Discord Role or User" ) - await self.bot.config.update() + await self.bot.blocklist.block_id( + user_id=user_or_role.id, + reason=reason, + expires_at=duration.dt if duration is not None else None, + blocked_by=ctx.author.id, + block_type=blocktype, + ) return await send_embed("Success", desc) @@ -1901,20 +1874,13 @@ async def send_embed(title: str, message: str): title, desc = "Error", f"{mention} is not blocked." - if isinstance(user_or_role, discord.Role): - if str(user_or_role.id) not in self.bot.blocked_roles: - return await send_embed(title, desc) - self.bot.blocked_roles.pop(str(user_or_role.id)) - elif isinstance(user_or_role, discord.User): - if str(user_or_role.id) not in self.bot.blocked_users: - return await send_embed(title, desc) - self.bot.blocked_users.pop(str(user_or_role.id)) - else: + if not isinstance(user_or_role, (discord.Role, discord.User)): return logger.warning( f"{__name__}: cannot unblock, user is neither an instance of Discord Role or User" ) - await self.bot.config.update() + if not await self.bot.blocklist.unblock_id(user_or_role.id): + return await send_embed(title, desc) return await send_embed("Success", f"{mention} has been unblocked.") diff --git a/cogs/utility.py b/cogs/utility.py index f4c1a5f663..135737dd8e 100644 --- a/cogs/utility.py +++ b/cogs/utility.py @@ -22,7 +22,7 @@ from discord.ext.commands.view import StringView from pkg_resources import parse_version -from core import checks, utils +from core import checks, migrations, utils from core.changelog import Changelog from core.models import HostingMethod, InvalidConfigError, PermissionLevel, UnseenFormatter, getLogger from core.paginator import EmbedPaginatorSession, MessagePaginatorSession @@ -1952,6 +1952,21 @@ async def autotrigger_list(self, ctx): await EmbedPaginatorSession(ctx, *embeds).run() + @commands.command() + @checks.has_permissions(PermissionLevel.OWNER) + async def migrate(self, ctx, migration: str): + """Perform a given database migration""" + if migration == "blocklist": + try: + await migrations.migrate_blocklist(self.bot) + except Exception as e: + await ctx.send( + embed=discord.Embed(title="Error", description=str(e), color=self.bot.error_color) + ) + raise e + + await ctx.send(embed=discord.Embed(title="Success", color=self.bot.main_color)) + @commands.command() @checks.has_permissions(PermissionLevel.OWNER) @checks.github_token_required() diff --git a/core/blocklist.py b/core/blocklist.py new file mode 100644 index 0000000000..47295a9756 --- /dev/null +++ b/core/blocklist.py @@ -0,0 +1,198 @@ +import datetime +import enum +from dataclasses import dataclass +from typing import Optional, Self, Tuple + +import discord +import isodate +from bson import CodecOptions +from motor.core import AgnosticCollection + + +class BlockType(enum.IntEnum): + USER = 0 + ROLE = 1 + + +class BlockReason(enum.StrEnum): + GUILD_AGE = "guild_age" + ACCOUNT_AGE = "account_age" + BLOCKED_ROLE = "blocked_role" + BLOCKED_USER = "blocked_user" + + +@dataclass(frozen=True) +class BlocklistEntry: + # _id: ObjectId + # May be role or user id + id: int + expires_at: Optional[datetime.datetime] + reason: Optional[str] + timestamp: datetime.datetime + blocking_user_id: int + # specifies if the id is a role or user id + type: BlockType + + @staticmethod + def from_dict(data: dict): + return BlocklistEntry( + id=data["id"], + expires_at=data["expires_at"], + reason=data["reason"], + timestamp=data["timestamp"], + blocking_user_id=data["blocking_user_id"], + type=data["type"], + ) + + +class Blocklist: + blocklist_collection: AgnosticCollection + + def __init__(self: Self, bot) -> None: + self.blocklist_collection = bot.api.db.blocklist.with_options( + codec_options=CodecOptions(tz_aware=True, tzinfo=datetime.timezone.utc) + ) + self.bot = bot + + async def setup(self): + await self.blocklist_collection.create_index("id") + await self.blocklist_collection.create_index("expires_at", expireAfterSeconds=0) + + async def add_block(self, block: BlocklistEntry) -> None: + await self.blocklist_collection.insert_one(block.__dict__) + + async def block_id( + self, + user_id: int, + expires_at: Optional[datetime.datetime], + reason: str, + blocked_by: int, + block_type: BlockType, + ) -> None: + now = datetime.datetime.utcnow() + + await self.add_block( + block=BlocklistEntry( + id=user_id, + expires_at=expires_at, + reason=reason, + timestamp=now, + blocking_user_id=blocked_by, + type=block_type, + ) + ) + + async def unblock_id(self, user_or_role_id: int) -> bool: + result = await self.blocklist_collection.delete_one({"id": user_or_role_id}) + if result.deleted_count == 0: + return False + return True + + async def is_id_blocked(self, user_or_role_id: int) -> Tuple[bool, Optional[BlocklistEntry]]: + """ + Checks if the given ID is blocked + + This method only checks to see if there is an active manual block for the given ID. + It does not do any checks for whitelisted users or roles, account age, or guild age. + + Parameters + ---------- + user_or_role_id + + Returns + ------- + + True if there is an active block for the given ID + + """ + result = await self.blocklist_collection.find_one({"id": user_or_role_id}) + if result is None: + return False, None + return True, BlocklistEntry.from_dict(result) + + async def get_all_blocks(self) -> list[BlocklistEntry]: + """ + Returns a list of all active blocks + + THIS RETRIEVES ALL ITEMS FROM THE DATABASE COLLECTION, USE WITH CAUTION + + Returns + ------- + + A list of BlockListItems + + """ + dict_list = await self.blocklist_collection.find().to_list(length=None) + dataclass_list: list[BlocklistEntry] = [] + for i in dict_list: + dataclass_list.append(BlocklistEntry.from_dict(i)) + return dataclass_list + + # TODO we will probably want to cache these + async def is_user_blocked(self, member: discord.Member) -> Tuple[bool, Optional[BlockReason]]: + """ + Side effect free version of is_blocked + + Parameters + ---------- + member + + Returns + ------- + True if the user is blocked + """ + # + + if str(member.id) in self.bot.blocked_whitelisted_users: + return False, None + + blocked = await self.blocklist_collection.find_one({"id": member.id}) + if blocked is not None: + return True, BlockReason.BLOCKED_USER + + roles = member.roles + + blocked = await self.blocklist_collection.find_one(filter={"id": {"$in": [r.id for r in roles]}}) + if blocked is not None: + return True, BlockReason.BLOCKED_ROLE + + if not self.is_valid_account_age(member): + return True, BlockReason.ACCOUNT_AGE + if not self.is_valid_guild_age(member): + return True, BlockReason.GUILD_AGE + + return False, None + + def is_valid_account_age(self, author: discord.User) -> bool: + account_age = self.bot.config.get("account_age") + + if account_age is None or account_age == isodate.Duration(): + return True + + now = discord.utils.utcnow() + + min_account_age = author.created_at + account_age + + if min_account_age < now: + # User account has not reached the required time + return False + return True + + def is_valid_guild_age(self, author: discord.Member) -> bool: + guild_age = self.bot.config.get("guild_age") + + if guild_age is None or guild_age == isodate.Duration(): + return True + + now = discord.utils.utcnow() + + if not hasattr(author, "joined_at"): + self.bot.logger.warning("Not in guild, cannot verify guild_age, %s.", author.name) + return False + + min_guild_age = author.joined_at + guild_age + + if min_guild_age > now: + # User has not stayed in the guild for long enough + return False + return True diff --git a/core/migrations.py b/core/migrations.py new file mode 100644 index 0000000000..87ddbbe766 --- /dev/null +++ b/core/migrations.py @@ -0,0 +1,137 @@ +import datetime +import re +from typing import Optional + +from core import blocklist +from core.models import getLogger + +logger = getLogger(__name__) + +old_format_matcher = re.compile("by (\w*#\d{1,4})(?: until )?.") + + +def _convert_legacy_dict_block_format( + k, v: dict, block_type: blocklist.BlockType +) -> Optional[blocklist.BlocklistEntry]: + """ + Converts a legacy dict based blocklist entry to the new dataclass format + + Returns None if the block has expired + """ + + blocked_by = int(v["blocked_by"]) + if "until" in v: + # todo make sure this is the correct from format + blocked_until = datetime.datetime.fromisoformat(v["until"]) + # skip if blocked_until occurred in the past + if blocked_until < datetime.datetime.now(datetime.timezone.utc): + return None + else: + blocked_until = None + + if "reason" in v: + reason = v["reason"] + else: + reason = None + + blocked_ts = datetime.datetime.fromisoformat(v["blocked_at"]) + + return blocklist.BlocklistEntry( + id=int(k), + expires_at=blocked_until, + reason=reason, + timestamp=blocked_ts, + blocking_user_id=blocked_by, + type=block_type, + ) + + +def _convert_legacy_block_format( + k, v: str, block_type: blocklist.BlockType +) -> Optional[blocklist.BlocklistEntry]: + """ + Converts a legacy string based blocklist entry to the new dataclass format + + Returns None if the block has expired + """ + + match = old_format_matcher.match(v) + blocked_until = match.group(2) + if blocked_until is not None: + blocked_until = datetime.datetime.fromtimestamp(int(blocked_until), tz=datetime.timezone.utc) + # skip if blocked_until occurred in the past + if blocked_until < datetime.datetime.now(datetime.timezone.utc): + return None + + return blocklist.BlocklistEntry( + id=int(k), + expires_at=blocked_until, + reason=f"migrated from old format `{v}`", + timestamp=datetime.datetime.utcnow(), + # I'm not bothering to fetch the user object here, discords username migrations will have broken all of them + blocking_user_id=0, + type=block_type, + ) + + +async def _convert_legacy_block_list( + foo: dict, blocklist_batch: list[blocklist.BlocklistEntry], block_type: blocklist.BlockType, bot +) -> int: + skipped = 0 + + for k, v in foo.items(): + # handle new block format + if type(v) is dict: + block = _convert_legacy_dict_block_format(k, v, block_type=block_type) + if block is None: + logger.debug("skipping expired block entry") + skipped += 1 + continue + logger.debug(f"migrating new format {k}: {v}") + else: + block = _convert_legacy_block_format(k, v, block_type=block_type) + if block is None: + logger.debug("skipping expired block entry") + skipped += 1 + continue + logger.debug(f"migrating legacy format {k}: {v}") + + blocklist_batch.append(block) + + if len(blocklist_batch) >= 100: + await bot.api.db.blocklist.insert_many([x.__dict__ for x in blocklist_batch]) + blocklist_batch.clear() + + return skipped + + +async def migrate_blocklist(bot): + start_time = datetime.datetime.utcnow() + + blocked_users = bot.blocked_users + logger.info("preparing to migrate blocklist") + skipped = 0 + + blocklist_batch: list[blocklist.BlocklistEntry] = [] + logger.info(f"preparing to process {len(blocked_users)} blocked users") + skipped += await _convert_legacy_block_list( + foo=blocked_users, blocklist_batch=blocklist_batch, block_type=blocklist.BlockType.USER, bot=bot + ) + logger.info("processed blocked users") + logger.info(f"preparing to process {len(bot.blocked_roles)} blocked roles") + skipped += await _convert_legacy_block_list( + foo=bot.blocked_roles, blocklist_batch=blocklist_batch, block_type=blocklist.BlockType.ROLE, bot=bot + ) + logger.info("processed blocked roles") + + await bot.api.db.blocklist.insert_many([x.__dict__ for x in blocklist_batch]) + blocklist_batch.clear() + + logger.info("clearing old blocklists") + bot.blocked_users.clear() + bot.blocked_roles.clear() + await bot.config.update() + + logger.info(f"Migration complete! skipped {skipped} entries") + logger.info(f"migrated in {datetime.datetime.utcnow() - start_time}") + return