diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 2e9087bb55..644d1e8584 100644 --- a/CHANGELOG.rst +++ b/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) -------------------- diff --git a/docs/command-line-interface.rst b/docs/command-line-interface.rst index 044fc289f6..216fae52d8 100644 --- a/docs/command-line-interface.rst +++ b/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` ---------------------------------------------- diff --git a/scanpipe/management/commands/check-compliance.py b/scanpipe/management/commands/check-compliance.py index 7af6bf6f8f..342b5b55e4 100644 --- a/scanpipe/management/commands/check-compliance.py +++ b/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 diff --git a/scanpipe/models.py b/scanpipe/models.py index 754c9f419d..d763b9c036 100644 --- a/scanpipe/models.py +++ b/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, diff --git a/scanpipe/pipes/output.py b/scanpipe/pipes/output.py index 8c481cc17a..30737d606c 100644 --- a/scanpipe/pipes/output.py +++ b/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 = [ diff --git a/scanpipe/tests/test_commands.py b/scanpipe/tests/test_commands.py index e41da2c036..6dcbd5d09c 100644 --- a/scanpipe/tests/test_commands.py +++ b/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])