Add a new check-compliance management command #1346 (#1364)

* Add compliance_issues QuerySet method #1346

Signed-off-by: tdruez <[email protected]>

* Add a new ``check-compliance`` management command #1346

Signed-off-by: tdruez <[email protected]>

---------

Signed-off-by: tdruez <[email protected]>

Author: tdruez

CHANGELOG.rst

@@ -47,6 +47,10 @@ v34.7.2 (unreleased)
 - Update link references of ownership from nexB to aboutcode-org
   https://github.com/aboutcode-org/scancode.io/issues/1350
 
+- Add a new ``check-compliance`` management command to check for compliance issues in
+  a project.
+  https://github.com/nexB/scancode.io/issues/1182
+
 v34.7.1 (2024-07-15)
 --------------------
 

docs/command-line-interface.rst

@@ -264,6 +264,19 @@ Optional arguments:
 Refer to :ref:`Mount projects workspace <mount_projects_workspace_volume>` to access
 your outputs on the host machine when running with Docker.
 
+`$ scanpipe check-compliance --project PROJECT`
+-----------------------------------------------
+
+Check for compliance issues in Project.
+Exit with a non-zero status if compliance issues are present in the project.
+The compliance alert indicates how the license expression complies with provided
+policies.
+
+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.
+
 `$ scanpipe archive-project --project PROJECT`
 ----------------------------------------------
 

scanpipe/management/commands/check-compliance.py

@@ -0,0 +1,92 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https://github.com/nexB/scancode.io
+# The ScanCode.io software is licensed under the Apache License version 2.0.
+# Data generated with ScanCode.io is provided as-is without warranties.
+# ScanCode is a trademark of nexB Inc.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# ScanCode.io is a free software code scanning tool from nexB Inc. and others.
+# Visit https://github.com/nexB/scancode.io for support and download.
+
+import sys
+from collections import defaultdict
+
+from scanpipe.management.commands import ProjectCommand
+from scanpipe.models import PACKAGE_URL_FIELDS
+
+
+class Command(ProjectCommand):
+    help = (
+        "Check for compliance issues in Project. Exit with a non-zero status if "
+        "compliance issues are present in the project."
+        "The compliance alert indicates how the license expression complies with "
+        "provided policies."
+    )
+
+    def add_arguments(self, parser):
+        super().add_arguments(parser)
+        parser.add_argument(
+            "--fail-level",
+            default="ERROR",
+            choices=["ERROR", "WARNING", "MISSING"],
+            help=(
+                "Compliance alert level that will cause the command to exit with a "
+                "non-zero status. Default is ERROR."
+            ),
+        )
+
+    def handle(self, *args, **options):
+        super().handle(*args, **options)
+        fail_level = options["fail_level"]
+        total_compliance_issues_count = 0
+
+        package_qs = self.project.discoveredpackages.compliance_issues(
+            severity=fail_level
+        ).only(*PACKAGE_URL_FIELDS, "compliance_alert")
+
+        resource_qs = self.project.codebaseresources.compliance_issues(
+            severity=fail_level
+        ).only("path", "compliance_alert")
+
+        queryset_mapping = {
+            "Package": package_qs,
+            "Resource": resource_qs,
+        }
+
+        results = {}
+        for label, queryset in queryset_mapping.items():
+            compliance_issues = defaultdict(list)
+            for instance in queryset:
+                compliance_issues[instance.compliance_alert].append(str(instance))
+                total_compliance_issues_count += 1
+            if compliance_issues:
+                results[label] = dict(compliance_issues)
+
+        if not total_compliance_issues_count:
+            sys.exit(0)
+
+        if self.verbosity > 0:
+            msg = [
+                f"{total_compliance_issues_count} compliance issues detected on "
+                f"this project."
+            ]
+            for label, issues in results.items():
+                msg.append(f"{label}:")
+                for severity, entries in issues.items():
+                    msg.append(f" - {severity}: {len(entries)}")
+
+            self.stderr.write("\n".join(msg))
+
+        sys.exit(1)

scanpipe/models.py

@@ -2061,6 +2061,33 @@ def profile(self, print_results=False):
                 print(output_str)
 
 
+class ComplianceAlertQuerySetMixin:
+    def compliance_issues(self, severity):
+        """
+        Retrieve compliance issues based on severity.
+        Supported values are 'error', 'warning', and 'missing'.
+        """
+        compliance = self.model.Compliance
+        severity = severity.lower()
+
+        severity_mapping = {
+            "error": [compliance.ERROR.value],
+            "warning": [compliance.ERROR.value, compliance.WARNING.value],
+            "missing": [
+                compliance.ERROR.value,
+                compliance.WARNING.value,
+                compliance.MISSING.value,
+            ],
+        }
+
+        if severity not in severity_mapping:
+            raise ValueError(
+                f"Supported severities are: {', '.join(severity_mapping.keys())}"
+            )
+
+        return self.filter(compliance_alert__in=severity_mapping[severity])
+
+
 def convert_glob_to_django_regex(glob_pattern):
     """
     Convert a glob pattern to an equivalent django regex pattern
@@ -2073,7 +2100,7 @@ def convert_glob_to_django_regex(glob_pattern):
     return escaped_pattern
 
 
-class CodebaseResourceQuerySet(ProjectRelatedQuerySet):
+class CodebaseResourceQuerySet(ComplianceAlertQuerySetMixin, ProjectRelatedQuerySet):
     def prefetch_for_serializer(self):
         """
         Optimized prefetching for a QuerySet to be consumed by the
@@ -2965,7 +2992,10 @@ def vulnerable(self):
 
 
 class DiscoveredPackageQuerySet(
-    VulnerabilityQuerySetMixin, PackageURLQuerySetMixin, ProjectRelatedQuerySet
+    VulnerabilityQuerySetMixin,
+    PackageURLQuerySetMixin,
+    ComplianceAlertQuerySetMixin,
+    ProjectRelatedQuerySet,
 ):
     def with_resources_count(self):
         count_subquery = Subquery(

scanpipe/tests/test_commands.py

@@ -44,6 +44,8 @@
 from scanpipe.models import Run
 from scanpipe.models import WebhookSubscription
 from scanpipe.pipes import purldb
+from scanpipe.tests import make_package
+from scanpipe.tests import make_resource_file
 
 scanpipe_app = apps.get_app_config("scanpipe")
 
@@ -925,6 +927,52 @@ def test_scanpipe_management_command_purldb_scan_queue_worker_continue_after_fai
             mock_post_call2.kwargs["data"]["scan_log"],
         )
 
+    def test_scanpipe_management_command_check_compliance(self):
+        project = Project.objects.create(name="my_project")
+
+        out = StringIO()
+        options = ["--project", project.name]
+        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("", out_value)
+
+        make_resource_file(
+            project,
+            path="warning",
+            compliance_alert=CodebaseResource.Compliance.WARNING,
+        )
+        make_package(
+            project,
+            package_url="pkg:generic/[email protected]",
+            compliance_alert=CodebaseResource.Compliance.ERROR,
+        )
+
+        out = StringIO()
+        options = ["--project", project.name]
+        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 = (
+            "1 compliance issues detected on this project." "\nPackage:\n - error: 1"
+        )
+        self.assertEqual(expected, out_value)
+
+        out = StringIO()
+        options = ["--project", project.name, "--fail-level", "WARNING"]
+        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 compliance issues detected on this project."
+            "\nPackage:\n - error: 1"
+            "\nResource:\n - warning: 1"
+        )
+        self.assertEqual(expected, out_value)
+
 
 class ScanPipeManagementCommandMixinTest(TestCase):
     class CreateProjectCommand(

scanpipe/tests/test_models.py

@@ -2436,6 +2436,29 @@ def test_scanpipe_codebase_resource_queryset_elfs(self):
         self.assertTrue("e" in paths)
         self.assertTrue("a" in paths)
 
+    def test_scanpipe_model_codebase_resource_compliance_alert_queryset_mixin(self):
+        severities = CodebaseResource.Compliance
+        make_resource_file(self.project1, path="none")
+        make_resource_file(self.project1, path="ok", compliance_alert=severities.OK)
+        warning = make_resource_file(
+            self.project1, path="warning", compliance_alert=severities.WARNING
+        )
+        error = make_resource_file(
+            self.project1, path="error", compliance_alert=severities.ERROR
+        )
+        missing = make_resource_file(
+            self.project1, path="missing", compliance_alert=severities.MISSING
+        )
+
+        qs = CodebaseResource.objects.order_by("path")
+        self.assertQuerySetEqual(qs.compliance_issues(severities.ERROR), [error])
+        self.assertQuerySetEqual(
+            qs.compliance_issues(severities.WARNING), [error, warning]
+        )
+        self.assertQuerySetEqual(
+            qs.compliance_issues(severities.MISSING), [error, missing, warning]
+        )
+
 
 class ScanPipeModelsTransactionTest(TransactionTestCase):
     """