Add --vulnerabilities and --strict options in verify-project (#1964)

Signed-off-by: tdruez <tdruez@aboutcode.org>

Author: tdruez

docs/command-line-interface.rst

@@ -803,29 +803,50 @@ See the :ref:`cli_output` for more information about supported output formats.
 
 Verifies the analysis results of a project against expected package and dependency
 counts.
-This command is designed to ensure that a project’s scan results meet specific
+This command is designed to ensure that a project's scan results meet specific
 expectations — for example, that a minimum number of packages or dependencies were
-discovered, and that no unexpected vulnerabilities were introduced.
+discovered, or that vulnerability counts match expected baselines.
 
 Optional arguments:
 
-- ``--packages`` Minimum number of discovered packages expected.
+- ``--packages`` Expected number of discovered packages.
 
-- ``--vulnerable-packages`` Minimum number of vulnerable packages expected.
+- ``--vulnerable-packages`` Expected number of vulnerable packages.
 
-- ``--dependencies`` Minimum number of discovered dependencies expected.
+- ``--dependencies`` Expected number of discovered dependencies.
 
-- ``--vulnerable-dependencies`` Minimum number of vulnerable dependencies expected.
+- ``--vulnerable-dependencies`` Expected number of vulnerable dependencies.
 
-If any of these expectations are not met, the command exits with a non-zero status
-and prints a summary of all issues found.
+- ``--vulnerabilities`` Expected number of unique vulnerabilities.
+  Combines vulnerabilities from both packages and dependencies.
+
+- ``--strict`` Assert on strict count equality instead of minimum threshold.
+  When not provided, the command checks that actual counts are at least the expected
+  values (greater than or equal). With ``--strict``, actual counts must match expected
+  values exactly.
+
+By default, the command verifies that actual counts meet or exceed the expected values.
+Only the metrics explicitly provided via command-line arguments are validated.
+
+If any expectations are not met, the command exits with a non-zero status and prints
+a summary of all issues found.
 
 Example usage:
 
-.. code-block:: bash
+1. Verify minimum thresholds (default behavior)::
+
+   $ scanpipe verify-project --project my_project --packages 100 --dependencies 50
+
+2. Verify exact counts with strict mode::
+
+   $ scanpipe verify-project --project my_project --vulnerabilities 14 --strict
+
+3. Verify only specific metrics::
 
-    $ scanpipe verify-project --project my_project --packages 100 --dependencies 50
+   $ scanpipe verify-project --project my_project --vulnerable-packages 5
 
 .. tip::
     This command is particularly useful for **CI/CD pipelines** that need to validate
-    SBOM or vulnerability scan results against known baselines.
+    SBOM or vulnerability scan results against known baselines. Use non-strict mode
+    to ensure minimum quality thresholds, or strict mode to detect unexpected changes
+    in scan results.

scanpipe/management/commands/verify-project.py

@@ -33,26 +33,40 @@ def add_arguments(self, parser):
         parser.add_argument(
             "--packages",
             type=int,
-            default=0,
-            help="Minimum number of packages expected (default: 0)",
+            default=None,
+            help="Expected number of packages",
         )
         parser.add_argument(
             "--vulnerable-packages",
             type=int,
-            default=0,
-            help="Minimum number of vulnerable packages expected (default: 0)",
+            default=None,
+            help="Expected number of vulnerable packages",
         )
         parser.add_argument(
             "--dependencies",
             type=int,
-            default=0,
-            help="Minimum number of dependencies expected (default: 0)",
+            default=None,
+            help="Expected number of dependencies",
         )
         parser.add_argument(
             "--vulnerable-dependencies",
             type=int,
-            default=0,
-            help="Minimum number of vulnerable dependencies expected (default: 0)",
+            default=None,
+            help="Expected number of vulnerable dependencies",
+        )
+        parser.add_argument(
+            "--vulnerabilities",
+            type=int,
+            default=None,
+            help=(
+                "Expected number of unique vulnerabilities. "
+                "Combines vulnerabilities from both packages and dependencies"
+            ),
+        )
+        parser.add_argument(
+            "--strict",
+            action="store_true",
+            help="Assert on strict count equality instead of minimum threshold",
         )
 
     def handle(self, *args, **options):
@@ -62,38 +76,66 @@ def handle(self, *args, **options):
         expected_vulnerable_packages = options["vulnerable_packages"]
         expected_dependencies = options["dependencies"]
         expected_vulnerable_dependencies = options["vulnerable_dependencies"]
+        expected_vulnerabilities = options["vulnerabilities"]
+        strict = options["strict"]
 
         project = self.project
         packages = project.discoveredpackages
-        package_count = packages.count()
-        vulnerable_package_count = packages.vulnerable().count()
-        dependencies = project.discovereddependencies.all()
-        dependency_count = dependencies.count()
-        vulnerable_dependency_count = dependencies.vulnerable().count()
+        dependencies = project.discovereddependencies
+        vulnerabilities = project.vulnerabilities
 
-        errors = []
+        # Check all counts (only if expected value is provided)
+        checks = [
+            (
+                packages.count(),
+                expected_packages,
+                "packages",
+            ),
+            (
+                packages.vulnerable().count(),
+                expected_vulnerable_packages,
+                "vulnerable packages",
+            ),
+            (
+                dependencies.count(),
+                expected_dependencies,
+                "dependencies",
+            ),
+            (
+                dependencies.vulnerable().count(),
+                expected_vulnerable_dependencies,
+                "vulnerable dependencies",
+            ),
+            (
+                len(vulnerabilities),
+                expected_vulnerabilities,
+                "vulnerabilities on the project",
+            ),
+        ]
 
-        if package_count < expected_packages:
-            errors.append(
-                f"Expected at least {expected_packages} packages, found {package_count}"
-            )
-        if vulnerable_package_count < expected_vulnerable_packages:
-            errors.append(
-                f"Expected at least {expected_vulnerable_packages} vulnerable packages,"
-                f" found {vulnerable_package_count}"
-            )
-        if dependency_count < expected_dependencies:
-            errors.append(
-                f"Expected at least {expected_dependencies} dependencies, "
-                f"found {dependency_count}"
-            )
-        if vulnerable_dependency_count < expected_vulnerable_dependencies:
-            errors.append(
-                f"Expected at least {expected_vulnerable_dependencies} "
-                f"vulnerable dependencies, found {vulnerable_dependency_count}"
-            )
+        errors = []
+        for actual, expected, label in checks:
+            if expected is not None:  # Only check if value was provided
+                if error := self.check_count(actual, expected, label, strict):
+                    errors.append(error)
 
         if errors:
             raise CommandError("Project verification failed:\n" + "\n".join(errors))
 
         self.stdout.write("Project verification passed.", self.style.SUCCESS)
+
+    @staticmethod
+    def check_count(actual, expected, label, strict):
+        """
+        Check if actual count meets expectations.
+
+        In strict mode, checks for exact equality.
+        Otherwise, checks if actual is at least the expected value.
+
+        Returns an error message string if check fails.
+        """
+        if strict and actual != expected:
+            return f"Expected exactly {expected} {label}, found {actual}"
+
+        if not strict and actual < expected:
+            return f"Expected at least {expected} {label}, found {actual}"

scanpipe/models.py

@@ -29,6 +29,7 @@
 from collections import Counter
 from collections import defaultdict
 from contextlib import suppress
+from itertools import chain
 from itertools import groupby
 from operator import itemgetter
 from pathlib import Path
@@ -1486,12 +1487,12 @@ def package_count(self):
     @cached_property
     def vulnerable_package_count(self):
         """Return the number of vulnerable packages related to this project."""
-        return self.discoveredpackages.vulnerable().count()
+        return self.vulnerable_packages.count()
 
     @cached_property
     def vulnerable_dependency_count(self):
         """Return the number of vulnerable dependencies related to this project."""
-        return self.discovereddependencies.vulnerable().count()
+        return self.vulnerable_dependencies.count()
 
     @cached_property
     def dependency_count(self):
@@ -1537,6 +1538,49 @@ def relation_count(self):
         """Return the number of relations related to this project."""
         return self.codebaserelations.count()
 
+    @cached_property
+    def vulnerable_packages(self):
+        """Return a QuerySet of vulnerable packages."""
+        return self.discoveredpackages.vulnerable()
+
+    @cached_property
+    def vulnerable_dependencies(self):
+        """Return a QuerySet of vulnerable dependencies."""
+        return self.discovereddependencies.vulnerable()
+
+    @property
+    def package_vulnerabilities(self):
+        """Return the list of package vulnerabilities."""
+        return self.vulnerable_packages.get_vulnerabilities_list()
+
+    @property
+    def dependency_vulnerabilities(self):
+        """Return the list of dependency vulnerabilities."""
+        return self.vulnerable_dependencies.get_vulnerabilities_list()
+
+    @property
+    def vulnerabilities(self):
+        """
+        Return a dict of all vulnerabilities affecting this project.
+
+        Combines package and dependency vulnerabilities, keyed by vulnerability_id.
+        Each vulnerability includes an "affects" list of all affected packages
+        and dependencies.
+        """
+        vulnerabilities_dict = {}
+        # Process both packages and dependencies
+        querysets = [self.vulnerable_packages, self.vulnerable_dependencies]
+
+        for queryset in querysets:
+            vulnerabilities = queryset.get_vulnerabilities_dict()
+            for vcid, vuln_data in vulnerabilities.items():
+                if vcid in vulnerabilities_dict:
+                    vulnerabilities_dict[vcid]["affects"].extend(vuln_data["affects"])
+                else:
+                    vulnerabilities_dict[vcid] = vuln_data
+
+        return vulnerabilities_dict
+
     @cached_property
     def has_single_resource(self):
         """
@@ -3284,16 +3328,70 @@ def is_vulnerable(self):
 
 
 class VulnerabilityQuerySetMixin:
+    AFFECTED_BY_FIELD = "affected_by_vulnerabilities"
+
     def vulnerable(self):
-        return self.filter(~Q(affected_by_vulnerabilities__in=EMPTY_VALUES))
+        return self.filter(~Q(**{f"{self.AFFECTED_BY_FIELD}__in": EMPTY_VALUES}))
 
     def vulnerable_ordered(self):
         return (
             self.vulnerable()
-            .only_package_url_fields(extra=["affected_by_vulnerabilities"])
+            .only_package_url_fields(extra=[self.AFFECTED_BY_FIELD])
             .order_by_package_url()
         )
 
+    def get_vulnerabilities_list(self):
+        """
+        Return a deduplicated, sorted flat list of all vulnerabilities from the
+        queryset.
+
+        Extracts and flattens the affected_by_vulnerabilities field from
+        all objects in the queryset. Removes duplicates based on vulnerability_id
+        while preserving the first occurrence of each unique vulnerability.
+        """
+        vulnerabilities_lists = self.values_list(self.AFFECTED_BY_FIELD, flat=True)
+        flatten_vulnerabilities = chain.from_iterable(vulnerabilities_lists)
+
+        # Deduplicate by vulnerability_id while preserving order
+        unique_vulnerabilities = {
+            vuln["vulnerability_id"]: vuln for vuln in flatten_vulnerabilities
+        }
+
+        return sorted(
+            unique_vulnerabilities.values(), key=itemgetter("vulnerability_id")
+        )
+
+    def get_vulnerabilities_dict(self):
+        """
+        Return a dict of vulnerabilities keyed by vulnerability_id.
+
+        Each vulnerability includes an "affects" list containing all
+        objects from this queryset affected by that vulnerability.
+
+        Returns:
+            dict: {
+                'VCID-1': {
+                    'vulnerability_id': 'VCID-1',
+                    'affects': [obj1, obj2, ...]
+                },
+                ...
+            }
+
+        """
+        vulnerabilities_dict = {}
+
+        for obj in self.vulnerable_ordered():
+            for vulnerability in obj.affected_by_vulnerabilities:
+                vcid = vulnerability.get("vulnerability_id")
+                if not vcid:
+                    continue
+
+                if vcid not in vulnerabilities_dict:
+                    vulnerabilities_dict[vcid] = {**vulnerability, "affects": []}
+                vulnerabilities_dict[vcid]["affects"].append(obj)
+
+        return vulnerabilities_dict
+
 
 class DiscoveredPackageQuerySet(
     VulnerabilityQuerySetMixin,
@@ -3324,7 +3422,7 @@ def only_package_url_fields(self, extra=None):
         if not extra:
             extra = []
 
-        return self.only("uuid", *PACKAGE_URL_FIELDS, *extra)
+        return self.only("uuid", *PACKAGE_URL_FIELDS, "project_id", *extra)
 
     def filter(self, *args, **kwargs):
         """Add support for using ``package_url`` as a field lookup."""
@@ -3945,7 +4043,7 @@ def only_package_url_fields(self, extra=None):
         if not extra:
             extra = []
 
-        return self.only("dependency_uid", *PACKAGE_URL_FIELDS, *extra)
+        return self.only("dependency_uid", *PACKAGE_URL_FIELDS, "project_id", *extra)
 
 
 class DiscoveredDependency(