Implement "greylisting"

src/mainframe/endpoints/report.py

@@ -13,6 +13,7 @@
 from mainframe.constants import mainframe_settings
 from mainframe.database import get_db
 from mainframe.dependencies import get_ms_graph_client, validate_token
+from mainframe.greylist import in_greylist
 from mainframe.json_web_token import AuthenticationData
 from mainframe.models.orm import Scan
 from mainframe.models.schemas import Error, PackageSpecifier
@@ -152,7 +153,7 @@ async def report_package(
             error = HTTPException(
                 409,
                 detail=(
-                    f"Only one version of a package may be reported at a time. "
+                    "Only one version of a package may be reported at a time. "
                     f"(`{row.name}@{row.version}` was already reported)"
                 ),
             )
@@ -191,7 +192,7 @@ async def report_package(
             error = HTTPException(
                 400,
                 detail=(
-                    f"additional_information is a required field as package "
+                    "additional_information is a required field as package "
                     f"`{name}@{version}` has no matched rules in the database"
                 ),
             )
@@ -202,6 +203,16 @@ async def report_package(
 
         rules_matched.extend(rule.name for rule in row.rules)
 
+    if in_greylist(name, row.rule_names, session):
+        raise HTTPException(
+            409,
+            detail=(
+                f"A previous version ({row.version}) of this package matched "
+                "all of the same rules. It is improbable that this new version "
+                "has become malicious without deviations in rule flagging"
+            ),
+        )
+
     additional_information = body.additional_information
 
     send_report_email(

src/mainframe/greylist.py

@@ -0,0 +1,37 @@
+from typing import Annotated
+
+from fastapi import Depends
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from mainframe.database import get_db
+from mainframe.models.orm import Scan, Status
+
+
+async def in_greylist(
+    package_name: str,
+    rules_matched: list[str],
+    session: Annotated[AsyncSession, Depends(get_db)],
+) -> bool:
+    """
+    Check if the rules we matched are the same as the last scan.
+
+    If this package is the first release we scanned, returns `False`.
+    """
+
+    if len(rules_matched) == 0:
+        return False
+
+    row = (
+        await session.scalars(
+            select(Scan.rules)
+            .where(Scan.name == package_name)
+            .where(Scan.status == Status.FINISHED)
+            .order_by(Scan.finished_at.desc())
+        )
+    ).first()
+
+    if row is None:
+        return False
+
+    return set(r.name for r in row) == set(rules_matched)