Merge branch 'main' into pdm

src/bot/__main__.py

@@ -27,7 +27,7 @@ def get_prefix(bot_: Bot, message_: discord.Message) -> Callable[[Bot, discord.M
 
 async def main() -> None:
     """Run the bot."""
-    async with ClientSession(headers={"Content-Type": "application/json"}, timeout=ClientTimeout(total=10)) as session:
+    async with ClientSession(headers={"Content-Type": "application/json"}, timeout=ClientTimeout(total=30)) as session:
         dragonfly_services = DragonflyServices(
             session=session,
             base_url=constants.Dragonfly.base_url,

src/bot/constants.py

@@ -33,6 +33,18 @@ class _Miscellaneous(EnvConfig):
 
 Miscellaneous = _Miscellaneous()
 
+
+class _ThreatIntelFeed(EnvConfig, env_prefix="tif_"):
+    """Threat Intelligence Feed Configuration."""
+
+    repository: str = "pypi/pypi-observation-reports-private"
+    interval: int = 60 * 60  # 1 hour
+    access_token: str = ""
+    channel_id: int = 1121471544355455058
+
+
+ThreatIntelFeed = _ThreatIntelFeed()
+
 FILE_LOGS = Miscellaneous.file_logs
 DEBUG_MODE = Miscellaneous.debug
 

src/bot/dragonfly_services.py

@@ -7,6 +7,7 @@
 from typing import Any, Self
 
 from aiohttp import ClientSession
+from pydantic import BaseModel
 
 
 class ScanStatus(Enum):
@@ -18,41 +19,29 @@ class ScanStatus(Enum):
     FAILED = "failed"
 
 
-@dataclass
-class PackageScanResult:
-    """A package scan result."""
+class Package(BaseModel):
+    """Model representing a package queried from the database."""
 
-    status: ScanStatus
-    inspector_url: str
-    queued_at: datetime
+    scan_id: str
+    name: str
+    version: str
+    status: ScanStatus | None
+    score: int | None
+    inspector_url: str | None
+    rules: list[str] = []
+    download_urls: list[str] = []
+    queued_at: datetime | None
+    queued_by: str | None
+    reported_at: datetime | None
+    reported_by: str | None
     pending_at: datetime | None
+    pending_by: str | None
     finished_at: datetime | None
-    reported_at: datetime | None
-    version: str
-    name: str
-    package_id: str
-    rules: list[str]
-    score: int
-
-    @classmethod
-    def from_dict(cls: type[Self], data: dict) -> Self:  # type: ignore[type-arg]
-        """Create a PackageScanResult from a dictionary."""
-        return cls(
-            status=ScanStatus(data["status"]),
-            inspector_url=data["inspector_url"],
-            queued_at=datetime.fromisoformat(data["queued_at"]),
-            pending_at=datetime.fromisoformat(p) if (p := data["pending_at"]) else None,
-            finished_at=datetime.fromisoformat(p) if (p := data["finished_at"]) else None,
-            reported_at=datetime.fromisoformat(p) if (p := data["reported_at"]) else None,
-            version=data["version"],
-            name=data["name"],
-            package_id=data["scan_id"],
-            rules=[d["name"] for d in data["rules"]],
-            score=int(data["score"]),
-        )
-
-    def __str__(self: Self) -> str:
-        """Return a string representation of the package scan result."""
+    finished_by: str | None
+    commit_hash: str | None
+
+    def __str__(self) -> str:
+        """Return package name and version."""
         return f"{self.name} {self.version}"
 
 
@@ -146,7 +135,7 @@ async def get_scanned_packages(
         name: str | None = None,
         version: str | None = None,
         since: datetime | None = None,
-    ) -> list[PackageScanResult]:
+    ) -> list[Package]:
         """Get a list of scanned packages."""
         params = {}
         if name:
@@ -159,7 +148,7 @@ async def get_scanned_packages(
             params["since"] = int(since.timestamp())  # type: ignore[assignment]
 
         data = await self.make_request("GET", "/package", params=params)
-        return [PackageScanResult.from_dict(dct) for dct in data]
+        return list(map(Package.model_validate, data))
 
     async def report_package(
         self: Self,

src/bot/exts/audit.py

@@ -10,7 +10,7 @@
 from discord.ext import commands
 
 from bot.bot import Bot
-from bot.dragonfly_services import PackageScanResult
+from bot.dragonfly_services import Package
 
 
 class PaginatorView(ui.View):
@@ -20,7 +20,7 @@ def __init__(
         self: Self,
         *,
         member: discord.Member | discord.User,
-        packages: list[PackageScanResult],
+        packages: list[Package],
         per: int = 15,
     ) -> None:
         """Initialize the paginator view."""
@@ -70,7 +70,7 @@ async def interaction_check(self: Self, interaction: discord.Interaction) -> boo
         await interaction.response.send_message("This paginator is not for you!", ephemeral=True)
         return False
 
-    def _build_embed(self: Self, packages: list[PackageScanResult], page: int, total: int) -> discord.Embed:
+    def _build_embed(self: Self, packages: list[Package], page: int, total: int) -> discord.Embed:
         """Build an embed for the given packages."""
         embed = discord.Embed(
             title="Package Audit",