diff --git a/Dockerfile b/Dockerfile index 55153b5eb6..75cd1d1214 100644 --- a/Dockerfile +++ b/Dockerfile @@ -149,9 +149,17 @@ RUN apt-get update \ libguestfs-tools \ linux-image-amd64 \ openjdk-21-jre-headless \ + nix-bin \ + zstd \ && apt-get clean \ && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* +# Single-user Nix for the "scan_nix_package" pipeline: the store is owned by the +# app user, and the build sandbox is not available in an unprivileged container. +RUN mkdir --parents /nix/store /nix/var/nix /etc/nix \ + && chown --recursive $APP_USER:$APP_USER /nix \ + && printf 'sandbox = false\nbuild-users-group =\n' > /etc/nix/nix.conf + # jadx (runs on the OpenJDK above) decompiles the Android .dex files into Java # sources for the "android_d2d" pipeline provided by android-inspector. ARG JADX_VERSION=1.5.6 diff --git a/docs/built-in-pipelines.rst b/docs/built-in-pipelines.rst index e5eb8bbf4b..7e7d743c18 100644 --- a/docs/built-in-pipelines.rst +++ b/docs/built-in-pipelines.rst @@ -299,6 +299,14 @@ Scan Maven Package :members: :member-order: bysource +.. _pipeline_scan_nix_package: + +Scan Nix Package +---------------- +.. autoclass:: scanpipe.pipelines.scan_nix_package.ScanNixPackage() + :members: + :member-order: bysource + Fetch Scores (addon) -------------------- .. warning:: diff --git a/docs/installation.rst b/docs/installation.rst index 798e8ec170..c28d6fba6e 100644 --- a/docs/installation.rst +++ b/docs/installation.rst @@ -50,7 +50,8 @@ Go to http://localhost/ on a web browser to **access the web UI**. .. note:: ``docker-compose.yml`` uses the ``full`` image of the ScanCode.io release, which - includes OpenJDK, jadx, and android-inspector for the Android analysis pipeline. + includes OpenJDK, jadx, and android-inspector for the Android analysis pipeline, + and Nix for the Nix package pipeline. An overview of the web application usage is available at :ref:`user_interface`. @@ -305,8 +306,9 @@ Restart the worker to apply them to pipeline runs:: .. note:: ``make run`` is a shortcut for ``docker compose -f docker-compose.yml -f docker-compose.dev.yml up``. - The development image is the ``core`` image: the VM image extraction and the - Android analysis features of the ``full`` image are not available. + The development image is the ``core`` image: the VM image extraction, the + Android analysis, and the Nix package features of the ``full`` image are not + available. Rebuild the development image after a dependency change with:: diff --git a/pyproject.toml b/pyproject.toml index 795fbaeb44..3b8e7dfd7d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -177,6 +177,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_nix_package = "scanpipe.pipelines.scan_nix_package:ScanNixPackage" scan_single_package = "scanpipe.pipelines.scan_single_package:ScanSinglePackage" scan_repo_health = "scanpipe.pipelines.scan_repo_health:ScanRepoHealth" diff --git a/scancodeio/static/js/project_form.js b/scancodeio/static/js/project_form.js index a23a13de69..1493d1c605 100644 --- a/scancodeio/static/js/project_form.js +++ b/scancodeio/static/js/project_form.js @@ -124,6 +124,7 @@ if (value.startsWith("docker:")) return "analyze_docker_image"; // Handle Package URL + if (value.startsWith("pkg:nix/")) return "scan_nix_package"; if (value.startsWith("pkg:")) return "scan_single_package"; // Handle SBOM file formats diff --git a/scanpipe/pipelines/scan_nix_package.py b/scanpipe/pipelines/scan_nix_package.py new file mode 100644 index 0000000000..97c2770788 --- /dev/null +++ b/scanpipe/pipelines/scan_nix_package.py @@ -0,0 +1,230 @@ +# 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 scanpipe.pipelines.deploy_to_develop import DeployToDevelop +from scanpipe.pipes import d2d +from scanpipe.pipes import nix + + +class ScanNixPackage(DeployToDevelop): + """ + Scan a nixpkgs package and compare its binary with its source. + + This pipeline takes a single nixpkgs PURL as input, for example: + ``pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux``. + The ``system`` qualifier is required. The nixpkgs commit is looked up from + the version, or can be provided with the ``commit`` qualifier. + + The source is fetched with the nixpkgs patches applied, and the binary is + downloaded from the Nix binary cache. A deployment to development analysis + runs between them, and the whole source is scanned to compare the detected + licenses with the license declared in nixpkgs. + + The package outputs holding only headers and documentation, such as ``dev``, + ``doc``, or ``man``, are not downloaded. A single output can be selected with + the ``output`` qualifier. + + When the source cannot be fetched, for a package without a ``src`` attribute, + the binary is analyzed alone and the license check is skipped. + + The binaries mapping is always enabled: ELF for a Linux ``system``, and + Mach-O for a macOS ``system``. The ELF mapping relies on the DWARF debug + paths, available only for the packages that provide a ``debug`` output. + + The source is evaluated for the Linux system running ScanCode.io: for a + macOS ``system``, the source does not include the macOS specific patches. + + Add the ``match_to_matchcode`` addon pipeline to the project to match the + source with MatchCode.io and find the code borrowed from other packages. + + The nixpkgs commit is checked with the GitHub API, limited to 60 requests + per hour without a token. Provide a token with the + ``SCANCODEIO_FETCH_HEADERS`` setting, for example: + ``api.github.com=Authorization=token ``. + + This pipeline requires the ``nix`` command, available in the full + ScanCode.io image. + """ + + download_inputs = False + + def __init__(self, run_instance): + super().__init__(run_instance) + binary_group = nix.get_binary_group(self.project) + if binary_group not in self.selected_groups: + self.selected_groups = [*self.selected_groups, binary_group] + + @classmethod + def get_availability(cls): + if missing_commands := nix.get_missing_commands(): + return ( + f"Missing commands: {', '.join(missing_commands)}. " + "Use the full ScanCode.io image." + ) + + @classmethod + def steps(cls): + excluded_steps = ( + cls.get_inputs, + cls.extract_inputs_to_codebase_directory, + cls.create_local_files_packages, + ) + d2d_steps = tuple( + step for step in super().steps() if step not in excluded_steps + ) + return ( + cls.get_nix_package, + cls.get_nix_package_metadata, + cls.fetch_nix_source, + cls.fetch_nix_binary, + *d2d_steps, + cls.create_nix_package, + cls.check_nix_package_license, + cls.collect_nix_garbage, + ) + + def get_nix_package(self): + """Validate the input Nix PURL and get its nixpkgs commit.""" + self.purl = nix.get_nix_purl(self.project) + self.commit = nix.get_nixpkgs_commit(self.purl) + self.log(f"nixpkgs commit: {self.commit}") + + def get_nix_package_metadata(self): + """Get the package metadata and binary store paths from nixpkgs.""" + self.metadata = nix.get_package_metadata( + attribute_path=self.purl.name, + commit=self.commit, + system=self.purl.qualifiers["system"], + cache_directory=self.project.tmp_path, + ) + self.selected_outputs = nix.get_selected_outputs( + outputs=self.metadata["outputs"], + selected_output=self.purl.qualifiers.get("output"), + ) + + version = self.metadata["version"] + if self.purl.version and self.purl.version != version: + self.project.add_warning( + description=( + f"The PURL version {self.purl.version} differs from the " + f"nixpkgs version {version} at commit {self.commit}." + ), + model=self.pipeline_name, + ) + + def fetch_nix_source(self): + """Fetch the source to ``from/``, with the nixpkgs patches applied.""" + try: + patch_error = nix.fetch_source( + attribute_path=self.purl.name, + commit=self.commit, + destination=self.project.codebase_path / d2d.FROM, + cache_directory=self.project.tmp_path, + ) + except nix.NixCommandError as error: + self.project.add_warning( + description=( + "The source cannot be fetched, the binary is analyzed alone " + "and the license check is skipped." + ), + model=self.pipeline_name, + details={"error": str(error)}, + ) + return + + if patch_error: + self.project.add_warning( + description=( + "The source with the nixpkgs patches cannot be built, the " + "upstream source is used instead." + ), + model=self.pipeline_name, + details={"error": patch_error}, + ) + + def fetch_nix_binary(self): + """Download the package outputs from the Nix binary cache to ``to/``.""" + missing_outputs = [] + for output, store_path in self.selected_outputs.items(): + is_fetched = nix.fetch_binary( + store_path=store_path, + destination=self.project.codebase_path / d2d.TO / output, + download_directory=self.project.tmp_path, + ) + if not is_fetched: + missing_outputs.append(output) + + if missing_outputs: + self.project.add_warning( + description=( + f"Outputs not available in the Nix binary cache: " + f"{', '.join(missing_outputs)}" + ), + model=self.pipeline_name, + ) + + def scan_mapped_from_for_files(self): + """ + Scan the ``from/`` files for copyrights, licenses, emails, and urls, + including the files not deployed in the binary. + """ + super().scan_mapped_from_for_files() + nix.scan_undeployed_from_files(self.project, progress_logger=self.log) + + def create_nix_package(self): + """Create the Nix package and assign the codebase resources to it.""" + self.package = nix.create_nix_package( + self.project, self.purl, self.metadata, self.commit + ) + nix.add_resources_to_package( + self.package, self.project.codebaseresources.files() + ) + + def check_nix_package_license(self): + """Store the licenses detected in the source but not declared in nixpkgs.""" + has_source = self.project.codebaseresources.from_codebase().files().exists() + if not has_source: + return + + declared_license = self.package.declared_license_expression + if not declared_license: + self.project.add_warning( + description="No declared license in nixpkgs, license check skipped.", + model=self.pipeline_name, + ) + return + + uncovered_licenses = nix.get_uncovered_license_keys( + self.project, declared_license + ) + self.package.update_extra_data({"uncovered_licenses": uncovered_licenses}) + + def collect_nix_garbage(self): + """Delete the Nix store paths not in use, such as the ones of this scan.""" + try: + nix.collect_garbage() + except nix.NixCommandError as error: + self.project.add_warning( + description="The Nix store garbage collection failed.", + model=self.pipeline_name, + details={"error": str(error)}, + ) diff --git a/scanpipe/pipes/nix.py b/scanpipe/pipes/nix.py new file mode 100644 index 0000000000..ad1b982eb7 --- /dev/null +++ b/scanpipe/pipes/nix.py @@ -0,0 +1,714 @@ +# 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 json +import os +import re +import shutil +import stat +import subprocess +from contextlib import suppress +from itertools import islice +from pathlib import Path +from urllib.parse import urlencode + +import requests +from license_expression import Licensing +from packageurl import PackageURL + +from scanpipe import pipes +from scanpipe.pipes import fetch +from scanpipe.pipes import flag +from scanpipe.pipes import scancode +from scanpipe.pipes.resolve import convert_spdx_expression + +REQUIRED_COMMANDS = ["nix-build", "nix-instantiate", "nix-store", "xz", "zstd"] +NIX_CACHE_URL = "https://cache.nixos.org" +DEVBOX_SEARCH_URL = "https://search.devbox.sh/v2/pkg" +NIXPKGS_API_URL = "https://api.github.com/repos/NixOS/nixpkgs" + +NIX_EVAL_TIMEOUT = 600 +NIX_BUILD_TIMEOUT = 1800 +DOWNLOAD_CHUNK_SIZE = 1024 * 1024 + +GITHUB_RATE_LIMIT_ERROR = ( + "The GitHub API rate limit is reached. Provide a GitHub token for " + "api.github.com with the SCANCODEIO_FETCH_HEADERS setting." +) + +# The PURL values end up in Nix expressions arguments, URLs, and file paths: +# only a strict set of characters is accepted. +ATTRIBUTE_PATH_REGEX = re.compile(r"[a-zA-Z0-9_][a-zA-Z0-9_\-+.]*") +COMMIT_REGEX = re.compile(r"[0-9a-f]{40}") +SYSTEM_REGEX = re.compile(r"[a-z0-9_]+-[a-z0-9_]+") +OUTPUT_REGEX = re.compile(r"[a-z][a-z0-9]*") + +# The nixpkgs licenses without SPDX id, such as "LicenseRef-nixos-unfree", are not +# known by ScanCode: a single one would turn the whole expression into "unknown". +NIXOS_LICENSE_REF_REGEX = re.compile(r"LicenseRef-nixos-[\w.+-]+") +UNKNOWN_LICENSE_REF = "LicenseRef-scancode-unknown-license-reference" + +# A ScanCode "or later" license key: a versioned license key followed by "-plus", +# such as "gpl-3.0-plus" for "gpl-3.0". The "m-plus" font license is not one. +OR_LATER_LICENSE_KEY_REGEX = re.compile(r"(?P.+-\d+(\.\d+)*)-plus") + +# The package outputs holding headers and documentation, not the deployed code. +NON_DEPLOYED_OUTPUTS = {"dev", "devdoc", "devman", "doc", "info", "man"} + +DECOMPRESS_COMMANDS = { + "none": ["cat"], + "xz": ["xz", "--decompress", "--stdout"], + "zstd": ["zstd", "--decompress", "--stdout"], +} + +# Licenses detected in the source that are not useful to compare with the +# declared license. +IGNORED_LICENSE_KEYS = { + "free-unknown", + "unknown", + "unknown-license-reference", + "unknown-spdx", +} + +# The user input is passed as `--argstr` values, never interpolated in the +# expression. `fetchTarball` downloads nixpkgs at the given commit, no local +# clone is required. +NIXPKGS_EXPRESSION = """ +{ commit, attributePath, system ? builtins.currentSystem }: +let + nixpkgs = fetchTarball "https://github.com/NixOS/nixpkgs/archive/${commit}.tar.gz"; + pkgs = import nixpkgs { + inherit system; + config = { + allowUnfree = true; + allowBroken = true; + allowUnsupportedSystem = true; + allowInsecurePredicate = _: true; + }; + }; + package = pkgs.lib.getAttrFromPath (pkgs.lib.splitString "." attributePath) pkgs; +in +""" + +METADATA_EXPRESSION = ( + NIXPKGS_EXPRESSION + + """ +let + meta = package.meta or { }; + # Recent nixpkgs combine licenses with operators, such as "OR [ mit asl20 ]", + # converted by toSPDX. A license without SPDX id is a "LicenseRef-nixos-" one. + getLicenseExpression = license: + if builtins.isAttrs license && license ? licenseType + then pkgs.lib.licenses.toSPDX license + else if builtins.isAttrs license + then license.spdxId or "LicenseRef-nixos-${license.shortName or "unknown"}" + else if builtins.isString license then license + else null; + licenseExpressions = map getLicenseExpression (pkgs.lib.toList (meta.license or [ ])); + getOutputPath = output: { name = output; value = package.${output}.outPath; }; +in +{ + version = package.version or ""; + description = meta.description or ""; + homepage = meta.homepage or ""; + licenses = builtins.filter (expression: expression != null) licenseExpressions; + outputs = builtins.listToAttrs (map getOutputPath (package.outputs or [ "out" ])); +} +""" +) + +PATCHED_SOURCE_EXPRESSION = ( + NIXPKGS_EXPRESSION + + """ +pkgs.applyPatches { + name = "source"; + src = package.src; + patches = package.patches or [ ]; + prePatch = package.prePatch or ""; + postPatch = package.postPatch or ""; +} +""" +) + +UPSTREAM_SOURCE_EXPRESSION = NIXPKGS_EXPRESSION + "package.src" + + +class NixCommandError(Exception): + pass + + +def get_missing_commands(): + """Return the list of commands required by the Nix pipeline not installed.""" + return [command for command in REQUIRED_COMMANDS if not shutil.which(command)] + + +def get_nix_purl(project): + """Return the Nix PURL provided as the unique input of the ``project``.""" + input_sources = project.inputsources.all() + if len(input_sources) != 1: + raise ValueError("Only 1 Nix PURL is accepted as input.") + + purl = PackageURL.from_string(input_sources[0].download_url) + validate_nix_purl(purl) + return purl + + +def validate_nix_purl(purl): + """Raise a ValueError if the ``purl`` is not a supported Nix PURL.""" + if purl.type != "nix" or purl.namespace != "nixpkgs": + raise ValueError("Only pkg:nix/nixpkgs/ PURLs are supported.") + + if not ATTRIBUTE_PATH_REGEX.fullmatch(purl.name): + raise ValueError(f"Invalid Nix package name: {purl.name!r}") + + qualifiers = purl.qualifiers or {} + system = qualifiers.get("system", "") + if not SYSTEM_REGEX.fullmatch(system): + raise ValueError( + "A valid 'system' qualifier is required, for example: system=x86_64-linux" + ) + + commit = qualifiers.get("commit", "") + if commit and not COMMIT_REGEX.fullmatch(commit): + raise ValueError(f"Invalid commit: {commit!r}, a 40 characters SHA-1.") + + if not commit and not purl.version: + raise ValueError("A version or a 'commit' qualifier is required.") + + output = qualifiers.get("output", "") + if output and not OUTPUT_REGEX.fullmatch(output): + raise ValueError(f"Invalid output: {output!r}") + + +def get_binary_group(project): + """ + Return the optional steps group mapping the binaries of the ``project`` Nix + PURL input: "MacOS" for a macOS ``system`` qualifier, or else "Elf". The + input is validated later by the pipeline. + """ + input_source = project.inputsources.first() + if not input_source: + return "Elf" + + system = "" + with suppress(ValueError): + purl = PackageURL.from_string(input_source.download_url) + system = purl.qualifiers.get("system", "") + + if system.endswith("-darwin"): + return "MacOS" + return "Elf" + + +def get_nixpkgs_commit(purl): + """ + Return the nixpkgs commit of the ``purl``, from its qualifiers or from the + package releases listed by search.devbox.sh, after checking that the commit + belongs to the official NixOS/nixpkgs history. + """ + commit = purl.qualifiers.get("commit") or find_release_commit(purl) + + if not COMMIT_REGEX.fullmatch(commit): + raise ValueError(f"Invalid commit: {commit!r}") + + if not is_official_nixpkgs_commit(commit): + raise ValueError( + f"The commit {commit} is not part of the official NixOS/nixpkgs " + "master or release-* branches." + ) + + return commit + + +def get_devbox_releases(package_name): + """ + Return the ``package_name`` releases listed by search.devbox.sh, or an empty + list when the package is unknown. + """ + query = urlencode({"name": package_name}) + session = fetch.get_request_session(DEVBOX_SEARCH_URL) + response = session.get( + f"{DEVBOX_SEARCH_URL}?{query}", timeout=fetch.HTTP_REQUEST_TIMEOUT + ) + if response.status_code == 404: + return [] + + response.raise_for_status() + return response.json().get("releases") or [] + + +def find_release_commit(purl): + """Return the nixpkgs commit of the ``purl`` version and system on devbox.""" + system = purl.qualifiers["system"] + for release in get_devbox_releases(purl.name): + if release.get("version") != purl.version: + continue + for platform in release.get("platforms") or []: + if platform.get("system") == system: + return platform.get("commit_hash", "") + + raise ValueError( + f"No nixpkgs commit found for {purl.name}@{purl.version} on {system}. " + "Provide the nixpkgs commit with the 'commit' qualifier." + ) + + +def is_official_nixpkgs_commit(commit): + """ + Return True if the ``commit`` is reachable from the master branch, or from + a release-* branch for backports. + A fork commit can be downloaded through the official repository URL, so the + commit is checked against the official branches. + """ + session = fetch.get_request_session(NIXPKGS_API_URL) + if is_commit_on_branch(session, commit, "master"): + return True + + release_branches = get_release_branches(session) + return any( + is_commit_on_branch(session, commit, branch) for branch in release_branches + ) + + +def check_github_response(response): + """ + Raise an HTTPError for an error ``response`` of the GitHub API, with a hint + to provide a token when the rate limit is reached. + """ + is_rate_limited = ( + response.status_code == 429 + or response.headers.get("x-ratelimit-remaining") == "0" + ) + if is_rate_limited: + raise requests.HTTPError(GITHUB_RATE_LIMIT_ERROR, response=response) + + response.raise_for_status() + + +def is_commit_on_branch(session, commit, branch): + """Return True if the ``commit`` is an ancestor of, or equal to, the ``branch``.""" + compare_url = f"{NIXPKGS_API_URL}/compare/{commit}...{branch}" + response = session.get(compare_url, timeout=fetch.HTTP_REQUEST_TIMEOUT) + if response.status_code == 404: + return False + + check_github_response(response) + # "ahead" means the branch has commits after the commit. + return response.json().get("status") in ("ahead", "identical") + + +def get_release_branches(session): + """Return the nixpkgs release-* branch names, most recent first.""" + url = f"{NIXPKGS_API_URL}/git/matching-refs/heads/release-?per_page=100" + branches = [] + while url: + response = session.get(url, timeout=fetch.HTTP_REQUEST_TIMEOUT) + check_github_response(response) + branches.extend( + ref["ref"].removeprefix("refs/heads/") for ref in response.json() + ) + url = response.links.get("next", {}).get("url") + + return sorted(branches, reverse=True) + + +def run_nix_command(arguments, timeout, cache_directory=None): + """ + Run a Nix command and return its standard output. The nixpkgs tarballs are + cached in the ``cache_directory`` when provided, instead of the user cache. + """ + environment = None + if cache_directory: + environment = {**os.environ, "XDG_CACHE_HOME": str(cache_directory)} + + try: + completed_process = subprocess.run( # noqa: S603 + arguments, + capture_output=True, + text=True, + timeout=timeout, + env=environment, + ) + except subprocess.TimeoutExpired as error: + raise NixCommandError( + f"{arguments[0]} timed out after {timeout} seconds" + ) from error + + if completed_process.returncode: + error_lines = completed_process.stderr.strip().splitlines() + raise NixCommandError("\n".join(error_lines[-20:])) + + return completed_process.stdout + + +def get_nix_expression_arguments(attribute_path, commit, system=None): + """Return the ``--argstr`` arguments of the nixpkgs expressions.""" + arguments = [ + "--argstr", + "attributePath", + attribute_path, + "--argstr", + "commit", + commit, + ] + if system: + arguments.extend(["--argstr", "system", system]) + return arguments + + +def get_package_metadata(attribute_path, commit, system, cache_directory=None): + """ + Return the version, description, homepage, licenses SPDX expressions, and the + store path of each output of the package for the target ``system``. + """ + arguments = [ + "nix-instantiate", + "--eval", + "--strict", + "--json", + *get_nix_expression_arguments(attribute_path, commit, system), + "--expr", + METADATA_EXPRESSION, + ] + output = run_nix_command( + arguments, timeout=NIX_EVAL_TIMEOUT, cache_directory=cache_directory + ) + return json.loads(output) + + +def get_selected_outputs(outputs, selected_output=None): + """ + Return the ``outputs`` to download: the ``selected_output`` when provided, or + else the outputs holding deployed code, or else all the outputs. + Raise a ValueError if the ``selected_output`` is not one of the ``outputs``. + """ + if not selected_output: + deployed_outputs = { + output: store_path + for output, store_path in outputs.items() + if output not in NON_DEPLOYED_OUTPUTS + } + return deployed_outputs or outputs + + if selected_output not in outputs: + available_outputs = ", ".join(sorted(outputs)) + raise ValueError( + f"The output {selected_output!r} does not exist, " + f"available outputs: {available_outputs}" + ) + + return {selected_output: outputs[selected_output]} + + +def build_source(expression, attribute_path, commit, cache_directory=None): + """Build the source ``expression`` and return its store path.""" + arguments = [ + "nix-build", + "--no-out-link", + *get_nix_expression_arguments(attribute_path, commit), + "--expr", + expression, + ] + output = run_nix_command( + arguments, timeout=NIX_BUILD_TIMEOUT, cache_directory=cache_directory + ) + return output.strip().splitlines()[-1] + + +def fetch_source(attribute_path, commit, destination, cache_directory=None): + """ + Copy the package source, with the nixpkgs patches applied, to ``destination``. + Fall back to the upstream source when the patched source cannot be built, for + a patch or an evaluation error, and return this error, or an empty string. + + The source is evaluated for the system running ScanCode.io: the patches + are the same for most systems. + """ + patch_error = "" + try: + store_path = build_source( + PATCHED_SOURCE_EXPRESSION, attribute_path, commit, cache_directory + ) + except NixCommandError as error: + patch_error = str(error) + store_path = build_source( + UPSTREAM_SOURCE_EXPRESSION, attribute_path, commit, cache_directory + ) + + copy_store_path(store_path, destination) + return patch_error + + +def copy_store_path(store_path, destination): + """ + Copy the ``store_path`` content to ``destination``. A store path can be a + directory or a single file, such as a source archive. + """ + store_path = Path(store_path) + destination.mkdir(parents=True, exist_ok=True) + + if store_path.is_dir(): + shutil.copytree(store_path, destination, symlinks=True, dirs_exist_ok=True) + else: + archive_name = store_path.name.split("-", 1)[-1] + shutil.copyfile(store_path, destination / archive_name) + + make_writable(destination) + + +def make_writable(directory): + """ + Add the write permission on all the ``directory`` content, as the Nix store + files are read-only and the project work directory must remain deletable. + """ + for path in [directory, *directory.rglob("*")]: + if path.is_symlink(): + continue + path.chmod(path.stat().st_mode | stat.S_IWUSR) + + +def get_narinfo(store_path): + """ + Return the narinfo entries of the ``store_path`` from the Nix binary cache, + or an empty dict when the store path is not available in the cache. + """ + store_hash = Path(store_path).name.split("-", 1)[0] + narinfo_url = f"{NIX_CACHE_URL}/{store_hash}.narinfo" + session = fetch.get_request_session(narinfo_url) + response = session.get(narinfo_url, timeout=fetch.HTTP_REQUEST_TIMEOUT) + if response.status_code == 404: + return {} + + response.raise_for_status() + entries = [line.split(": ", 1) for line in response.text.splitlines()] + return {entry[0]: entry[1] for entry in entries if len(entry) == 2} + + +def fetch_binary(store_path, destination, download_directory): + """ + Download the ``store_path`` from the Nix binary cache and extract it to + ``destination``. Return False when the store path is not in the cache. + """ + narinfo = get_narinfo(store_path) + if not narinfo: + return False + + nar_location = Path(download_directory) / f"{Path(store_path).name}.nar" + download_file(f"{NIX_CACHE_URL}/{narinfo['URL']}", nar_location) + restore_nar(nar_location, narinfo.get("Compression", "none"), destination) + nar_location.unlink() + return True + + +def download_file(url, file_location): + """ + Download the ``url`` to ``file_location`` by chunks, as a NAR archive can be + larger than the available memory. + """ + session = fetch.get_request_session(url) + with session.get(url, stream=True, timeout=fetch.HTTP_REQUEST_TIMEOUT) as response: + response.raise_for_status() + chunks = response.iter_content(chunk_size=DOWNLOAD_CHUNK_SIZE) + with open(file_location, "wb") as downloaded_file: + downloaded_file.writelines(chunks) + + +def restore_nar(nar_location, compression, destination): + """Extract the ``compression`` compressed NAR archive to ``destination``.""" + decompress_command = DECOMPRESS_COMMANDS.get(compression) + if not decompress_command: + raise ValueError(f"Unsupported NAR compression: {compression}") + + destination.parent.mkdir(parents=True, exist_ok=True) + with ( + open(nar_location, "rb") as nar_file, + subprocess.Popen( # noqa: S603 + decompress_command, + stdin=nar_file, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) as decompress_process, + ): + try: + restore_process = subprocess.run( # noqa: S603 + ["nix-store", "--restore", str(destination)], # noqa: S607 + stdin=decompress_process.stdout, + capture_output=True, + text=True, + timeout=NIX_BUILD_TIMEOUT, + ) + except subprocess.TimeoutExpired as error: + decompress_process.kill() + raise NixCommandError( + f"Cannot extract {nar_location}: timed out after " + f"{NIX_BUILD_TIMEOUT} seconds" + ) from error + + # Closing the pipe stops the decompression when the restore failed. + decompress_process.stdout.close() + decompress_error = decompress_process.stderr.read().decode() + decompress_process.wait() + + if decompress_process.returncode or restore_process.returncode: + extract_errors = [decompress_error.strip(), restore_process.stderr.strip()] + extract_error = "\n".join(error for error in extract_errors if error) + raise NixCommandError(f"Cannot extract {nar_location}: {extract_error}") + + make_writable(destination) + + +def collect_garbage(): + """ + Delete the Nix store paths not used by a garbage collector root, such as the + nixpkgs and the sources fetched for a scan, so the disk usage does not grow + across scans. The installed Nix packages are kept, as their profiles are roots. + """ + run_nix_command(["nix-store", "--gc"], timeout=NIX_BUILD_TIMEOUT) + + +def combine_license_expressions(license_expressions): + """ + Return the SPDX ``license_expressions`` combined with AND, the compound ones + in parentheses. + """ + if len(license_expressions) == 1: + return license_expressions[0] + + return " AND ".join( + f"({expression})" if " " in expression else expression + for expression in license_expressions + ) + + +def convert_nixpkgs_license_expression(license_expression_spdx): + """ + Return the ScanCode license expression of the nixpkgs SPDX + ``license_expression_spdx``, the nixpkgs licenses without SPDX id converted + to unknown license references. + """ + known_license_expression_spdx = NIXOS_LICENSE_REF_REGEX.sub( + UNKNOWN_LICENSE_REF, license_expression_spdx + ) + return convert_spdx_expression(known_license_expression_spdx) + + +def create_nix_package(project, purl, metadata, commit): + """ + Create the Nix package of the ``project`` from the nixpkgs ``metadata``, with + the nixpkgs ``commit`` and the system it was evaluated for. + """ + declared_license_spdx = combine_license_expressions(metadata["licenses"]) + declared_license = "" + if declared_license_spdx: + declared_license = convert_nixpkgs_license_expression(declared_license_spdx) + + homepage = metadata["homepage"] + if isinstance(homepage, list): + homepage = homepage[0] if homepage else "" + + package_data = { + "type": purl.type, + "namespace": purl.namespace, + "name": purl.name, + "version": metadata["version"] or purl.version, + "declared_license_expression": declared_license, + "declared_license_expression_spdx": declared_license_spdx, + "homepage_url": homepage, + "description": metadata["description"], + "datasource_id": "nixpkgs_meta", + "extra_data": { + "nixpkgs_commit": commit, + "system": purl.qualifiers["system"], + }, + } + return pipes.update_or_create_package(project, package_data) + + +def add_resources_to_package(package, resource_qs, batch_size=2000): + """Assign the ``resource_qs`` resources to the ``package``, by batches.""" + through_model = package.codebase_resources.through + resource_ids = resource_qs.values_list("id", flat=True).iterator() + + while batch := list(islice(resource_ids, batch_size)): + through_model.objects.bulk_create( + [ + through_model( + discoveredpackage_id=package.pk, + codebaseresource_id=resource_id, + ) + for resource_id in batch + ], + ignore_conflicts=True, + ) + + +def scan_undeployed_from_files(project, progress_logger=None): + """ + Scan the ``from/`` files not deployed in the binary, keeping their + ``not-deployed`` status. The license check needs the whole source, including + the legal files that are usually not deployed. + """ + undeployed_files = ( + project.codebaseresources.files() + .from_codebase() + .filter(status=flag.NOT_DEPLOYED) + ) + undeployed_file_ids = list(undeployed_files.values_list("id", flat=True)) + + scancode.scan_for_files(project, undeployed_files, progress_logger=progress_logger) + project.codebaseresources.filter( + id__in=undeployed_file_ids, status=flag.SCANNED + ).update(status=flag.NOT_DEPLOYED) + + +def get_base_license_key(license_key): + """Return the ``license_key`` without its "or later" suffix.""" + if or_later_match := OR_LATER_LICENSE_KEY_REGEX.fullmatch(license_key): + return or_later_match.group("base_key") + return license_key + + +def get_uncovered_license_keys(project, declared_license_expression): + """ + Return the sorted license keys detected in the ``from/`` source that are not + part of the ``declared_license_expression``. A license and its "or later" + variant cover each other, as the GPL text of a COPYING file is detected + without the "or later" notice of the source headers. + """ + licensing = Licensing() + declared_keys = set(licensing.license_keys(declared_license_expression)) + + from_resources = project.codebaseresources.from_codebase().has_license_expression() + detected_expressions = from_resources.values_list( + "detected_license_expression", flat=True + ).distinct() + + detected_keys = set() + for detected_expression in detected_expressions.iterator(): + detected_keys.update(licensing.license_keys(detected_expression)) + + declared_base_keys = {get_base_license_key(key) for key in declared_keys} + return sorted( + key + for key in detected_keys - IGNORED_LICENSE_KEYS + if get_base_license_key(key) not in declared_base_keys + ) diff --git a/scanpipe/tests/data/nix/binary/doc/hello.c b/scanpipe/tests/data/nix/binary/doc/hello.c new file mode 100644 index 0000000000..3324f318c4 --- /dev/null +++ b/scanpipe/tests/data/nix/binary/doc/hello.c @@ -0,0 +1,7 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +#include + +int main(void) { + printf("Hello, world!\n"); + return 0; +} diff --git a/scanpipe/tests/data/nix/binary/hello b/scanpipe/tests/data/nix/binary/hello new file mode 100644 index 0000000000..a637a3a980 --- /dev/null +++ b/scanpipe/tests/data/nix/binary/hello @@ -0,0 +1 @@ +ELF-placeholder diff --git a/scanpipe/tests/data/nix/devbox-hello.json b/scanpipe/tests/data/nix/devbox-hello.json new file mode 100644 index 0000000000..71ffff2818 --- /dev/null +++ b/scanpipe/tests/data/nix/devbox-hello.json @@ -0,0 +1,46 @@ +{ + "name": "hello", + "summary": "Program that produces a familiar, friendly greeting", + "homepage_url": "https://www.gnu.org/software/hello/manual/", + "license": "GPL-3.0-or-later", + "releases": [ + { + "version": "2.12.3", + "last_updated": "2026-10-02T22:55:44Z", + "platforms": [ + { + "arch": "x86-64", + "os": "Linux", + "system": "x86_64-linux", + "attribute_path": "hello", + "commit_hash": "55ba7f49ef2962b42cbd126522b7df5f95037679", + "date": "2026-10-02T22:55:44Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/5z2yp3ysx8476c8g5w25b0smlgkjvaq3-hello-2.12.3", + "default": true + } + ] + }, + { + "arch": "arm64", + "os": "Linux", + "system": "aarch64-linux", + "attribute_path": "hello", + "commit_hash": "55ba7f49ef2962b42cbd126522b7df5f95037679", + "date": "2026-10-02T22:55:44Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/0xy6jlccm8kfpwmhkxabz3638lchyi74-hello-2.12.3", + "default": true + } + ] + } + ], + "platforms_summary": "Linux and macOS", + "outputs_summary": "" + } + ] +} diff --git a/scanpipe/tests/data/nix/hello.narinfo b/scanpipe/tests/data/nix/hello.narinfo new file mode 100644 index 0000000000..023c1d4680 --- /dev/null +++ b/scanpipe/tests/data/nix/hello.narinfo @@ -0,0 +1,10 @@ +StorePath: /nix/store/5z2yp3ysx8476c8g5w25b0smlgkjvaq3-hello-2.12.3 +URL: nar/1k684wya73qd387kbf1pdpzb986qalk7c7jbnh6ib5846i6q18yc.nar.zst +Compression: zstd +FileHash: sha256:13mnhcrnl350izxg8d614qhhwiwaa3acfn6yz0gxxqhy3xg593n2 +FileSize: 76021 +NarHash: sha256:1k684wya73qd387kbf1pdpzb986qalk7c7jbnh6ib5846i6q18yc +NarSize: 279592 +References: 5z2yp3ysx8476c8g5w25b0smlgkjvaq3-hello-2.12.3 h4wfwic161kxrr74jlzla5lsm28hgary-glibc-2.44-25 +Deriver: cnraiw0hbdl6jzdhbkn8dah2hqj7xgij-hello-2.12.3.drv +Sig: cache.nixos.org-1:6SW6CfeVfKkjds7I1NQ4vVuBfk+97bXw2pvGZjw+LNUptSvz70MfGHSAjbGQ48Mt++oobnSPZNEmxhE8gFb+CQ== diff --git a/scanpipe/tests/data/nix/source/src/hello.c b/scanpipe/tests/data/nix/source/src/hello.c new file mode 100644 index 0000000000..3324f318c4 --- /dev/null +++ b/scanpipe/tests/data/nix/source/src/hello.c @@ -0,0 +1,7 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +#include + +int main(void) { + printf("Hello, world!\n"); + return 0; +} diff --git a/scanpipe/tests/data/nix/source/src/helper.c b/scanpipe/tests/data/nix/source/src/helper.c new file mode 100644 index 0000000000..2c6f90b5e0 --- /dev/null +++ b/scanpipe/tests/data/nix/source/src/helper.c @@ -0,0 +1,4 @@ +// SPDX-License-Identifier: MIT +int add(int first, int second) { + return first + second; +} diff --git a/scanpipe/tests/pipes/test_nix.py b/scanpipe/tests/pipes/test_nix.py new file mode 100644 index 0000000000..2927a48c39 --- /dev/null +++ b/scanpipe/tests/pipes/test_nix.py @@ -0,0 +1,688 @@ +# 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 json +import os +import subprocess +import tempfile +from pathlib import Path +from unittest import mock + +from django.test import TestCase + +import requests +from packageurl import PackageURL + +from scanpipe.pipes import flag +from scanpipe.pipes import nix +from scanpipe.tests import make_project +from scanpipe.tests import make_resource_file + +COMMIT = "55ba7f49ef2962b42cbd126522b7df5f95037679" + + +def make_response(status_code=200, json_data=None, text="", links=None, headers=None): + response = mock.Mock( + status_code=status_code, text=text, links=links or {}, headers=headers or {} + ) + response.json.return_value = json_data + return response + + +class ScanPipeNixPipesTest(TestCase): + data = Path(__file__).parent.parent / "data" / "nix" + + def setUp(self): + self.project = make_project() + + @mock.patch("scanpipe.pipes.nix.shutil.which") + def test_scanpipe_nix_get_missing_commands(self, mock_which): + def which(command): + if command != "zstd": + return f"/usr/bin/{command}" + + mock_which.side_effect = which + self.assertEqual(["zstd"], nix.get_missing_commands()) + + def test_scanpipe_nix_get_nix_purl(self): + purl = f"pkg:nix/nixpkgs/hello@2.12.3?commit={COMMIT}&system=x86_64-linux" + self.project.add_input_source(download_url=purl) + self.assertEqual(PackageURL.from_string(purl), nix.get_nix_purl(self.project)) + + self.project.add_input_source(download_url="pkg:nix/nixpkgs/other@1.0") + with self.assertRaisesMessage(ValueError, "Only 1 Nix PURL is accepted"): + nix.get_nix_purl(self.project) + + def test_scanpipe_nix_get_binary_group(self): + self.assertEqual("Elf", nix.get_binary_group(self.project)) + + cases = [ + ("pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux", "Elf"), + ("pkg:nix/nixpkgs/hello@2.12.3?system=aarch64-darwin", "MacOS"), + ("pkg:nix/nixpkgs/hello@2.12.3", "Elf"), + ("not a purl", "Elf"), + ] + for download_url, expected in cases: + with self.subTest(download_url=download_url): + project = make_project() + project.add_input_source(download_url=download_url) + self.assertEqual(expected, nix.get_binary_group(project)) + + def test_scanpipe_nix_validate_nix_purl(self): + valid_purls = [ + "pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux", + f"pkg:nix/nixpkgs/haskellPackages.aeson?system=x86_64-linux&commit={COMMIT}", + "pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux&output=man", + ] + for purl in valid_purls: + with self.subTest(purl=purl): + nix.validate_nix_purl(PackageURL.from_string(purl)) + + invalid_purls = [ + ("pkg:npm/hello@1.0?system=x86_64-linux", "Only pkg:nix/nixpkgs/"), + ("pkg:nix/other/hello@1.0?system=x86_64-linux", "Only pkg:nix/nixpkgs/"), + ("pkg:nix/nixpkgs/.hello@1.0?system=x86_64-linux", "Invalid Nix package"), + ("pkg:nix/nixpkgs/hello@1.0", "A valid 'system' qualifier"), + ("pkg:nix/nixpkgs/hello@1.0?system=x86_64-linux%0A", "A valid 'system'"), + ( + "pkg:nix/nixpkgs/hello@1.0?system=x86_64-linux&commit=abc", + "Invalid commit", + ), + ("pkg:nix/nixpkgs/hello?system=x86_64-linux", "A version or a 'commit'"), + ( + "pkg:nix/nixpkgs/hello@1.0?system=x86_64-linux&output=..", + "Invalid output", + ), + ] + for purl, expected_message in invalid_purls: + with self.subTest(purl=purl): + with self.assertRaisesMessage(ValueError, expected_message): + nix.validate_nix_purl(PackageURL.from_string(purl)) + + @mock.patch("scanpipe.pipes.nix.is_official_nixpkgs_commit") + @mock.patch("scanpipe.pipes.nix.find_release_commit") + def test_scanpipe_nix_get_nixpkgs_commit(self, mock_find, mock_official): + mock_official.return_value = True + purl = PackageURL.from_string( + f"pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux&commit={COMMIT}" + ) + self.assertEqual(COMMIT, nix.get_nixpkgs_commit(purl)) + mock_find.assert_not_called() + + purl = PackageURL.from_string( + "pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux" + ) + mock_find.return_value = COMMIT + self.assertEqual(COMMIT, nix.get_nixpkgs_commit(purl)) + + mock_find.return_value = "not-a-commit" + with self.assertRaisesMessage(ValueError, "Invalid commit"): + nix.get_nixpkgs_commit(purl) + + mock_find.return_value = COMMIT + mock_official.return_value = False + with self.assertRaisesMessage(ValueError, "is not part of the official"): + nix.get_nixpkgs_commit(purl) + + @mock.patch("scanpipe.pipes.nix.fetch.get_request_session") + def test_scanpipe_nix_get_devbox_releases(self, mock_session): + devbox_data = json.loads((self.data / "devbox-hello.json").read_text()) + mock_session.return_value.get.return_value = make_response( + json_data=devbox_data + ) + self.assertEqual(devbox_data["releases"], nix.get_devbox_releases("hello")) + requested_url = mock_session.return_value.get.call_args.args[0] + self.assertEqual("https://search.devbox.sh/v2/pkg?name=hello", requested_url) + + mock_session.return_value.get.return_value = make_response(status_code=404) + self.assertEqual([], nix.get_devbox_releases("unknown")) + + @mock.patch("scanpipe.pipes.nix.get_devbox_releases") + def test_scanpipe_nix_find_release_commit(self, mock_releases): + devbox_data = json.loads((self.data / "devbox-hello.json").read_text()) + mock_releases.return_value = devbox_data["releases"] + + purl = PackageURL.from_string( + "pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux" + ) + self.assertEqual(COMMIT, nix.find_release_commit(purl)) + + purl = PackageURL.from_string("pkg:nix/nixpkgs/hello@1.0?system=x86_64-linux") + with self.assertRaisesMessage(ValueError, "No nixpkgs commit found"): + nix.find_release_commit(purl) + + mock_releases.return_value = [] + purl = PackageURL.from_string("pkg:nix/nixpkgs/unknown@1.0?system=x86_64-linux") + with self.assertRaisesMessage(ValueError, "No nixpkgs commit found"): + nix.find_release_commit(purl) + + @mock.patch("scanpipe.pipes.nix.get_release_branches") + @mock.patch("scanpipe.pipes.nix.is_commit_on_branch") + @mock.patch("scanpipe.pipes.nix.fetch.get_request_session") + def test_scanpipe_nix_is_official_nixpkgs_commit( + self, mock_session, mock_on_branch, mock_release_branches + ): + mock_on_branch.return_value = True + self.assertTrue(nix.is_official_nixpkgs_commit(COMMIT)) + mock_release_branches.assert_not_called() + + def on_release_branch(session, commit, branch): + return branch == "release-25.05" + + mock_on_branch.side_effect = on_release_branch + mock_release_branches.return_value = ["release-25.11", "release-25.05"] + self.assertTrue(nix.is_official_nixpkgs_commit(COMMIT)) + + mock_release_branches.return_value = ["release-25.11"] + self.assertFalse(nix.is_official_nixpkgs_commit(COMMIT)) + + def test_scanpipe_nix_check_github_response(self): + nix.check_github_response(make_response()) + + rate_limited_responses = [ + make_response(status_code=403, headers={"x-ratelimit-remaining": "0"}), + make_response(status_code=429), + ] + for response in rate_limited_responses: + with self.subTest(status_code=response.status_code): + with self.assertRaisesMessage( + requests.HTTPError, "SCANCODEIO_FETCH_HEADERS" + ): + nix.check_github_response(response) + response.raise_for_status.assert_not_called() + + response = make_response(status_code=500) + response.raise_for_status.side_effect = requests.HTTPError("500") + with self.assertRaisesMessage(requests.HTTPError, "500"): + nix.check_github_response(response) + + def test_scanpipe_nix_is_commit_on_branch(self): + session = mock.Mock() + cases = [ + (make_response(json_data={"status": "ahead"}), True), + (make_response(json_data={"status": "identical"}), True), + (make_response(json_data={"status": "diverged"}), False), + (make_response(json_data={"status": "behind"}), False), + (make_response(status_code=404), False), + ] + for response, expected in cases: + with self.subTest(response=response.json.return_value): + session.get.return_value = response + self.assertEqual( + expected, nix.is_commit_on_branch(session, COMMIT, "master") + ) + + compare_url = session.get.call_args.args[0] + expected_url = ( + f"https://api.github.com/repos/NixOS/nixpkgs/compare/{COMMIT}...master" + ) + self.assertEqual(expected_url, compare_url) + + def test_scanpipe_nix_get_release_branches(self): + session = mock.Mock() + session.get.side_effect = [ + make_response( + json_data=[{"ref": "refs/heads/release-24.11"}], + links={"next": {"url": "https://api.github.com/page-2"}}, + ), + make_response(json_data=[{"ref": "refs/heads/release-25.05"}]), + ] + self.assertEqual( + ["release-25.05", "release-24.11"], nix.get_release_branches(session) + ) + self.assertEqual( + "https://api.github.com/page-2", session.get.call_args_list[1].args[0] + ) + + @mock.patch("scanpipe.pipes.nix.subprocess.run") + def test_scanpipe_nix_run_nix_command(self, mock_run): + mock_run.return_value = subprocess.CompletedProcess( + args=[], returncode=0, stdout="output", stderr="" + ) + self.assertEqual("output", nix.run_nix_command(["nix-store", "--gc"], 10)) + self.assertIsNone(mock_run.call_args.kwargs["env"]) + + nix.run_nix_command(["nix-build"], 10, cache_directory=Path("/project/tmp")) + environment = mock_run.call_args.kwargs["env"] + self.assertEqual("/project/tmp", environment["XDG_CACHE_HOME"]) + self.assertEqual(os.environ["PATH"], environment["PATH"]) + + error_lines = [f"line {index}" for index in range(30)] + mock_run.return_value = subprocess.CompletedProcess( + args=[], returncode=1, stdout="", stderr="\n".join(error_lines) + ) + with self.assertRaises(nix.NixCommandError) as context: + nix.run_nix_command(["nix-store", "--gc"], 10) + self.assertEqual("\n".join(error_lines[-20:]), str(context.exception)) + + mock_run.side_effect = subprocess.TimeoutExpired(cmd="nix-build", timeout=10) + expected_message = "nix-build timed out after 10 seconds" + with self.assertRaisesMessage(nix.NixCommandError, expected_message): + nix.run_nix_command(["nix-build"], 10) + + def test_scanpipe_nix_get_nix_expression_arguments(self): + expected = ["--argstr", "attributePath", "hello", "--argstr", "commit", COMMIT] + self.assertEqual(expected, nix.get_nix_expression_arguments("hello", COMMIT)) + + expected += ["--argstr", "system", "x86_64-linux"] + self.assertEqual( + expected, nix.get_nix_expression_arguments("hello", COMMIT, "x86_64-linux") + ) + + @mock.patch("scanpipe.pipes.nix.run_nix_command") + def test_scanpipe_nix_get_package_metadata(self, mock_run): + metadata = { + "version": "2.12.3", + "description": "Program that produces a familiar, friendly greeting", + "homepage": "https://www.gnu.org/software/hello/manual/", + "licenses": ["GPL-3.0-or-later"], + "outputs": {"out": "/nix/store/5z2yp3ysx8476c8g5w25b0smlgkjvaq3-hello"}, + } + mock_run.return_value = json.dumps(metadata) + + cache_directory = Path("/project/tmp") + self.assertEqual( + metadata, + nix.get_package_metadata("hello", COMMIT, "x86_64-linux", cache_directory), + ) + arguments = mock_run.call_args.args[0] + self.assertEqual(["nix-instantiate", "--eval", "--strict"], arguments[:3]) + self.assertIn("x86_64-linux", arguments) + self.assertEqual(nix.METADATA_EXPRESSION, arguments[-1]) + self.assertEqual(cache_directory, mock_run.call_args.kwargs["cache_directory"]) + + def test_scanpipe_nix_get_selected_outputs(self): + outputs = { + "bin": "/nix/store/aaa-curl-bin", + "dev": "/nix/store/bbb-curl-dev", + "out": "/nix/store/ccc-curl", + "man": "/nix/store/ddd-curl-man", + "devdoc": "/nix/store/eee-curl-devdoc", + "debug": "/nix/store/fff-curl-debug", + } + expected = { + "bin": "/nix/store/aaa-curl-bin", + "out": "/nix/store/ccc-curl", + "debug": "/nix/store/fff-curl-debug", + } + self.assertEqual(expected, nix.get_selected_outputs(outputs)) + + expected = {"man": "/nix/store/ddd-curl-man"} + self.assertEqual(expected, nix.get_selected_outputs(outputs, "man")) + + expected_message = ( + "The output 'lib' does not exist, available outputs: " + "bin, debug, dev, devdoc, man, out" + ) + with self.assertRaisesMessage(ValueError, expected_message): + nix.get_selected_outputs(outputs, "lib") + + documentation_outputs = {"dev": "/nix/store/aaa-dev", "doc": "/nix/store/b"} + self.assertEqual( + documentation_outputs, nix.get_selected_outputs(documentation_outputs) + ) + + @mock.patch("scanpipe.pipes.nix.run_nix_command") + def test_scanpipe_nix_build_source(self, mock_run): + mock_run.return_value = "warning: unknown setting\n/nix/store/abc-source\n" + store_path = nix.build_source(nix.PATCHED_SOURCE_EXPRESSION, "hello", COMMIT) + self.assertEqual("/nix/store/abc-source", store_path) + self.assertEqual(["nix-build", "--no-out-link"], mock_run.call_args.args[0][:2]) + + @mock.patch("scanpipe.pipes.nix.copy_store_path") + @mock.patch("scanpipe.pipes.nix.build_source") + def test_scanpipe_nix_fetch_source(self, mock_build, mock_copy): + destination = Path("/codebase/from") + mock_build.return_value = "/nix/store/abc-source" + self.assertEqual("", nix.fetch_source("hello", COMMIT, destination)) + mock_copy.assert_called_once_with("/nix/store/abc-source", destination) + + mock_build.side_effect = [nix.NixCommandError("patch failed"), "/nix/store/src"] + cache_directory = Path("/project/tmp") + self.assertEqual( + "patch failed", + nix.fetch_source("hello", COMMIT, destination, cache_directory), + ) + mock_build.assert_called_with( + nix.UPSTREAM_SOURCE_EXPRESSION, "hello", COMMIT, cache_directory + ) + mock_copy.assert_called_with("/nix/store/src", destination) + + def test_scanpipe_nix_copy_store_path(self): + with tempfile.TemporaryDirectory() as temp_directory: + store_directory = Path(temp_directory) / "abc-source" + (store_directory / "src").mkdir(parents=True) + source_file = store_directory / "src" / "hello.c" + source_file.write_text("int main(void);") + source_file.chmod(0o444) + (store_directory / "src").chmod(0o555) + + destination = Path(temp_directory) / "codebase" / "from" + nix.copy_store_path(store_directory, destination) + copied_file = destination / "src" / "hello.c" + self.assertEqual("int main(void);", copied_file.read_text()) + self.assertTrue(os.access(copied_file, os.W_OK)) + self.assertTrue(os.access(destination / "src", os.W_OK)) + (store_directory / "src").chmod(0o755) + + store_archive = Path(temp_directory) / "abc123-hello-2.12.3.tar.gz" + store_archive.write_bytes(b"archive") + nix.copy_store_path(store_archive, destination) + self.assertEqual( + b"archive", (destination / "hello-2.12.3.tar.gz").read_bytes() + ) + + def test_scanpipe_nix_make_writable(self): + with tempfile.TemporaryDirectory() as temp_directory: + directory = Path(temp_directory) + read_only_file = directory / "file" + read_only_file.write_text("content") + read_only_file.chmod(0o444) + (directory / "link").symlink_to("/does/not/exist") + + nix.make_writable(directory) + self.assertTrue(os.access(read_only_file, os.W_OK)) + + @mock.patch("scanpipe.pipes.nix.fetch.get_request_session") + def test_scanpipe_nix_get_narinfo(self, mock_session): + narinfo_text = (self.data / "hello.narinfo").read_text() + mock_session.return_value.get.return_value = make_response(text=narinfo_text) + + store_path = "/nix/store/5z2yp3ysx8476c8g5w25b0smlgkjvaq3-hello-2.12.3" + narinfo = nix.get_narinfo(store_path) + self.assertEqual( + "nar/1k684wya73qd387kbf1pdpzb986qalk7c7jbnh6ib5846i6q18yc.nar.zst", + narinfo["URL"], + ) + self.assertEqual("zstd", narinfo["Compression"]) + requested_url = mock_session.return_value.get.call_args.args[0] + self.assertEqual( + "https://cache.nixos.org/5z2yp3ysx8476c8g5w25b0smlgkjvaq3.narinfo", + requested_url, + ) + + mock_session.return_value.get.return_value = make_response(status_code=404) + self.assertEqual({}, nix.get_narinfo(store_path)) + + @mock.patch("scanpipe.pipes.nix.restore_nar") + @mock.patch("scanpipe.pipes.nix.download_file") + @mock.patch("scanpipe.pipes.nix.get_narinfo") + def test_scanpipe_nix_fetch_binary(self, mock_narinfo, mock_fetch, mock_restore): + store_path = "/nix/store/5z2yp3ysx8476c8g5w25b0smlgkjvaq3-hello-2.12.3" + destination = Path("/codebase/to/out") + + mock_narinfo.return_value = {} + self.assertFalse(nix.fetch_binary(store_path, destination, "/downloads")) + mock_fetch.assert_not_called() + + with tempfile.TemporaryDirectory() as temp_directory: + nar_location = ( + Path(temp_directory) + / "5z2yp3ysx8476c8g5w25b0smlgkjvaq3-hello-2.12.3.nar" + ) + + def download_file(url, file_location): + file_location.write_bytes(b"nar") + + mock_fetch.side_effect = download_file + mock_narinfo.return_value = { + "URL": "nar/abc.nar.zst", + "Compression": "zstd", + } + + self.assertTrue(nix.fetch_binary(store_path, destination, temp_directory)) + mock_fetch.assert_called_once_with( + "https://cache.nixos.org/nar/abc.nar.zst", nar_location + ) + mock_restore.assert_called_once_with(nar_location, "zstd", destination) + self.assertFalse(nar_location.exists()) + + @mock.patch("scanpipe.pipes.nix.fetch.get_request_session") + def test_scanpipe_nix_download_file(self, mock_session): + response = mock.MagicMock() + response.iter_content.return_value = [b"first ", b"second"] + mock_session.return_value.get.return_value.__enter__.return_value = response + + with tempfile.TemporaryDirectory() as temp_directory: + file_location = Path(temp_directory) / "hello.nar" + nix.download_file("https://cache.nixos.org/nar/abc.nar", file_location) + self.assertEqual(b"first second", file_location.read_bytes()) + + self.assertTrue(mock_session.return_value.get.call_args.kwargs["stream"]) + response.iter_content.assert_called_once_with( + chunk_size=nix.DOWNLOAD_CHUNK_SIZE + ) + + @mock.patch("scanpipe.pipes.nix.make_writable") + @mock.patch("scanpipe.pipes.nix.subprocess.run") + @mock.patch("scanpipe.pipes.nix.subprocess.Popen") + def test_scanpipe_nix_restore_nar(self, mock_popen, mock_run, mock_writable): + with tempfile.TemporaryDirectory() as temp_directory: + nar_location = Path(temp_directory) / "hello.nar.xz" + nar_location.write_bytes(b"nar") + destination = Path(temp_directory) / "to" / "out" + + with self.assertRaisesMessage(ValueError, "Unsupported NAR compression"): + nix.restore_nar(nar_location, "lzip", destination) + + decompress_process = mock_popen.return_value.__enter__.return_value + decompress_process.returncode = 0 + decompress_process.stderr.read.return_value = b"" + mock_run.return_value = mock.Mock(returncode=0, stderr="") + nix.restore_nar(nar_location, "xz", destination) + self.assertEqual( + ["xz", "--decompress", "--stdout"], mock_popen.call_args.args[0] + ) + self.assertEqual( + ["nix-store", "--restore", str(destination)], + mock_run.call_args.args[0], + ) + mock_writable.assert_called_once_with(destination) + + mock_run.return_value = mock.Mock(returncode=1, stderr="bad archive") + with self.assertRaisesMessage(nix.NixCommandError, "bad archive"): + nix.restore_nar(nar_location, "xz", destination) + + decompress_process.returncode = 1 + decompress_process.stderr.read.return_value = b"xz: Unexpected end" + mock_run.return_value = mock.Mock(returncode=0, stderr="") + with self.assertRaisesMessage(nix.NixCommandError, "xz: Unexpected end"): + nix.restore_nar(nar_location, "xz", destination) + + mock_run.side_effect = subprocess.TimeoutExpired( + cmd="nix-store", timeout=nix.NIX_BUILD_TIMEOUT + ) + with self.assertRaisesMessage(nix.NixCommandError, "timed out"): + nix.restore_nar(nar_location, "xz", destination) + decompress_process.kill.assert_called_once() + + @mock.patch("scanpipe.pipes.nix.run_nix_command") + def test_scanpipe_nix_collect_garbage(self, mock_run): + nix.collect_garbage() + mock_run.assert_called_once_with( + ["nix-store", "--gc"], timeout=nix.NIX_BUILD_TIMEOUT + ) + + def test_scanpipe_nix_combine_license_expressions(self): + cases = [ + ([], ""), + (["Unlicense OR MIT"], "Unlicense OR MIT"), + (["Apache-2.0", "MIT"], "Apache-2.0 AND MIT"), + ( + ["Unlicense OR MIT", "GPL-3.0-or-later"], + "(Unlicense OR MIT) AND GPL-3.0-or-later", + ), + ] + for license_expressions, expected in cases: + with self.subTest(license_expressions=license_expressions): + self.assertEqual( + expected, nix.combine_license_expressions(license_expressions) + ) + + def test_scanpipe_nix_convert_nixpkgs_license_expression(self): + cases = [ + ("Unlicense OR MIT", "unlicense OR mit"), + ( + "LicenseRef-nixos-unfreeRedistributable", + "unknown-license-reference", + ), + ("MIT AND LicenseRef-nixos-free", "mit AND unknown-license-reference"), + ] + for license_expression_spdx, expected in cases: + with self.subTest(license_expression_spdx=license_expression_spdx): + self.assertEqual( + expected, + nix.convert_nixpkgs_license_expression(license_expression_spdx), + ) + + def test_scanpipe_nix_create_nix_package(self): + purl = PackageURL.from_string( + "pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux" + ) + metadata = { + "version": "2.12.3", + "description": "Program that produces a familiar, friendly greeting", + "homepage": ["https://www.gnu.org/software/hello/", "https://other"], + "licenses": ["GPL-3.0-or-later", "Unlicense OR MIT"], + "outputs": {}, + } + package = nix.create_nix_package(self.project, purl, metadata, COMMIT) + self.assertEqual("pkg:nix/nixpkgs/hello@2.12.3", package.purl) + self.assertEqual( + "gpl-3.0-plus AND (unlicense OR mit)", package.declared_license_expression + ) + self.assertEqual( + "GPL-3.0-or-later AND (Unlicense OR MIT)", + package.declared_license_expression_spdx, + ) + self.assertEqual("https://www.gnu.org/software/hello/", package.homepage_url) + self.assertEqual(metadata["description"], package.description) + self.assertEqual(["nixpkgs_meta"], package.datasource_ids) + expected_extra_data = {"nixpkgs_commit": COMMIT, "system": "x86_64-linux"} + self.assertEqual(expected_extra_data, package.extra_data) + + purl = PackageURL.from_string( + f"pkg:nix/nixpkgs/zlib?system=x86_64-linux&commit={COMMIT}" + ) + metadata.update(version="1.3.1", licenses=[], homepage="") + package = nix.create_nix_package(self.project, purl, metadata, COMMIT) + self.assertEqual("1.3.1", package.version) + self.assertEqual("", package.declared_license_expression) + + # The nixpkgs version is the one of the scanned code. + purl = PackageURL.from_string("pkg:nix/nixpkgs/curl@8.0?system=x86_64-linux") + metadata.update(version="8.16.0") + package = nix.create_nix_package(self.project, purl, metadata, COMMIT) + self.assertEqual("pkg:nix/nixpkgs/curl@8.16.0", package.purl) + + def test_scanpipe_nix_add_resources_to_package(self): + purl = PackageURL.from_string( + "pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux" + ) + metadata = {"version": "", "description": "", "homepage": "", "licenses": []} + package = nix.create_nix_package(self.project, purl, metadata, COMMIT) + make_resource_file(self.project, "from/hello.c") + make_resource_file(self.project, "to/bin/hello") + resource_qs = self.project.codebaseresources.files() + + nix.add_resources_to_package(package, resource_qs, batch_size=1) + self.assertEqual(2, package.codebase_resources.count()) + + nix.add_resources_to_package(package, resource_qs) + self.assertEqual(2, package.codebase_resources.count()) + + @mock.patch("scanpipe.pipes.nix.scancode.scan_for_files") + def test_scanpipe_nix_scan_undeployed_from_files(self, mock_scan): + make_resource_file(self.project, "from/COPYING", status=flag.NOT_DEPLOYED) + make_resource_file(self.project, "from/hello.c", status=flag.SCANNED) + make_resource_file(self.project, "to/COPYING", status=flag.NOT_DEPLOYED) + scanned_paths = [] + + def scan_for_files(project, resource_qs, progress_logger=None): + scanned_paths.extend(resource.path for resource in resource_qs) + resource_qs.update( + status=flag.SCANNED, detected_license_expression="gpl-3.0-plus" + ) + + mock_scan.side_effect = scan_for_files + + nix.scan_undeployed_from_files(self.project) + self.assertEqual(["from/COPYING"], scanned_paths) + + resource_statuses = dict( + self.project.codebaseresources.values_list("path", "status") + ) + expected_statuses = { + "from/COPYING": flag.NOT_DEPLOYED, + "from/hello.c": flag.SCANNED, + "to/COPYING": flag.NOT_DEPLOYED, + } + self.assertEqual(expected_statuses, resource_statuses) + copying = self.project.codebaseresources.get(path="from/COPYING") + self.assertEqual("gpl-3.0-plus", copying.detected_license_expression) + + def test_scanpipe_nix_get_uncovered_license_keys(self): + resources = [ + ("from/src/hello.c", "gpl-3.0-plus"), + ("from/lib/helper.c", "mit AND gpl-3.0-plus"), + ("from/lib/other.c", "mit"), + ("from/README", "unknown-license-reference"), + ("to/share/doc/LICENSE", "apache-2.0"), + ] + for path, license_expression in resources: + make_resource_file( + self.project, path, detected_license_expression=license_expression + ) + + self.assertEqual( + ["mit"], nix.get_uncovered_license_keys(self.project, "gpl-3.0-plus") + ) + self.assertEqual( + [], nix.get_uncovered_license_keys(self.project, "gpl-3.0-plus AND mit") + ) + + # The GPL text of a COPYING file is detected without "or later". + make_resource_file( + self.project, "from/COPYING", detected_license_expression="gpl-3.0" + ) + self.assertEqual( + ["mit"], nix.get_uncovered_license_keys(self.project, "gpl-3.0-plus") + ) + self.assertEqual( + ["mit"], nix.get_uncovered_license_keys(self.project, "gpl-3.0") + ) + self.assertEqual( + ["gpl-3.0", "gpl-3.0-plus"], + nix.get_uncovered_license_keys(self.project, "gpl-2.0-plus AND mit"), + ) + + def test_scanpipe_nix_get_base_license_key(self): + cases = [ + ("gpl-3.0-plus", "gpl-3.0"), + ("lgpl-2.1-plus", "lgpl-2.1"), + ("gpl-3.0", "gpl-3.0"), + ("m-plus", "m-plus"), + ("mit", "mit"), + ] + for license_key, expected in cases: + with self.subTest(license_key=license_key): + self.assertEqual(expected, nix.get_base_license_key(license_key)) diff --git a/scanpipe/tests/test_pipelines.py b/scanpipe/tests/test_pipelines.py index f335fd3606..c323ef2037 100644 --- a/scanpipe/tests/test_pipelines.py +++ b/scanpipe/tests/test_pipelines.py @@ -52,10 +52,12 @@ from scanpipe.pipelines import scan_single_package from scanpipe.pipelines.find_origins import FindOrigins from scanpipe.pipelines.find_vulnerabilities import FindVulnerabilities +from scanpipe.pipelines.scan_nix_package import ScanNixPackage from scanpipe.pipes import d2d from scanpipe.pipes import flag from scanpipe.pipes import matchcode from scanpipe.pipes import maven +from scanpipe.pipes import nix from scanpipe.pipes import output from scanpipe.pipes import scancode from scanpipe.pipes.input import copy_input @@ -185,6 +187,149 @@ def test_scanpipe_find_origins_get_availability( mock_scanoss_is_configured.return_value = True self.assertIsNone(FindOrigins.get_availability()) + @mock.patch("scanpipe.pipes.nix.get_missing_commands") + def test_scanpipe_scan_nix_package_get_availability(self, mock_missing_commands): + mock_missing_commands.return_value = ["nix-build", "zstd"] + self.assertEqual( + "Missing commands: nix-build, zstd. Use the full ScanCode.io image.", + ScanNixPackage.get_availability(), + ) + + mock_missing_commands.return_value = [] + self.assertIsNone(ScanNixPackage.get_availability()) + + def test_scanpipe_scan_nix_package_selected_groups(self): + project1 = make_project() + run = project1.add_pipeline("scan_nix_package") + self.assertEqual(["Elf"], run.make_pipeline_instance().selected_groups) + + run = project1.add_pipeline("scan_nix_package", selected_groups=["Python"]) + pipeline = run.make_pipeline_instance() + self.assertEqual(["Python", "Elf"], pipeline.selected_groups) + self.assertEqual(["Python"], run.selected_groups) + + steps = ScanNixPackage.get_steps(groups=pipeline.selected_groups) + self.assertIn(ScanNixPackage.map_elf, steps) + + project2 = make_project() + project2.add_input_source( + download_url="pkg:nix/nixpkgs/hello@2.12.3?system=aarch64-darwin" + ) + pipeline = project2.add_pipeline("scan_nix_package").make_pipeline_instance() + self.assertEqual(["MacOS"], pipeline.selected_groups) + steps = ScanNixPackage.get_steps(groups=pipeline.selected_groups) + self.assertIn(ScanNixPackage.map_macho, steps) + self.assertNotIn(ScanNixPackage.map_elf, steps) + + @mock.patch("scanpipe.pipes.nix.get_package_metadata") + def test_scanpipe_scan_nix_package_get_nix_package_metadata(self, mock_metadata): + project1 = make_project() + pipeline = project1.add_pipeline("scan_nix_package").make_pipeline_instance() + pipeline.purl = PackageURL.from_string( + "pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux" + ) + pipeline.commit = "55ba7f49ef2962b42cbd126522b7df5f95037679" + outputs = {"out": "/nix/store/aaa-hello", "man": "/nix/store/bbb-man"} + + mock_metadata.return_value = {"version": "2.12.3", "outputs": outputs} + pipeline.get_nix_package_metadata() + self.assertEqual({"out": "/nix/store/aaa-hello"}, pipeline.selected_outputs) + self.assertEqual(0, project1.projectmessages.count()) + + pipeline.purl = PackageURL.from_string( + "pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux&output=debug" + ) + with self.assertRaisesMessage(ValueError, "The output 'debug' does not"): + pipeline.get_nix_package_metadata() + + pipeline.purl = PackageURL.from_string( + "pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux" + ) + mock_metadata.return_value = {"version": "2.12.2", "outputs": outputs} + pipeline.get_nix_package_metadata() + message = project1.projectmessages.get() + self.assertEqual( + "The PURL version 2.12.3 differs from the nixpkgs version 2.12.2 at commit " + "55ba7f49ef2962b42cbd126522b7df5f95037679.", + message.description, + ) + + @mock.patch("scanpipe.pipes.nix.fetch_source") + def test_scanpipe_scan_nix_package_fetch_nix_source_error(self, mock_fetch): + project1 = make_project() + pipeline = project1.add_pipeline("scan_nix_package").make_pipeline_instance() + pipeline.purl = PackageURL.from_string( + "pkg:nix/nixpkgs/papertrail@0.11.3?system=x86_64-linux" + ) + pipeline.commit = "55ba7f49ef2962b42cbd126522b7df5f95037679" + mock_fetch.side_effect = nix.NixCommandError("attribute 'src' missing") + + pipeline.fetch_nix_source() + message = project1.projectmessages.get() + self.assertEqual( + "The source cannot be fetched, the binary is analyzed alone and the " + "license check is skipped.", + message.description, + ) + self.assertEqual({"error": "attribute 'src' missing"}, message.details) + + @mock.patch("scanpipe.pipes.nix.fetch_binary") + def test_scanpipe_scan_nix_package_fetch_nix_binary(self, mock_fetch): + project1 = make_project() + pipeline = project1.add_pipeline("scan_nix_package").make_pipeline_instance() + pipeline.selected_outputs = { + "out": "/nix/store/aaa-hello", + "debug": "/nix/store/bbb-debug", + } + + def fetch_binary(store_path, destination, download_directory): + return store_path.endswith("-hello") + + mock_fetch.side_effect = fetch_binary + + pipeline.fetch_nix_binary() + self.assertEqual(2, mock_fetch.call_count) + self.assertEqual( + project1.codebase_path / "to" / "out", + mock_fetch.call_args_list[0].kwargs["destination"], + ) + message = project1.projectmessages.get() + self.assertEqual( + "Outputs not available in the Nix binary cache: debug", + message.description, + ) + + @mock.patch("scanpipe.pipes.nix.collect_garbage") + def test_scanpipe_scan_nix_package_collect_nix_garbage(self, mock_collect): + project1 = make_project() + pipeline = project1.add_pipeline("scan_nix_package").make_pipeline_instance() + mock_collect.side_effect = nix.NixCommandError("store is locked") + + pipeline.collect_nix_garbage() + message = project1.projectmessages.get() + self.assertEqual( + "The Nix store garbage collection failed.", message.description + ) + self.assertEqual({"error": "store is locked"}, message.details) + + def test_scanpipe_scan_nix_package_check_nix_package_license(self): + project1 = make_project() + pipeline = project1.add_pipeline("scan_nix_package").make_pipeline_instance() + pipeline.package = make_package(project1, "pkg:nix/nixpkgs/hello@2.12.3") + + pipeline.check_nix_package_license() + self.assertEqual({}, pipeline.package.extra_data) + self.assertEqual(0, project1.projectmessages.count()) + + make_resource_file(project1, "from/src/hello.c") + pipeline.check_nix_package_license() + self.assertEqual({}, pipeline.package.extra_data) + message = project1.projectmessages.get() + self.assertEqual( + "No declared license in nixpkgs, license check skipped.", + message.description, + ) + def test_scanpipe_pipeline_class_get_summary(self): expected = "Do nothing, in 2 steps." self.assertEqual(expected, DoNothing.get_summary()) @@ -2487,3 +2632,141 @@ def test_scanpipe_benchmark_purls_pipeline_integration(self): ) expected_file = self.data / "benchmark" / "alpine-3.22.1-expected-benchmark.txt" self.assertEqual(expected_file.read_text(), result_file.read_text()) + + def run_scan_nix_package_pipeline(self, fetch_source_effect): + """ + Run the ``scan_nix_package`` pipeline on the hello test data, with the Nix + commands mocked, and return the pipeline run. + """ + nix_data = self.data / "nix" + metadata = { + "version": "2.12.3", + "description": "Program that produces a familiar, friendly greeting", + "homepage": "https://www.gnu.org/software/hello/manual/", + "licenses": ["GPL-3.0-or-later"], + "outputs": { + "out": "/nix/store/5z2yp3ysx8476c8g5w25b0smlgkjvaq3-hello-2.12.3", + "man": "/nix/store/00000000000000000000000000000000-hello-man", + "debug": "/nix/store/11111111111111111111111111111111-hello-debug", + }, + } + + def fetch_binary(store_path, destination, download_directory): + if store_path.endswith("-debug"): + return False + shutil.copytree(nix_data / "binary", destination) + return True + + project1 = make_project() + project1.add_input_source( + download_url="pkg:nix/nixpkgs/hello@2.12.3?system=x86_64-linux" + ) + run = project1.add_pipeline("scan_nix_package") + pipeline = run.make_pipeline_instance() + + with ( + mock.patch( + "scanpipe.pipes.nix.get_nixpkgs_commit", + return_value="55ba7f49ef2962b42cbd126522b7df5f95037679", + ), + mock.patch( + "scanpipe.pipes.nix.get_package_metadata", return_value=metadata + ) as mock_get_metadata, + mock.patch( + "scanpipe.pipes.nix.fetch_source", side_effect=fetch_source_effect + ), + mock.patch( + "scanpipe.pipes.nix.fetch_binary", side_effect=fetch_binary + ) as mock_fetch_binary, + mock.patch("scanpipe.pipes.nix.collect_garbage") as mock_collect_garbage, + ): + exitcode, out = pipeline.execute() + + self.assertEqual(0, exitcode, msg=out) + self.assertEqual("x86_64-linux", mock_get_metadata.call_args.kwargs["system"]) + mock_collect_garbage.assert_called_once() + + fetched_outputs = [ + call.kwargs["destination"].name for call in mock_fetch_binary.call_args_list + ] + self.assertEqual(["out", "debug"], fetched_outputs) + + return run + + @skipIf(sys.platform == "darwin", "Not supported on macOS") + def test_scanpipe_scan_nix_package_pipeline_integration(self): + nix_data = self.data / "nix" + + def fetch_source(attribute_path, commit, destination, cache_directory): + shutil.copytree(nix_data / "source", destination) + return "patch failed" + + run = self.run_scan_nix_package_pipeline(fetch_source) + project1 = run.project + self.assertIn("Step [map_elf] completed", run.log) + + relation = project1.codebaserelations.get() + self.assertEqual("sha1", relation.map_type) + self.assertEqual("from/src/hello.c", relation.from_resource.path) + self.assertEqual("to/out/doc/hello.c", relation.to_resource.path) + + helper = project1.codebaseresources.get(path="from/src/helper.c") + self.assertEqual(flag.NOT_DEPLOYED, helper.status) + self.assertEqual("mit", helper.detected_license_expression) + + package = project1.discoveredpackages.get() + self.assertEqual("pkg:nix/nixpkgs/hello@2.12.3", package.purl) + self.assertEqual("gpl-3.0-plus", package.declared_license_expression) + expected_extra_data = { + "nixpkgs_commit": "55ba7f49ef2962b42cbd126522b7df5f95037679", + "system": "x86_64-linux", + "uncovered_licenses": ["mit"], + } + self.assertEqual(expected_extra_data, package.extra_data) + self.assertEqual( + project1.codebaseresources.files().count(), + package.codebase_resources.count(), + ) + + warnings = sorted( + project1.projectmessages.values_list("description", flat=True) + ) + self.assertEqual( + [ + "Outputs not available in the Nix binary cache: debug", + "The source with the nixpkgs patches cannot be built, the upstream " + "source is used instead.", + ], + warnings, + ) + patch_warning = project1.projectmessages.get(details__has_key="error") + self.assertEqual({"error": "patch failed"}, patch_warning.details) + + @skipIf(sys.platform == "darwin", "Not supported on macOS") + def test_scanpipe_scan_nix_package_pipeline_integration_without_source(self): + source_error = nix.NixCommandError("attribute 'src' missing") + project1 = self.run_scan_nix_package_pipeline(source_error).project + + self.assertEqual(0, project1.codebaseresources.from_codebase().count()) + self.assertEqual(0, project1.codebaserelations.count()) + binary = project1.codebaseresources.get(path="to/out/hello") + self.assertEqual(flag.REQUIRES_REVIEW, binary.status) + + package = project1.discoveredpackages.get() + self.assertNotIn("uncovered_licenses", package.extra_data) + self.assertEqual( + project1.codebaseresources.files().count(), + package.codebase_resources.count(), + ) + + warnings = sorted( + project1.projectmessages.values_list("description", flat=True) + ) + self.assertEqual( + [ + "Outputs not available in the Nix binary cache: debug", + "The source cannot be fetched, the binary is analyzed alone and the " + "license check is skipped.", + ], + warnings, + )