Add anti-phishing, ref #111

This commit is contained in:
Zeva Rose 2022-03-17 15:20:04 -06:00
parent eba4f1fff0
commit 06ab5d105f

View file

@ -3,9 +3,12 @@ import re
import traceback
from datetime import datetime
from aiohttp import ClientSession
from dis_snek import Snake, listen
from dis_snek.api.events.discord import MessageCreate, MessageDelete, MessageUpdate
from dis_snek.client.utils.misc_utils import find, find_all
from dis_snek.ext.tasks.task import Task
from dis_snek.ext.tasks.triggers import TimeTrigger
from dis_snek.models.discord.channel import DMChannel
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.message import Message
@ -13,7 +16,7 @@ from dis_snek.models.discord.user import Member
from dis_snek.models.snek.context import Context, InteractionContext
from jarvis_core.db import q
from jarvis_core.db.models import Autopurge, Autoreact, Roleping, Setting, Warning
from jarvis_core.filters import invites
from jarvis_core.filters import invites, url
from jarvis_core.util import build_embed
from jarvis_core.util.ansi import RESET, Fore, Format, fmt
from pastypy import AsyncPaste as Paste
@ -41,9 +44,34 @@ CMD_FMT = fmt(Fore.GREEN, Format.BOLD)
class Jarvis(Snake):
def __init__(self, *args, **kwargs): # noqa: ANN002 ANN003
super().__init__(*args, **kwargs)
self.phishing_domains = set()
@Task.create(TimeTrigger())
async def _update_domains(self) -> None:
async with ClientSession(headers={"X-Identity": "Discord: zevaryx#5779"}) as session:
response = await session.get("https://phish.sinking.yachts/v2/recent/86415")
response.raise_for_status()
data = await response.json()
for update in data:
if update["type"] == "add":
self.phishing_domains.add(update["domain"])
elif update["type"] == "delete":
self.phishing_domains.discard(update["domain"])
async def _sync_domains(self) -> None:
async with ClientSession(headers={"X-Identity": "Discord: zevaryx#5779"}) as session:
response = await session.get("https://phish.sinking.yachts/v2/all")
response.raise_for_status()
self.phishing_domains = set(await response.json())
@listen()
async def on_ready(self) -> None:
"""Lepton on_ready override."""
await self._sync_domains()
self._update_domains.start()
print("Logged in as {}".format(self.user)) # noqa: T001
print("Connected to {} guild(s)".format(len(self.guilds))) # noqa: T001
print( # noqa: T001
@ -413,6 +441,34 @@ class Jarvis(Snake):
)
await message.channel.send(embed=embed)
async def phishing(self, message: Message) -> None:
"""Check if the message contains any known phishing domains."""
for match in url.findall(message.content):
if match in self.phishing_domains:
w = Warning(
active=True,
admin=self.user.id,
duration=24,
guild=message.guild.id,
reason="Phishing URL",
user=message.author.id,
)
await w.commit()
fields = [EmbedField(name="Reason", value="Phishing URL", inline=False)]
embed = build_embed(
title="Warning",
description=f"{message.author.mention} has been warned",
fields=fields,
)
embed.set_author(
name=message.author.nick if message.author.nick else message.author.name,
icon_url=message.author.display_avatar.url,
)
embed.set_footer(
text=f"{message.author.name}#{message.author.discriminator} | {message.author.id}"
)
await message.channel.send(embed=embed)
async def on_message(self, event: MessageCreate) -> None:
"""Handle on_message event. Calls other event handlers."""
message = event.message
@ -422,3 +478,4 @@ class Jarvis(Snake):
await self.roleping(message)
await self.autopurge(message)
await self.checks(message)
await self.phishing(message)