Add on-demand indexing for Nixpkgs in PurlDB

minecode/collectors/nix.py

@@ -0,0 +1,182 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# purldb is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/nexB/purldb for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import logging
+import shutil
+import subprocess
+import json
+from packageurl import PackageURL
+from minecode.miners.nix import build_packages, verify_url_existence
+from fetchcode import fetch_json_response
+
+from minecode import priority_router
+from packagedb.models import PackageContentType
+
+logger = logging.getLogger(__name__)
+handler = logging.StreamHandler()
+logger.addHandler(handler)
+logger.setLevel(logging.INFO)
+
+
+def map_nix_package(package_url, pipelines, priority=0):
+    """
+    Add a composer `package_url` to the PackageDB.
+    """
+    from minecode.model_utils import add_package_to_scan_queue, merge_or_create_package
+
+    # Check if the `nix` command is available on the system
+    have_nix = False
+    if shutil.which("nix") is not None:
+        have_nix = True
+
+    namespace = package_url.namespace
+    # We will only work with the official nixpkgs repository.
+    # It is impossible to handle all third‑party or custom repositories.
+    if not namespace.lower() == "nixpkgs":
+        return None
+
+    data = get_nix_package_data(package_url)
+
+    if not data and have_nix:
+        data = get_nix_package_data_via_cli(package_url)
+        clean_garbage()
+
+    if data:
+        packages = build_packages(data, package_url)
+
+        error = None
+        for package in packages:
+            package.extra_data["package_content"] = PackageContentType.BINARY
+            db_package, _, _, error = merge_or_create_package(package, visit_level=0)
+            if error:
+                break
+
+            if db_package:
+                add_package_to_scan_queue(
+                    package=db_package, pipelines=pipelines, priority=priority
+                )
+
+        return error
+    else:
+        return f"Failed to fetch package data for {package_url}"
+
+
+def get_nix_package_data(purl):
+    """
+    Fetch package data from https://search.devbox.sh/.
+    """
+
+    api_url = f"https://search.devbox.sh/v2/pkg?name={purl.name}"
+    if not verify_url_existence(api_url):
+        return None
+
+    return fetch_json_response(api_url)
+
+
+def parse_license(license_data):
+    """
+    Parse a license object and return a string representation of the license.
+    """
+    return (
+        license_data.get("spdxId")
+        or license_data.get("fullName")
+        or license_data.get("shortName")
+        or str(license_data)
+    )
+
+
+def get_nix_package_data_via_cli(package_url):
+    """
+    Fetch package metadata using the Nix CLI (mainly for packages that
+    Devbox doesn't index).
+    """
+    # Create a Nix expression to pull just the fields we want.
+    package_name = package_url.name
+    package_version = package_url.version
+
+    nix_exp = f"""
+        let
+            pkgs = import <nixpkgs> {{}};
+            pkg = pkgs.{package_name};
+            outputNames = if pkg ? outputs then pkg.outputs else ["out"];
+        in {{
+            name = pkg.pname or pkg.name or "";
+            version = pkg.version or "";
+            storePath = pkg.outPath or "";
+            outputs = builtins.listToAttrs (map (name: {{ name = name; value = pkg.${{name}}.outPath; }}) outputNames);
+            description = pkg.meta.description or "";
+            homepage = pkg.meta.homepage or "";
+            license = pkg.meta.license or {{}};
+            system = pkg.system or "";
+        }}
+    """
+
+    cmd = ["nix-instantiate", "--eval", "--json", "--strict", "-E", nix_exp]
+
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True, check=True)  # noqa: S603
+        pkg_data = json.loads(result.stdout)
+
+        version = pkg_data.get("version", "")
+        if package_version:
+            if not version or version != package_version:
+                print(f"Cannot find version: {package_version}, got {version}")
+                return None
+
+        output_list = [
+            {"name": pname, "path": ppath} for pname, ppath in pkg_data.get("outputs", {}).items()
+        ]
+
+        # Return a similar JSON layout as Devbox
+        return {
+            "summary": pkg_data.get("description", ""),
+            "homepage_url": pkg_data.get("homepage", ""),
+            "license": parse_license(pkg_data.get("license", {})),
+            "releases": [
+                {
+                    "version": version,
+                    "platforms": [{"system": pkg_data.get("system", ""), "outputs": output_list}],
+                }
+            ],
+        }
+    except Exception as e:
+        print(
+            f"Failed to fetch Nix package data for {package_url}: {e.stderr if hasattr(e, 'stderr') else e}"
+        )
+        return None
+
+
+def clean_garbage():
+    """
+    Delete all unreferenced downloaded tarballs and evaluation caches
+    """
+    nix_store = shutil.which("nix-store")
+    if nix_store:
+        subprocess.run([nix_store, "--gc"], capture_output=True)  # noqa: S603
+    else:
+        raise FileNotFoundError("nix-store not found in PATH")
+
+
+@priority_router.route("pkg:nix/nixpkgs/.*")
+def process_request(purl_str, **kwargs):
+    """
+    Process `priority_resource_uri` containing a nix Package URL (PURL).
+    """
+    from minecode.model_utils import DEFAULT_PIPELINES
+
+    addon_pipelines = kwargs.get("addon_pipelines", [])
+    pipelines = DEFAULT_PIPELINES + tuple(addon_pipelines)
+    priority = kwargs.get("priority", 0)
+
+    package_url = PackageURL.from_string(purl_str)
+
+    error_msg = map_nix_package(package_url, pipelines, priority)
+
+    if error_msg:
+        return error_msg

minecode/miners/nix.py

@@ -0,0 +1,127 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# purldb is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/nexB/purldb for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+from packagedcode import models as scan_models
+from packageurl import PackageURL
+
+import requests
+
+
+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 build_packages(metadata_dict, purl):
+    package_version = purl.version
+
+    description = metadata_dict.get("summary", "")
+    homepage_url = metadata_dict.get("homepage_url", "")
+    license = metadata_dict.get("license", [])
+    extracted_license_statement = license if isinstance(license, list) else [license]
+
+    releases = metadata_dict.get("releases", [])
+    for release in releases:
+        version = release.get("version", "")
+        if package_version and version != package_version:
+            continue
+        common_data = dict(
+            name=purl.name,
+            namespace=purl.namespace,
+            version=version,
+            description=description,
+            homepage_url=homepage_url,
+            extracted_license_statement=extracted_license_statement,
+        )
+
+        platforms = release.get("platforms", "")
+        for platform in platforms:
+            date = platform.get("date") or None
+            platform_system = platform.get("system", "")
+            commit = platform.get("commit_hash", "")
+            platform_outputs = platform.get("outputs") or None
+
+            if not platform_outputs:
+                continue
+            for platform_output in platform_outputs:
+                output_name = platform_output.get("name")
+                path = platform_output.get("path")
+                narinfo_url = path.replace("/nix/store/", "").split("-")[0]
+                download_url = get_nix_download_url(narinfo_url)
+                download_data = dict(
+                    datasource_id="nix_pkginfo",
+                    type="nix",
+                    download_url=download_url,
+                    release_date=date,
+                )
+                download_data.update(common_data)
+                package = scan_models.PackageData.from_data(download_data)
+                package.datasource_id = "nix_api_metadata"
+                qualifiers = {}
+                if platform_system:
+                    qualifiers["system"] = platform_system
+                if commit:
+                    qualifiers["commit"] = commit
+                if output_name:
+                    qualifiers["output"] = output_name
+                updated_purl = update_purl_with_version_qualifiers(purl, version, qualifiers)
+                package.set_purl(updated_purl)
+                yield package
+
+
+def update_purl_with_version_qualifiers(purl, version, qualifiers):
+    """
+    Update a PackageURL with the given version and qualifiers.
+    """
+    return PackageURL(
+        type=purl.type,
+        namespace=purl.namespace,
+        name=purl.name,
+        version=version,
+        qualifiers=qualifiers,
+        subpath=purl.subpath,
+    )
+
+
+def verify_url_existence(url):
+    """
+    Perform 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