Add a --fail-on-vulnerabilities in check-compliance command (#1702)

Signed-off-by: tdruez <tdruez@nexb.com>

Author: tdruez

CHANGELOG.rst

@@ -1,6 +1,15 @@
 Changelog
 =========
 
+v35.1.0 (unreleased)
+--------------------
+
+- Add a ``--fail-on-vulnerabilities`` option in ``check-compliance`` management command.
+  When this option is enabled, the command will exit with a non-zero status if known
+  vulnerabilities are detected in discovered packages and dependencies.
+  Requires the ``find_vulnerabilities`` pipeline to be executed beforehand.
+  https://github.com/aboutcode-org/scancode.io/pull/1702
+
 v35.0.0 (2025-06-23)
 --------------------
 

docs/command-line-interface.rst

@@ -497,6 +497,10 @@ Optional arguments:
 - ``--fail-level {ERROR,WARNING,MISSING}`` Compliance alert level that will cause the
   command to exit with a non-zero status. Default is ERROR.
 
+- ``--fail-on-vulnerabilities`` Exit with a non-zero status if known vulnerabilities
+  are detected in discovered packages and dependencies.
+  Requires the ``find_vulnerabilities`` pipeline to be executed beforehand.
+
 `$ scanpipe archive-project --project PROJECT`
 ----------------------------------------------
 

scanpipe/management/commands/check-compliance.py

@@ -45,31 +45,64 @@ def add_arguments(self, parser):
                 "non-zero status. Default is ERROR."
             ),
         )
+        parser.add_argument(
+            "--fail-on-vulnerabilities",
+            action="store_true",
+            help=(
+                "Exit with a non-zero status if known vulnerabilities are detected in "
+                "discovered packages and dependencies. "
+                "Requires the `find_vulnerabilities` pipeline to be executed "
+                "beforehand."
+            ),
+        )
 
     def handle(self, *args, **options):
         super().handle(*args, **options)
-        fail_level = options["fail_level"]
-        compliance_alerts = get_project_compliance_alerts(self.project, fail_level)
+        exit_code = 0
+
+        if self.check_compliance(options["fail_level"]):
+            exit_code = 1
+
+        if options["fail_on_vulnerabilities"] and self.check_vulnerabilities():
+            exit_code = 1
 
-        compliance_alerts_count = sum(
-            len(issues_by_severity)
-            for model_alerts in compliance_alerts.values()
-            for issues_by_severity in model_alerts.values()
+        sys.exit(exit_code)
+
+    def check_compliance(self, fail_level):
+        alerts = get_project_compliance_alerts(self.project, fail_level)
+        count = sum(
+            len(issues) for model in alerts.values() for issues in model.values()
         )
-        if not compliance_alerts_count:
-            sys.exit(0)
 
-        if self.verbosity > 0:
-            msg = [
-                f"{compliance_alerts_count} compliance issues detected on this project."
-            ]
-            for label, issues in compliance_alerts.items():
-                msg.append(f"[{label}]")
-                for severity, entries in issues.items():
-                    msg.append(f" > {severity.upper()}: {len(entries)}")
+        if count and self.verbosity > 0:
+            self.stderr.write(f"{count} compliance issues detected.")
+            for label, model in alerts.items():
+                self.stderr.write(f"[{label}]")
+                for severity, entries in model.items():
+                    self.stderr.write(f" > {severity.upper()}: {len(entries)}")
                     if self.verbosity > 1:
-                        msg.append("   " + "\n   ".join(entries))
+                        self.stderr.write("   " + "\n   ".join(entries))
+
+        return count > 0
 
-            self.stderr.write("\n".join(msg))
+    def check_vulnerabilities(self):
+        packages = self.project.discoveredpackages.vulnerable_ordered()
+        dependencies = self.project.discovereddependencies.vulnerable_ordered()
+
+        vulnerable_records = list(packages) + list(dependencies)
+        count = len(vulnerable_records)
+
+        if self.verbosity > 0:
+            if count:
+                self.stderr.write(f"{count} vulnerable records found:")
+                for entry in vulnerable_records:
+                    self.stderr.write(str(entry))
+                    vulnerability_ids = [
+                        vulnerability.get("vulnerability_id")
+                        for vulnerability in entry.affected_by_vulnerabilities
+                    ]
+                    self.stderr.write(" > " + ", ".join(vulnerability_ids))
+            else:
+                self.stdout.write("No vulnerabilities found")
 
-        sys.exit(1)
+        return count > 0

scanpipe/models.py

@@ -3150,6 +3150,13 @@ class VulnerabilityQuerySetMixin:
     def vulnerable(self):
         return self.filter(~Q(affected_by_vulnerabilities__in=EMPTY_VALUES))
 
+    def vulnerable_ordered(self):
+        return (
+            self.vulnerable()
+            .only_package_url_fields(extra=["affected_by_vulnerabilities"])
+            .order_by_package_url()
+        )
+
 
 class DiscoveredPackageQuerySet(
     VulnerabilityQuerySetMixin,

scanpipe/pipes/output.py

@@ -567,21 +567,11 @@ def to_xlsx(project):
 
 
 def add_vulnerabilities_sheet(workbook, project):
-    vulnerable_packages_queryset = (
-        DiscoveredPackage.objects.project(project)
-        .vulnerable()
-        .only_package_url_fields(extra=["affected_by_vulnerabilities"])
-        .order_by_package_url()
-    )
-    vulnerable_dependencies_queryset = (
-        DiscoveredDependency.objects.project(project)
-        .vulnerable()
-        .only_package_url_fields(extra=["affected_by_vulnerabilities"])
-        .order_by_package_url()
-    )
+    vulnerable_packages = project.discoveredpackages.vulnerable_ordered()
+    vulnerable_dependencies = project.discovereddependencies.vulnerable_ordered()
     vulnerable_querysets = [
-        vulnerable_packages_queryset,
-        vulnerable_dependencies_queryset,
+        vulnerable_packages,
+        vulnerable_dependencies,
     ]
 
     vulnerability_fields = [

scanpipe/tests/test_commands.py

@@ -49,6 +49,7 @@
 from scanpipe.pipes import flag
 from scanpipe.pipes import purldb
 from scanpipe.tests import filter_warnings
+from scanpipe.tests import make_dependency
 from scanpipe.tests import make_mock_response
 from scanpipe.tests import make_package
 from scanpipe.tests import make_project
@@ -1196,9 +1197,7 @@ def test_scanpipe_management_command_check_compliance(self):
             call_command("check-compliance", *options, stderr=out)
         self.assertEqual(cm.exception.code, 1)
         out_value = out.getvalue().strip()
-        expected = (
-            "1 compliance issues detected on this project.\n[packages]\n > ERROR: 1"
-        )
+        expected = "1 compliance issues detected.\n[packages]\n > ERROR: 1"
         self.assertEqual(expected, out_value)
 
         out = StringIO()
@@ -1208,12 +1207,46 @@ def test_scanpipe_management_command_check_compliance(self):
         self.assertEqual(cm.exception.code, 1)
         out_value = out.getvalue().strip()
         expected = (
-            "2 compliance issues detected on this project."
+            "2 compliance issues detected."
             "\n[packages]\n > ERROR: 1"
             "\n[resources]\n > WARNING: 1"
         )
         self.assertEqual(expected, out_value)
 
+    def test_scanpipe_management_command_check_compliance_vulnerabilities(self):
+        project = make_project(name="my_project")
+        package1 = make_package(project, package_url="pkg:generic/name@1.0")
+
+        out = StringIO()
+        options = ["--project", project.name, "--fail-on-vulnerabilities"]
+        with self.assertRaises(SystemExit) as cm:
+            call_command("check-compliance", *options, stdout=out)
+        self.assertEqual(cm.exception.code, 0)
+        out_value = out.getvalue().strip()
+        self.assertEqual("No vulnerabilities found", out_value)
+
+        vulnerability_data = [{"vulnerability_id": "VCID-cah8-awtr-aaad"}]
+        package1.update(affected_by_vulnerabilities=vulnerability_data)
+        make_dependency(
+            project,
+            dependency_uid="dependency1",
+            affected_by_vulnerabilities=vulnerability_data,
+        )
+        out = StringIO()
+        options = ["--project", project.name, "--fail-on-vulnerabilities"]
+        with self.assertRaises(SystemExit) as cm:
+            call_command("check-compliance", *options, stderr=out)
+        self.assertEqual(cm.exception.code, 1)
+        out_value = out.getvalue().strip()
+        expected = (
+            "2 vulnerable records found:\n"
+            "pkg:generic/name@1.0\n"
+            " > VCID-cah8-awtr-aaad\n"
+            "dependency1\n"
+            " > VCID-cah8-awtr-aaad"
+        )
+        self.assertEqual(expected, out_value)
+
     def test_scanpipe_management_command_report(self):
         label1 = "label1"
         project1 = make_project("project1", labels=[label1])