chore: minor fixes

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

Author: benmss

src/macaron/slsa_analyzer/analyzer.py

@@ -512,7 +512,7 @@ def run_single(
                 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(), all_package_registries
+                    parsed_purl, local_artifact_dirs, hashlib.sha256(), package_registries_info
                 )
                 if artifact_hash:
                     git_attestation_dict = git_service.api_client.get_attestation(
@@ -991,7 +991,7 @@ def get_artifact_hash(
         purl: PackageURL,
         cached_artifacts: list[str] | None,
         hash_algorithm: Any,
-        all_package_registries: list[PackageRegistryInfo],
+        package_registries_info: list[PackageRegistryInfo],
     ) -> str | None:
         """Get the hash of the artifact found from the passed PURL using local or remote files.
 
@@ -1003,7 +1003,7 @@ 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]
+        package_registries_info: list[PackageRegistryInfo]
             The list of package registry information.
 
         Returns
@@ -1049,7 +1049,7 @@ def get_artifact_hash(
             registry_info = next(
                 (
                     info
-                    for info in all_package_registries
+                    for info in package_registries_info
                     if info.package_registry == pypi_registry and info.build_tool_name in {"pip", "poetry"}
                 ),
                 None,

src/macaron/slsa_analyzer/checks/detect_malicious_metadata_check.py

@@ -32,7 +32,6 @@
 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.osv_dev import OSVDevService
-from macaron.slsa_analyzer.package_registry.pypi_registry import PyPIPackageJsonAsset, PyPIRegistry
 from macaron.slsa_analyzer.package_registry.pypi_registry import (
     PyPIPackageJsonAsset,
     PyPIRegistry,

src/macaron/slsa_analyzer/package_registry/maven_central_registry.py

@@ -274,15 +274,15 @@ def get_artifact_hash(self, purl: PackageURL, hash_algorithm: Any) -> str | None
         str | None
             The hash of the artifact, or None if not found.
         """
-        if not (purl.namespace and purl.version):
+        if not purl.namespace:
             return None
 
-        artifact_path = construct_maven_repository_path(purl.namespace, purl.name, purl.version)
         file_name = MavenCentralRegistry.get_artifact_file_name(purl)
-        if not file_name:
+        if not (purl.version and file_name):
             return None
 
         # Maven supports but does not require a sha256 hash of uploaded artifacts.
+        artifact_path = construct_maven_repository_path(purl.namespace, purl.name, purl.version)
         artifact_url = self.registry_url + "/" + artifact_path + "/" + file_name
         sha256_url = artifact_url + ".sha256"
         logger.debug("Search for artifact hash using URL: %s", [sha256_url, artifact_url])

tests/slsa_analyzer/package_registry/test_maven_central_registry.py

@@ -7,11 +7,14 @@
 import os
 import urllib.parse
 from datetime import datetime
+from hashlib import sha256
 from pathlib import Path
 
 import pytest
+from packageurl import PackageURL
 from pytest_httpserver import HTTPServer
 
+from macaron.artifact.maven import construct_maven_repository_path
 from macaron.config.defaults import load_defaults
 from macaron.errors import ConfigurationError, InvalidHTTPResponseError
 from macaron.slsa_analyzer.package_registry.maven_central_registry import MavenCentralRegistry
@@ -35,6 +38,28 @@ def maven_central_instance() -> MavenCentralRegistry:
     )
 
 
+@pytest.fixture(name="maven_service")
+def maven_service_(httpserver: HTTPServer, tmp_path: Path) -> None:
+    """Set up the Maven httpserver."""
+    base_url_parsed = urllib.parse.urlparse(httpserver.url_for(""))
+
+    user_config_input = f"""
+    [package_registry.maven_central]
+    request_timeout = 20
+    search_netloc = {base_url_parsed.netloc}
+    search_scheme = {base_url_parsed.scheme}
+    registry_url_netloc = {base_url_parsed.netloc}
+    registry_url_scheme = {base_url_parsed.scheme}
+    """
+    user_config_path = os.path.join(tmp_path, "config.ini")
+    with open(user_config_path, "w", encoding="utf-8") as user_config_file:
+        user_config_file.write(user_config_input)
+    # We don't have to worry about modifying the ``defaults`` object causing test
+    # pollution here, since we reload the ``defaults`` object before every test with the
+    # ``setup_test`` fixture.
+    load_defaults(user_config_path)
+
+
 def test_load_defaults(tmp_path: Path) -> None:
     """Test the ``load_defaults`` method."""
     user_config_path = os.path.join(tmp_path, "config.ini")
@@ -150,31 +175,14 @@ def test_is_detected(
 def test_find_publish_timestamp(
     resources_path: Path,
     httpserver: HTTPServer,
-    tmp_path: Path,
+    maven_service: dict,  # pylint: disable=unused-argument
     purl: str,
     mc_json_path: str,
     query_string: str,
     expected_timestamp: str,
 ) -> None:
     """Test that the function finds the timestamp correctly."""
-    base_url_parsed = urllib.parse.urlparse(httpserver.url_for(""))
-
     maven_central = MavenCentralRegistry()
-
-    # Set up responses of solrsearch endpoints using the httpserver plugin.
-    user_config_input = f"""
-    [package_registry.maven_central]
-    request_timeout = 20
-    search_netloc = {base_url_parsed.netloc}
-    search_scheme = {base_url_parsed.scheme}
-    """
-    user_config_path = os.path.join(tmp_path, "config.ini")
-    with open(user_config_path, "w", encoding="utf-8") as user_config_file:
-        user_config_file.write(user_config_input)
-    # We don't have to worry about modifying the ``defaults`` object causing test
-    # pollution here, since we reload the ``defaults`` object before every test with the
-    # ``setup_test`` fixture.
-    load_defaults(user_config_path)
     maven_central.load_defaults()
 
     with open(os.path.join(resources_path, "maven_central_files", mc_json_path), encoding="utf8") as page:
@@ -208,35 +216,19 @@ def test_find_publish_timestamp(
 def test_find_publish_timestamp_errors(
     resources_path: Path,
     httpserver: HTTPServer,
-    tmp_path: Path,
+    maven_service: dict,  # pylint: disable=unused-argument
     purl: str,
     mc_json_path: str,
     expected_msg: str,
 ) -> None:
     """Test that the function handles errors correctly."""
-    base_url_parsed = urllib.parse.urlparse(httpserver.url_for(""))
-
     maven_central = MavenCentralRegistry()
-
-    # Set up responses of solrsearch endpoints using the httpserver plugin.
-    user_config_input = f"""
-    [package_registry.maven_central]
-    request_timeout = 20
-    search_netloc = {base_url_parsed.netloc}
-    search_scheme = {base_url_parsed.scheme}
-    """
-    user_config_path = os.path.join(tmp_path, "config.ini")
-    with open(user_config_path, "w", encoding="utf-8") as user_config_file:
-        user_config_file.write(user_config_input)
-    # We don't have to worry about modifying the ``defaults`` object causing test
-    # pollution here, since we reload the ``defaults`` object before every test with the
-    # ``setup_test`` fixture.
-    load_defaults(user_config_path)
     maven_central.load_defaults()
 
     with open(os.path.join(resources_path, "maven_central_files", mc_json_path), encoding="utf8") as page:
         mc_json_response = json.load(page)
 
+    # Set up responses of solrsearch endpoints using the httpserver plugin.
     httpserver.expect_request(
         "/solrsearch/select",
         query_string="q=g:org.apache.logging.log4j+AND+a:log4j-core+AND+v:3.0.0-beta2&core=gav&rows=1&wt=json",
@@ -245,3 +237,67 @@ def test_find_publish_timestamp_errors(
     pat = f"^{expected_msg}"
     with pytest.raises(InvalidHTTPResponseError, match=pat):
         maven_central.find_publish_timestamp(purl=purl)
+
+
+def test_get_artifact_file_name() -> None:
+    """Test the artifact file name function."""
+    assert not MavenCentralRegistry().get_artifact_file_name(PackageURL.from_string("pkg:maven/test/example"))
+
+    assert (
+        MavenCentralRegistry().get_artifact_file_name(PackageURL.from_string("pkg:maven/text/example@1"))
+        == "example-1.jar"
+    )
+
+
+@pytest.mark.parametrize("purl_string", ["pkg:maven/example", "pkg:maven/example/test", "pkg:maven/example/test@1"])
+def test_get_artifact_hash_failures(
+    httpserver: HTTPServer, maven_service: dict, purl_string: str  # pylint: disable=unused-argument
+) -> None:
+    """Test failures of get artifact hash."""
+    purl = PackageURL.from_string(purl_string)
+
+    maven_registry = MavenCentralRegistry()
+    maven_registry.load_defaults()
+
+    if (
+        purl.namespace
+        and purl.version
+        and (file_name := MavenCentralRegistry().get_artifact_file_name(purl))
+        and file_name
+    ):
+        artifact_path = "/" + construct_maven_repository_path(purl.namespace, purl.name, purl.version) + "/" + file_name
+        hash_algorithm = sha256()
+        hash_algorithm.update(b"example_data")
+        expected_hash = hash_algorithm.hexdigest()
+        httpserver.expect_request(artifact_path + ".sha256").respond_with_data(expected_hash)
+        httpserver.expect_request(artifact_path).respond_with_data(b"example_data_2")
+
+    result = maven_registry.get_artifact_hash(purl, sha256())
+
+    assert not result
+
+
+def test_get_artifact_hash_success(
+    httpserver: HTTPServer, maven_service: dict  # pylint: disable=unused-argument
+) -> None:
+    """Test success of get artifact hash."""
+    purl = PackageURL.from_string("pkg:maven/example/test@1")
+    assert purl.namespace
+    assert purl.version
+
+    maven_registry = MavenCentralRegistry()
+    maven_registry.load_defaults()
+
+    file_name = MavenCentralRegistry().get_artifact_file_name(purl)
+    assert file_name
+
+    artifact_path = "/" + construct_maven_repository_path(purl.namespace, purl.name, purl.version) + "/" + file_name
+    hash_algorithm = sha256()
+    hash_algorithm.update(b"example_data")
+    expected_hash = hash_algorithm.hexdigest()
+    httpserver.expect_request(artifact_path + ".sha256").respond_with_data(expected_hash)
+    httpserver.expect_request(artifact_path).respond_with_data(b"example_data")
+
+    result = maven_registry.get_artifact_hash(purl, sha256())
+
+    assert result