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 Guild, Member, Message
from discord import utils as discordutils
from discord.ext.commands import Cog
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,
		default_value=False,
		brief='URL spam detection',
		description='Whether URLs posted soon after joining are flagged.',
	)
	SETTING_ACTION = CogSetting(
		'action',
		Literal['nothing', 'modwarn', 'delete', 'kick', 'ban'],
		default_value='nothing',
		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',
		timedelta,
		default_value=timedelta(minutes=15),
		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.',
		min_value=timedelta(seconds=0),
	)
	SETTING_DECEPTIVE_ACTION = CogSetting(
		'deceptiveaction',
		Literal['nothing', 'modwarn', 'modwarndelete', 'chatwarn', 'chatwarndelete', 'delete', 'kick', 'ban'],
		default_value='nothing',
		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'},
	)
	SETTING_IGNORED_DOMAINS = CogSetting(
		'ignoreddomains',
		str,
		default_value='tenor.com;giphy.com;imgur.com',
		brief='List of allowed domains to exclude from detecting',
		description='A semicolon-delimited list of domains that can be posted '
					'without triggering a warning. E.g. GIF hosting sites.'
	)

	def __init__(self, bot):
		super().__init__(
			bot,
			config_prefix='urlspam',
			short_description='Manages URL spam detection.',
		)
		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)
		self.add_setting(URLSpamCog.SETTING_IGNORED_DOMAINS)

	@Cog.listener()
	async def on_message(self, message: Message):
		"""Event listener"""
		if message.author is None or \
				message.guild is None or \
				message.channel is None or \
				message.content is None:
			self.__trace("Missing message data")
			return
		if message.author.bot:
			self.__trace("Message from bot")
			return
		if message.channel.permissions_for(message.author).ban_members:
			self.__trace("User exempt")
			return
		if not self.get_guild_setting(message.guild, self.SETTING_ENABLED):
			self.__trace("Cog disabled")
			return
		urls = self.__find_urls(message.content)
		if len(urls) == 0:
			self.__trace("Contains no URLs")
			return
		if self.__all_ignored_domains(message.guild, urls):
			self.__trace("All URLs are ignored domains")
			return
		await self.check_message_recency(message)
		await self.check_deceptive_links(message)

	def __all_ignored_domains(self, guild: Guild, urls: list[str]) -> bool:
		domain_list: str = self.get_guild_setting(guild, self.SETTING_IGNORED_DOMAINS)
		self.__trace(f"Domain list is {domain_list}")
		domains = domain_list.split(';')
		self.__trace(f"Split into {len(domains)} domains")
		for url in urls:
			if not self.__is_ignored_url(url, domains):
				return False
		return True

	def __is_ignored_url(self, url: str, ignored_domains: list[str]) -> bool:
		p = re.compile(r'https?://([\w\.]+)')
		domain = re.search(p, url).group(1)
		for ignored_domain in ignored_domains:
			if domain.endswith(ignored_domain.strip()):
				self.__trace(f"URL {url} matches ignored domain {ignored_domain}")
				return True
		self.__trace("URL {url} not ignored")
		return False

	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':
			self.__trace("Configured action is nothing")
			return
		join_age = message.created_at - message.author.joined_at
		join_age_str = describe_timedelta(join_age)
		if join_age > min_join_age:
			self.__trace("User has been member long enough")
			return
		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 = ':warning: Links with deceptive labels are prohibited :warning:'
			else:
				response = ':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 += '\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 += '\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='Rocketbot: User posted a deceptive link')
		elif action == 'ban':
			mod_text += '\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='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) -> bool:
		"""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) -> bool:
		"""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))

	def __trace(self, message: str):
		# print(f'URLSpamCog: {message}')
		pass

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

	@classmethod
	def __find_urls(cls, text: str) -> list[str]:
		p = re.compile(r'https?://\S+')
		return re.findall(p, text)