diff --git a/pyproject.toml b/pyproject.toml
index 350c59a0b7..c0c2df2c0d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -171,6 +171,7 @@ resolve_dependencies = "scanpipe.pipelines.resolve_dependencies:ResolveDependenc
scan_codebase = "scanpipe.pipelines.scan_codebase:ScanCodebase"
scan_for_virus = "scanpipe.pipelines.scan_for_virus:ScanForVirus"
scan_maven_package = "scanpipe.pipelines.scan_maven_package:ScanMavenPackage"
+scan_rust_package = "scanpipe.pipelines.scan_rust_package:ScanRustPackage"
scan_single_package = "scanpipe.pipelines.scan_single_package:ScanSinglePackage"
[tool.setuptools.packages.find]
diff --git a/scanpipe/pipelines/scan_rust_package.py b/scanpipe/pipelines/scan_rust_package.py
new file mode 100644
index 0000000000..99b9a486c6
--- /dev/null
+++ b/scanpipe/pipelines/scan_rust_package.py
@@ -0,0 +1,207 @@
+# 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.
+
+import shutil
+import tempfile
+from pathlib import Path
+
+from scanpipe.pipelines.deploy_to_develop import DeployToDevelop
+from scanpipe.pipelines.scan_codebase import ScanCodebase
+from scanpipe.pipelines.scan_single_package import ScanSinglePackage
+from scanpipe.pipes import d2d
+from scanpipe.pipes import flag
+from scanpipe.pipes import utils
+from scanpipe.pipes.rust import build_crates
+from scanpipe.pipes.rust import check_input_and_return_purl
+from scanpipe.pipes.rust import get_cargo_toml_path
+from scanpipe.pipes.rust import get_repository_value_from_cargo_toml
+
+
+class ScanRustPackage(ScanSinglePackage, DeployToDevelop, ScanCodebase):
+ """
+ Download the crate’s source, build it, and run a d2d comparison between
+ the compiled binary and the source crate to detect any discrepancies.
+
+ Identify the upstream source repository and verify that it matches the
+ contents of the source crate.
+
+ Scan the source crate and confirm that the detected license aligns with
+ the license declared in Cargo.toml.
+ """
+
+ download_inputs = False
+
+ @classmethod
+ def steps(cls):
+ return (
+ cls.check_input_and_return_purl,
+ cls.fetch_inputs,
+ cls.collect_input_info,
+ cls.extract_input_to_codebase_directory,
+ cls.check_docker_command,
+ cls.get_cargo_toml,
+ cls.build_crates,
+ cls.run_scan,
+ cls.load_inventory_from_toolkit_scan,
+ cls.add_from_to_tag,
+ cls.validate_package_license_integrity,
+ cls.identify_built_sources,
+ cls.flag_mapped_status,
+ cls.get_src_repo_download_url,
+ cls.download_src_repo,
+ cls.compare_src_repo_with_from_codebase,
+ cls.update_comparison_summary,
+ cls.make_summary_from_scan_results,
+ )
+
+ def check_input_and_return_purl(self):
+ """Validate the input is a PURL string and return the PURL object."""
+ self.purl = check_input_and_return_purl(self.project)
+
+ def fetch_inputs(self):
+ """Fetch the source of the given PURL."""
+ self.from_files = utils.fetch_inputs(self.purl)
+
+ def collect_input_info(self):
+ """Collect information about the input."""
+ self.input_path = self.from_files
+ self.collect_input_information()
+
+ def check_docker_command(self):
+ """Check if the Docker command is available."""
+ self.have_docker = False
+ if shutil.which("docker"):
+ self.have_docker = True
+
+ def get_cargo_toml(self):
+ """Get the Cargo.toml path from the codebase directory."""
+ self.cargo_toml_path = None
+ self.devel_codebase_dir = None
+ if self.have_docker:
+ codebase_dir = Path(self.project.codebase_path)
+ self.devel_codebase_dir = codebase_dir
+ self.cargo_toml_path = get_cargo_toml_path(codebase_dir)
+
+ def build_crates(self):
+ """
+ Build the Rust crate using Docker and put the built files under the
+ "to" directory.
+ """
+ self.d2d_enable = False
+ if self.cargo_toml_path:
+ codebase_dir = self.devel_codebase_dir
+ cargo_toml_path = self.cargo_toml_path
+ if build_crates(codebase_dir, cargo_toml_path):
+ self.d2d_enable = True
+ updated_path = cargo_toml_path.relative_to(codebase_dir)
+ self.cargo_toml_path = codebase_dir / "from" / updated_path
+ self.devel_codebase_dir = codebase_dir / "from"
+ else:
+ print("Docker command not found. Skipping crate build.")
+ else:
+ print("Cargo.toml is not found.")
+
+ def add_from_to_tag(self):
+ """Update 'from' and 'to' tag to resources based on their path."""
+ if self.d2d_enable:
+ d2d.update_from_to_tag(self.project)
+
+ def validate_package_license_integrity(self):
+ """
+ Validate the correctness of the package license compare with the
+ detected license from the codebase.
+ """
+ utils.validate_package_license_integrity(self.project)
+
+ def identify_built_sources(self):
+ """Identify the built sources from the '.d' file in the "to" directory."""
+ if self.d2d_enable:
+ d2d.map_rust_paths(self.project)
+
+ def flag_mapped_status(self):
+ """Flag the from codebase resources that were mapped."""
+ if self.d2d_enable:
+ flag.flag_mapped_resources(self.project)
+
+ def get_src_repo_download_url(self):
+ """
+ Get the source repository url from Cargo.toml and determine its
+ download url.
+ """
+ self.src_download_url = None
+ repository_url = get_repository_value_from_cargo_toml(self.cargo_toml_path)
+ if not repository_url:
+ self.project.add_warning(
+ description="No source repository URL found in Cargo.toml."
+ )
+ else:
+ self.src_download_url = utils.get_download_url(
+ repository_url, self.purl.version
+ )
+ if not self.src_download_url:
+ self.project.add_warning(
+ description=(
+ "Not able to determine the source repository download URL from "
+ "Cargo.toml."
+ )
+ )
+
+ def download_src_repo(self):
+ """Download the source from the source repo."""
+ self.src_repo_path = None
+ if self.src_download_url:
+ self.src_repo_path = utils.download_src_repo(self.src_download_url)
+ if not self.src_repo_path:
+ self.project.add_warning(
+ description=(
+ f"The source repository URL "
+ f"{self.src_download_url} "
+ f"could not be downloaded. Skipping the source "
+ f"crate and source repository comparison."
+ )
+ )
+
+ def compare_src_repo_with_from_codebase(self):
+ """Compare the downloaded source repo with the from codebase."""
+ self.matched_count = 0
+ self.mismatches = []
+ if self.src_repo_path:
+ with tempfile.TemporaryDirectory() as source_repo_path:
+ self.extract_archive(self.src_repo_path, source_repo_path)
+
+ self.matched_count, self.mismatches = utils.compare_directories(
+ self.devel_codebase_dir, source_repo_path
+ )
+
+ def update_comparison_summary(self):
+ """Update the comparison summary in the discovered package."""
+ if self.src_repo_path:
+ utils.update_comparison_summary(
+ self.project,
+ self.purl,
+ self.devel_codebase_dir,
+ self.src_download_url,
+ self.purl.name,
+ self.purl.version,
+ self.matched_count,
+ self.mismatches,
+ )
diff --git a/scanpipe/pipes/d2d.py b/scanpipe/pipes/d2d.py
index 2e7b2d18dd..cc3cf696d3 100644
--- a/scanpipe/pipes/d2d.py
+++ b/scanpipe/pipes/d2d.py
@@ -1737,20 +1737,25 @@ def map_paths_resource(
relations_to_create[rel_key] = relation
if paths_not_mapped:
to_resource.status = flag.REQUIRES_REVIEW
- logger(
- f"WARNING: #{len(paths_not_mapped)} {map_type} paths NOT mapped for: "
- f"{to_resource.path!r}"
- )
+ if logger:
+ logger(
+ f"WARNING: #{len(paths_not_mapped)} {map_type} paths NOT "
+ f" mapped for: {to_resource.path!r}"
+ )
to_resource.save()
if relations_to_create:
rels = CodebaseRelation.objects.bulk_create(relations_to_create.values())
- logger(
- f"Created {len(rels)} mappings using "
- f"{', '.join(map_types)} for: {to_resource.path!r}"
- )
+ if logger:
+ logger(
+ f"Created {len(rels)} mappings using "
+ f"{', '.join(map_types)} for: {to_resource.path!r}"
+ )
else:
- logger(f"No mappings using {', '.join(map_types)} for: {to_resource.path!r}")
+ if logger:
+ logger(
+ f"No mappings using {', '.join(map_types)} for: {to_resource.path!r}"
+ )
def process_paths_in_binary(
@@ -1940,6 +1945,120 @@ def map_go_paths(project, logger=None):
)
+def get_rust_file_paths(location):
+ """Retrieve Rust file paths."""
+ file_paths = {}
+ rust_lib_path, rust_file_paths = parse_d_file(location) or []
+ if rust_file_paths:
+ file_paths["rust_file_paths"] = rust_file_paths
+ return rust_lib_path, file_paths
+
+
+def parse_d_file(path):
+ """Parse the .d file from rust package."""
+ context = Path(path).read_text()
+ cleaned_context = context.replace("\\\n", " ")
+
+ # Invalid .d file
+ if ":" not in cleaned_context:
+ return []
+
+ rust_lib, dep_paths = cleaned_context.split(":", 1)
+ rust_lib_path = rust_lib.strip()
+
+ file_paths = []
+ for file_path in dep_paths.split():
+ file_path = file_path.strip()
+ if file_path:
+ file_paths.append(file_path)
+
+ return rust_lib_path, file_paths
+
+
+def map_rust_paths(project, logger=None):
+ """Map the path listed in the .d file to the source in ``project``."""
+ from_resources = project.codebaseresources.files().from_codebase()
+ # Fetch the .d files to extract data from
+ data_resources = (
+ project.codebaseresources.files()
+ .to_codebase()
+ .exclude(path__contains="/deps/")
+ .exclude(path__contains="/build/")
+ .filter(path__endswith=".d")
+ )
+ target_rlib_ids = []
+ for resource in data_resources:
+ try:
+ rlib_path_str, paths = get_rust_file_paths(resource.location_path)
+ rlib_path = Path(rlib_path_str)
+ rlib_resource = None
+ try:
+ if rlib_path_str.startswith("/codebase/"):
+ clean_rlib_path = str(rlib_path.relative_to("/codebase"))
+ else:
+ clean_rlib_path = str(rlib_path.relative_to(project.codebase_path))
+
+ rlib_resource = (
+ project.codebaseresources.files()
+ .to_codebase()
+ .filter(path=clean_rlib_path)
+ .first()
+ )
+ except ValueError:
+ pass
+
+ if rlib_resource:
+ rlib_resource.update_extra_data(paths)
+ target_rlib_ids.append(rlib_resource.id)
+ elif logger:
+ logger(
+ f"Warning: Could not find rlib file {rlib_path_str} in database."
+ )
+ except Exception as exception:
+ project.add_warning(
+ exception=exception,
+ object_instance=resource,
+ description=f"Cannot parse file at {resource.path}",
+ model="map_rust_paths",
+ details={"path": resource.path},
+ )
+
+ to_resources = project.codebaseresources.filter(id__in=target_rlib_ids)
+
+ if logger:
+ logger(
+ f"Mapping {to_resources.count():,d} to/ resources using paths "
+ f"with {from_resources.count():,d} from/ resources."
+ )
+
+ from_resources_index = pathmap.build_index(
+ from_resources.values_list("id", "path"), with_subpaths=True
+ )
+
+ if logger:
+ logger("Done building from/ resources index.")
+
+ resource_iterator = to_resources.iterator(chunk_size=2000)
+ progress = LoopProgress(to_resources.count(), logger)
+ for to_resource in progress.iter(resource_iterator):
+ map_paths_resource(
+ to_resource,
+ from_resources,
+ from_resources_index,
+ map_types=["rust_file_paths"],
+ logger=logger,
+ )
+
+
+def update_from_to_tag(project):
+ """Update 'from' or 'to' tag to resources based on their path."""
+ for resource in project.codebaseresources.files():
+ if resource.path.startswith("from/"):
+ resource.update(tag="from")
+ elif resource.path.startswith("to/"):
+ resource.update(tag="to")
+
+
RUST_BINARY_OPTIONS = ["Rust"]
ELF_BINARY_OPTIONS = ["Python", "Go", "Elf"]
MACHO_BINARY_OPTIONS = ["Rust", "Go", "MacOS"]
diff --git a/scanpipe/pipes/fetch.py b/scanpipe/pipes/fetch.py
index 3cbbb13200..68164ecfa9 100644
--- a/scanpipe/pipes/fetch.py
+++ b/scanpipe/pipes/fetch.py
@@ -82,6 +82,13 @@
def get_request_session(uri):
"""Return a Requests session setup with authentication and headers."""
session = requests.Session()
+
+ # Set a default User-Agent to avoid 403 Forbidden errors on strict
+ # registries like crates.io that block default python-requests headers.
+ session.headers.update(
+ {"User-Agent": "ScanCode.io (https://github.com/aboutcode-org/scancode.io)"}
+ )
+
netloc = urlparse(uri).netloc
if credentials := scanpipe_settings.FETCH_BASIC_AUTH.get(netloc):
diff --git a/scanpipe/pipes/flag.py b/scanpipe/pipes/flag.py
index f94d0cc82d..2d4b2edf51 100644
--- a/scanpipe/pipes/flag.py
+++ b/scanpipe/pipes/flag.py
@@ -66,6 +66,7 @@
REQUIRES_REVIEW = "requires-review"
REVIEW_DANGLING_LEGAL_FILE = "review-dangling-legal-file"
NOT_DEPLOYED = "not-deployed"
+LICENSE_ISSUE = "license-mismatch-declared-vs-detected"
# Target files that should be ignored during processing as those are related to the app
diff --git a/scanpipe/pipes/rust.py b/scanpipe/pipes/rust.py
new file mode 100644
index 0000000000..c50daee37e
--- /dev/null
+++ b/scanpipe/pipes/rust.py
@@ -0,0 +1,131 @@
+# 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.
+
+import logging
+import shutil
+import subprocess
+from pathlib import Path
+
+import tomllib
+from packageurl import PackageURL
+
+from scanpipe.pipes import run_command_safely
+
+logger = logging.getLogger(__name__)
+
+
+def build_crates(codebase_dir, cargo_toml_path):
+ """
+ Build the Rust crate from source in an isolated Docker container.
+ Use the official Rust image for the build process.
+ Return True if the build succeeds, False otherwise.
+ """
+ to_dir = codebase_dir / "to"
+ cargo_toml_path = Path(cargo_toml_path)
+ build_dir = Path(to_dir)
+
+ # Get the relative paths
+ relative_cargo_toml = cargo_toml_path.relative_to(codebase_dir).as_posix()
+ relative_build_dir = build_dir.relative_to(codebase_dir).as_posix()
+
+ container_cargo_toml = f"/codebase/{relative_cargo_toml}"
+ container_build_dir = f"/codebase/{relative_build_dir}"
+
+ # Since we will use the .d file for deployment and development file
+ # mapping, we will not require building with DWARF debug symbols. If we
+ # later decide to include DWARF, we can add the following to the
+ # command:
+ # "--env", "RUSTFLAGS=-C debuginfo=2",
+ cmd = [
+ "docker",
+ "run",
+ "--rm",
+ "--volume",
+ f"{codebase_dir}:/codebase",
+ "--workdir",
+ "/codebase",
+ "rust:latest",
+ "cargo",
+ "build",
+ "--release",
+ "--locked",
+ "--manifest-path",
+ container_cargo_toml,
+ "--target-dir",
+ container_build_dir,
+ ]
+
+ try:
+ run_command_safely(cmd)
+ except subprocess.SubprocessError as error:
+ logger.warning(f"Failed to build the Rust crate in Docker: {error}")
+ return False
+
+ # Move the development code under the /codebase/from/
+ from_dir = codebase_dir / "from"
+ from_dir.mkdir(exist_ok=True)
+ for item in codebase_dir.iterdir():
+ if item != to_dir and item != from_dir:
+ shutil.move(str(item), str(from_dir / item.name))
+ return True
+
+
+def check_input_and_return_purl(project):
+ """Validate the input and return a cargo PURL."""
+ input_sources = project.inputsources.all()
+ if len(input_sources) != 1:
+ error_msg = "Only 1 cargo purl is accepted."
+ raise ValueError(error_msg)
+ # Strip the qualifiers if present as this is not needed
+ project_input = str(input_sources[0]).split("?")[0]
+ input_purl = PackageURL.from_string(project_input)
+
+ if input_purl.type != "cargo":
+ error_msg = "Only cargo purl is supported."
+ raise ValueError(error_msg)
+ if not input_purl.version:
+ error_msg = "Version is required."
+ raise ValueError(error_msg)
+
+ return input_purl
+
+
+def get_repository_value_from_cargo_toml(cargo_toml_path):
+ """Get the repository value from Cargo.toml."""
+ path = Path(cargo_toml_path)
+ if not path.exists():
+ raise FileNotFoundError(f"{cargo_toml_path} not found")
+
+ with path.open("rb") as f:
+ data = tomllib.load(f)
+
+ return data.get("package", {}).get("repository", "")
+
+
+def get_cargo_toml_path(codebase_dir):
+ """Get the Cargo.toml path from the codebase directory."""
+ cargo_toml_path = None
+ # There is only one "Cargo.toml" per published package
+ for path in codebase_dir.rglob("Cargo.toml"):
+ cargo_toml_path = path
+ break
+ return cargo_toml_path
diff --git a/scanpipe/pipes/utils.py b/scanpipe/pipes/utils.py
new file mode 100644
index 0000000000..81a5b85bf3
--- /dev/null
+++ b/scanpipe/pipes/utils.py
@@ -0,0 +1,521 @@
+# 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.
+
+import hashlib
+import logging
+import os
+from fnmatch import fnmatch
+from pathlib import Path
+from urllib.parse import urlparse
+
+import requests
+from license_expression import Licensing
+from packageurl import PackageURL
+from packageurl.contrib.purl2url import get_repo_download_url_by_package_type
+
+from scanpipe.pipes import fetch
+from scanpipe.pipes import flag
+
+logger = logging.getLogger(__name__)
+
+
+def validate_package_license_integrity(project):
+ """Validate the correctness of the package license."""
+ # Patterns to ignore certain resources during license validation
+ ignore_patterns = [
+ "*test*",
+ "*.sh",
+ ]
+
+ for package in project.discoveredpackages.all():
+ package_lic = package.get_declared_license_expression()
+ if package_lic:
+ if package.type == "cargo":
+ # A single cargo package only has one Cargo.toml file
+ # meaning only one package is defined. Therefore, we don't
+ # need to check for the package_uid
+ # In addition, the package_uid is not populated to source files:
+ # https://github.com/aboutcode-org/scancode.io/issues/2169
+ # so we set package_uid to None to consider all resources
+ # in the codebase for license validation.
+ package_uid = None
+ else:
+ package_uid = package.package_uid
+ resources = project.codebaseresources.has_license_expression()
+ detected_lic_list = collect_detected_licenses(
+ resources, ignore_patterns, package_uid
+ )
+
+ if detected_lic_list:
+ lic_exp = " AND ".join(detected_lic_list)
+ detected_lic_exp = str(Licensing().dedup(lic_exp))
+
+ if detected_lic_exp != package_lic:
+ package_issues = package.extra_data.get("issues", [])
+
+ package_issues.append(
+ {
+ "issue_type": "License Mismatch",
+ "declared_license": package_lic,
+ "detected_codebase_license": detected_lic_exp,
+ }
+ )
+
+ package.update_extra_data({"issues": package_issues})
+
+ for datafile_path in package.datafile_paths:
+ if not datafile_path.startswith("https://"):
+ data_path = project.codebaseresources.get(
+ path=datafile_path
+ )
+ data_path.update(status=flag.LICENSE_ISSUE)
+
+ resource_issues = data_path.extra_data.get("issues", [])
+ resource_issues.append(
+ {
+ "issue_type": "License Mismatch",
+ "declared_license": package_lic,
+ "detected_codebase_license": detected_lic_exp,
+ }
+ )
+
+ data_path.update_extra_data({"issues": resource_issues})
+
+
+def contains_ignore_pattern(resource_path, ignore_patterns):
+ """Check if the resource path matches any of the ignore patterns."""
+ for pattern in ignore_patterns:
+ if fnmatch(resource_path, pattern):
+ return True
+ return False
+
+
+def filter_ignored_licenses(license_expression, licensing):
+ """Filter out ignored licenses from a license expression."""
+ # Some licenses are not useful for validating package license
+ # integrity, so we ignore them.
+ ignored_licenses = [
+ "free-unknown",
+ "unknown",
+ "unknown-license-reference",
+ "unknown-spdx",
+ ]
+
+ if license_expression is None:
+ return None
+
+ if isinstance(license_expression, licensing.Symbol):
+ if (
+ hasattr(license_expression, "key")
+ and license_expression.key in ignored_licenses
+ ):
+ return None
+ return license_expression
+
+ # Handle AND operations
+ if isinstance(license_expression, licensing.AND):
+ return handle_operator_expression(license_expression, licensing, licensing.AND)
+
+ # Handle OR operations
+ if isinstance(license_expression, licensing.OR):
+ return handle_operator_expression(license_expression, licensing, licensing.OR)
+
+ return license_expression
+
+
+def handle_operator_expression(expression, licensing, operator):
+ """
+ Process AND/OR operations in a license expression, filtering out
+ ignored licenses.
+ """
+ args = []
+ for arg in expression.args:
+ filtered_arg = filter_ignored_licenses(arg, licensing)
+ if filtered_arg is not None:
+ args.append(filtered_arg)
+ if not args:
+ return None
+ if len(args) == 1:
+ return args[0]
+
+ return operator(*args)
+
+
+def collect_detected_licenses(resources, ignore_patterns, package_uid=None):
+ """Collect detected licenses from resources, ignoring defined patterns."""
+ licensing = Licensing()
+ detected_lic_list = []
+
+ for resource in resources:
+ if contains_ignore_pattern(resource.path, ignore_patterns):
+ continue
+
+ # If a package_uid is provided, only consider resources linked to it
+ if package_uid and package_uid not in resource.for_packages:
+ continue
+
+ license_str = resource.detected_license_expression
+ if not license_str:
+ continue
+ try:
+ parsed_lic = licensing.parse(license_str)
+
+ # Filter out the ignored keys
+ filtered_license = filter_ignored_licenses(parsed_lic, licensing)
+
+ if filtered_license is not None:
+ final_lic = str(filtered_license)
+
+ if final_lic not in detected_lic_list:
+ # Apply parentheses so that the 'OR' expression will
+ # not be filtered out when doing deduplication later.
+ detected_lic_list.append(f"({final_lic})")
+
+ except Exception:
+ logger.warning(
+ "Failed to parse the license expression: %s at %s",
+ license_str,
+ resource.path,
+ )
+ return detected_lic_list
+
+
+def get_url_netloc_namespace_and_name(url):
+ """
+ Extract netloc, namespace, and name from a URL path.
+ - The last path component (except for web files) is considered the name.
+ - Everything between netloc and name is considered the namespace.
+ """
+ parsed = urlparse(url)
+ netloc = parsed.netloc
+ parts = parsed.path.strip("/").split("/")
+
+ if not parts or parts == [""]:
+ return netloc, None, None
+
+ if len(parts) > 1:
+ last_part = parts[-1].lower()
+ ignore_extensions = (".html", ".htm", ".php", ".jsp", ".asp", ".aspx")
+ if last_part.startswith("index.") or last_part.endswith(ignore_extensions):
+ parts.pop()
+
+ name = parts[-1]
+ namespace = "/".join(parts[:-1]) if len(parts) > 1 else None
+
+ return netloc, namespace, name
+
+
+def download_src_repo(download_url):
+ try:
+ return fetch.fetch_url(url=download_url).path
+ except (ValueError, requests.RequestException):
+ logger.warning("Failed to download source repository: %s", download_url)
+ return None
+
+
+def get_download_url(homepage_url, version):
+ netloc, namespace, name = get_url_netloc_namespace_and_name(homepage_url)
+ hostname = (urlparse(homepage_url).hostname or "").rstrip(".").lower()
+ if hostname == "github.io" or hostname.endswith(".github.io"):
+ github_page_url = github_pages_to_repo(homepage_url)
+ if github_page_url:
+ netloc, namespace, name = get_url_netloc_namespace_and_name(github_page_url)
+
+ if netloc in ("github.com", "gitlab.com", "bitbucket.org"):
+ if netloc.endswith(".com"):
+ package_type = netloc.removesuffix(".com")
+ # There is an issue where the version may have a different prefix.
+ # For example, version can have the following prefixes:
+ # ["v", "V", "release-", "RELEASE-", "v-", "V-"]
+ clarified_version = clarify_version_tag(
+ package_type, namespace, name, version
+ )
+ if clarified_version:
+ version = clarified_version
+ elif netloc.endswith(".org"):
+ package_type = netloc.removesuffix(".org")
+ download_url = get_repo_download_url_by_package_type(
+ type=package_type, namespace=namespace, name=name, version=version
+ )
+ return download_url
+ return None
+
+
+def clarify_version_tag(repo_type, namespace, name, version):
+ """Use github/gitlab API to verify the version tag"""
+ headers = {}
+ if repo_type == "github":
+ github_token = os.environ.get("GITHUB_TOKEN")
+ if github_token:
+ headers["Authorization"] = f"token {github_token}"
+ url_base = f"https://api.github.com/repos/{namespace}/{name}/git/refs/tags/{{}}"
+ elif repo_type == "gitlab":
+ gitlab_token = os.environ.get("GITLAB_TOKEN")
+ if gitlab_token:
+ headers["PRIVATE-TOKEN"] = gitlab_token
+ ns = namespace or ""
+ project_path = f"{ns}/{name}".strip("/").replace("/", "%2F")
+ url_base = (
+ f"https://gitlab.com/api/v4/projects/{project_path}/repository/tags/{{}}"
+ )
+ else:
+ return None
+
+ potential_prefixes = ["", "v", "V", "release-", "RELEASE-", "v-", "V-"]
+ for prefix in potential_prefixes:
+ potential_tag = f"{prefix}{version}"
+ url = url_base.format(potential_tag)
+
+ try:
+ response = requests.get(url, headers=headers, timeout=10)
+ except requests.RequestException:
+ continue
+ if response.status_code == 200:
+ return potential_tag
+ elif response.status_code in (403, 429):
+ print(
+ f"Rate limited by {repo_type} API while checking tag {potential_tag}."
+ )
+ return None
+
+ return None
+
+
+def github_pages_to_repo(url):
+ """
+ Try to map a GitHub Pages URL (https://{org}.github.io/{name}/)
+ to its corresponding GitHub repository (https://github.com/{org}/{name}).
+ Returns the repo URL if it exists, otherwise None.
+ """
+ parsed = urlparse(url)
+ host = parsed.netloc
+ parts = parsed.path.strip("/").split("/")
+
+ # Only handle {org}.github.io/{name} pattern
+ if not host.endswith(".github.io") or len(parts) < 1:
+ return None
+
+ org = host.replace(".github.io", "")
+ name = parts[0]
+
+ repo = f"https://github.com/{org}/{name}"
+
+ # Verify existence via GitHub API
+ api_url = f"https://api.github.com/repos/{org}/{name}"
+ try:
+ response = requests.get(api_url, timeout=10)
+ if response.status_code == 200:
+ return repo
+ except requests.RequestException:
+ return None
+
+ return None
+
+
+def compute_sha1(file_path):
+ """Compute the SHA1 hash of a file."""
+ try:
+ with open(file_path, "rb") as f:
+ return hashlib.file_digest(f, "sha1").hexdigest()
+ except OSError:
+ return None
+
+
+def get_all_files(base_dir):
+ """
+ Walk a directory and returns a dictionary mapping relative paths
+ to their filename and SHA1 hash.
+ """
+ file_map = {}
+ for root, _dirs, files in os.walk(base_dir):
+ for file in files:
+ full_path = os.path.join(root, file)
+ hash = compute_sha1(full_path)
+ rel_path = Path(full_path).relative_to(base_dir).as_posix()
+ file_map[rel_path] = {"name": file, "hash": hash}
+ return file_map
+
+
+def consolidate_unmatched(all_files, unmatched_files):
+ """
+ Consolidate the unmatched files. If all files in a directory are
+ unmatched, report the directory instead of individual files.
+ Returns a list of tuples: (path, is_directory)
+ """
+ matched_files = set(all_files) - set(unmatched_files)
+
+ # Find every parent directory that contains at least one matched file.
+ directories_with_matches = set()
+ for file_path in matched_files:
+ for parent in Path(file_path).parents:
+ parent_str = parent.as_posix()
+ if parent_str != ".":
+ directories_with_matches.add(parent_str)
+
+ consolidated_results = set()
+
+ # For each unmatched file, check if it belongs to a fully unmatched directory.
+ for file_path in unmatched_files:
+ path_obj = Path(file_path)
+ target_path = file_path
+ is_directory = False
+
+ # Check from top to bottom
+ for parent in reversed(path_obj.parents):
+ current_dir = parent.as_posix()
+ if current_dir == ".":
+ continue
+
+ if current_dir not in directories_with_matches:
+ target_path = current_dir
+ is_directory = True
+ break # Stop at the highest possible unmatched directory level
+
+ consolidated_results.add((target_path, is_directory))
+
+ # Convert the set to a list and sort it
+ results_list = list(consolidated_results)
+ results_list.sort()
+
+ return results_list
+
+
+def compare_directories(input_source, source_repo):
+ """
+ Compare two directories and return the count of matched files and a
+ dictionary of mismatches.
+ """
+ input_files = get_all_files(input_source)
+ repo_files = get_all_files(source_repo)
+
+ matched_count = 0
+
+ mismatches = {"mismatches": [], "input_source_only": [], "source_repo_only": []}
+
+ repo_unmatched = {path: data for path, data in repo_files.items()}
+ input_unmatched = {}
+
+ # Check for exact path and hash match
+ for input_path, input_data in input_files.items():
+ if input_path in repo_files:
+ if input_data["hash"] == repo_files[input_path]["hash"]:
+ matched_count += 1
+ else:
+ mismatches["mismatches"].append(f"[File] {input_path}")
+ del repo_unmatched[input_path]
+ else:
+ input_unmatched[input_path] = input_data
+
+ repo_by_hash_name = {}
+ for path, data in repo_unmatched.items():
+ key = (data["hash"], data["name"])
+ if key not in repo_by_hash_name:
+ repo_by_hash_name[key] = []
+ repo_by_hash_name[key].append(path)
+
+ still_unmatched_input = {}
+
+ # Check for files with the same hash and name but different paths
+ for input_path, input_data in input_unmatched.items():
+ hash_name_key = (input_data["hash"], input_data["name"])
+
+ if hash_name_key in repo_by_hash_name and repo_by_hash_name[hash_name_key]:
+ repo_match_path = repo_by_hash_name[hash_name_key].pop(0)
+ matched_count += 1
+ del repo_unmatched[repo_match_path]
+ else:
+ still_unmatched_input[input_path] = input_data
+
+ input_consolidated = consolidate_unmatched(
+ input_files.keys(), still_unmatched_input.keys()
+ )
+ for path, is_directory in input_consolidated:
+ item_type = "Directory" if is_directory else "File"
+ mismatches["input_source_only"].append(f"[{item_type}] {path}")
+
+ repo_consolidated = consolidate_unmatched(repo_files.keys(), repo_unmatched.keys())
+ for path, is_directory in repo_consolidated:
+ item_type = "Directory" if is_directory else "File"
+ mismatches["source_repo_only"].append(f"[{item_type}] {path}")
+
+ return matched_count, mismatches
+
+
+def count_total_files(directory):
+ """Recursively counts all files in a given directory."""
+ total_files = 0
+ for _, _, files in os.walk(directory):
+ total_files += len(files)
+ return total_files
+
+
+def update_comparison_summary(
+ project,
+ purl,
+ devel_codebase_dir,
+ src_repo_url,
+ package_name,
+ package_version,
+ matched_count,
+ mismatches,
+):
+ total_num_files = count_total_files(devel_codebase_dir)
+
+ package = project.discoveredpackages.filter(
+ name=package_name, version=package_version
+ ).first()
+
+ if package:
+ summary_dict = {
+ "input_source": str(purl),
+ "compare_source_repository_url": str(src_repo_url),
+ "total_matching_files": matched_count,
+ "total_files_in_source_crate": total_num_files,
+ "mismatches": mismatches,
+ }
+ package.update_extra_data({"comparison_summary": summary_dict})
+ else:
+ project.add_warning(
+ description=(
+ f"Could not find a discovered package matching {package_name} "
+ f"{package_version} to attach the summary."
+ )
+ )
+
+
+def fetch_inputs(purl):
+ """Fetch the source for the given input purl"""
+ purl_str = PackageURL.to_string(purl)
+ purl_src_path = fetch_path(purl_str)
+ if not purl_src_path:
+ err_msg = f"No source could be resolved for {purl}."
+ raise ValueError(err_msg)
+ return purl_src_path
+
+
+def fetch_path(purl):
+ """Fetch the purl and return the location of the fetched tarball"""
+ try:
+ return fetch.fetch_url(url=purl).path
+ except (ValueError, requests.RequestException) as e:
+ logger.warning("Failed to fetch package: %s - %s", purl, e)
+ return None
diff --git a/scanpipe/templates/scanpipe/package_list.html b/scanpipe/templates/scanpipe/package_list.html
index 90f917c245..206a66526e 100644
--- a/scanpipe/templates/scanpipe/package_list.html
+++ b/scanpipe/templates/scanpipe/package_list.html
@@ -34,6 +34,11 @@
{% endif %}
+ {% if package.extra_data.issues %}
+
+
+
+ {% endif %}
|
@@ -75,4 +80,4 @@
{% include 'scanpipe/includes/pagination.html' with page_obj=page_obj %}
{% endif %}
-{% endblock %}
\ No newline at end of file
+{% endblock %}
diff --git a/scanpipe/tests/pipes/test_rust.py b/scanpipe/tests/pipes/test_rust.py
new file mode 100644
index 0000000000..813383b18d
--- /dev/null
+++ b/scanpipe/tests/pipes/test_rust.py
@@ -0,0 +1,107 @@
+# 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 tempfile
+from pathlib import Path
+from unittest import mock
+
+from django.test import TestCase
+
+from scanpipe.pipes import rust
+
+
+class ScanPipeRustPipesTest(TestCase):
+ @mock.patch("pathlib.Path.rglob")
+ def test_get_cargo_toml_path_found(self, mock_rglob):
+ mock_cargo_path = Path("/mock/Cargo.toml")
+ mock_rglob.return_value = [mock_cargo_path]
+
+ found_path = rust.get_cargo_toml_path(Path("/mock"))
+ self.assertEqual(found_path, mock_cargo_path)
+ mock_rglob.assert_called_once_with("Cargo.toml")
+
+ @mock.patch("pathlib.Path.rglob")
+ def test_get_cargo_toml_path_not_found(self, mock_rglob):
+ mock_rglob.return_value = []
+
+ found_path = rust.get_cargo_toml_path(Path("/mock"))
+ self.assertIsNone(found_path)
+
+ @mock.patch("pathlib.Path.exists", return_value=True)
+ @mock.patch("pathlib.Path.open", new_callable=mock.mock_open)
+ @mock.patch("scanpipe.pipes.rust.tomllib.load")
+ def test_get_repository_value_from_cargo_toml_success(
+ self, mock_tomllib_load, mock_file_open, mock_exists
+ ):
+ mock_tomllib_load.return_value = {
+ "package": {"repository": "https://github.com/owner/repo"}
+ }
+ repo_url = rust.get_repository_value_from_cargo_toml("Cargo.toml")
+ self.assertEqual(repo_url, "https://github.com/owner/repo")
+
+ @mock.patch("pathlib.Path.exists", return_value=False)
+ def test_get_repository_value_from_cargo_toml_missing(self, mock_exists):
+ with self.assertRaises(FileNotFoundError):
+ rust.get_repository_value_from_cargo_toml("/nonexistent/Cargo.toml")
+
+ def test_check_input_and_return_purl_success(self):
+ mock_project = mock.Mock()
+ mock_project.inputsources.all.return_value = ["pkg:cargo/test@1.0.0"]
+
+ purl = rust.check_input_and_return_purl(mock_project)
+ self.assertEqual(purl.type, "cargo")
+ self.assertEqual(purl.name, "test")
+ self.assertEqual(purl.version, "1.0.0")
+
+ def test_check_input_and_return_purl_invalid_type(self):
+ mock_project = mock.Mock()
+ mock_project.inputsources.all.return_value = ["pkg:pypi/test@1.0.0"]
+
+ with self.assertRaises(ValueError):
+ rust.check_input_and_return_purl(mock_project)
+
+ def test_check_input_and_return_purl_missing_version(self):
+ mock_project = mock.Mock()
+ mock_project.inputsources.all.return_value = ["pkg:cargo/test"]
+
+ with self.assertRaises(ValueError):
+ rust.check_input_and_return_purl(mock_project)
+
+ @mock.patch("scanpipe.pipes.rust.run_command_safely")
+ def test_build_crates_success(self, mock_run_cmd):
+ with tempfile.TemporaryDirectory() as temp_dir:
+ base_path = Path(temp_dir)
+ cargo_path = base_path / "Cargo.toml"
+ cargo_path.touch()
+ other_file = base_path / "src_file.rs"
+ other_file.touch()
+
+ # Since run_command_safely is mocked, Cargo won't create the
+ # 'to' dir automatically
+ (base_path / "to").mkdir()
+
+ success = rust.build_crates(base_path, cargo_path)
+ self.assertTrue(success)
+ self.assertTrue((base_path / "to").exists())
+ self.assertTrue((base_path / "from").exists())
+ self.assertTrue((base_path / "from" / "src_file.rs").exists())
diff --git a/scanpipe/tests/pipes/test_utils.py b/scanpipe/tests/pipes/test_utils.py
new file mode 100644
index 0000000000..f679b31c91
--- /dev/null
+++ b/scanpipe/tests/pipes/test_utils.py
@@ -0,0 +1,314 @@
+# 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 tempfile
+from pathlib import Path
+from unittest import mock
+
+from django.test import TestCase
+
+from license_expression import Licensing
+
+from scanpipe.pipes import flag
+from scanpipe.pipes import utils
+
+
+class ScanPipeUtilsTest(TestCase):
+ def setUp(self):
+ self.licensing = Licensing()
+
+ @mock.patch("scanpipe.models.CodebaseResource")
+ @mock.patch("scanpipe.models.DiscoveredPackage")
+ @mock.patch("scanpipe.models.Project")
+ def test_validate_package_license_integrity_mismatch(
+ self, mock_project_class, mock_package_class, mock_resource_class
+ ):
+ mock_project = mock_project_class()
+ mock_package = mock_package_class()
+
+ mock_package.type = "pypi"
+ mock_package.package_uid = "pkg:pypi/test@1.0"
+ mock_package.get_declared_license_expression.return_value = "mit"
+ mock_package.datafile_paths = ["src/main.py"]
+ mock_package.extra_data = {}
+
+ mock_project.discoveredpackages.all.return_value = [mock_package]
+
+ mock_resource = mock_resource_class()
+ mock_resource.path = "src/main.py"
+ mock_resource.for_packages = ["pkg:pypi/test@1.0"]
+ mock_resource.detected_license_expression = "gpl-3.0"
+
+ mock_project.codebaseresources.has_license_expression.return_value = [
+ mock_resource
+ ]
+
+ mock_data_path = mock_resource_class()
+ mock_data_path.extra_data = {}
+ mock_project.codebaseresources.get.return_value = mock_data_path
+
+ utils.validate_package_license_integrity(mock_project)
+
+ package_update_args = mock_package.update_extra_data.call_args.args[0]
+ self.assertEqual(
+ package_update_args["issues"][0]["issue_type"], "License Mismatch"
+ )
+ self.assertEqual(
+ package_update_args["issues"][0]["detected_codebase_license"], "gpl-3.0"
+ )
+
+ mock_data_path.update.assert_called_once_with(status=flag.LICENSE_ISSUE)
+
+ def test_contains_ignore_pattern(self):
+ ignore_patterns = ["*test*", "*.sh"]
+ self.assertTrue(
+ utils.contains_ignore_pattern("src/test_main.py", ignore_patterns)
+ )
+ self.assertTrue(
+ utils.contains_ignore_pattern("scripts/build.sh", ignore_patterns)
+ )
+ self.assertFalse(utils.contains_ignore_pattern("src/main.py", ignore_patterns))
+
+ def test_filter_ignored_licenses(self):
+ exp1 = self.licensing.parse("mit")
+ self.assertEqual(
+ str(utils.filter_ignored_licenses(exp1, self.licensing)), "mit"
+ )
+
+ exp2 = self.licensing.parse("unknown")
+ self.assertIsNone(utils.filter_ignored_licenses(exp2, self.licensing))
+
+ exp3 = self.licensing.parse("mit AND unknown")
+ self.assertEqual(
+ str(utils.filter_ignored_licenses(exp3, self.licensing)), "mit"
+ )
+
+ exp4 = self.licensing.parse("unknown-spdx OR free-unknown")
+ self.assertIsNone(utils.filter_ignored_licenses(exp4, self.licensing))
+
+ def test_collect_detected_licenses(self):
+ mock_resource1 = mock.Mock()
+ mock_resource1.path = "src/main.py"
+ mock_resource1.for_packages = ["pkg:pypi/test@1.0"]
+ mock_resource1.detected_license_expression = "mit AND unknown"
+
+ mock_resource2 = mock.Mock()
+ mock_resource2.path = "test/test_main.py"
+ mock_resource2.for_packages = ["pkg:pypi/test@1.0"]
+ mock_resource2.detected_license_expression = "gpl-3.0"
+
+ mock_resource3 = mock.Mock()
+ mock_resource3.path = "src/other.py"
+ mock_resource3.for_packages = ["pkg:pypi/test@2.0"]
+ mock_resource3.detected_license_expression = "apache-2.0"
+
+ resources = [mock_resource1, mock_resource2, mock_resource3]
+ ignore_patterns = ["*test*"]
+
+ result = utils.collect_detected_licenses(
+ resources, ignore_patterns, package_uid="pkg:pypi/test@1.0"
+ )
+
+ self.assertEqual(result, ["(mit)"])
+
+ def test_get_url_netloc_namespace_and_name(self):
+ url = "https://github.com/aboutcode-org/scancode.io/"
+ netloc, namespace, name = utils.get_url_netloc_namespace_and_name(url)
+ self.assertEqual(netloc, "github.com")
+ self.assertEqual(namespace, "aboutcode-org")
+ self.assertEqual(name, "scancode.io")
+
+ url_web = "https://example.com/ns/project/index.html"
+ netloc, namespace, name = utils.get_url_netloc_namespace_and_name(url_web)
+ self.assertEqual(netloc, "example.com")
+ self.assertEqual(namespace, "ns")
+ self.assertEqual(name, "project")
+
+ @mock.patch("scanpipe.pipes.utils.fetch.fetch_url")
+ def test_download_src_repo_success(self, mock_fetch):
+ mock_fetch.return_value.path = "/test/downloaded_repo"
+ result = utils.download_src_repo("https://example.com/repo.zip")
+ self.assertEqual(result, "/test/downloaded_repo")
+
+ @mock.patch("scanpipe.pipes.utils.fetch.fetch_url")
+ def test_download_src_repo_failure(self, mock_fetch):
+ mock_fetch.side_effect = ValueError("Invalid URL")
+ result = utils.download_src_repo("invalid_url")
+ self.assertIsNone(result)
+
+ @mock.patch("scanpipe.pipes.utils.get_repo_download_url_by_package_type")
+ @mock.patch("scanpipe.pipes.utils.clarify_version_tag")
+ def test_get_download_url(self, mock_clarify, mock_get_repo_url):
+ mock_clarify.return_value = "v1.0.0"
+ mock_get_repo_url.return_value = (
+ "https://github.com/namespace/repo_name/archive/v1.0.0.zip"
+ )
+
+ url = "https://github.com/namespace/repo_name"
+ result = utils.get_download_url(url, "1.0.0")
+
+ self.assertEqual(
+ result, "https://github.com/namespace/repo_name/archive/v1.0.0.zip"
+ )
+
+ @mock.patch("scanpipe.pipes.utils.requests.get")
+ def test_clarify_version_tag(self, mock_get):
+ # Simulate 2 requests which the first returns 404 and the second
+ # returns 200
+ mock_get.side_effect = [mock.Mock(status_code=404), mock.Mock(status_code=200)]
+
+ # The function tries prefixes in order: ["", "v", "V", "release-",
+ # "RELEASE-", "v-", "V-"]
+ result = utils.clarify_version_tag("github", "namespace", "name", "1.0.0")
+
+ self.assertEqual(result, "v1.0.0")
+ self.assertEqual(mock_get.call_count, 2)
+
+ @mock.patch("scanpipe.pipes.utils.requests.get")
+ def test_github_pages_to_repo(self, mock_get):
+ mock_get.return_value = mock.Mock(status_code=200)
+ url = "https://krumpetpirate.github.io/AAXtoMP3/"
+ result = utils.github_pages_to_repo(url)
+ self.assertEqual(result, "https://github.com/krumpetpirate/AAXtoMP3")
+
+ # Failed mapping
+ mock_get.return_value = mock.Mock(status_code=404)
+ result_failed = utils.github_pages_to_repo(url)
+ self.assertIsNone(result_failed)
+
+ def test_get_all_files_and_count(self):
+ with tempfile.TemporaryDirectory() as tmp_dir:
+ file1_path = Path(tmp_dir) / "file1.txt"
+ file2_path = Path(tmp_dir) / "tmp" / "file2.txt"
+
+ file2_path.parent.mkdir()
+ # Use touch() to create empty files
+ file1_path.touch()
+ file2_path.touch()
+
+ self.assertEqual(utils.count_total_files(tmp_dir), 2)
+
+ file_map = utils.get_all_files(tmp_dir)
+ self.assertIn("file1.txt", file_map)
+ self.assertIn("tmp/file2.txt", file_map)
+ self.assertEqual(file_map["file1.txt"]["name"], "file1.txt")
+ self.assertIsNotNone(file_map["file1.txt"]["hash"])
+
+ def test_consolidate_unmatched(self):
+ all_files = ["src/main.py", "src/utils.py", "docs/readme.md", "docs/install.md"]
+ unmatched_files = ["docs/readme.md", "docs/install.md", "src/utils.py"]
+
+ result = utils.consolidate_unmatched(all_files, unmatched_files)
+
+ # 'docs' directory is entirely unmatched.
+ # 'src/utils.py' is unmatched, but 'src' has a matched file ('main.py').
+ expected = [("docs", True), ("src/utils.py", False)]
+ self.assertEqual(result, sorted(expected))
+
+ @mock.patch("scanpipe.pipes.utils.get_all_files")
+ def test_compare_directories(self, mock_get_all_files):
+ mock_get_all_files.side_effect = [
+ # input_source
+ {
+ "hello.py": {"hash": "123", "name": "hello.py"},
+ "world.py": {"hash": "456", "name": "world.py"},
+ },
+ # source_repo
+ {
+ "hello.py": {"hash": "123", "name": "hello.py"},
+ "universe.py": {"hash": "789", "name": "universe.py"},
+ },
+ ]
+
+ matched_count, mismatches = utils.compare_directories("input", "repo")
+ self.assertEqual(matched_count, 1)
+ self.assertIn("[File] world.py", mismatches["input_source_only"])
+ self.assertIn("[File] universe.py", mismatches["source_repo_only"])
+
+ @mock.patch("scanpipe.models.DiscoveredPackage")
+ @mock.patch("scanpipe.models.Project")
+ def test_update_comparison_summary_package_found(
+ self, mock_project_class, mock_package_class
+ ):
+ mock_project = mock_project_class()
+ mock_package = mock_package_class()
+
+ mock_project.discoveredpackages.filter.return_value.first.return_value = (
+ mock_package
+ )
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ utils.update_comparison_summary(
+ project=mock_project,
+ purl="pkg:pypi/testg@1.0",
+ devel_codebase_dir=temp_dir,
+ src_repo_url="https://github.com/test/test",
+ package_name="test",
+ package_version="1.0",
+ matched_count=3,
+ mismatches={
+ "mismatches": [],
+ "input_source_only": [],
+ "source_repo_only": [],
+ },
+ )
+
+ called_data = mock_package.update_extra_data.call_args.args[0]
+
+ self.assertIn("comparison_summary", called_data)
+ self.assertEqual(
+ called_data["comparison_summary"]["total_matching_files"], 3
+ )
+ self.assertEqual(
+ called_data["comparison_summary"]["input_source"], "pkg:pypi/testg@1.0"
+ )
+
+ def test_handle_operator_expression_and(self):
+ expr = self.licensing.parse("mit AND apache-2.0")
+ result = utils.handle_operator_expression(
+ expr, self.licensing, self.licensing.AND
+ )
+ self.assertEqual(str(result), "mit AND apache-2.0")
+
+ def test_handle_operator_expression_or(self):
+ expr = self.licensing.parse("mit OR bsd-3-clause")
+ result = utils.handle_operator_expression(
+ expr, self.licensing, self.licensing.OR
+ )
+ self.assertEqual(str(result), "mit OR bsd-3-clause")
+
+ def test_handle_operator_expression_filters_to_single_arg(self):
+ # 'unknown' gets filtered out to None, leaving only 'mit' (len == 1)
+ expr = self.licensing.parse("mit AND unknown")
+ result = utils.handle_operator_expression(
+ expr, self.licensing, self.licensing.AND
+ )
+ self.assertEqual(str(result), "mit")
+
+ def test_handle_operator_expression_all_filtered_out(self):
+ # Both 'unknown' and 'free-unknown' get filtered out, leaving empty args
+ expr = self.licensing.parse("unknown AND free-unknown")
+ result = utils.handle_operator_expression(
+ expr, self.licensing, self.licensing.AND
+ )
+ self.assertIsNone(result)
diff --git a/scanpipe/views.py b/scanpipe/views.py
index 962fde218c..de8816ee0d 100644
--- a/scanpipe/views.py
+++ b/scanpipe/views.py
@@ -1761,6 +1761,7 @@ def get_queryset(self):
"compliance_alert",
"copyright",
"affected_by_vulnerabilities",
+ "extra_data",
)
.with_resources_count()
.order_by_package_url()
|