feat: improve vsa generation with digest for each subject

Signed-off-by: Nathan Nguyen <[email protected]>

Author: nathanwn

src/macaron/artifact/maven.py

@@ -0,0 +1,187 @@
+# Copyright (c) 2024 - 2024, 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 Maven artifacts."""
+
+import re
+from dataclasses import dataclass
+from enum import Enum
+from typing import NamedTuple, Self
+
+from packageurl import PackageURL
+
+from macaron.slsa_analyzer.provenance.intoto import InTotoPayload
+from macaron.slsa_analyzer.provenance.intoto.v01 import InTotoV01Subject
+from macaron.slsa_analyzer.provenance.intoto.v1 import InTotoV1ResourceDescriptor
+from macaron.slsa_analyzer.provenance.witness import (
+    extract_build_artifacts_from_witness_subjects,
+    is_witness_provenance_payload,
+    load_witness_verifier_config,
+)
+
+
+class _MavenArtifactType(NamedTuple):
+    filename_pattern: str
+    purl_qualifiers: dict[str, str]
+
+
+class MavenArtifactType(_MavenArtifactType, Enum):
+    """Maven artifact types that Macaron supports.
+
+    For reference, see:
+    - https://maven.apache.org/ref/3.9.6/maven-core/artifact-handlers.html
+    - https://github.com/package-url/purl-spec/blob/master/PURL-TYPES.rst#maven
+    """
+
+    # Enum with custom value type.
+    # See https://docs.python.org/3.10/library/enum.html#others.
+    JAR = _MavenArtifactType(
+        filename_pattern="{artifact_id}-{version}.jar",
+        purl_qualifiers={"type": "jar"},
+    )
+    POM = _MavenArtifactType(
+        filename_pattern="{artifact_id}-{version}.pom",
+        purl_qualifiers={"type": "pom"},
+    )
+    JAVADOC = _MavenArtifactType(
+        filename_pattern="{artifact_id}-{version}-javadoc.jar",
+        purl_qualifiers={"type": "javadoc"},
+    )
+    JAVA_SOURCE = _MavenArtifactType(
+        filename_pattern="{artifact_id}-{version}-sources.jar",
+        purl_qualifiers={"type": "sources"},
+    )
+
+
+@dataclass
+class MavenArtifact:
+    """A Maven artifact."""
+
+    group_id: str
+    artifact_id: str
+    version: str
+    artifact_type: MavenArtifactType
+
+    @property
+    def package_url(self) -> PackageURL:
+        """Get the PackageURL of this Maven artifact."""
+        return PackageURL(
+            type="maven",
+            namespace=self.group_id,
+            name=self.artifact_id,
+            version=self.version,
+            qualifiers=self.artifact_type.purl_qualifiers,
+        )
+
+    @classmethod
+    def from_package_url(cls, package_url: PackageURL) -> Self | None:
+        """Create a Maven artifact from a PackageURL.
+
+        Returns
+        -------
+        Self | None
+            A Maven artifact, or ``None`` if the PURL is not a valid Maven artifact PURL, or if
+            the artifact type is not supported.
+            For supported artifact types, see :class:`MavenArtifactType`.
+        """
+        if not package_url.namespace:
+            return None
+        if not package_url.version:
+            return None
+        if package_url.type != "maven":
+            return None
+        maven_artifact_type = None
+        for artifact_type in MavenArtifactType:
+            if artifact_type.purl_qualifiers == package_url.qualifiers:
+                maven_artifact_type = artifact_type
+                break
+        if not maven_artifact_type:
+            return None
+        return cls(
+            group_id=package_url.namespace,
+            artifact_id=package_url.name,
+            version=package_url.version,
+            artifact_type=maven_artifact_type,
+        )
+
+    @classmethod
+    def from_artifact_name(
+        cls,
+        artifact_name: str,
+        group_id: str,
+        version: str,
+    ) -> Self | None:
+        """Create a Maven artifact from a PackageURL.
+
+        Returns
+        -------
+        Self | None
+            A Maven artifact, or ``None`` if the PURL is not a valid Maven artifact PURL, or if
+            the artifact type is not supported.
+            For supported artifact types, see :class:`MavenArtifactType`.
+        """
+        for maven_artifact_type in MavenArtifactType:
+            pattern = maven_artifact_type.filename_pattern.format(
+                artifact_id="(.*)",
+                version=version,
+            )
+            match_result = re.search(pattern, artifact_name)
+            if not match_result:
+                continue
+            artifact_id = match_result.group(1)
+            return cls(
+                group_id=group_id,
+                artifact_id=artifact_id,
+                version=version,
+                artifact_type=maven_artifact_type,
+            )
+        return None
+
+
+class MavenSubjectPURLMatcher:
+    """A matcher matching a PURL identifying a Maven artifact to a provenance subject."""
+
+    @staticmethod
+    def get_subject_in_provenance_matching_purl(
+        provenance_payload: InTotoPayload, purl: PackageURL
+    ) -> InTotoV01Subject | InTotoV1ResourceDescriptor | None:
+        """Get the subject in the provenance matching the PURL.
+
+        In this case where the provenance is assumed to be built from a Java project,
+        the subject must be a Maven artifact.
+
+        Parameters
+        ----------
+        provenance_payload : InTotoPayload
+            The provenance payload.
+        purl : PackageURL
+            The PackageURL identifying the matching subject.
+
+        Returns
+        -------
+        InTotoV01Subject
+            The PackageURL identifying the matching subject.
+        """
+        if (maven_artifact := MavenArtifact.from_package_url(purl)) and is_witness_provenance_payload(
+            payload=provenance_payload,
+            predicate_types=load_witness_verifier_config().predicate_types,
+        ):
+            artifact_subjects = extract_build_artifacts_from_witness_subjects(provenance_payload)
+
+            maven_artifact_subject_pairs = []
+            for subject in artifact_subjects:
+                _, _, artifact_name = subject["name"].rpartition("/")
+                artifact = MavenArtifact.from_artifact_name(
+                    artifact_name=artifact_name,
+                    group_id=maven_artifact.group_id,
+                    version=maven_artifact.version,
+                )
+                if artifact is None:
+                    continue
+                maven_artifact_subject_pairs.append((artifact, subject))
+
+            for artifact, subject in maven_artifact_subject_pairs:
+                if artifact.package_url == purl:
+                    return subject
+
+        return None

src/macaron/database/table_definitions.py

@@ -15,7 +15,7 @@
 import string
 from datetime import datetime
 from pathlib import Path
-from typing import Any
+from typing import Any, Self
 
 from packageurl import PackageURL
 from sqlalchemy import (
@@ -32,9 +32,11 @@
 )
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+from macaron.artifact.maven import MavenSubjectPURLMatcher
 from macaron.database.database_manager import ORMBase
 from macaron.database.rfc3339_datetime import RFC3339DateTime
 from macaron.errors import InvalidPURLError
+from macaron.slsa_analyzer.provenance.intoto import InTotoPayload, ProvenanceSubjectPURLMatcher
 from macaron.slsa_analyzer.slsa_req import ReqName
 
 logger: logging.Logger = logging.getLogger(__name__)
@@ -168,6 +170,13 @@ class Component(PackageURLMixin, ORMBase):
         secondaryjoin=components_association_table.c.child_component == id,
     )
 
+    #: The optional one-to-one relationship with a provenance subject in case this
+    #: component represents a subject in a provenance.
+    provenance_subject: Mapped["ProvenanceSubject | None"] = relationship(
+        back_populates="component",
+        lazy="immediate",
+    )
+
     def __init__(self, purl: str, analysis: Analysis, repository: "Repository | None"):
         """
         Instantiate the software component using PURL identifier.
@@ -528,3 +537,71 @@ class HashDigest(ORMBase):
 
     #: The many-to-one relationship with artifacts.
     artifact: Mapped["ReleaseArtifact"] = relationship(back_populates="digests", lazy="immediate")
+
+
+class ProvenanceSubject(ORMBase):
+    """A subject in a provenance that matches the user-provided PackageURL.
+
+    This subject may be later populated in VSAs during policy verification.
+    """
+
+    __tablename__ = "_provenance_subject"
+
+    #: The primary key.
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)  # noqa: A003
+
+    #: The component id of the provenance subject.
+    component_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("_component.id"),
+        nullable=False,
+    )
+
+    #: The required one-to-one relationship with a component.
+    component: Mapped[Component] = relationship(
+        back_populates="provenance_subject",
+        lazy="immediate",
+    )
+
+    #: The SHA256 hash of the subject.
+    sha256: Mapped[str] = mapped_column(String, nullable=False)
+
+    @classmethod
+    def from_purl_and_provenance(
+        cls,
+        purl: PackageURL,
+        provenance_payload: InTotoPayload,
+    ) -> Self | None:
+        """Create a ``ProvenanceSubject`` entry if there is a provenance subject matching the PURL.
+
+        Parameters
+        ----------
+        purl : PackageURL
+            The PackageURL identifying the software component being analyzed.
+        provenance_payload : InTotoPayload
+            The provenance payload.
+
+        Returns
+        -------
+        Self | None
+            A ``ProvenanceSubject`` entry with the SHA256 of the provenance subject matching the
+            given PURL.
+        """
+        subject_artifact_types: list[ProvenanceSubjectPURLMatcher] = [MavenSubjectPURLMatcher]
+
+        for subject_artifact_type in subject_artifact_types:
+            subject = subject_artifact_type.get_subject_in_provenance_matching_purl(
+                provenance_payload,
+                purl,
+            )
+            if subject is None:
+                return None
+            digest = subject["digest"]
+            if digest is None:
+                return None
+            sha256 = digest.get("sha256")
+            if not sha256:
+                return None
+            return cls(sha256=sha256)
+
+        return None

src/macaron/slsa_analyzer/analyzer.py

@@ -19,7 +19,7 @@
 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
-from macaron.database.table_definitions import Analysis, Component, Repository
+from macaron.database.table_definitions import Analysis, Component, ProvenanceSubject, Repository
 from macaron.dependency_analyzer import DependencyAnalyzer, DependencyInfo
 from macaron.errors import (
     CloneError,
@@ -332,7 +332,12 @@ def run_single(
         # Create the component.
         component = None
         try:
-            component = self.add_component(analysis, analysis_target, existing_records)
+            component = self.add_component(
+                analysis,
+                analysis_target,
+                existing_records,
+                provenance_payload,
+            )
         except PURLNotFoundError as error:
             logger.error(error)
             return Record(
@@ -484,6 +489,7 @@ def add_component(
         analysis: Analysis,
         analysis_target: AnalysisTarget,
         existing_records: dict[str, Record] | None = None,
+        provenance_payload: InTotoPayload | None = None,
     ) -> Component:
         """Add a software component if it does not exist in the DB already.
 
@@ -547,18 +553,30 @@ def add_component(
                 raise PURLNotFoundError(
                     f"The repository {analysis_target.repo_path} is not available and no PURL is provided from the user."
                 )
-
-            repo_snapshot_purl = PackageURL(
+            purl = PackageURL(
                 type=repository.type,
                 namespace=repository.owner,
                 name=repository.name,
                 version=repository.commit_sha,
             )
-            return Component(purl=str(repo_snapshot_purl), analysis=analysis, repository=repository)
+        else:
+            # If the PURL is available, we always create the software component with it whether the repository is
+            # available or not.
+            purl = analysis_target.parsed_purl
+
+        component = Component(
+            purl=purl.to_string(),
+            analysis=analysis,
+            repository=repository,
+        )
+
+        if provenance_payload:
+            component.provenance_subject = ProvenanceSubject.from_purl_and_provenance(
+                purl=purl,
+                provenance_payload=provenance_payload,
+            )
 
-        # If the PURL is available, we always create the software component with it whether the repository is
-        # available or not.
-        return Component(purl=str(analysis_target.parsed_purl), analysis=analysis, repository=repository)
+        return component
 
     @staticmethod
     def parse_purl(config: Configuration) -> PackageURL | None:

src/macaron/slsa_analyzer/checks/provenance_available_check.py

@@ -57,7 +57,7 @@ class ProvenanceAvailableFacts(CheckFacts):
     id: Mapped[int] = mapped_column(ForeignKey("_check_facts.id"), primary_key=True)  # noqa: A003
 
     #: The provenance asset name.
-    asset_name: Mapped[str] = mapped_column(String, nullable=False, info={"justification": JustificationType.TEXT})
+    asset_name: Mapped[str] = mapped_column(String, nullable=True, info={"justification": JustificationType.TEXT})
 
     #: The URL for the provenance asset.
     asset_url: Mapped[str] = mapped_column(String, nullable=True, info={"justification": JustificationType.HREF})
@@ -504,6 +504,12 @@ def run_check(self, ctx: AnalyzeContext) -> CheckResultData:
         CheckResultData
             The result of the check.
         """
+        if ctx.dynamic_data["provenance"]:
+            return CheckResultData(
+                result_tables=[ProvenanceAvailableFacts(confidence=Confidence.HIGH)],
+                result_type=CheckResultType.PASSED,
+            )
+
         provenance_extensions = defaults.get_list(
             "slsa.verifier",
             "provenance_extensions",

src/macaron/slsa_analyzer/checks/provenance_l3_content_check.py

@@ -58,6 +58,12 @@ def run_check(self, ctx: AnalyzeContext) -> CheckResultData:
             logger.info("%s check was unable to find any expectations.", self.check_info.check_id)
             return CheckResultData(result_tables=[], result_type=CheckResultType.UNKNOWN)
 
+        if ctx.dynamic_data["provenance"] and expectation.validate(ctx.dynamic_data["provenance"]):
+            return CheckResultData(
+                result_tables=[expectation],
+                result_type=CheckResultType.PASSED,
+            )
+
         package_registry_info_entries = ctx.dynamic_data["package_registries"]
         ci_services = ctx.dynamic_data["ci_services"]