feat: Add support for multiple matching algorithm versions

infra/production.nix

@@ -105,6 +105,7 @@ in
       EMAIL_USE_SSL = true;
       EMAIL_HOST_USER = "noreply-securitytracker@nixos.org";
       DEFAULT_FROM_EMAIL = "noreply-securitytracker@nixos.org";
+      ACTIVE_MATCHING_ALGORITHM_VERSION = 1;
     };
 
     secrets = {

infra/staging.nix

@@ -114,6 +114,7 @@ in
       GH_SECURITY_TEAM = "sectracker-testing-security";
       GH_COMMITTERS_TEAM = "sectracker-testing-committers";
       EMAIL_BACKEND = "django.core.mail.backends.console.EmailBackend";
+      ACTIVE_MATCHING_ALGORITHM_VERSION = 1;
     };
 
     secrets = {

src/project/settings.py

@@ -169,6 +169,25 @@ class DjangoSettings(BaseModel):
             """,
             default=1_000,
         )
+        ACTIVE_MATCHING_ALGORITHM_VERSION: int = Field(
+            description="""
+            Controls which registered matching algorithm version is used when
+            linking CVEs to derivations. Must match a VERSION defined in
+            shared/listeners/algorithms/. Bump this setting to activate a new
+            algorithm version without changing code.
+            """,
+            default=1,
+        )
+        CANDIDATE_MATCHING_ALGORITHM_VERSION: int | None = Field(
+            description="""
+            Optional. When set, identifies a new algorithm version being evaluated
+            in parallel. The candidate version does not run automatically — it is
+            only invoked by the test-run management command, which generates proposals
+            tagged with this version number for later metric comparison.
+            Set to None when no candidate is under evaluation.
+            """,
+            default=None,
+        )
         SHOW_DEMO_DISCLAIMER: bool = Field(
             description="""
             When set to True, the application will display a disclaimer about

src/shared/listeners/algorithms/__init__.py

@@ -0,0 +1,57 @@
+"""
+Algorithm registry for CVE-to-derivation matching.
+
+Each algorithm version lives in its own module (v1.py, v2.py, ...) and must expose:
+  - VERSION
+  - build_new_links
+
+Modules register themselves by calling `register()`. The listener in
+"""
+
+from typing import Protocol
+
+from django.conf import settings
+
+from shared.models.cve import Container
+
+
+class MatchingAlgorithm(Protocol):
+    VERSION: int
+
+    def build_new_links(self, container: Container) -> bool: ...
+
+
+_registry: dict[int, MatchingAlgorithm] = {}
+
+
+def register(module: MatchingAlgorithm) -> None:
+    """Register an algorithm module under its VERSION."""
+    _registry[module.VERSION] = module
+
+
+def _resolve(version: int) -> MatchingAlgorithm:
+    """Return a resigister alogirithm that match the version."""
+    if not _registry:
+        raise RuntimeError("No matching algorithm registered.")
+    try:
+        return _registry[version]
+    except KeyError:
+        raise KeyError(f"No matching algorithm registered for version {version}.")
+
+
+def current_algorithm() -> MatchingAlgorithm:
+    """Return the active algorithm (ACTIVE_MATCHING_ALGORITHM_VERSION)."""
+    return _resolve(settings.ACTIVE_MATCHING_ALGORITHM_VERSION)
+
+
+# TODO (@adekoder) will be used when we create the process to run the inactive new algorithm
+# verison
+def candidate_algorithm() -> MatchingAlgorithm | None:
+    """
+    Return the candidate algorithm (CANDIDATE_MATCHING_ALGORITHM_VERSION),
+    or None if no candidate is configured.
+    """
+    version = settings.CANDIDATE_MATCHING_ALGORITHM_VERSION
+    if version is None:
+        return None
+    return _resolve(version)

src/shared/listeners/algorithms/v1.py

@@ -0,0 +1,196 @@
+"""
+Matching algorithm version 1.
+
+Candidates are found by matching package name / product name (case-insensitive
+substring) against derivation names in the latest completed evaluation of each
+major channel. No version constraint checking is applied at this stage.
+"""
+
+import logging
+import sys
+
+from django.conf import settings
+from django.db.models import (
+    Case,
+    Exists,
+    F,
+    IntegerField,
+    OuterRef,
+    Q,
+    QuerySet,
+    Value,
+    When,
+    Window,
+)
+from django.db.models.functions import RowNumber
+
+from shared.models.cve import AffectedProduct, Container, Cpe
+from shared.models.linkage import CVEDerivationClusterProposal, ProvenanceFlags
+from shared.models.nix_evaluation import MAJOR_CHANNELS, NixDerivation, NixEvaluation
+
+from . import register
+
+VERSION: int = 1
+
+logger = logging.getLogger(__name__)
+
+
+def produce_linkage_candidates(
+    container: Container,
+    filtered_affected: QuerySet[AffectedProduct],
+) -> dict[NixDerivation, ProvenanceFlags]:
+    # FIXME(@fricklerhandwerk): This will fall apart when we obtain the channel structure dynamically [ref:channel-structure]
+    active_channels_q = Q()
+    for ch in MAJOR_CHANNELS:
+        active_channels_q |= Q(channel__channel_branch__contains=ch)
+
+    latest_complete_channels = (
+        NixEvaluation.objects.filter(
+            active_channels_q,
+            state=NixEvaluation.EvaluationState.COMPLETED,
+        )
+        .annotate(
+            row_num=Window(
+                expression=RowNumber(),
+                partition_by=[F("channel")],
+                order_by=F("updated_at").desc(),
+            ),
+        )
+        .filter(row_num=1)
+    )
+
+    package_names = (
+        filtered_affected.exclude(package_name__isnull=True)
+        .values_list("package_name", flat=True)
+        .distinct()
+    )
+    products = (
+        filtered_affected.exclude(product__isnull=True)
+        .values_list("product", flat=True)
+        .distinct()
+    )
+
+    package_q = Q()
+    for name in package_names:
+        package_q |= Q(name__icontains=name)
+
+    product_q = Q()
+    for product in products:
+        product_q |= Q(name__icontains=product)
+
+    if not package_q | product_q:
+        return {}
+
+    annotations = {}
+    if package_q:
+        annotations["package_match"] = Case(
+            When(package_q, then=Value(ProvenanceFlags.PACKAGE_NAME_MATCH)),
+            default=Value(0),
+            output_field=IntegerField(),
+        )
+    if product_q:
+        annotations["product_match"] = Case(
+            When(product_q, then=Value(ProvenanceFlags.PRODUCT_MATCH)),
+            default=Value(0),
+            output_field=IntegerField(),
+        )
+
+    candidates: dict[NixDerivation, ProvenanceFlags] = {}
+    matches = NixDerivation.objects.filter(
+        package_q | product_q,
+        parent_evaluation__in=list(latest_complete_channels),
+    ).annotate(**annotations)
+    for drv in matches.iterator():
+        flags = getattr(drv, "package_match", 0) | getattr(drv, "product_match", 0)
+        candidates[drv] = ProvenanceFlags(flags)
+
+    return candidates
+
+
+def build_new_links(container: Container) -> bool:
+    if container.cve.triaged:
+        logger.info(
+            "Container received for '%s', but already triaged, skipping linkage.",
+            container.cve,
+        )
+        return False
+
+    if CVEDerivationClusterProposal.objects.filter(
+        cve=container.cve, algorithm_version=VERSION
+    ).exists():
+        logger.info("Suggestion already exists for '%s', skipping", container.cve)
+        return False
+
+    if container.tags.filter(value="exclusively-hosted-service").exists():
+        logger.info(
+            "Container for '%s' is exclusively-hosted-service, rejecting without match.",
+            container.cve,
+        )
+        CVEDerivationClusterProposal.objects.create(
+            cve=container.cve,
+            status=CVEDerivationClusterProposal.Status.REJECTED,
+            rejection_reason=CVEDerivationClusterProposal.RejectionReason.EXCLUSIVELY_HOSTED_SERVICE,
+            algorithm_version=VERSION,
+        )
+        return True
+
+    has_any_cpe = Exists(Cpe.objects.filter(affectedproduct=OuterRef("pk")))
+    has_non_hardware_cpe = Exists(
+        Cpe.objects.filter(affectedproduct=OuterRef("pk")).exclude(
+            name__istartswith="cpe:2.3:h:"
+        )
+    )
+    filtered_affected = container.affected.exclude(has_any_cpe & ~has_non_hardware_cpe)
+
+    if container.affected.exists() and not filtered_affected.exists():
+        logger.info(
+            "Container for '%s' has only hardware CPEs, rejecting without match.",
+            container.cve,
+        )
+        CVEDerivationClusterProposal.objects.create(
+            cve=container.cve,
+            status=CVEDerivationClusterProposal.Status.REJECTED,
+            rejection_reason=CVEDerivationClusterProposal.RejectionReason.HARDWARE_ONLY_CPE,
+            algorithm_version=VERSION,
+        )
+        return True
+
+    drvs = produce_linkage_candidates(container, filtered_affected)
+    if not drvs:
+        logger.info("No derivations matching '%s', ignoring", container.cve)
+        return False
+
+    if len(drvs) > settings.MAX_MATCHES:
+        logger.warning(
+            "More than '%d' derivations matching '%s', ignoring",
+            settings.MAX_MATCHES,
+            container.cve,
+        )
+        return False
+
+    proposal = CVEDerivationClusterProposal.objects.create(
+        cve=container.cve,
+        algorithm_version=VERSION,
+    )
+
+    drvs_throughs = [
+        CVEDerivationClusterProposal.derivations.through(
+            proposal_id=proposal.pk, derivation_id=drv.pk, provenance_flags=flags
+        )
+        for drv, flags in drvs.items()
+    ]
+
+    CVEDerivationClusterProposal.derivations.through.objects.bulk_create(drvs_throughs)
+
+    if drvs_throughs:
+        logger.info(
+            "Matching suggestion for '%s': %d derivations found.",
+            container.cve,
+            len(drvs_throughs),
+        )
+
+    return True
+
+
+# Self-register when imported
+register(sys.modules[__name__])  # type: ignore[arg-type]