diff --git a/CHANGELOG.rst b/CHANGELOG.rst index cc9fe557f3..68094c88a4 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -9,6 +9,7 @@ Unreleased ---------- * feat: expose the ``is_addon`` attribute in the pipelines REST API +* feat: [code-genetics] determine the origin of files from combined origin matches #1929 by @tdruez v38.2.0 (2026-10-22) -------------------- diff --git a/docs/data-models.rst b/docs/data-models.rst index 76c55593b6..b6edbff143 100644 --- a/docs/data-models.rst +++ b/docs/data-models.rst @@ -59,6 +59,15 @@ OriginMatch :undoc-members: :member-order: groupwise +.. _data_models_origin_determination: + +OriginDetermination +------------------- +.. autoclass:: scanpipe.models.OriginDetermination() + :members: + :undoc-members: + :member-order: groupwise + .. _data_models_project_message: ProjectMessage diff --git a/scanpipe/admin.py b/scanpipe/admin.py index 9d17fdc4db..1cd1ea633a 100644 --- a/scanpipe/admin.py +++ b/scanpipe/admin.py @@ -27,6 +27,7 @@ from scanpipe.models import CodebaseResource from scanpipe.models import DiscoveredDependency from scanpipe.models import DiscoveredPackage +from scanpipe.models import OriginDetermination from scanpipe.models import OriginMatch from scanpipe.models import Project @@ -192,6 +193,30 @@ def has_delete_permission(self, request, obj=None): return False +class OriginDeterminationAdmin(ScanPipeBaseAdmin): + list_display = [ + "resource", + "purl", + "status", + "rule", + "project", + ] + search_fields = [ + "uuid", + "purl", + "resource__path", + ] + list_filter = ["project", "status", "rule"] + list_select_related = ["resource", "project"] + ordering = ["project", "resource__path"] + + def has_change_permission(self, request, obj=None): + return False + + def has_delete_permission(self, request, obj=None): + return False + + class ScanCodeIOAdminSite(admin.AdminSite): site_header = "ScanCode.io administration" site_title = "ScanCode.io administration" @@ -203,3 +228,4 @@ class ScanCodeIOAdminSite(admin.AdminSite): admin_site.register(DiscoveredPackage, DiscoveredPackageAdmin) admin_site.register(DiscoveredDependency, DiscoveredDependencyAdmin) admin_site.register(OriginMatch, OriginMatchAdmin) +admin_site.register(OriginDetermination, OriginDeterminationAdmin) diff --git a/scanpipe/migrations/0085_origindetermination.py b/scanpipe/migrations/0085_origindetermination.py new file mode 100644 index 0000000000..c30b173770 --- /dev/null +++ b/scanpipe/migrations/0085_origindetermination.py @@ -0,0 +1,32 @@ +# Generated by Django 6.1.2 on 2026-10-27 07:18 + +import django.db.models.deletion +import scanpipe.models +import uuid +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('scanpipe', '0084_originmatch'), + ] + + operations = [ + migrations.CreateModel( + name='OriginDetermination', + fields=[ + ('uuid', models.UUIDField(db_index=True, default=uuid.uuid4, editable=False, primary_key=True, serialize=False, verbose_name='UUID')), + ('purl', models.CharField(blank=True, help_text='Package URL (PURL) of the origin package. Empty when no open source origin is found, or when the origin needs review and no candidate is preferred.', max_length=2048)), + ('status', models.CharField(choices=[('auto-confirmed', 'Auto Confirmed'), ('needs-review', 'Needs Review'), ('confirmed', 'Confirmed'), ('no-origin', 'No Origin')], help_text='Determination status: confirmed automatically by a rule, needs a review, confirmed by a user, or no open source origin.', max_length=20)), + ('rule', models.CharField(blank=True, help_text='Name of the determination rule that set the status. Empty when the status is set by a user.', max_length=50)), + ('project', models.ForeignKey(editable=False, on_delete=django.db.models.deletion.CASCADE, related_name='%(class)ss', to='scanpipe.project')), + ('resource', models.OneToOneField(editable=False, help_text='The file or directory whose origin is determined.', on_delete=django.db.models.deletion.CASCADE, related_name='origin_determination', to='scanpipe.codebaseresource')), + ], + options={ + 'indexes': [models.Index(fields=['purl'], name='scanpipe_or_purl_e28f4e_idx'), models.Index(fields=['status'], name='scanpipe_or_status_9ab697_idx')], + 'constraints': [models.CheckConstraint(condition=models.Q(models.Q(('purl', ''), ('status', 'no-origin')), ('status', 'needs-review'), models.Q(('status__in', ['auto-confirmed', 'confirmed']), models.Q(('purl', ''), _negated=True)), _connector='OR'), name='scanpipe_origindetermination_purl_consistent_with_status')], + }, + bases=(scanpipe.models.UpdateMixin, models.Model), + ), + ] diff --git a/scanpipe/models.py b/scanpipe/models.py index 0a31cb77f1..8627a5f60f 100644 --- a/scanpipe/models.py +++ b/scanpipe/models.py @@ -706,6 +706,7 @@ def delete_related_objects(self, keep_input=False, keep_webhook=False): self.discovereddependencies, self.discoveredlicenses, self.originmatchs, + self.origindeterminations, self.codebaseresources, self.runs, ] @@ -5068,8 +5069,8 @@ class OriginMatch(UUIDPKModel, ProjectRelatedModel, ExtraDataFieldMixin, AdminUR An origin tool can report several candidates for the same resource, and several tools can report candidates for the same resource. A candidate is not - a conclusion: only the origin selected after review becomes a discovered - package. Origin matches are not included in the output files yet. + a conclusion: only the confirmed origin, automatically or after review, becomes + a discovered package. Origin matches are not included in the output files yet. """ class MatchType(models.TextChoices): @@ -5155,3 +5156,87 @@ class Meta: def __str__(self): return f"{self.purl} ({self.match_type} match by {self.tool})" + + +class OriginDeterminationQuerySet(ProjectRelatedQuerySet): + def automatic(self): + return self.filter(status__in=self.model.AUTOMATIC_STATUSES) + + def confirmed(self): + return self.filter(status__in=self.model.CONFIRMED_STATUSES) + + +class OriginDetermination(UUIDPKModel, ProjectRelatedModel): + """ + The origin determined for a resource, selected among its origin matches by a + determination rule or by a user. + + The origin is stored as a PURL, independently of the origin matches, so a + user decision is kept when the origin tools are run again. + """ + + class Status(models.TextChoices): + AUTO_CONFIRMED = "auto-confirmed" + NEEDS_REVIEW = "needs-review" + CONFIRMED = "confirmed" + NO_ORIGIN = "no-origin" + + CONFIRMED_STATUSES = [Status.AUTO_CONFIRMED, Status.CONFIRMED] + # Set by the determination rules, and replaced each time the rules run. + AUTOMATIC_STATUSES = [Status.AUTO_CONFIRMED, Status.NEEDS_REVIEW] + + resource = models.OneToOneField( + CodebaseResource, + related_name="origin_determination", + on_delete=models.CASCADE, + editable=False, + help_text=_("The file or directory whose origin is determined."), + ) + purl = models.CharField( + max_length=2048, + blank=True, + help_text=_( + "Package URL (PURL) of the origin package. Empty when no open source " + "origin is found, or when the origin needs review and no candidate is " + "preferred." + ), + ) + status = models.CharField( + max_length=20, + choices=Status.choices, + help_text=_( + "Determination status: confirmed automatically by a rule, needs a " + "review, confirmed by a user, or no open source origin." + ), + ) + rule = models.CharField( + max_length=50, + blank=True, + help_text=_( + "Name of the determination rule that set the status. Empty when the " + "status is set by a user." + ), + ) + + objects = OriginDeterminationQuerySet.as_manager() + + class Meta: + indexes = [ + models.Index(fields=["purl"]), + models.Index(fields=["status"]), + ] + constraints = [ + models.CheckConstraint( + condition=( + Q(status="no-origin", purl="") + | Q(status="needs-review") + | (Q(status__in=["auto-confirmed", "confirmed"]) & ~Q(purl="")) + ), + name="%(app_label)s_%(class)s_purl_consistent_with_status", + ), + ] + + def __str__(self): + if self.purl: + return f"{self.purl} ({self.status})" + return self.status diff --git a/scanpipe/pipelines/find_origins.py b/scanpipe/pipelines/find_origins.py index 815e2b34e1..9e57b0fc74 100644 --- a/scanpipe/pipelines/find_origins.py +++ b/scanpipe/pipelines/find_origins.py @@ -24,6 +24,7 @@ from scanpipe.pipelines import Pipeline from scanpipe.pipes import matchcode +from scanpipe.pipes import origin from scanpipe.pipes import scanoss @@ -32,10 +33,11 @@ class FindOrigins(Pipeline): Find the origin of the codebase files using origin detection tools. The candidate origin packages reported by each configured tool are stored as - origin matches, to be combined and reviewed. A tool that is not configured is - skipped. + origin matches, then combined to determine the origin of each matched file and + directory. A tool that is not configured is skipped. Supported tools: + - MatchCode.io: the codebase scan, including file and directory fingerprints, is sent to the configured MatchCode.io server. - SCANOSS: the codebase files are fingerprinted locally, then their paths, @@ -51,6 +53,7 @@ def steps(cls): return ( cls.find_matchcode_origin_matches, cls.find_scanoss_origin_matches, + cls.determine_origins, ) @classmethod @@ -97,3 +100,17 @@ def find_scanoss_origin_matches(self): self.project, results_location ) self.log(f"{origin_match_count:,d} SCANOSS origin matches processed.") + + def determine_origins(self): + """ + Determine the origin of each matched file and directory from the origin + matches of all the tools. An origin is confirmed automatically when two + tools report the same package, with at least one exact match. Otherwise, + the origin needs a review. The origins confirmed by a user are kept. + The files and directories of a confirmed origin are assigned to the origin + package, created from the tools data when the codebase scan did not + detect it. + """ + determination_count, package_count = origin.determine_origins(self.project) + self.log(f"{determination_count:,d} origin determinations created.") + self.log(f"{package_count:,d} origin packages assigned.") diff --git a/scanpipe/pipes/origin.py b/scanpipe/pipes/origin.py new file mode 100644 index 0000000000..9517a1e62c --- /dev/null +++ b/scanpipe/pipes/origin.py @@ -0,0 +1,557 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# http://nexb.com and https://github.com/aboutcode-org/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/aboutcode-org/scancode.io for support and download. + +from contextlib import suppress +from dataclasses import dataclass +from itertools import batched +from itertools import groupby +from operator import attrgetter +from pathlib import PurePosixPath + +from django.db import transaction +from django.db.models import Q + +from packageurl import PackageURL + +from scanpipe.models import CodebaseResource +from scanpipe.models import DiscoveredPackage +from scanpipe.models import OriginDetermination +from scanpipe.models import OriginMatch +from scanpipe.pipes.resolve import convert_spdx_expression + +CONFLICTING_CANDIDATES_RULE = "conflicting-candidates" +CONFLICTING_VERSIONS_RULE = "conflicting-versions" +NO_EXACT_MATCH_RULE = "no-exact-match" +SINGLE_TOOL_RULE = "single-tool" +CONFIRMED_BY_TWO_TOOLS_RULE = "confirmed-by-two-tools" + +# Identifies the packages created from an origin determination, as opposed to +# the packages detected by the codebase scan. +ORIGIN_DATASOURCE_ID = "origin_determination" + +# The paths of the resources assigned to a package detected by the codebase scan, +# kept in its extra data to unassign them on a re-run without touching the +# resources assigned by the scan. +ORIGIN_RESOURCE_PATHS_KEY = "origin_resource_paths" + +# The SCANOSS license sources of the whole component. The other sources, such as +# "file_header", are the licenses of the matched file, already detected by the +# codebase scan. +SCANOSS_COMPONENT_LICENSE_SOURCES = ["component_declared", "license_file"] + + +def get_package_purl(purl, version=None): + """ + Return the ``purl`` without qualifiers and subpath, with the ``version`` when + provided. + """ + package_url = PackageURL.from_string(purl) + return PackageURL( + type=package_url.type, + namespace=package_url.namespace, + name=package_url.name, + version=version or package_url.version, + ).to_string() + + +def get_versioned_purls(purls, version): + """ + Return the valid ``purls`` with the ``version``, such as the versionless + equivalent PURLs reported by SCANOSS. + """ + versioned_purls = set() + for purl in purls: + with suppress(ValueError): + versioned_purls.add(get_package_purl(purl, version)) + return versioned_purls + + +def get_equivalent_purls(origin_match): + """ + Return the PURLs identifying the ``origin_match`` package: its own PURL and + the equivalent PURLs reported by the tool, such as the SCANOSS ``purls``. + """ + version = PackageURL.from_string(origin_match.purl).version + tool_purls = origin_match.extra_data.get("purls", []) + return get_versioned_purls([origin_match.purl, *tool_purls], version) + + +def get_versionless_purl(purl): + """Return the ``purl`` without version, qualifiers and subpath.""" + package_url = PackageURL.from_string(purl) + return PackageURL( + type=package_url.type, + namespace=package_url.namespace, + name=package_url.name, + ).to_string() + + +@dataclass +class Origin: + """ + A candidate origin package of a resource, combining the origin matches that + identify the same package, possibly with different but equivalent PURLs. + """ + + origin_matches: list + # Kept up to date on merge, as it is compared for each resource. + equivalent_purls: set + + @classmethod + def from_origin_match(cls, origin_match): + return cls([origin_match], get_equivalent_purls(origin_match)) + + @property + def tools(self): + return {origin_match.tool for origin_match in self.origin_matches} + + @property + def has_exact_match(self): + return any( + origin_match.match_type in OriginMatch.EXACT_MATCH_TYPES + for origin_match in self.origin_matches + ) + + @property + def purl(self): + """ + Return the first reported PURL that is equivalent for all the origin + matches, such as the "pypi" PURL when a tool reports the "github" one with + "pypi" as equivalent. + """ + reported_purls = [ + get_package_purl(origin_match.purl) for origin_match in self.origin_matches + ] + common_purls = set.intersection(*map(get_equivalent_purls, self.origin_matches)) + common_reported_purls = [ + reported_purl + for reported_purl in reported_purls + if reported_purl in common_purls + ] + return (common_reported_purls or reported_purls)[0] + + @property + def versionless_purls(self): + return {get_versionless_purl(purl) for purl in self.equivalent_purls} + + def is_same_package(self, other_origin): + return bool(self.equivalent_purls & other_origin.equivalent_purls) + + def merge(self, other_origin): + self.origin_matches.extend(other_origin.origin_matches) + self.equivalent_purls |= other_origin.equivalent_purls + + +def get_same_package_origins(origin, origins): + """Return the ``origins`` of the same package as ``origin``.""" + return [ + other_origin for other_origin in origins if origin.is_same_package(other_origin) + ] + + +def group_origin_matches(origin_matches): + """Return the list of distinct origins reported by the ``origin_matches``.""" + origins = [] + + for origin_match in origin_matches: + origin = Origin.from_origin_match(origin_match) + for same_package_origin in get_same_package_origins(origin, origins): + origin.merge(same_package_origin) + origins.remove(same_package_origin) + origins.append(origin) + + return origins + + +def iter_resource_origins(origin_matches): + """Yield (resource, origins) for the ``origin_matches`` ordered by resource.""" + get_resource = attrgetter("resource") + for resource, resource_origin_matches in groupby(origin_matches, key=get_resource): + yield resource, group_origin_matches(resource_origin_matches) + + +def get_ancestor_paths(path): + """Return the ancestor directory paths of ``path``, closest first.""" + return [str(parent) for parent in PurePosixPath(path).parents[:-1]] + + +def remove_ancestor_directory_origins(origins_by_directory_path): + """ + Remove from each directory the origins also found in one of its + subdirectories, as the deepest directory is the most precise match. + """ + for path, origins in origins_by_directory_path.items(): + for ancestor_path in get_ancestor_paths(path): + ancestor_origins = origins_by_directory_path.get(ancestor_path) + if not ancestor_origins: + continue + + origins_by_directory_path[ancestor_path] = [ + ancestor_origin + for ancestor_origin in ancestor_origins + if not get_same_package_origins(ancestor_origin, origins) + ] + + +def get_closest_directory_origins(path, origins_by_directory_path): + """Return the origins of the closest ancestor directory of ``path``.""" + for ancestor_path in get_ancestor_paths(path): + if ancestor_origins := origins_by_directory_path.get(ancestor_path): + return ancestor_origins + return [] + + +def is_directory_origin(file_origins, directory_origins): + """Return True when all the ``file_origins`` are ``directory_origins``.""" + return all( + get_same_package_origins(file_origin, directory_origins) + for file_origin in file_origins + ) + + +def is_same_package_versions(origins): + """Return True when the ``origins`` are versions of the same package.""" + return bool(set.intersection(*(origin.versionless_purls for origin in origins))) + + +def get_rule(origins): + """Return the determination rule of the candidate ``origins`` of a resource.""" + if len(origins) > 1: + if is_same_package_versions(origins): + return CONFLICTING_VERSIONS_RULE + return CONFLICTING_CANDIDATES_RULE + + origin = origins[0] + if not origin.has_exact_match: + return NO_EXACT_MATCH_RULE + + # A single tool can miss the other candidates: MatchCode.io stops at the + # first package matched, and SCANOSS reports one component per file. + if len(origin.tools) < 2: + return SINGLE_TOOL_RULE + + return CONFIRMED_BY_TWO_TOOLS_RULE + + +def make_origin_determination(project, resource, origins): + """Return an unsaved OriginDetermination for the ``resource`` ``origins``.""" + rule = get_rule(origins) + status = OriginDetermination.Status.NEEDS_REVIEW + if rule == CONFIRMED_BY_TWO_TOOLS_RULE: + status = OriginDetermination.Status.AUTO_CONFIRMED + + has_single_candidate = len(origins) == 1 + return OriginDetermination( + project=project, + resource=resource, + purl=origins[0].purl if has_single_candidate else "", + status=status, + rule=rule, + ) + + +def iter_origin_determinations(project, origin_matches): + """ + Yield unsaved OriginDetermination instances from the ``origin_matches``. + The files of a matched directory are determined with the directory, unless + they are matched to another package. + """ + directory_origin_matches = origin_matches.filter( + resource__type=CodebaseResource.Type.DIRECTORY + ) + directories_by_path = {} + origins_by_directory_path = {} + for resource, origins in iter_resource_origins(directory_origin_matches): + directories_by_path[resource.path] = resource + origins_by_directory_path[resource.path] = origins + + remove_ancestor_directory_origins(origins_by_directory_path) + + file_origin_matches = origin_matches.exclude( + resource__type=CodebaseResource.Type.DIRECTORY + ) + file_origin_matches = file_origin_matches.iterator(chunk_size=2000) + for resource, file_origins in iter_resource_origins(file_origin_matches): + directory_origins = get_closest_directory_origins( + resource.path, origins_by_directory_path + ) + if not is_directory_origin(file_origins, directory_origins): + yield make_origin_determination(project, resource, file_origins) + continue + + # The file matches count for the directory: an exact file match by one tool + # confirms the directory match by another tool. + for file_origin in file_origins: + get_same_package_origins(file_origin, directory_origins)[0].merge( + file_origin + ) + + for path, origins in origins_by_directory_path.items(): + if origins: + yield make_origin_determination(project, directories_by_path[path], origins) + + +def get_first_extra_data_value(origin_matches, key): + """Return the first non-empty ``key`` value of the ``origin_matches`` extra data.""" + for origin_match in origin_matches: + if value := origin_match.extra_data.get(key): + return value + return "" + + +def get_license_names(origin_matches): + """ + Return the distinct component license names of the ``origin_matches`` extra + data. + """ + license_names = [ + license_data["name"] + for origin_match in origin_matches + for license_data in origin_match.extra_data.get("licenses", []) + if license_data.get("source") in SCANOSS_COMPONENT_LICENSE_SOURCES + ] + return list(dict.fromkeys(license_names)) + + +def get_origin_license_data(origin_matches): + """ + Return the declared license fields of the ``origin_matches`` package: the + MatchCode.io license expression, or else the component license names reported + by SCANOSS, joined with AND. + """ + license_expression = get_first_extra_data_value( + origin_matches, "declared_license_expression" + ) + if license_expression: + return {"declared_license_expression": license_expression} + + license_names = get_license_names(origin_matches) + if not license_names: + return {} + + license_expression_spdx = " AND ".join(license_names) + return { + "declared_license_expression": convert_spdx_expression(license_expression_spdx), + "declared_license_expression_spdx": license_expression_spdx, + } + + +def get_origin_package_data(origin_matches): + """Return the package data reported by the origin tools ``origin_matches``.""" + return { + "download_url": get_first_extra_data_value(origin_matches, "download_url"), + "homepage_url": get_first_extra_data_value(origin_matches, "url"), + "release_date": get_first_extra_data_value(origin_matches, "release_date"), + **get_origin_license_data(origin_matches), + } + + +def get_purl_origin_matches(project, purl, resources): + """Return the ``resources`` origin matches equivalent to the ``purl``.""" + origin_matches = project.originmatchs.filter(resource__in=resources).only( + "purl", "extra_data" + ) + return [ + origin_match + for origin_match in origin_matches + if purl in get_equivalent_purls(origin_match) + ] + + +def get_origin_packages(project): + """Return the ``project`` packages created from an origin determination.""" + return project.discoveredpackages.filter(datasource_ids=[ORIGIN_DATASOURCE_ID]) + + +def get_or_create_origin_package(project, purl, resources): + """ + Return the ``project`` package of ``purl``, such as a package detected by the + codebase scan, or else create it from the ``resources`` origin matches. + """ + purl_data = PackageURL.from_string(purl).to_dict() + lookup_purl_data = DiscoveredPackage.extract_purl_data(purl_data) + if package := project.discoveredpackages.filter(**lookup_purl_data).first(): + return package + + origin_matches = get_purl_origin_matches(project, purl, resources) + package_data = { + **purl_data, + **get_origin_package_data(origin_matches), + "datasource_ids": [ORIGIN_DATASOURCE_ID], + } + return DiscoveredPackage.create_from_data(project, package_data) + + +def assign_origin_package_resources(package, resources): + """ + Assign the ``resources`` to the origin ``package``. On a package detected by + the codebase scan, the paths of the newly assigned resources are kept in its + extra data. + """ + is_scan_package = ORIGIN_DATASOURCE_ID not in package.datasource_ids + if not is_scan_package: + package.add_resources(resources) + return + + new_resources = list(resources.exclude(discovered_packages=package)) + if not new_resources: + return + + package.add_resources(new_resources) + new_resource_paths = [resource.path for resource in new_resources] + package.update_extra_data({ORIGIN_RESOURCE_PATHS_KEY: new_resource_paths}) + + +def unassign_scan_package_resources(project): + """ + Remove the resources assigned by a previous run to the ``project`` packages + detected by the codebase scan. + """ + scan_packages = project.discoveredpackages.filter( + extra_data__has_key=ORIGIN_RESOURCE_PATHS_KEY + ) + for package in scan_packages: + origin_resource_paths = package.extra_data.pop(ORIGIN_RESOURCE_PATHS_KEY) + package.codebase_resources.through.objects.filter( + discoveredpackage=package, + codebaseresource__path__in=origin_resource_paths, + ).delete() + package.update(extra_data=package.extra_data) + + +def get_descendants_filter(directory_paths): + """Return a filter of the resources under any of the ``directory_paths``.""" + # Matches no resource when there is no directory. + descendants_filter = Q(pk__in=[]) + for directory_path in directory_paths: + descendants_filter |= Q(path__startswith=f"{directory_path}/") + return descendants_filter + + +def get_determined_directory_paths(project): + """Return the paths of the ``project`` directories that have a determination.""" + directory_determinations = project.origindeterminations.filter( + resource__type=CodebaseResource.Type.DIRECTORY + ) + return list(directory_determinations.values_list("resource__path", flat=True)) + + +def get_determined_resources(project, determinations, determined_directory_paths): + """ + Return the resources of the ``determinations``, including the resources of + the determined directories that have no determination of their own and are + not under a determined subdirectory of ``determined_directory_paths``. + """ + resource_ids = [determination.resource_id for determination in determinations] + resources_filter = Q(pk__in=resource_ids) + + for determination in determinations: + if not determination.resource.is_dir: + continue + + directory_path = determination.resource.path + subdirectory_paths = [ + path + for path in determined_directory_paths + if path.startswith(f"{directory_path}/") + ] + resources_filter |= ( + get_descendants_filter([directory_path]) + & ~get_descendants_filter(subdirectory_paths) + & Q(origin_determination__isnull=True) + ) + + return project.codebaseresources.filter(resources_filter) + + +def assign_origin_packages(project): + """ + Assign the resources of the confirmed ``project`` determinations to their + origin package, created from the origin matches data when missing. + Return the number of origin packages. + """ + confirmed_determinations = ( + project.origindeterminations.confirmed() + .select_related("resource") + .only("purl", "resource__path", "resource__type") + .order_by("purl") + .iterator(chunk_size=2000) + ) + determined_directory_paths = get_determined_directory_paths(project) + get_purl = attrgetter("purl") + package_count = 0 + + for purl, purl_determinations in groupby(confirmed_determinations, key=get_purl): + resources = get_determined_resources( + project, list(purl_determinations), determined_directory_paths + ) + package = get_or_create_origin_package(project, purl, resources) + assign_origin_package_resources(package, resources) + package_count += 1 + + return package_count + + +def determine_origins(project, batch_size=2000): + """ + Determine the origin of the ``project`` resources from their origin matches, + and assign the resources of a confirmed origin to the origin package. + The previous automatic determinations, origin packages and resources assigned + to packages detected by the codebase scan are replaced, while the + determinations made by a user are kept unchanged. A directory determined + by a user covers its content: the origin matches under it are not used. + Return the number of created determinations and the number of origin packages. + """ + determination_count = 0 + + with transaction.atomic(): + get_origin_packages(project).delete() + unassign_scan_package_resources(project) + project.origindeterminations.automatic().delete() + + # Only the user determinations are left. + user_directory_paths = get_determined_directory_paths(project) + user_covered_resources = project.codebaseresources.filter( + get_descendants_filter(user_directory_paths) + ) + origin_matches = ( + project.originmatchs.filter(resource__origin_determination__isnull=True) + .exclude(resource__in=user_covered_resources) + .select_related("resource") + .only( + "purl", + "match_type", + "tool", + "extra_data", + "resource__path", + "resource__type", + ) + .order_by("resource__path", "tool") + ) + determinations = iter_origin_determinations(project, origin_matches) + for determination_batch in batched(determinations, batch_size): + OriginDetermination.objects.bulk_create(determination_batch) + determination_count += len(determination_batch) + + package_count = assign_origin_packages(project) + + return determination_count, package_count diff --git a/scanpipe/tests/pipes/test_origin.py b/scanpipe/tests/pipes/test_origin.py new file mode 100644 index 0000000000..26b7aec208 --- /dev/null +++ b/scanpipe/tests/pipes/test_origin.py @@ -0,0 +1,884 @@ +# 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 json +import shutil +from pathlib import Path + +from django.test import TestCase + +from scanpipe.models import OriginDetermination +from scanpipe.models import OriginMatch +from scanpipe.pipes import collect_and_create_codebase_resources +from scanpipe.pipes import matchcode +from scanpipe.pipes import origin +from scanpipe.pipes import scanoss +from scanpipe.tests import make_package +from scanpipe.tests import make_project +from scanpipe.tests import make_resource_directory +from scanpipe.tests import make_resource_file + +SCANOSS_SIX_PURLS = ["pkg:github/benjaminp/six", "pkg:pypi/six", "pkg:conda/six"] + + +def make_unsaved_origin_match(purl, tool="matchcode", match_type="file", purls=None): + extra_data = {"purls": purls} if purls else {} + return OriginMatch( + purl=purl, tool=tool, match_type=match_type, extra_data=extra_data + ) + + +def make_origin(purl, tool="matchcode", match_type="file", purls=None): + origin_match = make_unsaved_origin_match(purl, tool, match_type, purls) + return origin.Origin.from_origin_match(origin_match) + + +def make_scanoss_license_match(*license_names_and_sources): + licenses = [ + {"name": license_name, "source": source} + for license_name, source in license_names_and_sources + ] + return OriginMatch(tool="scanoss", extra_data={"licenses": licenses}) + + +def make_scanoss_six_origin(): + return make_origin( + "pkg:github/benjaminp/six@1.17.0", tool="scanoss", purls=SCANOSS_SIX_PURLS + ) + + +class ScanPipeOriginPipesTest(TestCase): + data = Path(__file__).parent.parent / "data" / "code-genetics" + + def setUp(self): + self.project1 = make_project() + + def make_origin_match( + self, resource, purl, tool="matchcode", match_type="file", extra_data=None + ): + return OriginMatch.objects.create( + project=self.project1, + resource=resource, + purl=purl, + match_type=match_type, + tool=tool, + extra_data=extra_data or {}, + ) + + def load_data_set_case(self, case_name, tools=None): + """ + Return a project with the codebase and the results of the ``tools`` of a + data set case, all the tools by default. + """ + tools = tools or [matchcode.TOOL_NAME, scanoss.TOOL_NAME] + case_path = self.data / case_name + project = make_project() + shutil.copytree( + case_path / "codebase", project.codebase_path, dirs_exist_ok=True + ) + collect_and_create_codebase_resources(project) + + if matchcode.TOOL_NAME in tools: + results_location = case_path / "matchcode-results.json" + match_results = json.loads(results_location.read_text()) + matchcode.load_matchcode_results(project, match_results) + + if scanoss.TOOL_NAME in tools: + results_location = case_path / "scanoss-results.json" + scanoss.load_scanoss_results(project, results_location) + + return project + + def get_expected_purls_by_path(self, case_name): + """Return the expected origin PURL of each file and package directory.""" + expected_location = self.data / case_name / "expected.json" + expected_data = json.loads(expected_location.read_text()) + expected_purls_by_path = { + package["path"]: package["purl"] + for package in expected_data.get("packages", []) + } + for resource in expected_data["resources"]: + if resource["origin"]: + expected_purls_by_path[resource["path"]] = resource["origin"]["purl"] + return expected_purls_by_path + + def test_scanpipe_pipes_origin_get_package_purl(self): + purl = "pkg:pypi/six@1.17.0?file_name=six-1.17.0.tar.gz#six.py" + self.assertEqual("pkg:pypi/six@1.17.0", origin.get_package_purl(purl, "1.17.0")) + self.assertEqual( + "pkg:npm/ms@2.1.3", origin.get_package_purl("pkg:npm/ms", "2.1.3") + ) + self.assertEqual("pkg:pypi/six@1.17.0", origin.get_package_purl(purl)) + + def test_scanpipe_pipes_origin_get_versioned_purls(self): + purls = ["pkg:npm/ms", "pkg:github/vercel/ms", "not-a-purl"] + expected = {"pkg:npm/ms@2.1.3", "pkg:github/vercel/ms@2.1.3"} + self.assertEqual(expected, origin.get_versioned_purls(purls, "2.1.3")) + self.assertEqual(set(), origin.get_versioned_purls([], "2.1.3")) + + def test_scanpipe_pipes_origin_get_equivalent_purls(self): + origin_match = make_unsaved_origin_match( + "pkg:github/benjaminp/six@1.17.0", tool="scanoss", purls=SCANOSS_SIX_PURLS + ) + expected = { + "pkg:github/benjaminp/six@1.17.0", + "pkg:pypi/six@1.17.0", + "pkg:conda/six@1.17.0", + } + self.assertEqual(expected, origin.get_equivalent_purls(origin_match)) + + origin_match = make_unsaved_origin_match("pkg:pypi/six@1.17.0?file_name=six.py") + expected = {"pkg:pypi/six@1.17.0"} + self.assertEqual(expected, origin.get_equivalent_purls(origin_match)) + + def test_scanpipe_pipes_origin_get_versionless_purl(self): + purl = "pkg:pypi/six@1.17.0?file_name=six-1.17.0.tar.gz#six.py" + self.assertEqual("pkg:pypi/six", origin.get_versionless_purl(purl)) + + def test_scanpipe_pipes_origin_origin_from_origin_match(self): + origin_match = make_unsaved_origin_match( + "pkg:github/benjaminp/six@1.17.0", tool="scanoss", purls=SCANOSS_SIX_PURLS + ) + six_origin = origin.Origin.from_origin_match(origin_match) + + self.assertEqual([origin_match], six_origin.origin_matches) + expected_purls = { + "pkg:github/benjaminp/six@1.17.0", + "pkg:pypi/six@1.17.0", + "pkg:conda/six@1.17.0", + } + self.assertEqual(expected_purls, six_origin.equivalent_purls) + self.assertEqual({"scanoss"}, six_origin.tools) + self.assertTrue(six_origin.has_exact_match) + + snippet_origin = make_origin("pkg:npm/lodash@4.17.21", match_type="snippet") + self.assertFalse(snippet_origin.has_exact_match) + + def test_scanpipe_pipes_origin_origin_purl(self): + scanoss_origin = make_scanoss_six_origin() + self.assertEqual("pkg:github/benjaminp/six@1.17.0", scanoss_origin.purl) + + scanoss_origin.merge(make_origin("pkg:pypi/six@1.17.0")) + self.assertEqual("pkg:pypi/six@1.17.0", scanoss_origin.purl) + + def test_scanpipe_pipes_origin_origin_versionless_purls(self): + expected = {"pkg:github/benjaminp/six", "pkg:pypi/six", "pkg:conda/six"} + self.assertEqual(expected, make_scanoss_six_origin().versionless_purls) + + def test_scanpipe_pipes_origin_origin_is_same_package(self): + scanoss_origin = make_scanoss_six_origin() + self.assertTrue( + scanoss_origin.is_same_package(make_origin("pkg:pypi/six@1.17.0")) + ) + self.assertFalse( + scanoss_origin.is_same_package(make_origin("pkg:pypi/six@1.16.0")) + ) + + def test_scanpipe_pipes_origin_origin_merge(self): + matchcode_origin = make_origin("pkg:npm/ms@2.1.3", match_type="directory") + scanoss_origin = make_origin( + "pkg:github/vercel/ms@2.1.3", + tool="scanoss", + purls=["pkg:github/vercel/ms", "pkg:npm/ms"], + ) + matchcode_origin.merge(scanoss_origin) + + self.assertEqual(2, len(matchcode_origin.origin_matches)) + expected_equivalent_purls = {"pkg:npm/ms@2.1.3", "pkg:github/vercel/ms@2.1.3"} + self.assertEqual(expected_equivalent_purls, matchcode_origin.equivalent_purls) + self.assertEqual("pkg:npm/ms@2.1.3", matchcode_origin.purl) + self.assertEqual({"matchcode", "scanoss"}, matchcode_origin.tools) + self.assertTrue(matchcode_origin.has_exact_match) + + def test_scanpipe_pipes_origin_get_same_package_origins(self): + six_origin = make_origin("pkg:pypi/six@1.17.0") + scanoss_origin = make_scanoss_six_origin() + urllib3_origin = make_origin("pkg:pypi/urllib3@1.26.20") + + same_package_origins = origin.get_same_package_origins( + six_origin, [scanoss_origin, urllib3_origin] + ) + self.assertEqual([scanoss_origin], same_package_origins) + self.assertEqual([], origin.get_same_package_origins(six_origin, [])) + + def test_scanpipe_pipes_origin_group_origin_matches(self): + origin_matches = [ + make_unsaved_origin_match("pkg:pypi/six@1.17.0"), + make_unsaved_origin_match("pkg:pypi/urllib3@1.26.20"), + make_unsaved_origin_match( + "pkg:github/benjaminp/six@1.17.0", + tool="scanoss", + purls=SCANOSS_SIX_PURLS, + ), + ] + + origins = origin.group_origin_matches(origin_matches) + origin_purls = [grouped_origin.purl for grouped_origin in origins] + self.assertEqual( + ["pkg:pypi/urllib3@1.26.20", "pkg:pypi/six@1.17.0"], origin_purls + ) + self.assertEqual({"matchcode", "scanoss"}, origins[1].tools) + + def test_scanpipe_pipes_origin_iter_resource_origins(self): + six_resource = make_resource_file(self.project1, path="six.py") + ini_resource = make_resource_file(self.project1, path="ini.c") + self.make_origin_match(six_resource, "pkg:pypi/six@1.17.0") + self.make_origin_match(six_resource, "pkg:pypi/six@1.16.0") + self.make_origin_match(ini_resource, "pkg:github/benhoyt/inih@r62") + + origin_matches = self.project1.originmatchs.order_by("resource__path") + resource_origins = [ + (resource.path, len(origins)) + for resource, origins in origin.iter_resource_origins(origin_matches) + ] + self.assertEqual([("ini.c", 1), ("six.py", 2)], resource_origins) + + def test_scanpipe_pipes_origin_get_ancestor_paths(self): + expected = ["vendor/ms", "vendor"] + self.assertEqual(expected, origin.get_ancestor_paths("vendor/ms/index.js")) + self.assertEqual([], origin.get_ancestor_paths("six.py")) + + def test_scanpipe_pipes_origin_remove_ancestor_directory_origins(self): + vendor_ms_origin = make_origin("pkg:npm/ms@2.1.3", match_type="directory") + vendor_lodash_origin = make_origin( + "pkg:npm/lodash@4.17.21", match_type="directory" + ) + origins_by_directory_path = { + "vendor": [vendor_ms_origin, vendor_lodash_origin], + "vendor/ms": [make_origin("pkg:npm/ms@2.1.3", match_type="directory")], + } + + origin.remove_ancestor_directory_origins(origins_by_directory_path) + self.assertEqual([vendor_lodash_origin], origins_by_directory_path["vendor"]) + self.assertEqual(1, len(origins_by_directory_path["vendor/ms"])) + + def test_scanpipe_pipes_origin_get_closest_directory_origins(self): + vendor_origin = make_origin("pkg:npm/ms@2.1.3", match_type="directory") + origins_by_directory_path = {"vendor": [vendor_origin], "vendor/ms": []} + + closest_directory_origins = origin.get_closest_directory_origins( + "vendor/ms/index.js", origins_by_directory_path + ) + self.assertEqual([vendor_origin], closest_directory_origins) + + closest_directory_origins = origin.get_closest_directory_origins( + "src/index.js", origins_by_directory_path + ) + self.assertEqual([], closest_directory_origins) + + def test_scanpipe_pipes_origin_is_directory_origin(self): + directory_origins = [make_origin("pkg:npm/ms@2.1.3", match_type="directory")] + ms_origin = make_origin("pkg:npm/ms@2.1.3") + lodash_origin = make_origin("pkg:npm/lodash@4.17.21") + + self.assertTrue(origin.is_directory_origin([ms_origin], directory_origins)) + self.assertFalse( + origin.is_directory_origin([ms_origin, lodash_origin], directory_origins) + ) + self.assertFalse(origin.is_directory_origin([ms_origin], [])) + + def test_scanpipe_pipes_origin_is_same_package_versions(self): + six_17_origin = make_scanoss_six_origin() + six_16_origin = make_origin("pkg:pypi/six@1.16.0") + urllib3_origin = make_origin("pkg:pypi/urllib3@1.26.20") + + self.assertTrue(origin.is_same_package_versions([six_17_origin, six_16_origin])) + self.assertFalse( + origin.is_same_package_versions([six_16_origin, urllib3_origin]) + ) + + def test_scanpipe_pipes_origin_get_rule(self): + six_origin = make_origin("pkg:pypi/six@1.17.0") + six_16_origin = make_origin("pkg:pypi/six@1.16.0") + urllib3_origin = make_origin("pkg:pypi/urllib3@1.26.20") + snippet_origin = make_origin("pkg:npm/lodash@4.17.21", match_type="snippet") + + rule = origin.get_rule([six_origin, urllib3_origin]) + self.assertEqual(origin.CONFLICTING_CANDIDATES_RULE, rule) + rule = origin.get_rule([six_origin, six_16_origin]) + self.assertEqual(origin.CONFLICTING_VERSIONS_RULE, rule) + rule = origin.get_rule([snippet_origin]) + self.assertEqual(origin.NO_EXACT_MATCH_RULE, rule) + self.assertEqual(origin.SINGLE_TOOL_RULE, origin.get_rule([six_origin])) + + six_origin.merge(make_scanoss_six_origin()) + rule = origin.get_rule([six_origin]) + self.assertEqual(origin.CONFIRMED_BY_TWO_TOOLS_RULE, rule) + + def test_scanpipe_pipes_origin_make_origin_determination(self): + resource = make_resource_file(self.project1, path="six.py") + six_origin = make_origin("pkg:pypi/six@1.17.0") + urllib3_origin = make_origin("pkg:pypi/urllib3@1.26.20") + + determination = origin.make_origin_determination( + self.project1, resource, [six_origin] + ) + self.assertEqual(resource, determination.resource) + self.assertEqual("pkg:pypi/six@1.17.0", determination.purl) + self.assertEqual("needs-review", determination.status) + self.assertEqual(origin.SINGLE_TOOL_RULE, determination.rule) + + determination = origin.make_origin_determination( + self.project1, resource, [six_origin, urllib3_origin] + ) + self.assertEqual("", determination.purl) + self.assertEqual("needs-review", determination.status) + self.assertEqual(origin.CONFLICTING_CANDIDATES_RULE, determination.rule) + + six_origin.merge(make_scanoss_six_origin()) + determination = origin.make_origin_determination( + self.project1, resource, [six_origin] + ) + self.assertEqual("pkg:pypi/six@1.17.0", determination.purl) + self.assertEqual("auto-confirmed", determination.status) + + def test_scanpipe_pipes_origin_iter_origin_determinations(self): + vendor = make_resource_directory(self.project1, path="vendor") + vendor_ms = make_resource_directory(self.project1, path="vendor/ms") + index_js = make_resource_file(self.project1, path="vendor/ms/index.js") + debounce_js = make_resource_file(self.project1, path="vendor/ms/debounce.js") + self.make_origin_match(vendor, "pkg:npm/ms@2.1.3", match_type="directory") + self.make_origin_match(vendor_ms, "pkg:npm/ms@2.1.3", match_type="directory") + self.make_origin_match(index_js, "pkg:npm/ms@2.1.3", tool="scanoss") + self.make_origin_match(debounce_js, "pkg:npm/lodash@4.17.21") + + origin_matches = self.project1.originmatchs.order_by("resource__path") + determinations = origin.iter_origin_determinations( + self.project1, origin_matches + ) + values = [ + (determination.resource.path, determination.purl, determination.rule) + for determination in determinations + ] + expected = [ + ( + "vendor/ms/debounce.js", + "pkg:npm/lodash@4.17.21", + origin.SINGLE_TOOL_RULE, + ), + ("vendor/ms", "pkg:npm/ms@2.1.3", origin.CONFIRMED_BY_TWO_TOOLS_RULE), + ] + self.assertEqual(expected, values) + + def test_scanpipe_pipes_origin_determine_origins(self): + six_resource = make_resource_file(self.project1, path="six.py") + ini_resource = make_resource_file(self.project1, path="ini.c") + self.make_origin_match(six_resource, "pkg:pypi/six@1.17.0") + self.make_origin_match(ini_resource, "pkg:github/benhoyt/inih@498f34b") + user_determination = OriginDetermination.objects.create( + project=self.project1, + resource=ini_resource, + purl="pkg:github/benhoyt/inih@r62", + status=OriginDetermination.Status.CONFIRMED, + ) + + self.assertEqual((1, 1), origin.determine_origins(self.project1)) + self.assertEqual((1, 1), origin.determine_origins(self.project1)) + + determinations = self.project1.origindeterminations.order_by("resource__path") + values = determinations.values_list("resource__path", "purl", "status", "rule") + expected = [ + ("ini.c", "pkg:github/benhoyt/inih@r62", "confirmed", ""), + ("six.py", "pkg:pypi/six@1.17.0", "needs-review", origin.SINGLE_TOOL_RULE), + ] + self.assertEqual(expected, list(values)) + self.assertEqual(user_determination, determinations[0]) + + package = self.project1.discoveredpackages.get() + self.assertEqual("pkg:github/benhoyt/inih@r62", package.purl) + self.assertQuerySetEqual(package.codebase_resources.all(), [ini_resource]) + + def test_scanpipe_pipes_origin_determine_origins_user_directory(self): + vendor_ms = make_resource_directory(self.project1, path="vendor/ms") + index_js = make_resource_file(self.project1, path="vendor/ms/index.js") + self.make_origin_match(index_js, "pkg:npm/lodash@4.17.21") + OriginDetermination.objects.create( + project=self.project1, + resource=vendor_ms, + purl="pkg:npm/ms@2.1.3", + status=OriginDetermination.Status.CONFIRMED, + ) + + self.assertEqual((0, 1), origin.determine_origins(self.project1)) + self.assertEqual(1, self.project1.origindeterminations.count()) + package = self.project1.discoveredpackages.get() + self.assertEqual("pkg:npm/ms@2.1.3", package.purl) + self.assertQuerySetEqual( + package.codebase_resources.order_by("path"), [vendor_ms, index_js] + ) + + def test_scanpipe_pipes_origin_determine_origins_scan_package(self): + scan_package = make_package(self.project1, "pkg:pypi/six@1.17.0") + setup_py = make_resource_file(self.project1, path="setup.py") + scan_package.add_resources([setup_py]) + six_resource = make_resource_file(self.project1, path="six.py") + self.make_origin_match(six_resource, "pkg:pypi/six@1.17.0") + self.make_origin_match(six_resource, "pkg:pypi/six@1.17.0", tool="scanoss") + + self.assertEqual((1, 1), origin.determine_origins(self.project1)) + self.assertEqual((1, 1), origin.determine_origins(self.project1)) + self.assertQuerySetEqual(self.project1.discoveredpackages.all(), [scan_package]) + package_resources = scan_package.codebase_resources.order_by("path") + self.assertQuerySetEqual(package_resources, [setup_py, six_resource]) + + self.project1.originmatchs.all().delete() + self.assertEqual((0, 0), origin.determine_origins(self.project1)) + self.assertQuerySetEqual(scan_package.codebase_resources.all(), [setup_py]) + scan_package.refresh_from_db() + self.assertEqual({}, scan_package.extra_data) + + def test_scanpipe_pipes_origin_assign_origin_package_resources(self): + setup_py = make_resource_file(self.project1, path="setup.py") + six_resource = make_resource_file(self.project1, path="six.py") + resources = self.project1.codebaseresources.all() + scan_package = make_package(self.project1, "pkg:pypi/six@1.17.0") + scan_package.add_resources([setup_py]) + + origin.assign_origin_package_resources(scan_package, resources) + package_resources = scan_package.codebase_resources.order_by("path") + self.assertQuerySetEqual(package_resources, [setup_py, six_resource]) + expected = {origin.ORIGIN_RESOURCE_PATHS_KEY: ["six.py"]} + self.assertEqual(expected, scan_package.extra_data) + + origin_package = make_package( + self.project1, + "pkg:npm/ms@2.1.3", + datasource_ids=[origin.ORIGIN_DATASOURCE_ID], + ) + origin.assign_origin_package_resources(origin_package, resources) + self.assertEqual(2, origin_package.codebase_resources.count()) + self.assertEqual({}, origin_package.extra_data) + + def test_scanpipe_pipes_origin_unassign_scan_package_resources(self): + setup_py = make_resource_file(self.project1, path="setup.py") + six_resource = make_resource_file(self.project1, path="six.py") + scan_package = make_package( + self.project1, + "pkg:pypi/six@1.17.0", + extra_data={origin.ORIGIN_RESOURCE_PATHS_KEY: ["six.py"]}, + ) + scan_package.add_resources([setup_py, six_resource]) + + origin.unassign_scan_package_resources(self.project1) + self.assertQuerySetEqual(scan_package.codebase_resources.all(), [setup_py]) + scan_package.refresh_from_db() + self.assertEqual({}, scan_package.extra_data) + + def test_scanpipe_pipes_origin_get_first_extra_data_value(self): + origin_matches = [ + make_unsaved_origin_match("pkg:npm/ms@2.1.3"), + OriginMatch(extra_data={"release_date": "2020-12-08"}), + ] + value = origin.get_first_extra_data_value(origin_matches, "release_date") + self.assertEqual("2020-12-08", value) + value = origin.get_first_extra_data_value(origin_matches, "download_url") + self.assertEqual("", value) + + def test_scanpipe_pipes_origin_get_license_names(self): + origin_matches = [ + make_scanoss_license_match(("MIT", "component_declared")), + make_scanoss_license_match( + ("MIT", "license_file"), + ("ISC", "license_file"), + ("GPL-2.0-only", "file_header"), + ), + OriginMatch(extra_data={}), + ] + self.assertEqual(["MIT", "ISC"], origin.get_license_names(origin_matches)) + + def test_scanpipe_pipes_origin_get_origin_license_data(self): + scanoss_match = make_scanoss_license_match( + ("MIT", "license_file"), ("BSD-2-Clause", "license_file") + ) + expected = { + "declared_license_expression": "mit AND bsd-simplified", + "declared_license_expression_spdx": "MIT AND BSD-2-Clause", + } + self.assertEqual(expected, origin.get_origin_license_data([scanoss_match])) + + matchcode_match = OriginMatch(extra_data={"declared_license_expression": "mit"}) + expected = {"declared_license_expression": "mit"} + license_data = origin.get_origin_license_data([scanoss_match, matchcode_match]) + self.assertEqual(expected, license_data) + + self.assertEqual({}, origin.get_origin_license_data([OriginMatch()])) + + def test_scanpipe_pipes_origin_get_origin_package_data(self): + origin_matches = [ + OriginMatch( + extra_data={ + "download_url": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "declared_license_expression": "mit", + } + ), + OriginMatch( + extra_data={ + "url": "https://github.com/vercel/ms", + "release_date": "2020-12-08", + } + ), + ] + expected = { + "download_url": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "homepage_url": "https://github.com/vercel/ms", + "release_date": "2020-12-08", + "declared_license_expression": "mit", + } + self.assertEqual(expected, origin.get_origin_package_data(origin_matches)) + + def test_scanpipe_pipes_origin_get_purl_origin_matches(self): + resource = make_resource_file(self.project1, path="six.py") + matchcode_match = self.make_origin_match(resource, "pkg:pypi/six@1.17.0") + scanoss_match = self.make_origin_match( + resource, + "pkg:github/benjaminp/six@1.17.0", + tool="scanoss", + extra_data={"purls": SCANOSS_SIX_PURLS}, + ) + self.make_origin_match(resource, "pkg:pypi/six@1.16.0") + + resources = self.project1.codebaseresources.all() + origin_matches = origin.get_purl_origin_matches( + self.project1, "pkg:pypi/six@1.17.0", resources + ) + self.assertEqual({matchcode_match, scanoss_match}, set(origin_matches)) + + def test_scanpipe_pipes_origin_get_origin_packages(self): + make_package(self.project1, "pkg:npm/ms@2.1.3") + origin_package = make_package( + self.project1, + "pkg:pypi/six@1.17.0", + datasource_ids=[origin.ORIGIN_DATASOURCE_ID], + ) + self.assertQuerySetEqual( + origin.get_origin_packages(self.project1), [origin_package] + ) + + def test_scanpipe_pipes_origin_get_or_create_origin_package(self): + scan_package = make_package(self.project1, "pkg:npm/ms@2.1.3") + resources = self.project1.codebaseresources.all() + package = origin.get_or_create_origin_package( + self.project1, "pkg:npm/ms@2.1.3", resources + ) + self.assertEqual(scan_package, package) + + resource = make_resource_file(self.project1, path="six.py") + self.make_origin_match( + resource, + "pkg:pypi/six@1.17.0", + extra_data={"declared_license_expression": "mit"}, + ) + package = origin.get_or_create_origin_package( + self.project1, "pkg:pypi/six@1.17.0", resources + ) + self.assertEqual("pkg:pypi/six@1.17.0", package.purl) + self.assertEqual("mit", package.declared_license_expression) + self.assertEqual([origin.ORIGIN_DATASOURCE_ID], package.datasource_ids) + + def test_scanpipe_pipes_origin_get_determined_resources(self): + vendor_ms = make_resource_directory(self.project1, path="vendor/ms") + index_js = make_resource_file(self.project1, path="vendor/ms/index.js") + debounce_js = make_resource_file(self.project1, path="vendor/ms/debounce.js") + make_resource_file(self.project1, path="vendor/other.js") + directory_determination = OriginDetermination.objects.create( + project=self.project1, + resource=vendor_ms, + purl="pkg:npm/ms@2.1.3", + status=OriginDetermination.Status.AUTO_CONFIRMED, + ) + OriginDetermination.objects.create( + project=self.project1, + resource=debounce_js, + status=OriginDetermination.Status.NEEDS_REVIEW, + ) + + resources = origin.get_determined_resources( + self.project1, [directory_determination], ["vendor/ms"] + ) + self.assertQuerySetEqual(resources.order_by("path"), [vendor_ms, index_js]) + + def test_scanpipe_pipes_origin_get_determined_resources_nested_directories(self): + vendor = make_resource_directory(self.project1, path="vendor") + foo_js = make_resource_file(self.project1, path="vendor/foo.js") + vendor_ms = make_resource_directory(self.project1, path="vendor/ms") + make_resource_file(self.project1, path="vendor/ms/index.js") + vendor_determination = OriginDetermination.objects.create( + project=self.project1, + resource=vendor, + purl="pkg:npm/foo@1.0.0", + status=OriginDetermination.Status.AUTO_CONFIRMED, + ) + OriginDetermination.objects.create( + project=self.project1, + resource=vendor_ms, + status=OriginDetermination.Status.NEEDS_REVIEW, + ) + + resources = origin.get_determined_resources( + self.project1, [vendor_determination], ["vendor", "vendor/ms"] + ) + self.assertQuerySetEqual(resources.order_by("path"), [vendor, foo_js]) + + def test_scanpipe_pipes_origin_get_descendants_filter(self): + make_resource_directory(self.project1, path="vendor") + foo_js = make_resource_file(self.project1, path="vendor/foo.js") + index_js = make_resource_file(self.project1, path="src/index.js") + resources = self.project1.codebaseresources.order_by("path") + + descendants_filter = origin.get_descendants_filter(["vendor", "src"]) + self.assertQuerySetEqual( + resources.filter(descendants_filter), [index_js, foo_js] + ) + self.assertQuerySetEqual( + resources.filter(origin.get_descendants_filter([])), [] + ) + self.assertEqual( + 3, resources.exclude(origin.get_descendants_filter([])).count() + ) + + def test_scanpipe_pipes_origin_get_determined_directory_paths(self): + vendor = make_resource_directory(self.project1, path="vendor") + make_resource_directory(self.project1, path="src") + six_resource = make_resource_file(self.project1, path="six.py") + for resource in [vendor, six_resource]: + OriginDetermination.objects.create( + project=self.project1, + resource=resource, + status=OriginDetermination.Status.NEEDS_REVIEW, + ) + + paths = origin.get_determined_directory_paths(self.project1) + self.assertEqual(["vendor"], paths) + + def test_scanpipe_pipes_origin_assign_origin_packages(self): + vendor_ms = make_resource_directory(self.project1, path="vendor/ms") + index_js = make_resource_file(self.project1, path="vendor/ms/index.js") + six_resource = make_resource_file(self.project1, path="six.py") + ini_resource = make_resource_file(self.project1, path="ini.c") + OriginDetermination.objects.create( + project=self.project1, + resource=vendor_ms, + purl="pkg:npm/ms@2.1.3", + status=OriginDetermination.Status.AUTO_CONFIRMED, + ) + OriginDetermination.objects.create( + project=self.project1, + resource=six_resource, + purl="pkg:pypi/six@1.17.0", + status=OriginDetermination.Status.CONFIRMED, + ) + OriginDetermination.objects.create( + project=self.project1, + resource=ini_resource, + purl="pkg:github/benhoyt/inih@r62", + status=OriginDetermination.Status.NEEDS_REVIEW, + ) + + self.assertEqual(2, origin.assign_origin_packages(self.project1)) + packages = self.project1.discoveredpackages.order_by("type") + package_resources = [ + ( + package.purl, + sorted(package.codebase_resources.values_list("path", flat=True)), + ) + for package in packages + ] + expected = [ + ("pkg:npm/ms@2.1.3", [vendor_ms.path, index_js.path]), + ("pkg:pypi/six@1.17.0", [six_resource.path]), + ] + self.assertEqual(expected, package_resources) + + def test_scanpipe_pipes_origin_determine_origins_code_genetics_data_set(self): + expected_by_case = { + "exact-copy": [ + ( + "myapp/_vendor/six.py", + "pkg:pypi/six@1.17.0", + "auto-confirmed", + origin.CONFIRMED_BY_TWO_TOOLS_RULE, + ), + ], + "vendored-copy": [ + ( + "myapp/_vendor/urllib3/packages/six.py", + "", + "needs-review", + origin.CONFLICTING_CANDIDATES_RULE, + ), + ], + "modified-copy": [ + ( + "src/inih/ini.c", + "", + "needs-review", + origin.CONFLICTING_VERSIONS_RULE, + ), + ( + "src/inih/ini.h", + "pkg:github/benhoyt/inih@r62", + "auto-confirmed", + origin.CONFIRMED_BY_TWO_TOOLS_RULE, + ), + ], + "snippet": [ + ( + "src/scheduler.js", + "pkg:npm/lodash@4.17.21", + "needs-review", + origin.NO_EXACT_MATCH_RULE, + ), + ], + "embedded-package": [ + ( + "vendor/ms", + "pkg:npm/ms@2.1.3", + "auto-confirmed", + origin.CONFIRMED_BY_TWO_TOOLS_RULE, + ), + ], + "no-origin": [], + } + + expected_packages_by_case = { + "exact-copy": [("pkg:pypi/six@1.17.0", "mit", ["myapp/_vendor/six.py"])], + "modified-copy": [ + ("pkg:github/benhoyt/inih@r62", "bsd-new", ["src/inih/ini.h"]) + ], + "embedded-package": [ + ( + "pkg:npm/ms@2.1.3", + "mit", + [ + "vendor/ms", + "vendor/ms/index.js", + "vendor/ms/license.md", + "vendor/ms/package.json", + "vendor/ms/readme.md", + ], + ) + ], + } + + for case_name, expected in expected_by_case.items(): + with self.subTest(case_name): + project = self.load_data_set_case(case_name) + origin.determine_origins(project) + + packages = [ + ( + package.purl, + package.declared_license_expression, + sorted( + package.codebase_resources.values_list("path", flat=True) + ), + ) + for package in project.discoveredpackages.all() + ] + self.assertEqual(expected_packages_by_case.get(case_name, []), packages) + + determinations = project.origindeterminations.order_by("resource__path") + values = determinations.values_list( + "resource__path", "purl", "status", "rule" + ) + self.assertEqual(expected, list(values)) + + # Every automatic confirmation matches the ground truth. + expected_purls_by_path = self.get_expected_purls_by_path(case_name) + auto_confirmed = determinations.filter(status="auto-confirmed") + for path, purl in auto_confirmed.values_list("resource__path", "purl"): + self.assertEqual(expected_purls_by_path[path], purl) + + def test_scanpipe_pipes_origin_determine_origins_data_set_single_tool(self): + expected_by_tool = { + matchcode.TOOL_NAME: { + "exact-copy": [ + ("myapp/_vendor/six.py", "pkg:pypi/six@1.17.0", "single-tool"), + ], + # The vendoring package is reported instead of the six origin. + "vendored-copy": [ + ( + "myapp/_vendor/urllib3/packages/six.py", + "pkg:pypi/urllib3@1.26.20", + "single-tool", + ), + ], + "modified-copy": [ + ("src/inih/ini.c", "pkg:github/benhoyt/inih@r62", "no-exact-match"), + ("src/inih/ini.h", "pkg:github/benhoyt/inih@r62", "single-tool"), + ], + "snippet": [ + ("src/scheduler.js", "pkg:npm/lodash@4.17.21", "no-exact-match"), + ], + "embedded-package": [ + ("vendor/ms", "pkg:npm/ms@2.1.3", "no-exact-match"), + ], + "no-origin": [], + }, + scanoss.TOOL_NAME: { + "exact-copy": [ + ( + "myapp/_vendor/six.py", + "pkg:github/benjaminp/six@1.17.0", + "single-tool", + ), + ], + "vendored-copy": [ + ( + "myapp/_vendor/urllib3/packages/six.py", + "pkg:github/piotrmasek/anime365@0.1-Ryucon", + "single-tool", + ), + ], + "modified-copy": [ + ( + "src/inih/ini.c", + "pkg:github/benhoyt/inih@498f34b", + "no-exact-match", + ), + ("src/inih/ini.h", "pkg:github/benhoyt/inih@r62", "single-tool"), + ], + "snippet": [], + "embedded-package": [ + ("vendor/ms/index.js", "pkg:github/vercel/ms@2.1.3", "single-tool"), + ], + "no-origin": [], + }, + } + + for tool, expected_by_case in expected_by_tool.items(): + for case_name, expected in expected_by_case.items(): + with self.subTest(tool=tool, case_name=case_name): + project = self.load_data_set_case(case_name, tools=[tool]) + origin.determine_origins(project) + + determinations = project.origindeterminations.order_by( + "resource__path" + ) + values = determinations.values_list( + "resource__path", "purl", "rule" + ) + self.assertEqual(expected, list(values)) + # A single tool can miss a conflicting candidate. + statuses = set(determinations.values_list("status", flat=True)) + self.assertNotIn("auto-confirmed", statuses) + self.assertFalse(project.discoveredpackages.exists()) diff --git a/scanpipe/tests/test_models.py b/scanpipe/tests/test_models.py index 4a24767e4d..9ee27e1ab6 100644 --- a/scanpipe/tests/test_models.py +++ b/scanpipe/tests/test_models.py @@ -62,6 +62,7 @@ from scanpipe.models import DiscoveredDependency from scanpipe.models import DiscoveredPackage from scanpipe.models import DiscoveredPackageScore +from scanpipe.models import OriginDetermination from scanpipe.models import OriginMatch from scanpipe.models import Project from scanpipe.models import ProjectMessage @@ -184,6 +185,13 @@ def test_scanpipe_project_model_delete_related_objects(self): match_type=OriginMatch.MatchType.FILE, tool="matchcode", ) + OriginDetermination.objects.create( + project=self.project1, + resource=resource, + purl="pkg:pypi/six@1.17.0", + status=OriginDetermination.Status.NEEDS_REVIEW, + rule="single-tool", + ) delete_log = self.project1.delete_related_objects() expected = { @@ -194,6 +202,7 @@ def test_scanpipe_project_model_delete_related_objects(self): "scanpipe.DiscoveredPackage": 1, "scanpipe.DiscoveredPackage_codebase_resources": 1, "scanpipe.InputSource": 0, + "scanpipe.OriginDetermination": 1, "scanpipe.OriginMatch": 1, "scanpipe.ProjectMessage": 0, "scanpipe.Run": 1, @@ -3115,6 +3124,52 @@ def test_scanpipe_origin_match_queryset_methods(self): origin_matches.partial_matches(), [directory_match, snippet_match] ) + def test_scanpipe_origin_determination_queryset_methods(self): + six_resource = make_resource_file(self.project1, path="six.py") + ini_resource = make_resource_file(self.project1, path="ini.c") + urllib3_resource = make_resource_file(self.project1, path="urllib3.py") + six_determination = OriginDetermination.objects.create( + project=self.project1, + resource=six_resource, + purl="pkg:pypi/six@1.17.0", + status=OriginDetermination.Status.AUTO_CONFIRMED, + rule="confirmed-by-two-tools", + ) + ini_determination = OriginDetermination.objects.create( + project=self.project1, + resource=ini_resource, + purl="pkg:github/benhoyt/inih@r62", + status=OriginDetermination.Status.CONFIRMED, + ) + urllib3_determination = OriginDetermination.objects.create( + project=self.project1, + resource=urllib3_resource, + status=OriginDetermination.Status.NEEDS_REVIEW, + rule="conflicting-candidates", + ) + + determinations = self.project1.origindeterminations.order_by("resource__path") + self.assertQuerySetEqual( + determinations.automatic(), [six_determination, urllib3_determination] + ) + self.assertQuerySetEqual( + determinations.confirmed(), [ini_determination, six_determination] + ) + + def test_scanpipe_origin_determination_model_str(self): + resource = make_resource_file(self.project1, path="six.py") + determination = OriginDetermination( + project=self.project1, + resource=resource, + purl="pkg:pypi/six@1.17.0", + status=OriginDetermination.Status.CONFIRMED, + ) + self.assertEqual("pkg:pypi/six@1.17.0 (confirmed)", str(determination)) + + determination.purl = "" + determination.status = OriginDetermination.Status.NO_ORIGIN + self.assertEqual("no-origin", str(determination)) + class ScanPipeModelsTransactionTest(TransactionTestCase): """ @@ -3497,3 +3552,25 @@ def test_scanpipe_origin_match_model_similarity_max_100(self): tool="scanoss", similarity=101, ) + + def test_scanpipe_origin_determination_model_purl_consistent_with_status(self): + project1 = make_project("Analysis") + resource = make_resource_file(project1, path="six.py") + determination_data = {"project": project1, "resource": resource} + + with self.assertRaises(IntegrityError): + OriginDetermination.objects.create( + **determination_data, status=OriginDetermination.Status.CONFIRMED + ) + + with self.assertRaises(IntegrityError): + OriginDetermination.objects.create( + **determination_data, + purl="pkg:pypi/six@1.17.0", + status=OriginDetermination.Status.NO_ORIGIN, + ) + + OriginDetermination.objects.create( + **determination_data, status=OriginDetermination.Status.NEEDS_REVIEW + ) + self.assertEqual("needs-review", resource.origin_determination.status) diff --git a/scanpipe/tests/test_pipelines.py b/scanpipe/tests/test_pipelines.py index 81e6624b5a..df541e6f6f 100644 --- a/scanpipe/tests/test_pipelines.py +++ b/scanpipe/tests/test_pipelines.py @@ -1733,6 +1733,22 @@ def write_scanoss_results(codebase_path, output_location): self.assertEqual(expected, list(values)) self.assertEqual(1, len(list(project1.output_path.glob("matchcode-*.json")))) + determinations = project1.origindeterminations.order_by("resource__path") + expected_determinations = [ + (six_path, "pkg:pypi/six@1.17.0", "auto-confirmed"), + ( + "modified-copy/src/inih/ini.c", + "pkg:github/benhoyt/inih@498f34b", + "needs-review", + ), + ] + values = determinations.values_list("resource__path", "purl", "status") + self.assertEqual(expected_determinations, list(values)) + + package = project1.discoveredpackages.get() + self.assertEqual("pkg:pypi/six@1.17.0", package.purl) + self.assertEqual(six_path, package.codebase_resources.get().path) + @mock.patch("scanpipe.pipes.scanoss.scan_codebase") @mock.patch("scanpipe.pipes.scanoss.get_uvx_location") @mock.patch("scanpipe.pipes.scanoss.is_configured")