python-app-rocketbot/rocketbot/cogs/logcog.py

"""
Cog for detecting large numbers of guild joins in a short period of time.
"""
import weakref
from collections.abc import Sequence
from datetime import datetime, timedelta
from discord import Emoji, Guild, GuildChannel, GuildSticker, Invite, Member, Message, Role, Thread, User
from discord.ext import commands
from typing import List, Union

from config import CONFIG
from rocketbot.cogs.basecog import BaseCog, BotMessage, BotMessageReaction, CogSetting
from rocketbot.collections import AgeBoundList
from rocketbot.storage import Storage

class LogCog(BaseCog, name='Logging'):
	"""
	Cog for logging notable events to a designated logging channel.
	"""
	SETTING_ENABLED = CogSetting('enabled', bool,
			brief='logging',
			description='Whether this cog is enabled for a guild.')
	SETTING_EDITS_ENABLED = CogSetting('edits_enabled', bool,
			brief='post edits',
			description='Whether to log when users edit their posts.')
	SETTING_JOINS_ENABLED = CogSetting('joins_enabled', bool,
			brief='joins',
			description='Whether to log when new users join the server.')
	SETTING_LEAVES_ENABLED = CogSetting('leaves_enabled', bool,
			brief='leaves',
			description='Whether to log when users leave the server.')

	def __init__(self, bot):
		super().__init__(bot)
		self.add_setting(LogCog.SETTING_ENABLED)
		self.add_setting(LogCog.SETTING_EDITS_ENABLED)
		self.add_setting(LogCog.SETTING_JOINS_ENABLED)
		self.add_setting(LogCog.SETTING_LEAVES_ENABLED)

	@commands.group(
		brief='Manages event logging',
	)
	@commands.has_permissions(ban_members=True)
	@commands.guild_only()
	async def log(self, context: commands.Context):
		'Logging command group'
		if context.invoked_subcommand is None:
			await context.send_help()

	# Events - Channels

	@commands.Cog.listener()
	async def on_guild_channel_delete(self, channel: GuildChannel) -> None:
		pass

	@commands.Cog.listener()
	async def on_guild_channel_create(self, channel: GuildChannel) -> None:
		pass

	@commands.Cog.listener()
	async def on_guild_channel_update(self, before: GuildChannel, after: GuildChannel) -> None:
		pass

	# Events - Guilds

	@commands.Cog.listener()
	async def on_guild_available(self, guild: Guild) -> None:
		pass

	@commands.Cog.listener()
	async def on_guild_unavailable(self, guild: Guild) -> None:
		pass

	@commands.Cog.listener()
	async def on_guild_update(self, before: Guild, after: Guild) -> None:
		pass

	@commands.Cog.listener()
	async def on_guild_emojis_update(self, guild: Guild, before: Sequence[Emoji], after: Sequence[Emoji]) -> None:
		pass

	@commands.Cog.listener()
	async def on_guild_stickers_update(self, guild: Guild, before: Sequence[GuildSticker], after: Sequence[GuildSticker]) -> None:
		pass

	@commands.Cog.listener()
	async def on_invite_create(self, invite: Invite) -> None:
		pass

	@commands.Cog.listener()
	async def on_invite_delete(self, invite: Invite) -> None:
		pass

	# Events - Members

	@commands.Cog.listener()
	async def on_member_join(self, member: Member) -> None:
		pass

	@commands.Cog.listener()
	async def on_member_remove(self, member: Member) -> None:
		pass

	@commands.Cog.listener()
	async def on_member_update(self, before: Member, after: Member) -> None:
		pass

	@commands.Cog.listener()
	async def on_user_update(self, before: User, after: User) -> None:
		pass

	@commands.Cog.listener()
	async def on_member_ban(self, user: Union[User, Member]) -> None:
		pass

	@commands.Cog.listener()
	async def on_member_unban(self, guild: Guild, user: Union[User, Member]) -> None:
		pass

	# Events - Messages

	@commands.Cog.listener()
	async def on_message_edit(self, before: Message, after: Message) -> None:
		pass

	@commands.Cog.listener()
	async def on_message_delete(self, message: Message) -> None:
		pass

	@commands.Cog.listener()
	async def on_bulk_message_delete(self, messages: List[Message]) -> None:
		pass

	# Events - Roles

	@commands.Cog.listener()
	async def on_guild_role_create(self, role: Role) -> None:
		pass

	@commands.Cog.listener()
	async def on_guild_role_delete(self, role: Role) -> None:
		pass

	@commands.Cog.listener()
	async def on_guild_role_update(self, before: Role, after: Role) -> None:
		pass

	# Events - Threads

	@commands.Cog.listener()
	async def on_thread_create(self, thread: Thread) -> None:
		pass

	@commands.Cog.listener()
	async def on_thread_update(self, before: Thread, after: Thread) -> None:
		pass

	@commands.Cog.listener()
	async def on_thread_delete(self, thread: Thread) -> None:
		pass


	def remove_me():
		pass


	@commands.Cog.listener()
	async def on_member_join(self, member: Member) -> None:
		'Event handler'
		guild: Guild = member.guild
		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
			return
		min_count = self.get_guild_setting(guild, self.SETTING_JOIN_COUNT)
		seconds = self.get_guild_setting(guild, self.SETTING_JOIN_TIME)
		timespan: timedelta = timedelta(seconds=seconds)

		last_raid: JoinRaidContext = Storage.get_state_value(guild, self.STATE_KEY_LAST_RAID)
		recent_joins: AgeBoundList = Storage.get_state_value(guild, self.STATE_KEY_RECENT_JOINS)
		if recent_joins is None:
			recent_joins = AgeBoundList(timespan, lambda i, member : member.joined_at)
			Storage.set_state_value(guild, self.STATE_KEY_RECENT_JOINS, recent_joins)
		if last_raid:
			if member.joined_at - last_raid.last_join_time() > timespan:
				# Last raid is over
				Storage.set_state_value(guild, self.STATE_KEY_LAST_RAID, None)
				recent_joins.append(member)
				return
			# Add join to existing raid
			last_raid.join_members.append(member)
			self.record_warning(member)
			if len(last_raid.banned_members) > 0:
				self.log(guild, f'Banning as part of last join raid: {member.name}')
				await member.ban(
					reason='Rocketbot: Part of join raid.',
					delete_message_days=0)
				last_raid.banned_members.add(member)
			elif len(last_raid.kicked_members) > 0:
				self.log(guild, f'Kicking as part of last join raid: {member.name}')
				await member.kick(
					reason='Rocketbot: Part of join raid.')
				last_raid.kicked_members.add(member)
			await self.__update_warning_message(last_raid)
		else:
			# Add join to the general, non-raid recent join list
			recent_joins.append(member)
			if len(recent_joins) >= min_count:
				self.log(guild, '\u0007Join raid detected')
				last_raid = JoinRaidContext(recent_joins)
				Storage.set_state_value(guild, self.STATE_KEY_LAST_RAID, last_raid)
				recent_joins.clear()
				msg = BotMessage(guild,
						text='',
						type=BotMessage.TYPE_MOD_WARNING,
						context=last_raid)
				self.record_warnings(recent_joins)
				last_raid.warning_message_ref = weakref.ref(msg)
				await self.__update_warning_message(last_raid)
				await self.post_message(msg)

	async def on_setting_updated(self, guild: Guild, setting: CogSetting) -> None:
		if setting is self.SETTING_JOIN_TIME:
			seconds = self.get_guild_setting(guild, self.SETTING_JOIN_TIME)
			timespan: timedelta = timedelta(seconds=seconds)
			recent_joins: AgeBoundList = Storage.get_state_value(guild,
				self.STATE_KEY_RECENT_JOINS)
			if recent_joins:
				recent_joins.max_age = timespan
				recent_joins.purge_old_elements()