From c74910a1ecab7d302a32f8bc5011b9a9ed8f6d51 Mon Sep 17 00:00:00 2001 From: Chin Yeung Li Date: Thu, 18 Jun 2026 17:30:46 +0800 Subject: [PATCH 01/10] Add nix support #201 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * There are two main functions: get_download_url (system‑specific binary) and get_src_download_url (source tarball). * Avoid using nix to retrieve data whenever possible, but provide a fallback to nix if the non‑Nix method does not yield results. Signed-off-by: Chin Yeung Li --- src/fetchcode/nix.py | 610 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 610 insertions(+) create mode 100644 src/fetchcode/nix.py diff --git a/src/fetchcode/nix.py b/src/fetchcode/nix.py new file mode 100644 index 0000000..2d9238a --- /dev/null +++ b/src/fetchcode/nix.py @@ -0,0 +1,610 @@ +# fetchcode is a free software tool from nexB Inc. and others. +# Visit https://github.com/aboutcode-org/fetchcode for support and download. +# +# Copyright (c) nexB Inc. and others. All rights reserved. +# http://nexb.com and http://aboutcode.org +# +# This software is licensed under the Apache License version 2.0. +# +# 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. + +import json +import shutil +import subprocess +import sys +from urllib.parse import urlparse + +import requests +from packageurl import PackageURL +from packageurl.contrib.purl2url import build_hackage_download_url +from packageurl.contrib.purl2url import get_repo_download_url_by_package_type + +from fetchcode import fetch_json_response + + +class Nix: + """ + Handle Nix Package URL (PURL) resolution and download URL retrieval. + """ + + purl_pattern = "pkg:nix/nixpkgs/.*" + + @classmethod + def get_package_data(cls, purl): + """ + Fetch package data from https://search.devbox.sh/. + + Args: + purl: A Package URL string (e.g., "pkg:nix/nixpkgs/Axel@2.17.14") + + Returns: + The full JSON response from search.devbox.sh. + """ + parsed_purl = PackageURL.from_string(purl) + + api_url = f"https://search.devbox.sh/v2/pkg?name={parsed_purl.name}" + if not verify_url_existence(api_url): + return None + + return fetch_json_response(api_url) + + @classmethod + def get_download_url(cls, purl): + """ + Get a single direct download URL of a system-specific binary. + + Args: + purl: A Package URL string (e.g., "pkg:nix/nixpkgs/Axel@2.17.14?system=x86_64-linux") + + Returns: + The download URL, or None if not found. + """ + purl_data = PackageURL.from_string(purl) + have_nix = False + + if shutil.which("nix") is not None: + have_nix = True + + namespace = purl_data.namespace + name = purl_data.name + version = purl_data.version + qualifiers = purl_data.qualifiers + commit_hash = "" + system = "" + output = "" + + if "system" in qualifiers: + system = qualifiers.get("system") + else: + raise Exception( + f"The 'system' qualifier is required to resolve system-specific binaries." + ) + if "commit" in qualifiers: + commit_hash = qualifiers.get("commit") + if "output" in qualifiers: + output = qualifiers.get("output") + else: + # Default 'out' if no output is defined + output = "out" + + data = cls.get_package_data(purl) + path = "" + + if data: + path = get_nix_store_path(data, system, output, commit_hash, version) + if not data or not path: + if have_nix and commit_hash: + path = get_nix_store_path_with_nix(name, system, output, commit_hash) + clean_garbage() + + if path: + narinfo_url = path.replace("/nix/store/", "").split("-")[0] + return get_nix_download_url(narinfo_url) + else: + return None + + @classmethod + def get_src_download_url(cls, purl): + """ + Get a single direct source download URL. + + If no version is specified in the PURL, fetches the latest version. + + Args: + purl: A Package URL string (e.g., "pkg:nix/nixpkgs/Axel@2.17.14") + + Returns: + The download URL, or None if not found. + """ + # Since Nix is not a central repository, packages are hosted in + # different locations, making it difficult to determine their + # direct source download URLs. We will provide a few well‑known + # repositories and construct direct download URLs when packages are + # hosted there. For all other hosts, the download_url will + # currently return None. Additional repositories might be supported + # progressively. + purl_data = PackageURL.from_string(purl) + have_nix = False + download_url = None + + if shutil.which("nix") is not None: + have_nix = True + + namespace = "" + name = purl_data.name + version = purl_data.version + + data = cls.get_package_data(purl) + + if data: + download_url = construct_url_based_on_netloc(namespace, name, version, data) + if not download_url and have_nix: + download_url = retrieve_src_download_url_with_nix(name, version) + # Try to use the commit_hash appraoch to get the download_url + if not download_url and version: + commit_hash = get_commit_hash(data, version) + download_url = retrieve_src_download_url_with_nix(name, version, commit_hash) + else: + # Use the `nix` command/utility to retrieve data. + # However, this only fetches the latest version. + if have_nix: + download_url = retrieve_src_download_url_with_nix(name, version) + + if not download_url and not have_nix: + print("Install `nix` and re-run to let `nix` determine the download URL.") + + if have_nix: + clean_garbage() + + return download_url + + +def get_nix_store_path(data, system, output, commit_hash=None, version=None): + """ + Find and return the store path (/nix/store/) based on the qualifiers + """ + releases = data.get("releases") + for release in releases: + if commit_hash: + platforms = release.get("platforms") + for platform in platforms: + if system == platform.get("system") and commit_hash == platform.get("commit_hash"): + platform_outputs = platform.get("outputs") + for platform_output in platform.get("outputs"): + if output == platform_output.get("name"): + path = platform_output.get("path") + return path + return None + elif not commit_hash and version: + if version == release.get("version"): + platforms = release.get("platforms") + for platform in platforms: + if system == platform.get("system"): + platform_outputs = platform.get("outputs") + for platform_output in platform_outputs: + if output == platform_output.get("name"): + path = platform_output.get("path") + return path + return None + else: + # No version and no commit_hash. + # Default to the latest version. + platforms = release.get("platforms") + for platform in platforms: + if system == platform.get("system"): + platform_outputs = platform.get("outputs") + for platform_output in platform_outputs: + if output == platform_output.get("name"): + path = platform_output.get("path") + return path + return None + return None + + +def get_nix_store_path_with_nix(name, system, output, commit_hash): + """ + Find and return the store path (/nix/store/) based on the + qualifiers using 'nix' + """ + system_config = f'system = "{system}";' if system else "" + output_modifier = f".{output}" if output else "" + + nix_expression = ( + f'(import (fetchTarball "https://github.com/NixOS/nixpkgs/archive/{commit_hash}.tar.gz") ' + f"{{ {system_config} }}).{name}{output_modifier}.outPath" + ) + + cmd = ["nix-instantiate", "--eval", "--raw", "-E", nix_expression] + + try: + result = subprocess.run(cmd, capture_output=True, text=True, check=True) + return result.stdout.strip() + except subprocess.CalledProcessError as e: + print(f"Error evaluating attribute for package '{name}'", file=sys.stderr) + print(e.stderr, file=sys.stderr) + return None + + +def retrieve_src_download_url_with_nix(name, version, commit_hash=None): + """ + Find and return the source download url using 'nix' + """ + info = get_src_info(name, commit_hash) + + download_url = None + # We don't need to check for version if we have the commit_hash + if commit_hash: + for url in info["urls"]: + if url.startswith("mirror:"): + download_url = covert_mirror_url(url) + else: + if verify_url_existence(url): + download_url = url + else: + if not version: + for url in info["urls"]: + if url.startswith("mirror:"): + download_url = covert_mirror_url(url) + else: + if verify_url_existence(url): + download_url = url + else: + # Attempt to replace the version and validate that the + # URL exists. Return None if the URL is invalid. + latest_version = info["version"] + for url in info["urls"]: + if url.startswith("mirror:"): + converted_url = covert_mirror_url(url) + updated_version_url = converted_url.replace(latest_version, version) + else: + updated_version_url = url.replace(latest_version, version) + if verify_url_existence(updated_version_url): + download_url = updated_version_url + + return download_url + + +def get_src_info(attr_path, commit_hash=None): + """ + Use the `nix-instantiate` command together with the nix_expression to + retrieve the package’s version and download URL. Return a dictionary + with "version" and "urls" keys. + """ + # Determine the repository entry point definition + if commit_hash: + nixpkgs_import = ( + 'import (fetchTarball "https://github.com/NixOS/nixpkgs/archive/' + f'{commit_hash}.tar.gz") {{}}' + ) + else: + nixpkgs_import = "import {}" + + nix_expression = f""" + let + pkg = ({nixpkgs_import}).{attr_path}; + extract_urls = src: + if builtins.hasAttr "url" src then [src.url] + else if builtins.hasAttr "urls" src then src.urls + else if builtins.hasAttr "src" src then extract_urls src.src + else []; + in + {{ + version = if builtins.hasAttr "version" pkg then pkg.version else null; + urls = if builtins.hasAttr "src" pkg then extract_urls pkg.src else []; + }} + """ + cmd = [ + "nix-instantiate", + "--eval", + "--json", + "--strict", + "-E", + nix_expression, + ] + try: + result = subprocess.run(cmd, capture_output=True, text=True, check=True) + return json.loads(result.stdout) + except subprocess.CalledProcessError as e: + print(f"Error evaluating attribute '{attr_path}':", file=sys.stderr) + print(e.stderr, file=sys.stderr) + return {"version": None, "urls": []} + + +def construct_url_based_on_netloc(namespace, name, version, data): + """ + Determine and return the download url based on the homepage, name and + version + """ + homepage_url = data.get("homepage_url", None) + # Get the latest version if no version is provided + if not version: + version = data.get("releases")[0].get("version") + + netloc, namespace, name = get_url_netloc_namespace_and_name(homepage_url) + if netloc.endswith("github.io"): + github_page_url = github_pages_to_repo(homepage_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") + 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") + # There is an issue where the version may have a 'v' prefix, which + # makes the constructed download URL invalid. For example, in + # https://github.com/containers/aardvark-dns/, the constructed + # download_url is + # https://github.com/containers/aardvark-dns/archive/1.8.0.tar.gz, but + # this returns a 404 because the actual download URL should be + # https://github.com/containers/aardvark-dns/archive/v1.8.0.tar.gz. + + # Users are responsible for providing the correct version string + # (including any 'v' prefix). However, from a mining situation, the + # source data we collect may omit the prefix, so we need to handle that + # case. For example, versions from + # https://search.devbox.sh/v2/pkg?name=CuboCore.corepins do not include + # the 'v' prefix, but the actual versions from the download site + # include it in the tag field: + # https://gitlab.com/api/v4/projects/cubocore%2Fcoreapps%2Fcorepins/repository/tags + + # There are also cases that use other prefix such as release-{version} + # See https://search.devbox.sh/v2/pkg?name=SDL2_mixer where one of + # the versions is 2.8.2 while the tag from the github is + # release-2.8.2 + # (https://github.com/libsdl-org/SDL_mixer/releases/tag/release-2.8.2) + + # We want to validate whether the returned download URL is + # accessible. If it is not, insert a common prefix and try to + # validate again. + common_prefixes = ["v", "V", "v-", "V-", "release-", "RELEASE-"] + download_url = get_repo_download_url_by_package_type( + type=package_type, namespace=namespace, name=name, version=version + ) + if not verify_url_existence(download_url): + for prefix in common_prefixes: + prefix_version = prefix + version + download_url = get_repo_download_url_by_package_type( + type=package_type, namespace=namespace, name=name, version=prefix_version + ) + if verify_url_existence(download_url): + return download_url + else: + return download_url + elif netloc == "hackage.haskell.org": + pname = name.strip("haskellPackages.") + purl = "pkg:hackage/" + pname + "@" + version + download_url = build_hackage_download_url(purl) + if verify_url_existence(download_url): + return download_url + else: + candidates = [] + # CRAN (R Packages) + if netloc == "cran.r-project.org": + candidates = [ + f"https://cran.r-project.org/src/contrib/{name}_{version}.tar.gz", + f"https://cran.r-project.org/src/contrib/Archive/{name}/{name}_{version}.tar.gz", + ] + # PyPI (Python Packages) + elif netloc == "pypi.org" or netloc == "pypi.python.org": + first_letter = name[0] + candidates = [ + f"https://files.pythonhosted.org/packages/source/" + f"{first_letter}/{name}/{name}-{version}.tar.gz" + ] + # Bioconductor (R Biology Packages) + elif netloc == "bioconductor.org": + candidates = [ + f"https://bioconductor.org/packages/release/bioc/" + f"src/contrib/{name}_{version}.tar.gz" + ] + + # Test candidates to verify existence + for url in candidates: + if verify_url_existence(url): + return url + 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=5) + if response.status_code == 200: + return repo + except requests.RequestException: + return None + + return None + + +def get_url_netloc_namespace_and_name(url): + """ + Extract netloc, namespace, and name from a URL path. + - The last path component 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 + + name = parts[-1] + namespace = "/".join(parts[:-1]) if len(parts) > 1 else None + + return netloc, namespace, name + + +def get_mirrors_map(): + """ + Get the mirror map from mirrors.nix and export it in JSON format. + """ + nix_expression = ( + "builtins.removeAttrs " + "(import ) " + '[ "hashedMirrors" ]' + ) + + cmd = [ + "nix-instantiate", + "--eval", + "--json", + "--strict", + "-E", + nix_expression, + ] + + try: + result = subprocess.run(cmd, capture_output=True, text=True, check=True) + return json.loads(result.stdout) + except subprocess.CalledProcessError as e: + print("Error exporting mirrors mapping:", file=sys.stderr) + print(e.stderr, file=sys.stderr) + return {} + + +def covert_mirror_url(url): + """ + Convert a mirror:// URL into its actual direct URL + """ + from urllib.parse import urljoin + + mirror_map = get_mirrors_map() + mirror = url[len("mirror://") :] + # Split by "/" to separate components: the first is the type, and the rest form the path. + # For example, "mirror://sourceforge/enlightenment/imlib2-1.12.6.tar.xz": + # type = "sourceforge" + # path = "enlightenment/imlib2-1.12.6.tar.xz" + type = mirror.split("/")[0] + path = "/".join(mirror.split("/")[1:]) + mirror_urls = mirror_map.get(type, []) + for mirror_url in mirror_urls: + converted_url = urljoin(mirror_url, path) + if verify_url_existence(converted_url): + return converted_url + return None + + +def get_commit_hash(data, version): + """ + Get the commit hash. + """ + commit_hash = "" + releases = data.get("releases") + for release in releases: + if release["version"] == version: + commit_hash = release.get("platforms")[0].get("commit_hash") + break + return commit_hash + + +def get_nix_download_url(path): + """ + Construct a download url from cache.nixos.org based on the /nix/store/ + path + """ + narinfo_hash = path.replace("/nix/store/", "").split("-")[0] + narinfo_url = f"https://cache.nixos.org/{narinfo_hash}.narinfo" + url_path = get_narinfo_url(narinfo_url) + return f"https://cache.nixos.org/{url_path}" + + +def get_narinfo_url(narinfo_url): + """ + Visit the narinfo url, parsed and return the URL value + """ + # Fetch the narinfo file + response = requests.get(narinfo_url) + response.raise_for_status() + + # Parse line by line + for line in response.text.splitlines(): + if line.startswith("URL:"): + # Strip off "URL:" and any whitespace + return line.split(":", 1)[1].strip() + return None + + +def verify_url_existence(url): + """ + Performs a fast HTTP HEAD request to check if a generated URL is valid. + """ + try: + response = requests.head(url, allow_redirects=True, timeout=5) + if response.status_code == 200: + return True + elif response.status_code in (403, 429, 409): # forbidden, rate limit, conflict + return True # resource exists but not accessible + else: + return False + except Exception: + return False + + +def clarify_version_tag(type, namespace, name, version): + """ + Use github/gitlan API to verify the version tag + """ + normalized_version = version.lstrip("vV") + tags = [] + try: + if type == "github": + url = f"https://api.github.com/repos/{namespace}/{name}/tags" + + elif type == "gitlab": + project_path = f"{namespace}/{name}".strip("/").replace("/", "%2F") + url = f"https://gitlab.com/api/v4/projects/{project_path}/repository/tags" + else: + return None + response = requests.get(url) + response.raise_for_status() + tags = response.json() + except (requests.RequestException, ValueError): + return None + + for tag in tags: + tag_name = tag.get("name", "") + normalized_tag = tag_name + for prefix in ("release-", "RELEASE-", "v", "V"): + if normalized_tag.startswith(prefix): + normalized_tag = normalized_tag[len(prefix) :] + break + if normalized_version == normalized_tag: + return tag_name + return None + + +def clean_garbage(): + """ + Delete all unreferenced downloaded tarballs and evaluation caches + """ + subprocess.run(["nix-store", "--gc"], capture_output=True) From b8efdd174f1a27735e424e48b74a7d4249cea19e Mon Sep 17 00:00:00 2001 From: Chin Yeung Li Date: Thu, 18 Jun 2026 17:42:54 +0800 Subject: [PATCH 02/10] Pinned to a full-length commit SHA for the actions Signed-off-by: Chin Yeung Li --- .github/workflows/docs-ci.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/docs-ci.yml b/.github/workflows/docs-ci.yml index 57c86e6..cc6ca50 100644 --- a/.github/workflows/docs-ci.yml +++ b/.github/workflows/docs-ci.yml @@ -14,10 +14,10 @@ jobs: steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 + uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0 with: python-version: ${{ matrix.python-version }} From 47c57ff0d69faf0771df0b8d2796c22a9dfa466d Mon Sep 17 00:00:00 2001 From: Chin Yeung Li Date: Thu, 16 Jul 2026 17:40:00 +0800 Subject: [PATCH 03/10] Add tests and code enhancement #201 Signed-off-by: Chin Yeung Li --- src/fetchcode/download_urls.py | 3 +- src/fetchcode/nix.py | 293 +++++++++++++------------- tests/data/nix/hstr.json | 370 +++++++++++++++++++++++++++++++++ tests/test_download_urls.py | 1 + tests/test_nix.py | 115 ++++++++++ 5 files changed, 642 insertions(+), 140 deletions(-) create mode 100644 tests/data/nix/hstr.json create mode 100644 tests/test_nix.py diff --git a/src/fetchcode/download_urls.py b/src/fetchcode/download_urls.py index c0e89b3..0610153 100644 --- a/src/fetchcode/download_urls.py +++ b/src/fetchcode/download_urls.py @@ -22,8 +22,9 @@ from fetchcode.cran import CRAN from fetchcode.huggingface import Huggingface from fetchcode.pypi import Pypi +from fetchcode.nix import Nix -package_registry = [Pypi, CRAN, CPAN, Huggingface, Composer] +package_registry = [Pypi, CRAN, CPAN, Huggingface, Composer, Nix] router = Router() diff --git a/src/fetchcode/nix.py b/src/fetchcode/nix.py index 2d9238a..c0a7930 100644 --- a/src/fetchcode/nix.py +++ b/src/fetchcode/nix.py @@ -39,31 +39,20 @@ class Nix: def get_package_data(cls, purl): """ Fetch package data from https://search.devbox.sh/. - - Args: - purl: A Package URL string (e.g., "pkg:nix/nixpkgs/Axel@2.17.14") - - Returns: - The full JSON response from search.devbox.sh. """ parsed_purl = PackageURL.from_string(purl) api_url = f"https://search.devbox.sh/v2/pkg?name={parsed_purl.name}" - if not verify_url_existence(api_url): + try: + return fetch_json_response(api_url) + except Exception as e: + print(f"Failed to fetch package data for {purl}: {e}") return None - return fetch_json_response(api_url) - @classmethod def get_download_url(cls, purl): """ Get a single direct download URL of a system-specific binary. - - Args: - purl: A Package URL string (e.g., "pkg:nix/nixpkgs/Axel@2.17.14?system=x86_64-linux") - - Returns: - The download URL, or None if not found. """ purl_data = PackageURL.from_string(purl) have_nix = False @@ -72,26 +61,25 @@ def get_download_url(cls, purl): have_nix = True namespace = purl_data.namespace + # We will only work with the official nixpkgs repository, at least + # for now. + if not namespace or namespace.lower() != 'nixpkgs': + raise Exception( + "Only official nixpkgs repository is supported (i.e. namespace=nixpkgs)." + ) name = purl_data.name version = purl_data.version - qualifiers = purl_data.qualifiers - commit_hash = "" - system = "" - output = "" + qualifiers = purl_data.qualifiers or {} if "system" in qualifiers: - system = qualifiers.get("system") + system = qualifiers.get("system", "") else: raise Exception( - f"The 'system' qualifier is required to resolve system-specific binaries." + "The 'system' qualifier is required to resolve system-specific binaries." ) - if "commit" in qualifiers: - commit_hash = qualifiers.get("commit") - if "output" in qualifiers: - output = qualifiers.get("output") - else: - # Default 'out' if no output is defined - output = "out" + commit_hash = qualifiers.get("commit", "") + # Default 'out' if no output is defined + output = qualifiers.get("output", "out") data = cls.get_package_data(purl) path = "" @@ -104,31 +92,26 @@ def get_download_url(cls, purl): clean_garbage() if path: - narinfo_url = path.replace("/nix/store/", "").split("-")[0] - return get_nix_download_url(narinfo_url) + return get_nix_download_url(path) else: return None @classmethod - def get_src_download_url(cls, purl): + def get_upstream_src_download_url(cls, purl): """ - Get a single direct source download URL. - - If no version is specified in the PURL, fetches the latest version. - - Args: - purl: A Package URL string (e.g., "pkg:nix/nixpkgs/Axel@2.17.14") - - Returns: - The download URL, or None if not found. + Get a single upstream direct source download URL. """ - # Since Nix is not a central repository, packages are hosted in - # different locations, making it difficult to determine their - # direct source download URLs. We will provide a few well‑known - # repositories and construct direct download URLs when packages are - # hosted there. For all other hosts, the download_url will - # currently return None. Additional repositories might be supported - # progressively. + # Nix does not host a central repository of source code packages. + # Instead, each package definition (.nix file such as default.nix + # or package.nix) specifies how to fetch the upstream source (e.g. + # from GitHub, GitLab, SourceForge etc.) and then applies any + # patches or configuration during the build phases. The sources + # fetched are always the original, unmodified upstream archives. + # There is no direct URL for the patched or configured sources. + # This function is intended to return the direct upstream + # source download_url. Note that this may not represent the + # complete build input, since patches and configuration are applied + # later in the Nix build process. purl_data = PackageURL.from_string(purl) have_nix = False download_url = None @@ -136,25 +119,27 @@ def get_src_download_url(cls, purl): if shutil.which("nix") is not None: have_nix = True - namespace = "" + namespace = purl_data.namespace + # We will only work with the official nixpkgs repository, at least + # for now. + if not namespace or namespace.lower() != 'nixpkgs': + raise Exception( + "Only official nixpkgs repository is supported (i.e. namespace=nixpkgs)." + ) name = purl_data.name version = purl_data.version data = cls.get_package_data(purl) if data: - download_url = construct_url_based_on_netloc(namespace, name, version, data) - if not download_url and have_nix: - download_url = retrieve_src_download_url_with_nix(name, version) - # Try to use the commit_hash appraoch to get the download_url - if not download_url and version: - commit_hash = get_commit_hash(data, version) + download_url = construct_url_based_on_homepage_url(purl_data, data) + + if not download_url and have_nix: + download_url = retrieve_src_download_url_with_nix(name, version) + if not download_url and data and version: + commit_hash = get_commit_hash(data, version) + if commit_hash: download_url = retrieve_src_download_url_with_nix(name, version, commit_hash) - else: - # Use the `nix` command/utility to retrieve data. - # However, this only fetches the latest version. - if have_nix: - download_url = retrieve_src_download_url_with_nix(name, version) if not download_url and not have_nix: print("Install `nix` and re-run to let `nix` determine the download URL.") @@ -169,41 +154,24 @@ def get_nix_store_path(data, system, output, commit_hash=None, version=None): """ Find and return the store path (/nix/store/) based on the qualifiers """ - releases = data.get("releases") + releases = data.get("releases") or [] + + # Filter the list for specific version (no commit_hash provided) + if not commit_hash and version: + releases = [r for r in releases if r.get("version") == version] + for release in releases: - if commit_hash: - platforms = release.get("platforms") - for platform in platforms: - if system == platform.get("system") and commit_hash == platform.get("commit_hash"): - platform_outputs = platform.get("outputs") - for platform_output in platform.get("outputs"): - if output == platform_output.get("name"): - path = platform_output.get("path") - return path - return None - elif not commit_hash and version: - if version == release.get("version"): - platforms = release.get("platforms") - for platform in platforms: - if system == platform.get("system"): - platform_outputs = platform.get("outputs") - for platform_output in platform_outputs: - if output == platform_output.get("name"): - path = platform_output.get("path") - return path - return None - else: - # No version and no commit_hash. - # Default to the latest version. - platforms = release.get("platforms") - for platform in platforms: - if system == platform.get("system"): - platform_outputs = platform.get("outputs") - for platform_output in platform_outputs: - if output == platform_output.get("name"): - path = platform_output.get("path") - return path - return None + release_version = release.get("version", "") + if version and release_version != version: + continue + for platform in release.get("platforms", []): + if platform.get("system") != system: + continue + if commit_hash and platform.get("commit_hash") != commit_hash: + continue + for out in platform.get("outputs", []): + if out.get("name") == output: + return out.get("path") return None @@ -231,41 +199,54 @@ def get_nix_store_path_with_nix(name, system, output, commit_hash): return None -def retrieve_src_download_url_with_nix(name, version, commit_hash=None): +def retrieve_src_download_url_with_nix(name, version=None, commit_hash=None): """ Find and return the source download url using 'nix' """ info = get_src_info(name, commit_hash) + urls = info.get("urls", []) download_url = None # We don't need to check for version if we have the commit_hash if commit_hash: - for url in info["urls"]: + for url in urls: if url.startswith("mirror:"): - download_url = covert_mirror_url(url) + download_url = convert_mirror_url(url) else: if verify_url_existence(url): download_url = url + if download_url: + break else: if not version: - for url in info["urls"]: + for url in urls: if url.startswith("mirror:"): - download_url = covert_mirror_url(url) + download_url = convert_mirror_url(url) else: if verify_url_existence(url): download_url = url + if download_url: + break else: - # Attempt to replace the version and validate that the - # URL exists. Return None if the URL is invalid. - latest_version = info["version"] - for url in info["urls"]: + # Attempt to replace the fetched version with the input version + # and validate that the URL exists. Return None if the URL is + # invalid. + fetched_latest_version = info.get("version") + for url in urls: if url.startswith("mirror:"): - converted_url = covert_mirror_url(url) - updated_version_url = converted_url.replace(latest_version, version) + converted_url = convert_mirror_url(url) + if converted_url and fetched_latest_version: + updated_version_url = version.join(converted_url.rsplit(fetched_latest_version, 1)) + else: + continue else: - updated_version_url = url.replace(latest_version, version) + if not fetched_latest_version: + continue + updated_version_url = version.join(url.rsplit(fetched_latest_version, 1)) + if verify_url_existence(updated_version_url): download_url = updated_version_url + break return download_url @@ -310,26 +291,39 @@ def get_src_info(attr_path, commit_hash=None): try: result = subprocess.run(cmd, capture_output=True, text=True, check=True) return json.loads(result.stdout) - except subprocess.CalledProcessError as e: + except (subprocess.CalledProcessError, json.JSONDecodeError) as e: print(f"Error evaluating attribute '{attr_path}':", file=sys.stderr) print(e.stderr, file=sys.stderr) return {"version": None, "urls": []} -def construct_url_based_on_netloc(namespace, name, version, data): +def construct_url_based_on_homepage_url(input_purl, data): """ - Determine and return the download url based on the homepage, name and + Determine and return the download url based on the homepage and version """ homepage_url = data.get("homepage_url", None) + if not homepage_url: + return None + + netloc = "" # Get the latest version if no version is provided + if not input_purl.version: + releases = data.get("releases") + if not releases: + return None + version = releases[0].get("version") + else: + version = input_purl.version + if not version: - version = data.get("releases")[0].get("version") + return None netloc, namespace, name = get_url_netloc_namespace_and_name(homepage_url) if netloc.endswith("github.io"): github_page_url = github_pages_to_repo(homepage_url) - netloc, namespace, name = get_url_netloc_namespace_and_name(github_page_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"): @@ -436,7 +430,7 @@ def github_pages_to_repo(url): # Verify existence via GitHub API api_url = f"https://api.github.com/repos/{org}/{name}" try: - response = requests.get(api_url, timeout=5) + response = requests.get(api_url, timeout=10) if response.status_code == 200: return repo except requests.RequestException: @@ -486,13 +480,13 @@ def get_mirrors_map(): try: result = subprocess.run(cmd, capture_output=True, text=True, check=True) return json.loads(result.stdout) - except subprocess.CalledProcessError as e: + except (subprocess.CalledProcessError, json.JSONDecodeError) as e: print("Error exporting mirrors mapping:", file=sys.stderr) print(e.stderr, file=sys.stderr) return {} -def covert_mirror_url(url): +def convert_mirror_url(url): """ Convert a mirror:// URL into its actual direct URL """ @@ -500,13 +494,14 @@ def covert_mirror_url(url): mirror_map = get_mirrors_map() mirror = url[len("mirror://") :] - # Split by "/" to separate components: the first is the type, and the rest form the path. + # Split by "/" to separate components: the first is the mirror_type, + # and the rest form the path. # For example, "mirror://sourceforge/enlightenment/imlib2-1.12.6.tar.xz": - # type = "sourceforge" + # mirror_type = "sourceforge" # path = "enlightenment/imlib2-1.12.6.tar.xz" - type = mirror.split("/")[0] + mirror_type = mirror.split("/")[0] path = "/".join(mirror.split("/")[1:]) - mirror_urls = mirror_map.get(type, []) + mirror_urls = mirror_map.get(mirror_type, []) for mirror_url in mirror_urls: converted_url = urljoin(mirror_url, path) if verify_url_existence(converted_url): @@ -518,13 +513,13 @@ def get_commit_hash(data, version): """ Get the commit hash. """ - commit_hash = "" - releases = data.get("releases") + releases = data.get("releases") or [] for release in releases: - if release["version"] == version: - commit_hash = release.get("platforms")[0].get("commit_hash") - break - return commit_hash + if release.get("version") == version: + platforms = release.get("platforms") or [] + if platforms: + return platforms[0].get("commit_hash") + return None def get_nix_download_url(path): @@ -532,9 +527,15 @@ def get_nix_download_url(path): Construct a download url from cache.nixos.org based on the /nix/store/ path """ - narinfo_hash = path.replace("/nix/store/", "").split("-")[0] + base_name = path.rstrip("/").split("/")[-1] + narinfo_hash = base_name.split("-")[0] + narinfo_url = f"https://cache.nixos.org/{narinfo_hash}.narinfo" url_path = get_narinfo_url(narinfo_url) + + if not url_path: + return None + return f"https://cache.nixos.org/{url_path}" @@ -543,14 +544,18 @@ def get_narinfo_url(narinfo_url): Visit the narinfo url, parsed and return the URL value """ # Fetch the narinfo file - response = requests.get(narinfo_url) - response.raise_for_status() + try: + response = requests.get(narinfo_url, timeout=10) + response.raise_for_status() + except requests.exceptions.RequestException: + return None # Parse line by line for line in response.text.splitlines(): if line.startswith("URL:"): # Strip off "URL:" and any whitespace return line.split(":", 1)[1].strip() + return None @@ -559,7 +564,7 @@ def verify_url_existence(url): Performs a fast HTTP HEAD request to check if a generated URL is valid. """ try: - response = requests.head(url, allow_redirects=True, timeout=5) + response = requests.head(url, allow_redirects=True, timeout=10) if response.status_code == 200: return True elif response.status_code in (403, 429, 409): # forbidden, rate limit, conflict @@ -570,22 +575,23 @@ def verify_url_existence(url): return False -def clarify_version_tag(type, namespace, name, version): +def clarify_version_tag(repo_type, namespace, name, version): """ Use github/gitlan API to verify the version tag """ - normalized_version = version.lstrip("vV") + normalized_version = normalize_string(version) tags = [] try: - if type == "github": + if repo_type == "github": url = f"https://api.github.com/repos/{namespace}/{name}/tags" - elif type == "gitlab": - project_path = f"{namespace}/{name}".strip("/").replace("/", "%2F") + elif repo_type == "gitlab": + ns = namespace if namespace else "" + project_path = f"{ns}/{name}".strip("/").replace("/", "%2F") url = f"https://gitlab.com/api/v4/projects/{project_path}/repository/tags" else: return None - response = requests.get(url) + response = requests.get(url, timeout=10) response.raise_for_status() tags = response.json() except (requests.RequestException, ValueError): @@ -593,18 +599,27 @@ def clarify_version_tag(type, namespace, name, version): for tag in tags: tag_name = tag.get("name", "") - normalized_tag = tag_name - for prefix in ("release-", "RELEASE-", "v", "V"): - if normalized_tag.startswith(prefix): - normalized_tag = normalized_tag[len(prefix) :] - break + normalized_tag = normalize_string(tag_name) if normalized_version == normalized_tag: return tag_name return None +def normalize_string(version_string): + """ + Normalize common prefix version string + """ + for prefix in ("release-", "RELEASE-", "v", "V"): + if version_string.startswith(prefix): + return version_string[len(prefix):] + return version_string + + def clean_garbage(): """ Delete all unreferenced downloaded tarballs and evaluation caches """ - subprocess.run(["nix-store", "--gc"], capture_output=True) + try: + subprocess.run(["nix-store", "--gc"], capture_output=True) + except FileNotFoundError: + pass diff --git a/tests/data/nix/hstr.json b/tests/data/nix/hstr.json new file mode 100644 index 0000000..2d65678 --- /dev/null +++ b/tests/data/nix/hstr.json @@ -0,0 +1,370 @@ +{ + "name": "hstr", + "summary": "Shell history suggest box - easily view, navigate, search and use your command history", + "homepage_url": "https://github.com/dvorka/hstr", + "license": "Apache-2.0", + "releases": [ + { + "version": "3.2", + "last_updated": "2026-06-27T07:37:20Z", + "platforms": [ + { + "arch": "x86-64", + "os": "Linux", + "system": "x86_64-linux", + "attribute_path": "hstr", + "commit_hash": "3d46470bb3030020f7e1361f33514854f5bfa86d", + "date": "2026-06-27T07:37:20Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/i8dbqy1jr94q3m3bmxx1d5g26jm74jc7-hstr-3.2", + "default": true + } + ] + }, + { + "arch": "x86-64", + "os": "macOS", + "system": "x86_64-darwin", + "attribute_path": "hstr", + "commit_hash": "3d46470bb3030020f7e1361f33514854f5bfa86d", + "date": "2026-06-27T07:37:20Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/r910fm5w0iqywcwflp9fx1dsiwf8kqnx-hstr-3.2", + "default": true + } + ] + }, + { + "arch": "arm64", + "os": "Linux", + "system": "aarch64-linux", + "attribute_path": "hstr", + "commit_hash": "3d46470bb3030020f7e1361f33514854f5bfa86d", + "date": "2026-06-27T07:37:20Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/a43k2mgjnkcr101kriy3sy5xv7604prn-hstr-3.2", + "default": true + } + ] + }, + { + "arch": "arm64", + "os": "macOS", + "system": "aarch64-darwin", + "attribute_path": "hstr", + "commit_hash": "3d46470bb3030020f7e1361f33514854f5bfa86d", + "date": "2026-06-27T07:37:20Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/b074ik3wwddnax8vrb00q3xn6hcgnh5h-hstr-3.2", + "default": true + } + ] + } + ], + "platforms_summary": "Linux and macOS", + "outputs_summary": "" + }, + { + "version": "3.1", + "last_updated": "2026-05-21T08:15:18Z", + "platforms": [ + { + "arch": "x86-64", + "os": "Linux", + "system": "x86_64-linux", + "attribute_path": "hstr", + "commit_hash": "4a29d733e8a7d5b824c3d8c958a946a9867b3eb2", + "date": "2026-05-21T08:15:18Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/6vi4la60chs9pqh1hmr3i4mhrzv3kcsm-hstr-3.1", + "default": true + } + ] + }, + { + "arch": "x86-64", + "os": "macOS", + "system": "x86_64-darwin", + "attribute_path": "hstr", + "commit_hash": "4a29d733e8a7d5b824c3d8c958a946a9867b3eb2", + "date": "2026-05-21T08:15:18Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/vcildmf4v1jkci82ny3bfkhn6nlnf6nn-hstr-3.1", + "default": true + } + ] + }, + { + "arch": "arm64", + "os": "Linux", + "system": "aarch64-linux", + "attribute_path": "hstr", + "commit_hash": "4a29d733e8a7d5b824c3d8c958a946a9867b3eb2", + "date": "2026-05-21T08:15:18Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/11hr7jpg2cciyfabhkwpxwldglil4lnx-hstr-3.1", + "default": true + } + ] + }, + { + "arch": "arm64", + "os": "macOS", + "system": "aarch64-darwin", + "attribute_path": "hstr", + "commit_hash": "4a29d733e8a7d5b824c3d8c958a946a9867b3eb2", + "date": "2026-05-21T08:15:18Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/vh8m653ps0n96lg68w90sf4xbi9n2nvx-hstr-3.1", + "default": true + } + ] + } + ], + "platforms_summary": "Linux and macOS", + "outputs_summary": "" + }, + { + "version": "2.6", + "last_updated": "2023-04-13T03:55:09Z", + "platforms": [ + { + "arch": "x86-64", + "os": "Linux", + "system": "x86_64-linux", + "attribute_path": "hstr", + "commit_hash": "96ba1c52e54e74c3197f4d43026b3f3d92e83ff9", + "date": "2023-04-13T03:55:09Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/wrq1ywzjh0zmmywrm28xvrp75k7z3986-hstr-2.6", + "default": true, + "nar": "nar/0ij9ia9sdjrjzq4c8c6501661aj47jdns4g9rcrld3p6slvbg7p8.nar.xz" + } + ] + }, + { + "arch": "x86-64", + "os": "macOS", + "system": "x86_64-darwin", + "attribute_path": "hstr", + "commit_hash": "96ba1c52e54e74c3197f4d43026b3f3d92e83ff9", + "date": "2023-04-13T03:55:09Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/6fikr863aq1pjrzmqf4mi4fbzdmsr03n-hstr-2.6", + "default": true, + "nar": "nar/00f87phwxliq9pscsvmwh3r842ifrgjlghaix7mkx6i4gmqxmqjr.nar.xz" + } + ] + }, + { + "arch": "arm64", + "os": "macOS", + "system": "aarch64-darwin", + "attribute_path": "hstr", + "commit_hash": "96ba1c52e54e74c3197f4d43026b3f3d92e83ff9", + "date": "2023-04-13T03:55:09Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/82hrwkfgp2vsjsh38lqzrk3sls8gfhjy-hstr-2.6", + "default": true, + "nar": "nar/1dbypgck46balmkwzccbkdjwzkk67wm7183dr4yr7yzsfxh3jzh9.nar.xz" + } + ] + }, + { + "arch": "arm64", + "os": "Linux", + "system": "aarch64-linux", + "attribute_path": "hstr", + "commit_hash": "96ba1c52e54e74c3197f4d43026b3f3d92e83ff9", + "date": "2023-04-13T03:55:09Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/bb0jc868vmnxhdc5p00kfzcc7917hf0n-hstr-2.6", + "default": true, + "nar": "nar/11v0k6w7y22crvdsh7c23amai4i240vffg9spnw5wgswzj3w03mj.nar.xz" + } + ] + } + ], + "platforms_summary": "Linux and macOS", + "outputs_summary": "" + }, + { + "version": "2.5", + "last_updated": "2022-12-17T09:19:40Z", + "platforms": [ + { + "arch": "x86-64", + "os": "Linux", + "system": "x86_64-linux", + "attribute_path": "hstr", + "commit_hash": "80c24eeb9ff46aa99617844d0c4168659e35175f", + "date": "2022-12-17T09:19:40Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/7cn897g25ybypr2mds7jglhc48w6dvmk-hstr-2.5", + "default": true, + "nar": "nar/103fp6ka35wr4ib0dz5myk1nmql9fxy3m4bh86j18idv4ky6jj3r.nar.xz" + } + ] + }, + { + "arch": "x86-64", + "os": "macOS", + "system": "x86_64-darwin", + "attribute_path": "hstr", + "commit_hash": "80c24eeb9ff46aa99617844d0c4168659e35175f", + "date": "2022-12-17T09:19:40Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/pxzpdihf8q9axxv8gic02g8ag910rnj8-hstr-2.5", + "default": true, + "nar": "nar/1gwahs8fpbfphr1wznryx4yml640gn6vkr8nwqysvayp89nng6s7.nar.xz" + } + ] + }, + { + "arch": "arm64", + "os": "Linux", + "system": "aarch64-linux", + "attribute_path": "hstr", + "commit_hash": "80c24eeb9ff46aa99617844d0c4168659e35175f", + "date": "2022-12-17T09:19:40Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/3143m8ihxzqg2ibhl02kvkry5kz9q8lv-hstr-2.5", + "default": true, + "nar": "nar/0pggi3p3gzq59dx95w3182wa6v2knaspwap7z0v710428w8sg0km.nar.xz" + } + ] + }, + { + "arch": "arm64", + "os": "macOS", + "system": "aarch64-darwin", + "attribute_path": "hstr", + "commit_hash": "80c24eeb9ff46aa99617844d0c4168659e35175f", + "date": "2022-12-17T09:19:40Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/b69pqvxlsc9wyx9cyyk482sbfmhl2rvp-hstr-2.5", + "default": true, + "nar": "nar/1xcrmpf2bb93wnz9cwij0dw82xwi9iaa0an8ps8jg18vzf89h6k5.nar.xz" + } + ] + } + ], + "platforms_summary": "Linux and macOS", + "outputs_summary": "" + }, + { + "version": "2.3", + "last_updated": "2022-05-04T19:02:07Z", + "platforms": [ + { + "arch": "arm64", + "os": "macOS", + "system": "aarch64-darwin", + "attribute_path": "hstr", + "commit_hash": "b4cc9cd38f05f1a764e21bfb1b14e89be76068b0", + "date": "2022-05-04T19:02:07Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/4hkmy0rn7mz7zr1jid9hbbq54j7q3z9c-hstr-2.3", + "default": true, + "nar": "nar/1q49y1kdnjx5rahjy4a0v8flpk2dx4p0yqsbl5w6sc45d28nv5wn.nar.xz" + } + ] + }, + { + "arch": "x86-64", + "os": "macOS", + "system": "x86_64-darwin", + "attribute_path": "hstr", + "commit_hash": "b4cc9cd38f05f1a764e21bfb1b14e89be76068b0", + "date": "2022-05-04T19:02:07Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/6z3pc1xd7qib4zvwwrwrrppj0m5d2vkb-hstr-2.3", + "default": true, + "nar": "nar/1fmqrq7xhm0b7nzxcvvs7px06rgiwqnd81smcm9rlxyridf4p6qx.nar.xz" + } + ] + }, + { + "arch": "arm64", + "os": "Linux", + "system": "aarch64-linux", + "attribute_path": "hstr", + "commit_hash": "24c33ab7952544ad355d0677c9eea931b23f371c", + "date": "2022-05-03T02:40:48Z", + "outputs": [ + { + "name": "out", + "path": "/nix/store/pxchya4wfjr2ij3m78knyghcq65plsz3-hstr-2.3", + "default": true, + "nar": "nar/15sr6brwan6kvgybka8hgjiifyqyc1h3hy52k10fklhkrp8qj3k6.nar.xz" + } + ] + }, + { + "arch": "x86-64", + "os": "Linux", + "system": "x86_64-linux", + "attribute_path": "hstr", + "commit_hash": "24c33ab7952544ad355d0677c9eea931b23f371c", + "date": "2022-05-03T02:40:48Z", + "outputs": null + } + ], + "platforms_summary": "Linux and macOS", + "outputs_summary": "" + }, + { + "version": "2.2", + "last_updated": "2020-11-18T16:15:21Z", + "platforms": [ + { + "arch": "x86-64", + "os": "Linux", + "system": "x86_64-linux", + "attribute_path": "hstr", + "commit_hash": "6625284c397b44bc9518a5a1567c1b5aae455c08", + "date": "2020-11-18T16:15:21Z", + "outputs": null + } + ], + "platforms_summary": "Linux", + "outputs_summary": "" + } + ] +} diff --git a/tests/test_download_urls.py b/tests/test_download_urls.py index 464de85..ec1d0ed 100644 --- a/tests/test_download_urls.py +++ b/tests/test_download_urls.py @@ -29,6 +29,7 @@ def test_right_class_being_called_for_the_purls(): "pkg:cpan/EXAMPLE/Some-Module@1.2.3", "pkg:composer/laravel/framework@10.0.0", "pkg:cran/dplyr@1.0.0", + "pkg:nix/nixpkgs/test@1.2.3", ] with patch("fetchcode.download_urls.Router.process") as mock_fetch: diff --git a/tests/test_nix.py b/tests/test_nix.py new file mode 100644 index 0000000..e162320 --- /dev/null +++ b/tests/test_nix.py @@ -0,0 +1,115 @@ +# fetchcode is a free software tool from nexB Inc. and others. +# Visit https://github.com/aboutcode-org/fetchcode for support and download. +# +# Copyright (c) nexB Inc. and others. All rights reserved. +# http://nexb.com and http://aboutcode.org +# +# This software is licensed under the Apache License version 2.0. +# +# 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. + +import json +import unittest +from pathlib import Path +from unittest.mock import patch + +from fetchcode import nix +from fetchcode.nix import Nix + +from packageurl import PackageURL + +DATA_DIR = Path(__file__).parent / "data" / "nix" + + +def load_fixture(filename): + with open(DATA_DIR / filename) as f: + return json.load(f) + + +HSTR_RESPONSE = load_fixture("hstr.json") + + +def mock_fetch_json_response(url): + """Return appropriate fixture based on request URL.""" + if url == "https://search.devbox.sh/v2/pkg?name=hstr": + return HSTR_RESPONSE + raise ValueError(f"Unexpected URL: {url}") + + +@patch("fetchcode.nix.fetch_json_response", side_effect=mock_fetch_json_response) +class TestNix(unittest.TestCase): + def test_get_package_data(self, mock_fetch): + purl = "pkg:nix/nixpkgs/hstr" + result = Nix.get_package_data(purl) + self.assertEqual(result["name"], "hstr") + self.assertEqual(result["license"], "Apache-2.0") + + def test_get_nix_store_path_no_version(self, mock_fetch): + purl = "pkg:nix/nixpkgs/hstr?system=x86_64-darwin" + data = Nix.get_package_data(purl) + system = "x86_64-darwin" + output = "out" + path = nix.get_nix_store_path(data, system, output) + self.assertEqual(path, "/nix/store/r910fm5w0iqywcwflp9fx1dsiwf8kqnx-hstr-3.2") + + def test_get_nix_store_path_with_version(self, mock_fetch): + purl = "pkg:nix/nixpkgs/hstr@3.1?system=x86_64-darwin" + data = Nix.get_package_data(purl) + system = "x86_64-darwin" + output = "out" + commit_hash = None + version = "3.1" + path = nix.get_nix_store_path(data, system, output, commit_hash, version) + self.assertEqual(path, "/nix/store/vcildmf4v1jkci82ny3bfkhn6nlnf6nn-hstr-3.1") + + def test_get_nix_store_path_with_commit_hash(self, mock_fetch): + purl = "pkg:nix/nixpkgs/hstr?system=x86_64-darwin%commit_hash=4a29d733e8a7d5b824c3d8c958a946a9867b3eb2" + data = Nix.get_package_data(purl) + system = "aarch64-darwin" + output = "out" + commit_hash = "4a29d733e8a7d5b824c3d8c958a946a9867b3eb2" + version = None + path = nix.get_nix_store_path(data, system, output, commit_hash, version) + self.assertEqual(path, "/nix/store/vh8m653ps0n96lg68w90sf4xbi9n2nvx-hstr-3.1") + + def test_get_nix_store_path_version_commit_hash_not_match(self, mock_fetch): + purl = "pkg:nix/nixpkgs/hstr@3.2?system=x86_64-darwin&commit_hash=4a29d733e8a7d5b824c3d8c958a946a9867b3eb2" + data = Nix.get_package_data(purl) + system = "aarch64-darwin" + output = "out" + commit_hash = "4a29d733e8a7d5b824c3d8c958a946a9867b3eb2" + version = "3.2" + path = nix.get_nix_store_path(data, system, output, commit_hash, version) + self.assertEqual(path, None) + + def test_construct_url_based_on_homepage_url(self, mock_fetch): + purl_str = "pkg:nix/nixpkgs/hstr?system=x86_64-darwin%commit_hash=4a29d733e8a7d5b824c3d8c958a946a9867b3eb2" + purl = PackageURL.from_string(purl_str) + data = Nix.get_package_data(purl_str) + result = nix.construct_url_based_on_homepage_url(purl, data) + self.assertEqual(result, "https://github.com/dvorka/hstr/archive/v3.2.tar.gz") + + def test_get_url_netloc_namespace_and_name(self, mock_fetch): + url = "https://github.com/dvorka/hstr" + netloc, namespace, name = nix.get_url_netloc_namespace_and_name(url) + self.assertEqual(netloc, "github.com") + self.assertEqual(namespace, "dvorka") + self.assertEqual(name, "hstr") + + def test_get_commit_hash(self, mock_fetch): + purl_str = "pkg:nix/nixpkgs/hstr@2.6?system=x86_64-darwin" + data = Nix.get_package_data(purl_str) + version = "2.6" + commit_hash = nix.get_commit_hash(data, version) + self.assertEqual(commit_hash, "96ba1c52e54e74c3197f4d43026b3f3d92e83ff9") + + def test_normalize_string(self, mock_fetch): + version = nix.normalize_string("release-1.2.3") + self.assertEqual(version, "1.2.3") + From b49c23722ff18500f85d338a7b2379aa3550e12c Mon Sep 17 00:00:00 2001 From: Chin Yeung Li Date: Tue, 21 Jul 2026 17:31:31 +0800 Subject: [PATCH 04/10] Code enhancement #201 * version is now required * better error handling * etc... Signed-off-by: Chin Yeung Li --- src/fetchcode/download_urls.py | 2 +- src/fetchcode/nix.py | 208 +++++++++++++++++++-------------- tests/test_nix.py | 119 +++++++++++++++---- 3 files changed, 222 insertions(+), 107 deletions(-) diff --git a/src/fetchcode/download_urls.py b/src/fetchcode/download_urls.py index 0610153..45e903d 100644 --- a/src/fetchcode/download_urls.py +++ b/src/fetchcode/download_urls.py @@ -21,8 +21,8 @@ from fetchcode.cpan import CPAN from fetchcode.cran import CRAN from fetchcode.huggingface import Huggingface -from fetchcode.pypi import Pypi from fetchcode.nix import Nix +from fetchcode.pypi import Pypi package_registry = [Pypi, CRAN, CPAN, Huggingface, Composer, Nix] diff --git a/src/fetchcode/nix.py b/src/fetchcode/nix.py index c0a7930..8473cfb 100644 --- a/src/fetchcode/nix.py +++ b/src/fetchcode/nix.py @@ -15,6 +15,7 @@ # specific language governing permissions and limitations under the License. import json +import os import shutil import subprocess import sys @@ -63,12 +64,14 @@ def get_download_url(cls, purl): namespace = purl_data.namespace # We will only work with the official nixpkgs repository, at least # for now. - if not namespace or namespace.lower() != 'nixpkgs': + if not namespace or namespace.lower() != "nixpkgs": raise Exception( "Only official nixpkgs repository is supported (i.e. namespace=nixpkgs)." ) name = purl_data.name version = purl_data.version + if not version: + raise Exception("Version is requierd.") qualifiers = purl_data.qualifiers or {} if "system" in qualifiers: @@ -85,14 +88,22 @@ def get_download_url(cls, purl): path = "" if data: - path = get_nix_store_path(data, system, output, commit_hash, version) + path = get_nix_store_path(data, system, output, version, commit_hash) if not data or not path: - if have_nix and commit_hash: + if have_nix: + if not commit_hash: + print( + "Please provide a 'commit' qualifier " + "in the PURL for Nix to determine the download URL." + ) + return None path = get_nix_store_path_with_nix(name, system, output, commit_hash) - clean_garbage() if path: - return get_nix_download_url(path) + try: + return get_nix_download_url(path) + finally: + delete_nix_store_path(path) else: return None @@ -122,13 +133,14 @@ def get_upstream_src_download_url(cls, purl): namespace = purl_data.namespace # We will only work with the official nixpkgs repository, at least # for now. - if not namespace or namespace.lower() != 'nixpkgs': + if not namespace or namespace.lower() != "nixpkgs": raise Exception( "Only official nixpkgs repository is supported (i.e. namespace=nixpkgs)." ) name = purl_data.name version = purl_data.version - + if not version: + raise Exception("Version is requierd.") data = cls.get_package_data(purl) if data: @@ -144,25 +156,21 @@ def get_upstream_src_download_url(cls, purl): if not download_url and not have_nix: print("Install `nix` and re-run to let `nix` determine the download URL.") - if have_nix: - clean_garbage() - return download_url -def get_nix_store_path(data, system, output, commit_hash=None, version=None): +def get_nix_store_path(data, system, output, version, commit_hash=None): """ Find and return the store path (/nix/store/) based on the qualifiers """ releases = data.get("releases") or [] # Filter the list for specific version (no commit_hash provided) - if not commit_hash and version: - releases = [r for r in releases if r.get("version") == version] + releases = [r for r in releases if r.get("version") == version] for release in releases: release_version = release.get("version", "") - if version and release_version != version: + if release_version != version: continue for platform in release.get("platforms", []): if platform.get("system") != system: @@ -182,16 +190,17 @@ def get_nix_store_path_with_nix(name, system, output, commit_hash): """ system_config = f'system = "{system}";' if system else "" output_modifier = f".{output}" if output else "" + config_str = "config = { allowBroken = true; allowUnfree = true; };" nix_expression = ( f'(import (fetchTarball "https://github.com/NixOS/nixpkgs/archive/{commit_hash}.tar.gz") ' - f"{{ {system_config} }}).{name}{output_modifier}.outPath" + f"{{ {system_config} {config_str} }}).{name}{output_modifier}.outPath" ) cmd = ["nix-instantiate", "--eval", "--raw", "-E", nix_expression] try: - result = subprocess.run(cmd, capture_output=True, text=True, check=True) + result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=300) return result.stdout.strip() except subprocess.CalledProcessError as e: print(f"Error evaluating attribute for package '{name}'", file=sys.stderr) @@ -232,17 +241,17 @@ def retrieve_src_download_url_with_nix(name, version=None, commit_hash=None): # and validate that the URL exists. Return None if the URL is # invalid. fetched_latest_version = info.get("version") + if not fetched_latest_version: + return download_url for url in urls: if url.startswith("mirror:"): converted_url = convert_mirror_url(url) - if converted_url and fetched_latest_version: - updated_version_url = version.join(converted_url.rsplit(fetched_latest_version, 1)) + if converted_url: + updated_version_url = converted_url.replace(fetched_latest_version, version) else: continue else: - if not fetched_latest_version: - continue - updated_version_url = version.join(url.rsplit(fetched_latest_version, 1)) + updated_version_url = url.replace(fetched_latest_version, version) if verify_url_existence(updated_version_url): download_url = updated_version_url @@ -257,14 +266,16 @@ def get_src_info(attr_path, commit_hash=None): retrieve the package’s version and download URL. Return a dictionary with "version" and "urls" keys. """ + config_str = "config = { allowBroken = true; allowUnfree = true; };" + # Determine the repository entry point definition if commit_hash: nixpkgs_import = ( 'import (fetchTarball "https://github.com/NixOS/nixpkgs/archive/' - f'{commit_hash}.tar.gz") {{}}' + f'{commit_hash}.tar.gz") {{ {config_str} }}' ) else: - nixpkgs_import = "import {}" + nixpkgs_import = "import {{ {config_str} }}" nix_expression = f""" let @@ -289,7 +300,7 @@ def get_src_info(attr_path, commit_hash=None): nix_expression, ] try: - result = subprocess.run(cmd, capture_output=True, text=True, check=True) + result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=300) return json.loads(result.stdout) except (subprocess.CalledProcessError, json.JSONDecodeError) as e: print(f"Error evaluating attribute '{attr_path}':", file=sys.stderr) @@ -307,17 +318,17 @@ def construct_url_based_on_homepage_url(input_purl, data): return None netloc = "" - # Get the latest version if no version is provided - if not input_purl.version: - releases = data.get("releases") - if not releases: - return None - version = releases[0].get("version") - else: - version = input_purl.version + version = input_purl.version + qualifiers = input_purl.qualifiers or {} + commit_hash = qualifiers.get("commit", "") if not version: - return None + if not commit_hash: + return None + else: + version = get_version_from_commit_hash(data, commit_hash) + if not version: + return None netloc, namespace, name = get_url_netloc_namespace_and_name(homepage_url) if netloc.endswith("github.io"): @@ -333,32 +344,24 @@ def construct_url_based_on_homepage_url(input_purl, data): version = clarified_version elif netloc.endswith(".org"): package_type = netloc.removesuffix(".org") - # There is an issue where the version may have a 'v' prefix, which - # makes the constructed download URL invalid. For example, in - # https://github.com/containers/aardvark-dns/, the constructed - # download_url is - # https://github.com/containers/aardvark-dns/archive/1.8.0.tar.gz, but - # this returns a 404 because the actual download URL should be - # https://github.com/containers/aardvark-dns/archive/v1.8.0.tar.gz. - - # Users are responsible for providing the correct version string - # (including any 'v' prefix). However, from a mining situation, the - # source data we collect may omit the prefix, so we need to handle that - # case. For example, versions from - # https://search.devbox.sh/v2/pkg?name=CuboCore.corepins do not include - # the 'v' prefix, but the actual versions from the download site - # include it in the tag field: + # There is an issue where the version may have a different prefix. + # For example, versions from + # https://search.devbox.sh/v2/pkg?name=CuboCore.corepins do not + # include the 'v' prefix, but the actual versions from the download + # site include it in the tag field: # https://gitlab.com/api/v4/projects/cubocore%2Fcoreapps%2Fcorepins/repository/tags - # There are also cases that use other prefix such as release-{version} - # See https://search.devbox.sh/v2/pkg?name=SDL2_mixer where one of - # the versions is 2.8.2 while the tag from the github is - # release-2.8.2 + # There are also cases that use other prefixes such as + # release-{version} See + # https://search.devbox.sh/v2/pkg?name=SDL2_mixer where one of the + # versions is 2.8.2 while the tag from the github is release-2.8.2 # (https://github.com/libsdl-org/SDL_mixer/releases/tag/release-2.8.2) # We want to validate whether the returned download URL is # accessible. If it is not, insert a common prefix and try to # validate again. + # This is actually purely for bitbucket.org as the version is + # already checked for github and gitlab at clarify_version_tag() common_prefixes = ["v", "V", "v-", "V-", "release-", "RELEASE-"] download_url = get_repo_download_url_by_package_type( type=package_type, namespace=namespace, name=name, version=version @@ -380,6 +383,7 @@ def construct_url_based_on_homepage_url(input_purl, data): if verify_url_existence(download_url): return download_url else: + # This list can be improved and added over time. candidates = [] # CRAN (R Packages) if netloc == "cran.r-project.org": @@ -442,7 +446,7 @@ def github_pages_to_repo(url): def get_url_netloc_namespace_and_name(url): """ Extract netloc, namespace, and name from a URL path. - - The last path component is considered the name. + - The last path component (except for web files) is considered the name. - Everything between netloc and name is considered the namespace. """ parsed = urlparse(url) @@ -452,6 +456,12 @@ def get_url_netloc_namespace_and_name(url): 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 @@ -478,7 +488,7 @@ def get_mirrors_map(): ] try: - result = subprocess.run(cmd, capture_output=True, text=True, check=True) + result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=300) return json.loads(result.stdout) except (subprocess.CalledProcessError, json.JSONDecodeError) as e: print("Error exporting mirrors mapping:", file=sys.stderr) @@ -522,6 +532,21 @@ def get_commit_hash(data, version): return None +def get_version_from_commit_hash(data, commit_hash): + """ + Get the version. + """ + releases = data.get("releases") or [] + for release in releases: + release_version = release.get("version", "") + platforms = release.get("platforms") or [] + for platform in platforms: + platform_hash = platform.get("commit_hash", "") + if platform_hash == commit_hash: + return release_version + return None + + def get_nix_download_url(path): """ Construct a download url from cache.nixos.org based on the /nix/store/ @@ -563,6 +588,8 @@ def verify_url_existence(url): """ Performs a fast HTTP HEAD request to check if a generated URL is valid. """ + if not url: + return False try: response = requests.head(url, allow_redirects=True, timeout=10) if response.status_code == 200: @@ -577,49 +604,58 @@ def verify_url_existence(url): def clarify_version_tag(repo_type, namespace, name, version): """ - Use github/gitlan API to verify the version tag - """ - normalized_version = normalize_string(version) - tags = [] - try: - if repo_type == "github": - url = f"https://api.github.com/repos/{namespace}/{name}/tags" + 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}" + elif repo_type == "gitlab": + gitlab_token = os.environ.get("GITLAB_TOKEN") + if gitlab_token: + headers["PRIVATE-TOKEN"] = gitlab_token + else: + return None + + potential_prefixes = ["", "v", "V", "release-", "RELEASE-", "v-", "V-"] + for prefix in potential_prefixes: + potential_tag = f"{prefix}{version}" + if repo_type == "github": + url = f"https://api.github.com/repos/{namespace}/{name}/git/refs/tags/{potential_tag}" elif repo_type == "gitlab": ns = namespace if namespace else "" project_path = f"{ns}/{name}".strip("/").replace("/", "%2F") - url = f"https://gitlab.com/api/v4/projects/{project_path}/repository/tags" - else: - return None - response = requests.get(url, timeout=10) - response.raise_for_status() - tags = response.json() - except (requests.RequestException, ValueError): - return None + url = ( + f"https://gitlab.com/api/v4/projects/{project_path}/repository/tags/{potential_tag}" + ) + + try: + response = requests.get(url, headers=headers, timeout=10) + 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 + except requests.RequestException: + continue - for tag in tags: - tag_name = tag.get("name", "") - normalized_tag = normalize_string(tag_name) - if normalized_version == normalized_tag: - return tag_name return None -def normalize_string(version_string): +def delete_nix_store_path(store_path): """ - Normalize common prefix version string + Delete a specific path from the Nix store. """ - for prefix in ("release-", "RELEASE-", "v", "V"): - if version_string.startswith(prefix): - return version_string[len(prefix):] - return version_string - + if not store_path or not store_path.startswith("/nix/store/"): + return -def clean_garbage(): - """ - Delete all unreferenced downloaded tarballs and evaluation caches - """ try: - subprocess.run(["nix-store", "--gc"], capture_output=True) + subprocess.run( + ["nix-store", "--delete", store_path], capture_output=True, check=True, timeout=15 + ) + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: + print(f"Warning: Failed to delete Nix store path {store_path}: {e}") except FileNotFoundError: pass diff --git a/tests/test_nix.py b/tests/test_nix.py index e162320..ab32171 100644 --- a/tests/test_nix.py +++ b/tests/test_nix.py @@ -17,13 +17,14 @@ import json import unittest from pathlib import Path +from unittest.mock import Mock from unittest.mock import patch +from packageurl import PackageURL + from fetchcode import nix from fetchcode.nix import Nix -from packageurl import PackageURL - DATA_DIR = Path(__file__).parent / "data" / "nix" @@ -50,14 +51,6 @@ def test_get_package_data(self, mock_fetch): self.assertEqual(result["name"], "hstr") self.assertEqual(result["license"], "Apache-2.0") - def test_get_nix_store_path_no_version(self, mock_fetch): - purl = "pkg:nix/nixpkgs/hstr?system=x86_64-darwin" - data = Nix.get_package_data(purl) - system = "x86_64-darwin" - output = "out" - path = nix.get_nix_store_path(data, system, output) - self.assertEqual(path, "/nix/store/r910fm5w0iqywcwflp9fx1dsiwf8kqnx-hstr-3.2") - def test_get_nix_store_path_with_version(self, mock_fetch): purl = "pkg:nix/nixpkgs/hstr@3.1?system=x86_64-darwin" data = Nix.get_package_data(purl) @@ -65,17 +58,17 @@ def test_get_nix_store_path_with_version(self, mock_fetch): output = "out" commit_hash = None version = "3.1" - path = nix.get_nix_store_path(data, system, output, commit_hash, version) + path = nix.get_nix_store_path(data, system, output, version, commit_hash) self.assertEqual(path, "/nix/store/vcildmf4v1jkci82ny3bfkhn6nlnf6nn-hstr-3.1") def test_get_nix_store_path_with_commit_hash(self, mock_fetch): - purl = "pkg:nix/nixpkgs/hstr?system=x86_64-darwin%commit_hash=4a29d733e8a7d5b824c3d8c958a946a9867b3eb2" + purl = "pkg:nix/nixpkgs/hstr@3.1?system=x86_64-darwin&commit_hash=4a29d733e8a7d5b824c3d8c958a946a9867b3eb2" data = Nix.get_package_data(purl) system = "aarch64-darwin" output = "out" commit_hash = "4a29d733e8a7d5b824c3d8c958a946a9867b3eb2" - version = None - path = nix.get_nix_store_path(data, system, output, commit_hash, version) + version = "3.1" + path = nix.get_nix_store_path(data, system, output, version, commit_hash) self.assertEqual(path, "/nix/store/vh8m653ps0n96lg68w90sf4xbi9n2nvx-hstr-3.1") def test_get_nix_store_path_version_commit_hash_not_match(self, mock_fetch): @@ -89,11 +82,11 @@ def test_get_nix_store_path_version_commit_hash_not_match(self, mock_fetch): self.assertEqual(path, None) def test_construct_url_based_on_homepage_url(self, mock_fetch): - purl_str = "pkg:nix/nixpkgs/hstr?system=x86_64-darwin%commit_hash=4a29d733e8a7d5b824c3d8c958a946a9867b3eb2" + purl_str = "pkg:nix/nixpkgs/hstr?system=x86_64-darwin&commit=4a29d733e8a7d5b824c3d8c958a946a9867b3eb2" purl = PackageURL.from_string(purl_str) data = Nix.get_package_data(purl_str) result = nix.construct_url_based_on_homepage_url(purl, data) - self.assertEqual(result, "https://github.com/dvorka/hstr/archive/v3.2.tar.gz") + self.assertEqual(result, "https://github.com/dvorka/hstr/archive/3.1.tar.gz") def test_get_url_netloc_namespace_and_name(self, mock_fetch): url = "https://github.com/dvorka/hstr" @@ -109,7 +102,93 @@ def test_get_commit_hash(self, mock_fetch): commit_hash = nix.get_commit_hash(data, version) self.assertEqual(commit_hash, "96ba1c52e54e74c3197f4d43026b3f3d92e83ff9") - def test_normalize_string(self, mock_fetch): - version = nix.normalize_string("release-1.2.3") - self.assertEqual(version, "1.2.3") - + @patch("subprocess.run") + def test_get_nix_store_path_with_nix(self, mock_subproc_run, mock_fetch): + mock_subproc_run.return_value.stdout = "/nix/store/111111111111111-test_package-1.0\n" + + result = nix.get_nix_store_path_with_nix( + "test_package", "x86_64-linux", "out", "xxxxxxxxxxxxxxx" + ) + + self.assertEqual(result, "/nix/store/111111111111111-test_package-1.0") + mock_subproc_run.assert_called_once() + self.assertIn("nix-instantiate", mock_subproc_run.call_args[0][0]) + + @patch("subprocess.run") + def test_get_src_info(self, mock_subproc_run, mock_fetch): + mock_subproc_run.return_value.stdout = ( + '{"version": "1.0.0", "urls": ["https://example.com/source.tar.gz"]}' + ) + + result = nix.get_src_info("python3Packages.requests", "xxxxxxxxxxxxxxx") + + self.assertEqual(result["version"], "1.0.0") + self.assertEqual(result["urls"][0], "https://example.com/source.tar.gz") + + @patch("subprocess.run") + def test_get_mirrors_map(self, mock_subproc_run, mock_fetch): + mock_subproc_run.return_value.stdout = '{"sourceforge": ["https://sourceforge.net/"]}' + + result = nix.get_mirrors_map() + self.assertIn("sourceforge", result) + + @patch("fetchcode.nix.get_mirrors_map") + @patch("fetchcode.nix.verify_url_existence") + def test_convert_mirror_url(self, mock_verify, mock_get_mirrors, mock_fetch): + mock_get_mirrors.return_value = {"sourceforge": ["https://sourceforge.net/"]} + mock_verify.return_value = True + + url = "mirror://sourceforge/test/sample-1.12.6.tar.xz" + result = nix.convert_mirror_url(url) + self.assertEqual(result, "https://sourceforge.net/test/sample-1.12.6.tar.xz") + + @patch("requests.get") + def test_github_pages_to_repo(self, mock_get, mock_fetch): + mock_get.return_value.status_code = 200 + + url = "https://iovisor.github.io/bcc/" + result = nix.github_pages_to_repo(url) + self.assertEqual(result, "https://github.com/iovisor/bcc") + + @patch("requests.get") + def test_get_narinfo_url(self, mock_get, mock_fetch): + mock_get.return_value.text = "URL: 00000.nar.xz" + + result = nix.get_narinfo_url("https://cache.nixos.org/123.narinfo") + self.assertEqual(result, "00000.nar.xz") + + @patch("fetchcode.nix.get_narinfo_url") + def test_get_nix_download_url(self, mock_get_narinfo, mock_fetch): + mock_get_narinfo.return_value = "nar/00000.nar.xz" + + result = nix.get_nix_download_url("/nix/store/1234567890abcdef-hstr-3.1") + self.assertEqual(result, "https://cache.nixos.org/nar/00000.nar.xz") + + @patch("requests.get") + def test_clarify_version_tag_github(self, mock_get, mock_fetch): + # 404 for the first check (prefix ""), 200 for the second check (prefix "v") + mock_get.side_effect = [Mock(status_code=404), Mock(status_code=200)] + + result = nix.clarify_version_tag("github", "project_namespace", "project_name", "1.0.0") + + self.assertEqual(result, "v1.0.0") + + def test_get_version_from_commit_hash(self, mock_fetch): + mock_data = { + "releases": [ + { + "version": "2.6", + "platforms": [ + { + "system": "x86_64-darwin", + "commit_hash": "96ba1c52e54e74c3197f4d43026b3f3d92e83ff9", + } + ], + } + ] + } + + version = nix.get_version_from_commit_hash( + mock_data, "96ba1c52e54e74c3197f4d43026b3f3d92e83ff9" + ) + self.assertEqual(version, "2.6") From b3a626bd8a93e83c889aef0d5a02e0348663cef5 Mon Sep 17 00:00:00 2001 From: Chin Yeung Li Date: Wed, 22 Jul 2026 07:36:34 +0800 Subject: [PATCH 05/10] Add nix to the supported ecosystems in the README #201 Signed-off-by: Chin Yeung Li --- README.rst | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/README.rst b/README.rst index cd2139e..6c5960a 100644 --- a/README.rst +++ b/README.rst @@ -89,6 +89,7 @@ Ecosystems supported for fetching a purl from fetchcode: - hex - luarocks - maven +- nix - npm - nuget - pub @@ -110,19 +111,19 @@ This project is funded, supported and sponsored by: - Generous support and contributions from users like you! - the European Commission NGI programme -- the NLnet Foundation +- the NLnet Foundation - the Swiss State Secretariat for Education, Research and Innovation (SERI) - Google, including the Google Summer of Code and the Google Seasons of Doc programmes - Mercedes-Benz Group - Microsoft and Microsoft Azure - AboutCode ASBL -- nexB Inc. +- nexB Inc. -|europa| |dgconnect| +|europa| |dgconnect| -|ngi| |nlnet| +|ngi| |nlnet| |aboutcode| |nexb| @@ -136,7 +137,7 @@ Communications Networks, Content and Technology under grant agreement No 1010929 This project was funded through the NGI0 Entrust Fund, a fund established by NLnet with financial support from the European Commission's Next Generation Internet programme, under the aegis of DG -Communications Networks, Content and Technology under grant agreement No 101069594. +Communications Networks, Content and Technology under grant agreement No 101069594. |ngizeroentrust| https://nlnet.nl/project/Back2source/ @@ -150,7 +151,7 @@ Communications Networks, Content and Technology under grant agreement No 1010929 This project was funded through the NGI0 Entrust Fund, a fund established by NLnet with financial support from the European Commission's Next Generation Internet programme, under the aegis of DG -Communications Networks, Content and Technology under grant agreement No 101069594. +Communications Networks, Content and Technology under grant agreement No 101069594. |ngizeroentrust| https://nlnet.nl/project/purl2all/ From 091c595e4be8fdc71091d5dea5272d2c65bf03bb Mon Sep 17 00:00:00 2001 From: Chin Yeung Li Date: Wed, 22 Jul 2026 07:37:54 +0800 Subject: [PATCH 06/10] Add the missing f-string #201 Signed-off-by: Chin Yeung Li --- src/fetchcode/nix.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/fetchcode/nix.py b/src/fetchcode/nix.py index 8473cfb..e25bdcf 100644 --- a/src/fetchcode/nix.py +++ b/src/fetchcode/nix.py @@ -275,7 +275,7 @@ def get_src_info(attr_path, commit_hash=None): f'{commit_hash}.tar.gz") {{ {config_str} }}' ) else: - nixpkgs_import = "import {{ {config_str} }}" + nixpkgs_import = f"import {{ {config_str} }}" nix_expression = f""" let From f1142621f23125cc8261501b77d8d78cd38f8e8d Mon Sep 17 00:00:00 2001 From: Chin Yeung Li Date: Wed, 22 Jul 2026 07:51:42 +0800 Subject: [PATCH 07/10] Replaced actions/checkout@v4, actions/setup-python@v5, actions/upload-artifact@v7 and actions/download-artifact@v8 with a full-length commit SHA. Signed-off-by: Chin Yeung Li --- .github/workflows/pypi-release.yml | 10 +++++----- .github/workflows/run-unit-tests.yml | 4 ++-- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/.github/workflows/pypi-release.yml b/.github/workflows/pypi-release.yml index 3846827..f711c81 100644 --- a/.github/workflows/pypi-release.yml +++ b/.github/workflows/pypi-release.yml @@ -24,9 +24,9 @@ jobs: runs-on: ubuntu-24.04 steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 - name: Set up Python - uses: actions/setup-python@v5 + uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0 with: python-version: 3.13 @@ -37,7 +37,7 @@ jobs: run: python -m build --sdist --wheel --outdir dist/ - name: Upload built archives - uses: actions/upload-artifact@v7 + uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f with: name: pypi_archives path: dist/* @@ -51,7 +51,7 @@ jobs: steps: - name: Download built archives - uses: actions/download-artifact@v8 + uses: actions/download-artifact@37930b1c2abaa49bbe596cd826c3c89aef350131 with: name: pypi_archives path: dist @@ -75,7 +75,7 @@ jobs: steps: - name: Download built archives - uses: actions/download-artifact@v8 + uses: actions/download-artifact@37930b1c2abaa49bbe596cd826c3c89aef350131 with: name: pypi_archives path: dist/ diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index b062493..82df810 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -18,10 +18,10 @@ jobs: steps: - name: Checkout - uses: actions/checkout@v4 + uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 + uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0 with: python-version: ${{ matrix.python-version }} From fb3e79243825a26f000a80bc3464c15148235f09 Mon Sep 17 00:00:00 2001 From: Chin Yeung Li Date: Wed, 22 Jul 2026 07:54:30 +0800 Subject: [PATCH 08/10] Update AUTHORS.rst Signed-off-by: Chin Yeung Li --- AUTHORS.rst | 1 + 1 file changed, 1 insertion(+) diff --git a/AUTHORS.rst b/AUTHORS.rst index de5421c..1ad9940 100644 --- a/AUTHORS.rst +++ b/AUTHORS.rst @@ -7,3 +7,4 @@ The following organizations or individuals have contributed to this repo: - Tushar Goel @ TG1999 - Thomas Druez @ tdruez - Keshav Priyadarshi @ keshav-space +- Chin Yeung Li @ chinyeungli From 21f8c9ad4893080db5e2507ec8e2c1fa69caa3ae Mon Sep 17 00:00:00 2001 From: Chin Yeung Li Date: Wed, 22 Jul 2026 08:32:37 +0800 Subject: [PATCH 09/10] Better error handling #201 Signed-off-by: Chin Yeung Li --- src/fetchcode/nix.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/fetchcode/nix.py b/src/fetchcode/nix.py index e25bdcf..411efc0 100644 --- a/src/fetchcode/nix.py +++ b/src/fetchcode/nix.py @@ -559,6 +559,7 @@ def get_nix_download_url(path): url_path = get_narinfo_url(narinfo_url) if not url_path: + print(f"{narinfo_url} is not accessible.") return None return f"https://cache.nixos.org/{url_path}" From 9eb54ffa2a900c3b2df15b863c1755fcd7320849 Mon Sep 17 00:00:00 2001 From: Chin Yeung Li Date: Wed, 22 Jul 2026 16:30:17 +0800 Subject: [PATCH 10/10] Add flakeref support #201 * Correct typo * Add tests Signed-off-by: Chin Yeung Li --- src/fetchcode/nix.py | 93 ++++++++++++++++++++++++++++++++++++-------- tests/test_nix.py | 58 +++++++++++++++++++++++++++ 2 files changed, 135 insertions(+), 16 deletions(-) diff --git a/src/fetchcode/nix.py b/src/fetchcode/nix.py index 411efc0..2ba6ed4 100644 --- a/src/fetchcode/nix.py +++ b/src/fetchcode/nix.py @@ -71,7 +71,7 @@ def get_download_url(cls, purl): name = purl_data.name version = purl_data.version if not version: - raise Exception("Version is requierd.") + raise Exception("Version is required.") qualifiers = purl_data.qualifiers or {} if "system" in qualifiers: @@ -130,28 +130,30 @@ def get_upstream_src_download_url(cls, purl): if shutil.which("nix") is not None: have_nix = True - namespace = purl_data.namespace - # We will only work with the official nixpkgs repository, at least - # for now. - if not namespace or namespace.lower() != "nixpkgs": - raise Exception( - "Only official nixpkgs repository is supported (i.e. namespace=nixpkgs)." - ) name = purl_data.name + namespace = purl_data.namespace version = purl_data.version + qualifiers = purl_data.qualifiers or {} + commit_hash = qualifiers.get("commit", "") + flakeref = qualifiers.get("flakeref", "") + if not version: raise Exception("Version is requierd.") - data = cls.get_package_data(purl) + + if namespace != "nixpkgs" and not flakeref: + raise Exception( + "Only official nixpkgs repository is supported, " + "or please provide the flakeref qualifier." + ) + data = cls.get_package_data(purl) if not flakeref else None if data: download_url = construct_url_based_on_homepage_url(purl_data, data) if not download_url and have_nix: - download_url = retrieve_src_download_url_with_nix(name, version) - if not download_url and data and version: + if not commit_hash and data: commit_hash = get_commit_hash(data, version) - if commit_hash: - download_url = retrieve_src_download_url_with_nix(name, version, commit_hash) + download_url = retrieve_src_download_url_with_nix(name, version, commit_hash, flakeref) if not download_url and not have_nix: print("Install `nix` and re-run to let `nix` determine the download URL.") @@ -208,11 +210,11 @@ def get_nix_store_path_with_nix(name, system, output, commit_hash): return None -def retrieve_src_download_url_with_nix(name, version=None, commit_hash=None): +def retrieve_src_download_url_with_nix(name, version, commit_hash=None, flakeref=None): """ Find and return the source download url using 'nix' """ - info = get_src_info(name, commit_hash) + info = get_src_info(name, version, commit_hash, flakeref) urls = info.get("urls", []) download_url = None @@ -260,7 +262,7 @@ def retrieve_src_download_url_with_nix(name, version=None, commit_hash=None): return download_url -def get_src_info(attr_path, commit_hash=None): +def get_src_info(attr_path, version, commit_hash=None, flakeref=None): """ Use the `nix-instantiate` command together with the nix_expression to retrieve the package’s version and download URL. Return a dictionary @@ -268,6 +270,27 @@ def get_src_info(attr_path, commit_hash=None): """ config_str = "config = { allowBroken = true; allowUnfree = true; };" + if flakeref: + if not flakeref.startswith("github"): + print("Only flakeref for github is supported at the moment.") + return {"version": None, "urls": []} + + rest = flakeref.partition(":")[2] + parts = rest.split("/") + if len(parts) < 2: + return {"version": None, "urls": []} + owner, repo = parts[0], parts[1] + git_repo = f"https://github.com/{owner}/{repo}.git" + + version_tag = version + if not commit_hash: + commit_hash, version_tag = get_flakeref_version_commit_hash(git_repo, version) + if commit_hash and version_tag: + url = f"https://github.com/{owner}/{repo}/archive/{commit_hash}.tar.gz" + return {"version": version_tag, "urls": [url]} + else: + return {"version": None, "urls": []} + # Determine the repository entry point definition if commit_hash: nixpkgs_import = ( @@ -308,6 +331,44 @@ def get_src_info(attr_path, commit_hash=None): return {"version": None, "urls": []} +def get_flakeref_version_commit_hash(git_repo, version): + """ + Get the commit hash from the given version. + """ + if shutil.which("git") is None: + return None, None + cmd = [ + "git", + "ls-remote", + "--tags", + git_repo, + ] + try: + result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=300) + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: + print(f"Error running git ls-remote for {git_repo}: {e}", file=sys.stderr) + return None, None + + potential_version_matches = [ + version, + f"v{version}", + f"V{version}", + f"v-{version}", + f"V-{version}", + f"release-{version}", + f"RELEASE-{version}", + ] + for line in result.stdout.splitlines(): + parts = line.split("\t") + remote_hash = parts[0] + reference_tag = parts[1] + version_tag = reference_tag.replace("refs/tags/", "").replace("^{}", "") + + if version_tag in potential_version_matches: + return remote_hash, version_tag + return None, None + + def construct_url_based_on_homepage_url(input_purl, data): """ Determine and return the download url based on the homepage and diff --git a/tests/test_nix.py b/tests/test_nix.py index ab32171..02d7485 100644 --- a/tests/test_nix.py +++ b/tests/test_nix.py @@ -192,3 +192,61 @@ def test_get_version_from_commit_hash(self, mock_fetch): mock_data, "96ba1c52e54e74c3197f4d43026b3f3d92e83ff9" ) self.assertEqual(version, "2.6") + + @patch("fetchcode.nix.get_flakeref_version_commit_hash") + def test_get_src_info_with_flakeref_no_commit_hash(self, mock_get_flake_hash, mock_fetch): + mock_get_flake_hash.return_value = ("aaaaaaaaa", "v2.0") + + result = nix.get_src_info("package_name", "2.0", flakeref="github:owner/repo") + + self.assertEqual(result["version"], "v2.0") + self.assertEqual( + result["urls"][0], "https://github.com/owner/repo/archive/aaaaaaaaa.tar.gz" + ) + mock_get_flake_hash.assert_called_once_with("https://github.com/owner/repo.git", "2.0") + + def test_get_src_info_with_flakeref_and_commit_hash(self, mock_fetch): + result = nix.get_src_info( + "package_name", version="2.0", commit_hash="aaaaaaaaa", flakeref="github:owner/repo" + ) + + self.assertEqual(result["version"], "2.0") + self.assertEqual( + result["urls"][0], "https://github.com/owner/repo/archive/aaaaaaaaa.tar.gz" + ) + + @patch("subprocess.run") + @patch("shutil.which") + def test_get_flakeref_version_commit_hash_success( + self, mock_which, mock_subproc_run, mock_fetch + ): + mock_which.return_value = "/usr/bin/git" + + mock_subproc_run.return_value.stdout = ( + "1111111111111111111111111111111111111111\trefs/tags/1.0\n" + "2222222222222222222222222222222222222222\trefs/tags/v2.0\n" + "2222222222222222222222222222222222222222\trefs/tags/v3.0\n" + ) + + commit_hash, version_tag = nix.get_flakeref_version_commit_hash( + "https://github.com/owner/repo.git", "2.0" + ) + + self.assertEqual(commit_hash, "2222222222222222222222222222222222222222") + self.assertEqual(version_tag, "v2.0") + + @patch("subprocess.run") + @patch("shutil.which") + def test_get_flakeref_version_commit_hash_no_match( + self, mock_which, mock_subproc_run, mock_fetch + ): + mock_which.return_value = "/usr/bin/git" + mock_subproc_run.return_value.stdout = ( + "1111111111111111111111111111111111111111\trefs/tags/1.0\n" + ) + + commit_hash, version_tag = nix.get_flakeref_version_commit_hash( + "https://github.com/owner/repo.git", "1.1" + ) + self.assertIsNone(commit_hash) + self.assertIsNone(version_tag)