feat: add GitHub attestation discovery

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

Author: benmss

src/macaron/artifact/local_artifact.py

@@ -1,16 +1,21 @@
-# Copyright (c) 2024 - 2024, Oracle and/or its affiliates. All rights reserved.
+# Copyright (c) 2024 - 2025, Oracle and/or its affiliates. All rights reserved.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/.
 
 """This module declares types and utilities for handling local artifacts."""
 
 import fnmatch
 import glob
+import hashlib
+import logging
 import os
 
 from packageurl import PackageURL
 
 from macaron.artifact.maven import construct_maven_repository_path
 from macaron.errors import LocalArtifactFinderError
+from macaron.slsa_analyzer.package_registry import MavenCentralRegistry
+
+logger: logging.Logger = logging.getLogger(__name__)
 
 
 def construct_local_artifact_dirs_glob_pattern_maven_purl(maven_purl: PackageURL) -> list[str] | None:
@@ -247,3 +252,53 @@ def get_local_artifact_dirs(
         )
 
     raise LocalArtifactFinderError(f"Unsupported PURL type {purl_type}")
+
+
+def get_local_artifact_hash(purl: PackageURL, artifact_dirs: list[str], hash_algorithm_name: str) -> str | None:
+    """Compute the hash of the local artifact.
+
+    Parameters
+    ----------
+    purl: PackageURL
+        The PURL of the artifact being sought.
+    artifact_dirs: list[str]
+        The possible locations of the artifact.
+    hash_algorithm_name: str
+        The hash algorithm to use.
+
+    Returns
+    -------
+    str | None
+        The hash, or None if not found.
+    """
+    if not artifact_dirs:
+        logger.debug("No artifact directories provided.")
+        return None
+
+    if not purl.version:
+        logger.debug("PURL is missing version.")
+        return None
+
+    artifact_target = None
+    if purl.type == "maven":
+        artifact_target = MavenCentralRegistry.get_artifact_file_name(purl)
+
+    if not artifact_target:
+        logger.debug("PURL type not supported: %s", purl.type)
+        return None
+
+    for artifact_dir in artifact_dirs:
+        full_path = os.path.join(artifact_dir, artifact_target)
+        if not os.path.exists(full_path):
+            continue
+
+        with open(full_path, "rb") as file:
+            try:
+                hash_result = hashlib.file_digest(file, hash_algorithm_name)
+            except ValueError as error:
+                logger.debug("Error while hashing file: %s", error)
+                continue
+
+            return hash_result.hexdigest()
+
+    return None

src/macaron/slsa_analyzer/analyzer.py

@@ -4,11 +4,14 @@
 """This module handles the cloning and analyzing a Git repo."""
 
 import glob
+import hashlib
+import json
 import logging
 import os
 import re
 import sys
 import tempfile
+import urllib.parse
 from collections.abc import Mapping
 from datetime import datetime, timezone
 from pathlib import Path
@@ -20,7 +23,10 @@
 from sqlalchemy.orm import Session
 
 from macaron import __version__
-from macaron.artifact.local_artifact import get_local_artifact_dirs
+from macaron.artifact.local_artifact import (
+    get_local_artifact_dirs,
+    get_local_artifact_hash,
+)
 from macaron.config.global_config import global_config
 from macaron.config.target_config import Configuration
 from macaron.database.database_manager import DatabaseManager, get_db_manager, get_db_session
@@ -41,6 +47,7 @@
     ProvenanceError,
     PURLNotFoundError,
 )
+from macaron.json_tools import json_extract
 from macaron.output_reporter.reporter import FileReporter
 from macaron.output_reporter.results import Record, Report, SCMStatus
 from macaron.provenance import provenance_verifier
@@ -66,12 +73,14 @@
 from macaron.slsa_analyzer.checks import *  # pylint: disable=wildcard-import,unused-wildcard-import # noqa: F401,F403
 from macaron.slsa_analyzer.ci_service import CI_SERVICES
 from macaron.slsa_analyzer.database_store import store_analyze_context_to_db
-from macaron.slsa_analyzer.git_service import GIT_SERVICES, BaseGitService
+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
+from macaron.slsa_analyzer.package_registry import PACKAGE_REGISTRIES, MavenCentralRegistry
 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
+from macaron.slsa_analyzer.provenance.loader import load_provenance_payload
 from macaron.slsa_analyzer.provenance.slsa import SLSAProvenanceData
 from macaron.slsa_analyzer.registry import registry
 from macaron.slsa_analyzer.specs.ci_spec import CIInfo
@@ -403,6 +412,17 @@ def run_single(
                 status=SCMStatus.ANALYSIS_FAILED,
             )
 
+        local_artifact_dirs = None
+        if parsed_purl and parsed_purl.type in self.local_artifact_repo_mapper:
+            local_artifact_repo_path = self.local_artifact_repo_mapper[parsed_purl.type]
+            try:
+                local_artifact_dirs = get_local_artifact_dirs(
+                    purl=parsed_purl,
+                    local_artifact_repo_path=local_artifact_repo_path,
+                )
+            except LocalArtifactFinderError as error:
+                logger.debug(error)
+
         # Prepare the repo.
         git_obj = None
         commit_finder_outcome = CommitFinderInfo.NOT_USED
@@ -480,6 +500,37 @@ def run_single(
         git_service = self._determine_git_service(analyze_ctx)
         self._determine_ci_services(analyze_ctx, git_service)
         self._determine_build_tools(analyze_ctx, git_service)
+
+        # Try to find an attestation from GitHub, if applicable.
+        if parsed_purl and not provenance_payload and analysis_target.repo_path and isinstance(git_service, GitHub):
+            # Try to discover GitHub attestation for the target software component.
+            url = None
+            try:
+                url = urllib.parse.urlparse(analysis_target.repo_path)
+            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())
+                if artifact_hash:
+                    git_attestation_dict = git_service.api_client.get_attestation(
+                        analyze_ctx.component.repository.full_name, artifact_hash
+                    )
+                    if git_attestation_dict:
+                        git_attestation_list = json_extract(git_attestation_dict, ["attestations"], list)
+                        if git_attestation_list:
+                            git_attestation = git_attestation_list[0]
+
+                            with tempfile.TemporaryDirectory() as temp_dir:
+                                attestation_file = os.path.join(temp_dir, "attestation")
+                                with open(attestation_file, "w", encoding="UTF-8") as file:
+                                    json.dump(git_attestation, file)
+
+                                try:
+                                    payload = load_provenance_payload(attestation_file)
+                                    provenance_payload = payload
+                                except LoadIntotoAttestationError as error:
+                                    logger.debug("Failed to load provenance payload: %s", error)
+
         if parsed_purl is not None:
             self._verify_repository_link(parsed_purl, analyze_ctx)
         self._determine_package_registries(analyze_ctx, all_package_registries)
@@ -541,16 +592,8 @@ def run_single(
 
         analyze_ctx.dynamic_data["validate_malware"] = validate_malware
 
-        if parsed_purl and parsed_purl.type in self.local_artifact_repo_mapper:
-            local_artifact_repo_path = self.local_artifact_repo_mapper[parsed_purl.type]
-            try:
-                local_artifact_dirs = get_local_artifact_dirs(
-                    purl=parsed_purl,
-                    local_artifact_repo_path=local_artifact_repo_path,
-                )
-                analyze_ctx.dynamic_data["local_artifact_paths"].extend(local_artifact_dirs)
-            except LocalArtifactFinderError as error:
-                logger.debug(error)
+        if local_artifact_dirs:
+            analyze_ctx.dynamic_data["local_artifact_paths"].extend(local_artifact_dirs)
 
         analyze_ctx.check_results = registry.scan(analyze_ctx)
 
@@ -939,6 +982,54 @@ 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
+    ) -> str | None:
+        """Get the hash of the artifact found from the passed PURL using local or remote files.
+
+        Parameters
+        ----------
+        purl: PackageURL
+            The PURL of the artifact.
+        cached_artifacts: list[str] | None
+            The list of local files that match the PURL.
+        hash_algorithm: Any
+            The hash algorithm to use.
+
+        Returns
+        -------
+        str | None
+            The hash of the artifact, or None if not found.
+        """
+        if cached_artifacts:
+            # Try to get the hash from a local file.
+            artifact_hash = get_local_artifact_hash(purl, cached_artifacts, hash_algorithm.name)
+
+            if artifact_hash:
+                return artifact_hash
+
+        # Download the artifact.
+        if purl.type == "maven":
+            maven_registry = next(
+                (
+                    package_registry
+                    for package_registry in PACKAGE_REGISTRIES
+                    if isinstance(package_registry, MavenCentralRegistry)
+                ),
+                None,
+            )
+            if not maven_registry:
+                return None
+
+            return maven_registry.get_artifact_hash(purl, hash_algorithm)
+
+        if purl.type == "pypi":
+            # TODO implement
+            return None
+
+        logger.debug("Purl type '%s' not yet supported for GitHub attestation discovery.", purl.type)
+        return None
+
     def _determine_git_service(self, analyze_ctx: AnalyzeContext) -> BaseGitService:
         """Determine the Git service used by the software component."""
         remote_path = analyze_ctx.component.repository.remote_path if analyze_ctx.component.repository else None

src/macaron/slsa_analyzer/git_service/api_client.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2022 - 2024, Oracle and/or its affiliates. All rights reserved.
+# Copyright (c) 2022 - 2025, Oracle and/or its affiliates. All rights reserved.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/.
 
 """The module provides API clients for VCS services, such as GitHub."""
@@ -659,6 +659,25 @@ def download_asset(self, url: str, download_path: str) -> bool:
 
         return True
 
+    def get_attestation(self, full_name: str, artifact_hash: str) -> dict:
+        """Download and return the attestation associated with the passed artifact hash, if any.
+
+        Parameters
+        ----------
+        full_name : str
+            The full name of the repo.
+        artifact_hash: str
+            The SHA256 hash of an artifact.
+
+        Returns
+        -------
+        dict
+            The attestation data, or an empty dict if not found.
+        """
+        url = f"{GhAPIClient._REPO_END_POINT}/{full_name}/attestations/sha256:{artifact_hash}"
+        response_data = send_get_http(url, self.headers)
+        return response_data or {}
+
 
 def get_default_gh_client(access_token: str) -> GhAPIClient:
     """Return a GhAPIClient instance with default values.

src/macaron/slsa_analyzer/package_registry/maven_central_registry.py

@@ -6,10 +6,13 @@
 import logging
 import urllib.parse
 from datetime import datetime, timezone
+from typing import Any
 
 import requests
 from packageurl import PackageURL
+from requests import RequestException
 
+from macaron.artifact.maven import construct_maven_repository_path
 from macaron.config.defaults import defaults
 from macaron.errors import ConfigurationError, InvalidHTTPResponseError
 from macaron.slsa_analyzer.package_registry.package_registry import PackageRegistry
@@ -236,3 +239,72 @@ def find_publish_timestamp(self, purl: str) -> datetime:
                 raise InvalidHTTPResponseError(f"The timestamp returned by {url} is invalid") from error
 
         raise InvalidHTTPResponseError(f"Invalid response from Maven central for {url}.")
+
+    @staticmethod
+    def get_artifact_file_name(purl: PackageURL) -> str | None:
+        """Return the artifact file name of the passed PURL based on the Maven registry standard.
+
+        Parameters
+        ----------
+        purl: PackageURL
+            The PURL of the artifact.
+
+        Returns
+        -------
+        str | None
+            The artifact file name, or None if invalid.
+        """
+        if not purl.version:
+            return None
+
+        return purl.name + "-" + purl.version + ".jar"
+
+    def get_artifact_hash(self, purl: PackageURL, hash_algorithm: Any) -> str | None:
+        """Return the hash of the artifact found by the passed purl relevant to the registry's URL.
+
+        Parameters
+        ----------
+        purl: PackageURL
+            The purl of the artifact.
+        hash_algorithm: Any
+            The hash algorithm to use.
+
+        Returns
+        -------
+        str | None
+            The hash of the artifact, or None if not found.
+        """
+        if not (purl.namespace and purl.version):
+            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:
+            return None
+
+        artifact_url = self.registry_url + "/" + artifact_path + "/" + file_name
+        logger.debug("Search for artifact using URL: %s", artifact_url)
+
+        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:
+            return None
+
+        # Download file and compute hash as chunks are received.
+        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 target file: %s", error)
+            response.close()
+            return None
+
+        artifact_hash: str = hash_algorithm.hexdigest()
+        logger.debug("Computed hash of artifact: %s", artifact_hash)
+        return artifact_hash