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

"""
Cog for detecting URLs posted by new users.
"""
import re
from datetime import timedelta
from typing import Literal

from discord import Member, Message, utils as discordutils
from discord.ext import commands
from discord.utils import escape_markdown

from config import CONFIG
from rocketbot.cogs.basecog import BaseCog, BotMessage, BotMessageReaction, CogSetting
from rocketbot.utils import describe_timedelta

class URLSpamContext:
	"""
	Data about a suspected spam message containing a URL.
	"""
	def __init__(self, spam_message: Message):
		self.spam_message = spam_message
		self.is_deleted = False
		self.is_kicked = False
		self.is_banned = False

class URLSpamCog(BaseCog, name='URL Spam'):
	"""
	Detects users posting URLs who just joined recently: a common spam pattern.
	Can be configured to take immediate action or just warn the mods.
	"""

	SETTING_ENABLED = CogSetting('enabled', bool,
			brief='URL spam detection',
			description='Whether URLs posted soon after joining are flagged.')
	SETTING_ACTION = CogSetting('action', Literal['nothing', 'modwarn', 'delete', 'kick', 'ban'],
			brief='action to take on spam',
			description='The action to take on detected URL spam.',
			enum_values={'nothing', 'modwarn', 'delete', 'kick', 'ban'})
	SETTING_JOIN_AGE = CogSetting('joinage', float,
			brief='seconds since member joined',
			description='The minimum seconds since the user joined the ' + \
				'server before they can post URLs. URLs posted by users ' + \
				'who joined too recently will be flagged. Keep in mind ' + \
				'many servers have a minimum 10 minute cooldown before ' + \
				'new members can say anything. Setting to 0 effectively ' + \
				'disables URL spam detection.',
			usage='<seconds:int>',
			min_value=0)
	SETTING_DECEPTIVE_ACTION = CogSetting('deceptiveaction', Literal['nothing', 'modwarn', 'modwarndelete', 'chatwarn', 'chatwarndelete', 'delete', 'kick', 'ban'],
			brief='action to take on deceptive link markdown',
			description='The action to take on chat messages with links ' + \
				'where the text looks like a different URL than the actual link.',
			enum_values={'nothing', 'modwarn', 'modwarndelete',
				'chatwarn', 'chatwarndelete', 'delete', 'kick', 'ban'})

	def __init__(self, bot):
		super().__init__(bot, 'urlspam')
		self.add_setting(URLSpamCog.SETTING_ENABLED)
		self.add_setting(URLSpamCog.SETTING_ACTION)
		self.add_setting(URLSpamCog.SETTING_JOIN_AGE)
		self.add_setting(URLSpamCog.SETTING_DECEPTIVE_ACTION)

	@commands.group(
		brief='Manages URL spam detection',
	)
	@commands.has_permissions(ban_members=True)
	@commands.guild_only()
	async def urlspam(self, context: commands.Context):
		"""URL spam command group"""
		if context.invoked_subcommand is None:
			await context.send_help()

	@commands.Cog.listener()
	async def on_message(self, message: Message):
		"""Event listener"""
		if message.author is None or \
				message.author.bot or \
				message.guild is None or \
				message.channel is None or \
				message.content is None:
			return
		if not self.get_guild_setting(message.guild, self.SETTING_ENABLED):
			return
		await self.check_message_recency(message)
		await self.check_deceptive_links(message)

	async def check_message_recency(self, message: Message):
		"""Checks if the message was sent too recently by a new user"""
		action = self.get_guild_setting(message.guild, self.SETTING_ACTION)
		join_seconds = self.get_guild_setting(message.guild, self.SETTING_JOIN_AGE)
		min_join_age = timedelta(seconds=join_seconds)
		if action == 'nothing':
			return
		if not self.__contains_url(message.content):
			return
		join_age = message.created_at - message.author.joined_at
		join_age_str = describe_timedelta(join_age)
		if join_age < min_join_age:
			context = URLSpamContext(message)
			needs_attention = False
			if action == 'modwarn':
				needs_attention = not self.was_warned_recently(message.author)
				self.log(message.guild, f'New user {message.author.name} ' + \
					f'({message.author.id}) posted URL {join_age_str} after ' + \
					'joining.' + (' Mods alerted.' if needs_attention else ''))
			elif action == 'delete':
				await message.delete()
				context.is_deleted = True
				self.log(message.guild, f'New user {message.author.name} ' + \
					f'({message.author.id}) posted URL {join_age_str} after ' + \
					'joining. Message deleted.')
			elif action == 'kick':
				await message.delete()
				context.is_deleted = True
				await message.author.kick(
					reason=f'Rocketbot: Posted a link {join_age_str} after joining')
				context.is_kicked = True
				self.log(message.guild, f'New user {message.author.name} ' + \
					f'({message.author.id}) posted URL {join_age_str} after ' + \
					'joining. User kicked.')
			elif action == 'ban':
				await message.author.ban(
					reason=f'Rocketbot: User posted a link {join_age_str} after joining',
					delete_message_days=1)
				context.is_deleted = True
				context.is_kicked = True
				context.is_banned = True
				self.log(message.guild, f'New user {message.author.name} ' + \
					f'({message.author.id}) posted URL {join_age_str} after ' + \
					'joining. User banned.')
			bm = BotMessage(
					message.guild,
					f'User {message.author.mention} posted a URL ' + \
					f'{join_age_str} after joining: {message.jump_url}',
					type = BotMessage.TYPE_MOD_WARNING if needs_attention else BotMessage.TYPE_INFO,
					context = context)
			bm.quote = discordutils.remove_markdown(message.clean_content)
			await bm.set_reactions(BotMessageReaction.standard_set(
				did_delete=context.is_deleted,
				did_kick=context.is_kicked,
				did_ban=context.is_banned))
			await self.post_message(bm)
			if needs_attention:
				self.record_warning(message.author)

	async def check_deceptive_links(self, message: Message):
		"""
		Checks if the message contains deceptive URL Markdown, e.g.
		`[nicewebsite.com](https://evilwebsite.com)'`
		"""
		action = self.get_guild_setting(message.guild, self.SETTING_DECEPTIVE_ACTION)
		if action is None or action == 'nothing':
			return

		if not self.contains_deceptive_links(message.content):
			return
		mod_text = f'User {message.author.name} ({message.author.id}) posted a deceptive link. {message.jump_url}'
		quoted = '> ' + escape_markdown(message.content).replace('\n', '\n> ')
		mod_text += f'\n\n{quoted}'
		self.log(message.guild, f'{message.author.name} posted deceptive link - action: {action}')
		if 'modwarn' in action:
			if 'delete' in action:
				mod_text += '\n\nMessage deleted'
			else:
				mod_text += f'\n\n{message.jump_url}'
			bm = BotMessage(message.guild, mod_text, BotMessage.TYPE_MOD_WARNING, suppress_embeds=True)
			await self.post_message(bm)
			if 'delete' in action:
				await message.delete()
		elif 'chatwarn' in action:
			if 'delete' in action:
				response = f':warning: Links with deceptive labels are prohibited :warning:'
			else:
				response = f':warning: Message contains a deceptively labeled link! Click carefully. :warning:'
			await message.reply(response, mention_author=False)
			if 'delete' in action:
				await message.delete()
		elif action == 'delete':
			mod_text += f'\n\nDeleting message'
			bm = BotMessage(message.guild, mod_text, BotMessage.TYPE_INFO, suppress_embeds=True)
			await self.post_message(bm)
			await message.delete()
		elif action == 'kick':
			mod_text += f'\n\nUser kicked'
			bm = BotMessage(message.guild, mod_text, BotMessage.TYPE_MOD_WARNING, suppress_embeds=True)
			await self.post_message(bm)
			await message.delete()
			await message.author.kick(
				reason=f'Rocketbot: User posted a deceptive link')
		elif action == 'ban':
			mod_text += f'\n\nUser banned'
			bm = BotMessage(message.guild, mod_text, BotMessage.TYPE_MOD_WARNING, suppress_embeds=True)
			await self.post_message(bm)
			await message.author.ban(
				reason=f'Rocketbot: User posted a deceptive link',
				delete_message_days=1)

	def contains_deceptive_links(self, content: str) -> bool:
		# Strip Markdown that can safely contain URL sequences
		content = re.sub(r'`[^`]+`', '', content)  # `inline code`
		content = re.sub(r'```.+?```', '', content, re.DOTALL)  # ``` code block ```
		matches = re.findall(r'\[([^]]+)]\(([^)]+)\)', content)
		for match in matches:
			original_label: str = match[0].strip()
			original_link: str = match[1].strip()
			label: str = original_label
			link: str = original_link
			if link.startswith('<') and link.endswith('>'):
				link = link[1:-1]
			if self.is_url(label):
				if label != link:
					return True
			elif self.is_casual_url(label):
				# Trim www. for easier comparisons.
				if link.startswith('https://www.'):
					link = 'https://' + link[12:]
				if link.startswith('http://www.'):
					link = 'http://' + link[11:]
				if link.endswith('/'):
					link = link[:-1]
				if label.startswith('www.'):
					label = label[4:]
				if label.endswith('/'):
					label = label[:-1]
				if link.startswith('https://') and 'https://' + label != link:
					return True
				elif link.startswith('http://') and 'http://' + label != link:
					return True
		return False

	def is_url(self, s: str):
		"""Tests if a string is strictly a URL"""
		ipv6_host_pattern = r'\[[0-9a-fA-F:]+\]'
		ipv4_host_pattern = r'[0-9\.]+'
		hostname_pattern = r'[a-zA-Z0-9-]+\.[a-zA-Z0-9-\.]+'
		host_pattern = r'(?:' + ipv6_host_pattern + '|' + ipv4_host_pattern + '|' + hostname_pattern + ')'
		port_pattern = '(?::[0-9]+)?'
		path_pattern = r'(?:/[^ \]\)]*)?'
		pattern = r'^http[s]?://' + host_pattern + port_pattern + path_pattern + '$'
		return re.match(pattern, s, re.IGNORECASE) is not None

	def is_casual_url(self, s: str):
		"""Tests if a string is a "casual URL" with no scheme included"""
		ipv6_host_pattern = r'\[[0-9a-fA-F:]+\]'
		ipv4_host_pattern = r'[0-9\.]+'
		hostname_pattern = r'[a-zA-Z0-9-]+\.[a-zA-Z0-9-\.]+'
		host_pattern = r'(?:' + ipv6_host_pattern + '|' + ipv4_host_pattern + '|' + hostname_pattern + ')'
		port_pattern = '(?::[0-9]+)?'
		path_pattern = r'(?:/[^ \]\)]*)?'
		pattern = r'^' + host_pattern + port_pattern + path_pattern + '$'
		return re.match(pattern, s, re.IGNORECASE) is not None

	async def on_mod_react(self,
			bot_message: BotMessage,
			reaction: BotMessageReaction,
			reacted_by: Member) -> None:
		context: URLSpamContext = bot_message.context
		if context is None:
			return
		sm: Message = context.spam_message
		if reaction.emoji == CONFIG['trash_emoji']:
			if not context.is_deleted:
				await sm.delete()
				context.is_deleted = True
				self.log(sm.guild, f'URL spam by {sm.author.name} deleted ' + \
					f'by {reacted_by.name}')
		elif reaction.emoji == CONFIG['kick_emoji']:
			if not context.is_deleted:
				await sm.delete()
				context.is_deleted = True
			if not context.is_kicked:
				await sm.author.kick(
					reason=f'Rocketbot: Kicked for URL spam by {reacted_by.name}')
				context.is_kicked = True
				self.log(sm.guild, f'URL spammer {sm.author.name} kicked ' + \
					f'by {reacted_by.name}')
		elif reaction.emoji == CONFIG['ban_emoji']:
			if not context.is_banned:
				await sm.author.ban(
					reason=f'Rocketbot: Banned for URL spam by {reacted_by.name}',
					delete_message_days=1)
				context.is_deleted = True
				context.is_kicked = True
				context.is_banned = True
				self.log(sm.guild, f'URL spammer {sm.author.name} banned ' + \
					f'by {reacted_by.name}')
		else:
			return
		await bot_message.set_reactions(BotMessageReaction.standard_set(
			did_delete=context.is_deleted,
			did_kick=context.is_kicked,
			did_ban=context.is_banned))

	@classmethod
	def __contains_url(cls, text: str) -> bool:
		p = re.compile(r'https?://\S+')
		return p.search(text) is not None