chore: account for witness provenance

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

Author: benmss

src/macaron/database/db_custom_types.py

@@ -8,7 +8,12 @@
 
 from sqlalchemy import JSON, String, TypeDecorator
 
-from macaron.slsa_analyzer.provenance.intoto import InTotoPayload, validate_intoto_payload
+from macaron.slsa_analyzer.provenance.intoto import (
+    InTotoPayload,
+    InTotoV01Payload,
+    InTotoV1Payload,
+    validate_intoto_payload,
+)
 
 
 class RFC3339DateTime(TypeDecorator):  # pylint: disable=W0223
@@ -118,7 +123,8 @@ def process_bind_param(self, value: None | InTotoPayload, dialect: Any) -> None
         if not isinstance(value, InTotoPayload):
             raise TypeError("ProvenancePayload type expects an InTotoPayload.")
 
-        return value.statement.get("predicate")
+        payload_type = value.__class__.__name__
+        return {"payload_type": payload_type, "payload": value.statement}
 
     def process_result_value(self, value: None | dict, dialect: Any) -> None | InTotoPayload:
         """Process when loading an InTotoPayload object from the SQLite db.
@@ -132,4 +138,13 @@ def process_result_value(self, value: None | dict, dialect: Any) -> None | InTot
         if not isinstance(value, dict):
             raise TypeError("ProvenancePayload type expects a dict.")
 
+        if "payload_type" not in value or "payload" not in value:
+            raise TypeError("Missing keys in dict for ProvenancePayload type.")
+
+        payload = value["payload"]
+        if value["payload_type"] == "InTotoV01Payload":
+            return InTotoV01Payload(statement=payload)
+        if value["payload_type"] == "InTotoV1Payload":
+            return InTotoV1Payload(statement=payload)
+
         return validate_intoto_payload(value)

src/macaron/database/table_definitions.py

@@ -479,7 +479,7 @@ class Provenance(ORMBase):
     component: Mapped["Component"] = relationship(back_populates="provenance")
 
     #: The SLSA version.
-    version: Mapped[str] = mapped_column(String, nullable=True)
+    slsa_version: Mapped[str] = mapped_column(String, nullable=True)
 
     #: The SLSA level.
     slsa_level: Mapped[int] = mapped_column(Integer, default=0)

src/macaron/provenance/provenance_verifier.py

@@ -91,19 +91,21 @@ def verify_npm_provenance(purl: PackageURL, provenance: list[InTotoPayload]) ->
     found_signed_subject = None
     for signed_subject in signed_subjects:
         name = signed_subject.get("name")
-        if name and name == str(purl):
-            found_signed_subject = signed_subject
-            break
+        if not name or not check_purls_equivalent(purl, PackageURL.from_string(name)):
+            continue
+        found_signed_subject = signed_subject
+        break
 
     if not found_signed_subject:
         return False
 
     found_unsigned_subject = None
     for unsigned_subject in unsigned_subjects:
         name = unsigned_subject.get("name")
-        if name and name == str(purl):
-            found_unsigned_subject = unsigned_subject
-            break
+        if not name or not check_purls_equivalent(purl, PackageURL.from_string(name)):
+            continue
+        found_unsigned_subject = unsigned_subject
+        break
 
     if not found_unsigned_subject:
         return False
@@ -127,6 +129,19 @@ def verify_npm_provenance(purl: PackageURL, provenance: list[InTotoPayload]) ->
     return True
 
 
+def check_purls_equivalent(original_purl: PackageURL, new_purl: PackageURL) -> bool:
+    """Check if `new_purl` is equivalent to `original_purl`, excluding versions if the original has none."""
+    if (
+        original_purl.type != new_purl.type
+        or original_purl.name != new_purl.name
+        or original_purl.namespace != new_purl.namespace
+    ):
+        return False
+    if original_purl.version and original_purl.version != new_purl.version:
+        return False
+    return True
+
+
 def verify_ci_provenance(analyze_ctx: AnalyzeContext, ci_info: CIInfo, download_path: str) -> bool:
     """Try to verify the CI provenance in terms of SLSA level 3 requirements.
 
@@ -370,7 +385,9 @@ def determine_provenance_slsa_level(
     predicate = provenance_payload.statement.get("predicate")
     build_type = None
     if predicate:
-        build_type = json_extract(predicate, ["buildType"], str)
+        build_type = json_extract(predicate, ["buildDefinition", "buildType"], str)
+        if not build_type:
+            build_type = json_extract(predicate, ["buildType"], str)
 
     if build_type == "https://github.com/slsa-framework/slsa-github-generator/generic@v1" and verified_l3:
         # 3. Provenance is created by the SLSA GitHub generator and verified.

src/macaron/slsa_analyzer/analyzer.py

@@ -485,13 +485,15 @@ def run_single(
                     if verified and all(verified):
                         provenance_l3_verified = True
 
-        slsa_level = determine_provenance_slsa_level(
-            analyze_ctx, provenance_payload, provenance_is_verified, provenance_l3_verified
-        )
         slsa_version = None
         if provenance_payload:
+            analyze_ctx.dynamic_data["is_inferred_prov"] = False
             slsa_version = extract_predicate_version(provenance_payload)
 
+        slsa_level = determine_provenance_slsa_level(
+            analyze_ctx, provenance_payload, provenance_is_verified, provenance_l3_verified
+        )
+
         analyze_ctx.dynamic_data["provenance_info"] = table_definitions.Provenance(
             component=component,
             repository_url=provenance_repo_url,
@@ -502,8 +504,7 @@ def run_single(
             slsa_version=slsa_version,
             # TODO Add release tag, release digest.
         )
-        if provenance_payload:
-            analyze_ctx.dynamic_data["is_inferred_prov"] = False
+
         analyze_ctx.dynamic_data["validate_malware_switch"] = validate_malware_switch
 
         if parsed_purl and parsed_purl.type in self.local_artifact_repo_mapper:

src/macaron/slsa_analyzer/checks/provenance_verified_check.py

@@ -72,7 +72,7 @@ def run_check(self, ctx: AnalyzeContext) -> CheckResultData:
         if provenance_info and provenance_info.provenance_payload:
             predicate = provenance_info.provenance_payload.statement.get("predicate")
             if predicate:
-                build_type = json_extract(predicate, ["buildType"], str)
+                build_type = json_extract(predicate, ["buildDefinition", "buildType"], str)
 
         slsa_level = 0
         if provenance_info:

tests/integration/cases/urllib3_expectation_dir/policy.dl

@@ -20,7 +20,6 @@ Policy("test_policy", component_id, "") :-
     check_passed(component_id, "mcn_provenance_verified_1"),
     provenance_verified_check(_, build_level, _),
     build_level = 3,
-    check_failed(component_id, "mcn_infer_artifact_pipeline_1"),
     check_failed(component_id, "mcn_provenance_witness_level_one_1"),
     check_failed(component_id, "mcn_trusted_builder_level_three_1"),
     is_repo_url(component_id, "https://github.com/urllib3/urllib3"),