chore: add support for PyPI PURLs

Signed-off-by: Ben Selwyn-Smith <[email protected]>

Author: benmss

src/macaron/slsa_analyzer/analyzer.py

@@ -76,7 +76,8 @@
 from macaron.slsa_analyzer.git_service import GIT_SERVICES, BaseGitService, GitHub
 from macaron.slsa_analyzer.git_service.base_git_service import NoneGitService
 from macaron.slsa_analyzer.git_url import GIT_REPOS_DIR
-from macaron.slsa_analyzer.package_registry import PACKAGE_REGISTRIES, MavenCentralRegistry
+from macaron.slsa_analyzer.package_registry import PACKAGE_REGISTRIES, MavenCentralRegistry, PyPIRegistry
+from macaron.slsa_analyzer.package_registry.pypi_registry import find_or_create_pypi_asset
 from macaron.slsa_analyzer.provenance.expectations.expectation_registry import ExpectationRegistry
 from macaron.slsa_analyzer.provenance.intoto import InTotoPayload, InTotoV01Payload
 from macaron.slsa_analyzer.provenance.intoto.errors import LoadIntotoAttestationError
@@ -510,7 +511,9 @@ def run_single(
             except TypeError as error:
                 logger.debug("Failed to parse repository path as URL: %s", error)
             if url and url.hostname == "github.com":
-                artifact_hash = self.get_artifact_hash(parsed_purl, local_artifact_dirs, hashlib.sha256())
+                artifact_hash = self.get_artifact_hash(
+                    parsed_purl, local_artifact_dirs, hashlib.sha256(), all_package_registries
+                )
                 if artifact_hash:
                     git_attestation_dict = git_service.api_client.get_attestation(
                         analyze_ctx.component.repository.full_name, artifact_hash
@@ -983,7 +986,11 @@ def create_analyze_ctx(self, component: Component) -> AnalyzeContext:
         return analyze_ctx
 
     def get_artifact_hash(
-        self, purl: PackageURL, cached_artifacts: list[str] | None, hash_algorithm: Any
+        self,
+        purl: PackageURL,
+        cached_artifacts: list[str] | None,
+        hash_algorithm: Any,
+        all_package_registries: list[PackageRegistryInfo],
     ) -> str | None:
         """Get the hash of the artifact found from the passed PURL using local or remote files.
 
@@ -995,6 +1002,8 @@ def get_artifact_hash(
             The list of local files that match the PURL.
         hash_algorithm: Any
             The hash algorithm to use.
+        all_package_registries: list[PackageRegistryInfo]
+            The list of package registry information.
 
         Returns
         -------
@@ -1024,8 +1033,43 @@ def get_artifact_hash(
             return maven_registry.get_artifact_hash(purl, hash_algorithm)
 
         if purl.type == "pypi":
-            # TODO implement
-            return None
+            pypi_registry = next(
+                (
+                    package_registry
+                    for package_registry in PACKAGE_REGISTRIES
+                    if isinstance(package_registry, PyPIRegistry)
+                ),
+                None,
+            )
+            if not pypi_registry:
+                logger.debug("Missing registry for PyPI")
+                return None
+
+            registry_info = next(
+                (
+                    info
+                    for info in all_package_registries
+                    if info.package_registry == pypi_registry and info.build_tool_name in {"pip", "poetry"}
+                ),
+                None,
+            )
+            if not registry_info:
+                logger.debug("Missing registry information for PyPI")
+                return None
+
+            pypi_asset = find_or_create_pypi_asset(purl.name, purl.version, registry_info)
+            if not pypi_asset:
+                return None
+
+            pypi_asset.has_repository = True
+            if not pypi_asset.download(""):
+                return None
+
+            source_url = pypi_asset.get_sourcecode_url("bdist_wheel")
+            if not source_url:
+                return None
+
+            return pypi_registry.get_artifact_hash(source_url, hash_algorithm)
 
         logger.debug("Purl type '%s' not yet supported for GitHub attestation discovery.", purl.type)
         return None

src/macaron/slsa_analyzer/checks/detect_malicious_metadata_check.py

@@ -32,7 +32,11 @@
 from macaron.slsa_analyzer.checks.base_check import BaseCheck
 from macaron.slsa_analyzer.checks.check_result import CheckResultData, CheckResultType, Confidence, JustificationType
 from macaron.slsa_analyzer.package_registry.deps_dev import APIAccessError, DepsDevService
-from macaron.slsa_analyzer.package_registry.pypi_registry import PyPIPackageJsonAsset, PyPIRegistry
+from macaron.slsa_analyzer.package_registry.pypi_registry import (
+    PyPIPackageJsonAsset,
+    PyPIRegistry,
+    find_or_create_pypi_asset,
+)
 from macaron.slsa_analyzer.registry import registry
 from macaron.slsa_analyzer.specs.package_registry_spec import PackageRegistryInfo
 from macaron.util import send_post_http_raw
@@ -261,23 +265,16 @@ def run_check(self, ctx: AnalyzeContext) -> CheckResultData:
                 case PackageRegistryInfo(
                     build_tool_name="pip" | "poetry",
                     build_tool_purl_type="pypi",
-                    package_registry=PyPIRegistry() as pypi_registry,
+                    package_registry=PyPIRegistry(),
                 ) as pypi_registry_info:
-
-                    # Retrieve the pre-existing AssetLocator object for the PyPI package JSON object, if it exists.
-                    pypi_package_json = next(
-                        (asset for asset in pypi_registry_info.metadata if isinstance(asset, PyPIPackageJsonAsset)),
-                        None,
+                    # Retrieve the pre-existing asset, or create a new one.
+                    pypi_package_json = find_or_create_pypi_asset(
+                        ctx.component.name, ctx.component.version, pypi_registry_info
                     )
-                    if not pypi_package_json:
-                        # Create an AssetLocator object for the PyPI package JSON object.
-                        pypi_package_json = PyPIPackageJsonAsset(
-                            component_name=ctx.component.name,
-                            component_version=ctx.component.version,
-                            has_repository=ctx.component.repository is not None,
-                            pypi_registry=pypi_registry,
-                            package_json={},
-                        )
+                    if pypi_package_json is None:
+                        return CheckResultData(result_tables=[], result_type=CheckResultType.UNKNOWN)
+
+                    pypi_package_json.has_repository = ctx.component.repository is not None
 
                     pypi_registry_info.metadata.append(pypi_package_json)
 

src/macaron/slsa_analyzer/package_registry/pypi_registry.py

@@ -11,6 +11,7 @@
 import zipfile
 from dataclasses import dataclass
 from datetime import datetime
+from typing import Any
 
 import requests
 from bs4 import BeautifulSoup, Tag
@@ -21,6 +22,7 @@
 from macaron.json_tools import json_extract
 from macaron.malware_analyzer.datetime_parser import parse_datetime
 from macaron.slsa_analyzer.package_registry.package_registry import PackageRegistry
+from macaron.slsa_analyzer.specs.package_registry_spec import PackageRegistryInfo
 from macaron.util import send_get_http_raw
 
 logger: logging.Logger = logging.getLogger(__name__)
@@ -231,6 +233,45 @@ def fetch_sourcecode(self, src_url: str) -> dict[str, str] | None:
             logger.debug("Successfully fetch the source code from PyPI")
             return py_files_content
 
+    def get_artifact_hash(self, artifact_url: str, hash_algorithm: Any) -> str | None:
+        """Return the hash of the artifact found at the passed URL.
+
+        Parameters
+        ----------
+        artifact_url
+            The URL of the artifact.
+        hash_algorithm: Any
+            The hash algorithm to use.
+
+        Returns
+        -------
+        str | None
+            The hash of the artifact, or None if not found.
+        """
+        try:
+            response = requests.get(artifact_url, stream=True, timeout=40)
+            response.raise_for_status()
+        except requests.exceptions.HTTPError as http_err:
+            logger.debug("HTTP error occurred: %s", http_err)
+            return None
+
+        if response.status_code != 200:
+            logger.debug("Invalid response: %s", response.status_code)
+            return None
+
+        try:
+            for chunk in response.iter_content():
+                hash_algorithm.update(chunk)
+        except RequestException as error:
+            # Something went wrong with the request, abort.
+            logger.debug("Error while streaming source file: %s", error)
+            response.close()
+            return None
+
+        artifact_hash: str = hash_algorithm.hexdigest()
+        logger.debug("Computed artifact hash: %s", artifact_hash)
+        return artifact_hash
+
     def get_package_page(self, package_name: str) -> str | None:
         """Implement custom API to get package main page.
 
@@ -430,15 +471,19 @@ def get_latest_version(self) -> str | None:
         """
         return json_extract(self.package_json, ["info", "version"], str)
 
-    def get_sourcecode_url(self) -> str | None:
+    def get_sourcecode_url(self, package_type: str = "sdist") -> str | None:
         """Get the url of the source distribution.
 
+        Parameters
+        ----------
+        package_type: str
+            The package type to retrieve the URL of.
+
         Returns
         -------
         str | None
             The URL of the source distribution.
         """
-        urls: list | None = None
         if self.component_version:
             urls = json_extract(self.package_json, ["releases", self.component_version], list)
         else:
@@ -447,7 +492,7 @@ def get_sourcecode_url(self) -> str | None:
         if not urls:
             return None
         for distribution in urls:
-            if distribution.get("packagetype") != "sdist":
+            if distribution.get("packagetype") != package_type:
                 continue
             # We intentionally check if the url is None and use empty string if that's the case.
             source_url: str = distribution.get("url") or ""
@@ -497,3 +542,39 @@ def get_sourcecode(self) -> dict[str, str] | None:
             source_code: dict[str, str] | None = self.pypi_registry.fetch_sourcecode(url)
             return source_code
         return None
+
+
+def find_or_create_pypi_asset(
+    asset_name: str, asset_version: str | None, pypi_registry_info: PackageRegistryInfo
+) -> PyPIPackageJsonAsset | None:
+    """Find the asset in the provided package registry information, or create it.
+
+    Parameters
+    ----------
+    asset_name: str
+        The name of the asset.
+    asset_version: str | None
+        The version of the asset.
+    pypi_registry_info:
+        The package registry information.
+
+    Returns
+    -------
+    PyPIPackageJsonAsset | None
+        The asset, or None if not found.
+    """
+    pypi_package_json = next(
+        (asset for asset in pypi_registry_info.metadata if isinstance(asset, PyPIPackageJsonAsset)),
+        None,
+    )
+    if pypi_package_json:
+        return pypi_package_json
+
+    package_registry = pypi_registry_info.package_registry
+    if not isinstance(package_registry, PyPIRegistry):
+        logger.debug("Failed to create PyPIPackageJson asset.")
+        return None
+
+    asset = PyPIPackageJsonAsset(asset_name, asset_version, False, package_registry, {})
+    pypi_registry_info.metadata.append(asset)
+    return asset