Support LicenseDetection creation in all pipelines

Signed-off-by: Ayan Sinha Mahapatra <ayansmahapatra@gmail.com>

Author: AyanSinhaMahapatra

scanpipe/models.py

@@ -2726,6 +2726,16 @@ def with_resources_count(self):
         )
         return self.annotate(resources_count=count_subquery)
 
+    def has_license_detections(self):
+        return self.filter(
+            ~Q(license_detections=[]) | ~Q(other_license_detections=[])
+        )
+
+    def has_no_license_detections(self):
+        return self.filter(
+            Q(license_detections=[]) & Q(other_license_detections=[])
+        )
+
 
 class AbstractPackage(models.Model):
     """These fields should be kept in line with `packagedcode.models.PackageData`."""
@@ -3506,6 +3516,7 @@ class AbstractLicenseDetection(models.Model):
     )
 
     matches = models.JSONField(
+        _("Reference Matches"),
         default=list,
         blank=True,
         help_text=_('List of license matches combined in this detection.'),
@@ -3547,13 +3558,18 @@ class DiscoveredLicense(
     """
     license_expression_field = "license_expression"
 
+    # If this license was discovered in a extracted license statement
+    # this is True, and False if this was discovered in a file.
+    from_package = None
+
     detection_count = models.BigIntegerField(
         blank=True,
         null=True,
         help_text=_("Total number of this license detection discovered."),
     )
 
     file_regions = models.JSONField(
+        _("Detection Locations"),
         default=list,
         blank=True,
         help_text=_(
@@ -3622,6 +3638,20 @@ def create_from_data(cls, project, detection_data):
         discovered_license.save(save_error=False, capture_exception=False)
         return discovered_license
 
+    def update_with_file_region(self, file_region):
+        """
+        If the `file_region` is a new file region, include it in the
+        `file_regions` list and increase the `detection_count` by 1.
+        """
+        file_region_data = file_region.to_dict()
+        if not file_region_data in self.file_regions:
+            self.file_regions.append(file_region_data)
+            if not self.detection_count:
+                self.detection_count = 1
+            else:
+                self.detection_count += 1
+            self.save(update_fields=["detection_count", "file_regions"])
+
 
 class WebhookSubscription(UUIDPKModel, ProjectRelatedModel):
     target_url = models.URLField(_("Target URL"), max_length=1024)

scanpipe/pipelines/deploy_to_develop.py

@@ -83,6 +83,7 @@ def steps(cls):
             cls.remove_packages_without_resources,
             cls.scan_unmapped_to_files,
             cls.scan_mapped_from_for_files,
+            cls.collect_and_create_license_detections,
             cls.flag_deployed_from_resources_with_missing_license,
             cls.create_local_files_packages,
         )
@@ -286,6 +287,13 @@ def scan_mapped_from_for_files(self):
         scan_files = d2d.get_from_files_for_scanning(self.project.codebaseresources)
         scancode.scan_for_files(self.project, scan_files, progress_logger=self.log)
 
+    def collect_and_create_license_detections(self):
+        """
+        Collect and create unique license detections from resources and
+        package data.
+        """
+        scancode.collect_and_create_license_detections(project=self.project)
+
     def create_local_files_packages(self):
         """Create local-files packages for codebase resources not part of a package."""
         d2d.create_local_files_packages(self.project)

scanpipe/pipelines/docker.py

@@ -42,6 +42,7 @@ def steps(cls):
             cls.flag_ignored_resources,
             cls.scan_for_application_packages,
             cls.scan_for_files,
+            cls.collect_and_create_license_detections,
             cls.analyze_scanned_files,
             cls.flag_not_analyzed_codebase_resources,
         )

scanpipe/pipelines/docker_windows.py

@@ -45,6 +45,7 @@ def steps(cls):
             cls.flag_ignored_resources,
             cls.scan_for_application_packages,
             cls.scan_for_files,
+            cls.collect_and_create_license_detections,
             cls.analyze_scanned_files,
             cls.flag_data_files_with_no_clues,
             cls.flag_not_analyzed_codebase_resources,

scanpipe/pipelines/root_filesystem.py

@@ -45,6 +45,7 @@ def steps(cls):
             cls.scan_for_application_packages,
             cls.match_not_analyzed_to_system_packages,
             cls.scan_for_files,
+            cls.collect_and_create_license_detections,
             cls.analyze_scanned_files,
             cls.flag_not_analyzed_codebase_resources,
         )
@@ -123,6 +124,13 @@ def scan_for_files(self):
         """Scan unknown resources for copyrights, licenses, emails, and urls."""
         scancode.scan_for_files(self.project, progress_logger=self.log)
 
+    def collect_and_create_license_detections(self):
+        """
+        Collect and create unique license detections from resources and
+        package data.
+        """
+        scancode.collect_and_create_license_detections(project=self.project)
+
     def analyze_scanned_files(self):
         """Analyze single file scan results for completeness."""
         flag.analyze_scanned_files(self.project)

scanpipe/pipelines/scan_codebase.py

@@ -45,6 +45,7 @@ def steps(cls):
             cls.flag_ignored_resources,
             cls.scan_for_application_packages,
             cls.scan_for_files,
+            cls.collect_and_create_license_detections,
         )
 
     def copy_inputs_to_codebase_directory(self):
@@ -65,3 +66,10 @@ def scan_for_application_packages(self):
     def scan_for_files(self):
         """Scan unknown resources for copyrights, licenses, emails, and urls."""
         scancode.scan_for_files(self.project, progress_logger=self.log)
+
+    def collect_and_create_license_detections(self):
+        """
+        Collect and create unique license detections from resources and
+        package data.
+        """
+        scancode.collect_and_create_license_detections(project=self.project)

scanpipe/pipes/__init__.py

@@ -246,11 +246,18 @@ def update_or_create_dependency(
     return dependency
 
 
-def update_or_create_license_detection(project, detection_data):
+def update_or_create_license_detection(
+    project, detection_data, resource_path, from_package=False,
+):
     """
     Get, update or create a DiscoveredLicense object then return it.
     Use the `project` and `detection_data` mapping to lookup and creates the
     DiscoveredLicense using its detection identifier as a unique key.
+
+    Additonally if `resource_path` is passed, add the file region where
+    the license was detected to the DiscoveredLicense object, if not present
+    already. `from_package` is True if the license detection was in a
+    `extracted_license_statement` from a package metadata.
     """
     detection_identifier = detection_data["identifier"]
 
@@ -267,13 +274,22 @@ def update_or_create_license_detection(project, detection_data):
             detection_data,
         )
 
+    if resource_path:
+        file_region = scancode.get_file_region(
+            detection_data=detection_data,
+            resource_path=resource_path,
+        )
+        license_detection.update_with_file_region(file_region)
+
+    license_detection.from_package = from_package
     return license_detection
 
 
 def _clean_license_detection_data(detection_data):
     detection_data = detection_data.copy()
-    matches = detection_data.pop("sample_matches")
-    detection_data["matches"] = matches
+    if "sample_matches" in detection_data:
+        matches = detection_data.pop("sample_matches")
+        detection_data["matches"] = matches
     return detection_data
 
 

scanpipe/pipes/scancode.py

@@ -39,6 +39,7 @@
 from extractcode import api as extractcode_api
 from packagedcode import get_package_handler
 from packagedcode import models as packagedcode_models
+from licensedcode.detection import FileRegion
 from scancode import Scanner
 from scancode import api as scancode_api
 from scancode import cli as scancode_cli
@@ -411,6 +412,67 @@ def add_resource_to_package(package_uid, resource, project):
     resource.discovered_packages.add(package)
 
 
+def collect_and_create_license_detections(project):
+    """
+    Create instances of DiscoveredLicense for `project` from the parsed
+    license detections present in the CodebaseResources and
+    DiscoveredPackages of `project`.
+    """
+    logger.info(f"Project {project} collect_license_detections:")
+
+    for resource in project.codebaseresources.has_license_detections():
+        logger.info(f"  Processing: {resource.path} for licenses")
+
+        for detection_data in resource.license_detections:
+            pipes.update_or_create_license_detection(
+                project=project,
+                detection_data=detection_data,
+                resource_path=resource.path,
+            )
+
+    for resource in project.codebaseresources.has_package_data():
+
+        for package_mapping in resource.package_data:
+            package_data = packagedcode_models.PackageData.from_dict(
+                mapping=package_mapping,
+            )
+
+            for detection in package_data.license_detections:
+                pipes.update_or_create_license_detection(
+                    project=project,
+                    detection_data=detection,
+                    resource_path=resource.path,
+                    from_package=True,
+                )
+
+            for detection in package_data.other_license_detections:
+                pipes.update_or_create_license_detection(
+                    project=project,
+                    detection_data=detection,
+                    resource_path=resource.path,
+                    from_package=True,
+                )
+
+
+def get_file_region(detection_data, resource_path):
+    """
+    From a LicenseDetection mapping `detection_data`, create a FileRegion
+    object containing information about where this license was detected
+    exactly in a codebase, with `resource_path`, with start and end lines.
+    """
+    start_line = min(
+        [match['start_line'] for match in detection_data["matches"]]
+    )
+    end_line = max(
+        [match['end_line'] for match in detection_data["matches"]]
+    )
+    return FileRegion(
+        path=resource_path,
+        start_line=start_line,
+        end_line=end_line,
+    )
+
+
 def assemble_packages(project):
     """
     Create instances of DiscoveredPackage and DiscoveredDependency for `project`