diff --git a/src/bot/constants.py b/src/bot/constants.py index 052928e..febfc3d 100644 --- a/src/bot/constants.py +++ b/src/bot/constants.py @@ -34,6 +34,15 @@ class _Miscellaneous(EnvConfig): Miscellaneous = _Miscellaneous() +class _Pastebin(EnvConfig, env_prefix="pastebin_"): + """Pastebin configuration for large contents that don't fit in Discord.""" + + base_url: str | None = None + + +Pastebin = _Pastebin() + + class _ThreatIntelFeed(EnvConfig, env_prefix="tif_"): """Threat Intelligence Feed Configuration.""" diff --git a/src/bot/exts/dragonfly/dragonfly.py b/src/bot/exts/dragonfly/dragonfly.py index 2b7f77b..e13db63 100644 --- a/src/bot/exts/dragonfly/dragonfly.py +++ b/src/bot/exts/dragonfly/dragonfly.py @@ -14,6 +14,7 @@ from bot.bot import Bot from bot.constants import Channels, DragonflyConfig, Roles from bot.dragonfly_services import DragonflyServices, Package, PackageReport +from bot.utils.pastebin import PasteFile, PasteRequest, PasteResponse, paste log = getLogger(__name__) log.setLevel(logging.INFO) @@ -297,6 +298,19 @@ def _build_all_packages_scanned_embed(scan_results: list[Package]) -> discord.Em return discord.Embed(description="_No packages scanned_") +def _build_pastebin_embed(paste_response: PasteResponse) -> discord.Embed: + """Build the embed that links to a pastebin when the output would have otherwise been too long.""" + return discord.Embed( + title="Embed too large", + description=( + "This embed would have been too large, so the contents were uploaded to a pastebin instead." + f"Click [here]({paste_response.link}) to view the pastebin." + ), + color=discord.Color.orange(), + url=paste_response.link, + ) + + async def run( bot: Bot, *, @@ -316,7 +330,16 @@ async def run( view=ReportView(bot, result), ) - await logs_channel.send(embed=_build_all_packages_scanned_embed(scan_results)) + all_packages_scanned_embed = _build_all_packages_scanned_embed(scan_results) + if len(all_packages_scanned_embed) <= 4096: # noqa: PLR2004 + await logs_channel.send(embed=all_packages_scanned_embed) + else: + content = "\n".join(map(str, scan_results)) + paste_request = PasteRequest(expiry="1day", files=[PasteFile(lexer="text", content=content)]) + paste_response = await paste(paste_request, session=bot.http_session) + embed = _build_pastebin_embed(paste_response) + await logs_channel.send(embed=embed) + log.info("Package scan log embed would have exceeded size, sent in a pastebin instead") class Dragonfly(commands.Cog): diff --git a/src/bot/utils/pastebin.py b/src/bot/utils/pastebin.py new file mode 100644 index 0000000..2af5a61 --- /dev/null +++ b/src/bot/utils/pastebin.py @@ -0,0 +1,50 @@ +"""Utilities relating to Pastebin services.""" + +from typing import Literal + +import aiohttp +from pydantic import BaseModel + +from bot.constants import Pastebin + + +class PastebinNotConfiguredError(Exception): + """Raised when a paste was requested but no pastebin service is configured.""" + + def __init__(self) -> None: + super().__init__("A pastebin service is not configured.") + + +class PasteFile(BaseModel): + """Represents a single file as part of a paste request.""" + + name: str | None = None + lexer: str + content: str + + +class PasteRequest(BaseModel): + """Represents a paste request.""" + + expiry: Literal["1day", "7days", "30days"] + files: list[PasteFile] + + +class PasteResponse(BaseModel): + """Represents a paste response.""" + + link: str + removal: str + + +async def paste(payload: PasteRequest, *, session: aiohttp.ClientSession) -> PasteResponse: + """Create a paste using the configured pastebin service. Raise an error if no service is configured.""" + if not (base_url := Pastebin.base_url): + raise PastebinNotConfiguredError + + url = base_url + "/api/v1/paste" + json = payload.model_dump() + + async with session.post(url, json=json) as response: + response_json = await response.json() + return PasteResponse.model_validate(response_json)