Support embedded public key

Author: jobselko

pulp_python/app/provenance.py

@@ -8,15 +8,15 @@
 from cryptography.hazmat.primitives.asymmetric import padding as crypto_padding
 from cryptography.x509 import load_der_x509_certificate
 from django.conf import settings
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import Base64Bytes, BaseModel, ConfigDict, Field, model_validator
 from pydantic.alias_generators import to_snake
 from pypi_attestations import Attestation as _UpstreamAttestation
 from pypi_attestations import (
     Distribution,
     Envelope,  # noqa - needed in module namespace for Pydantic model rebuild
     Publisher,
     VerificationError,
-    VerificationMaterial,
+    VerificationMaterial as _UpstreamVerificationMaterial,
 )
 from sigstore.dsse import Envelope as DSSEEnvelope
 from sigstore.dsse import _pae
@@ -28,6 +28,23 @@
 SLSA_PROVENANCE_V02 = "https://slsa.dev/provenance/v0.2"
 
 
+class VerificationMaterial(_UpstreamVerificationMaterial):
+    """Extended verification material that supports optional certificate and public key.
+
+    PEP 740 requires a certificate, but this extension allows attestations signed
+    with a custom key where the certificate is absent.
+    """
+
+    certificate: Base64Bytes | None = None
+    public_key: str | None = None
+
+    @model_validator(mode="after")
+    def _require_certificate_or_public_key(self):
+        if self.certificate is None and self.public_key is None:
+            raise ValueError("verification_material must contain either a certificate or a public_key")
+        return self
+
+
 class _PermissivePolicy:
     """A permissive verification policy that always succeeds."""
 
@@ -110,6 +127,8 @@ def _has_valid_certificate(attestation):
         if vm is None:
             return False
         cert_bytes = vm.certificate
+        if cert_bytes is None:
+            return False
         load_der_x509_certificate(cert_bytes)
         return True
     except (ValueError, Exception):
@@ -180,6 +199,29 @@ def _verify_signature(attestation, public_key):
         raise VerificationError(f"signature verification failed: {e}")
 
 
+def _verify_embedded_key(attestation, server_key):
+    """Verify that the embedded public key matches the server-configured key."""
+    vm = attestation.verification_material
+    if vm is None or vm.public_key is None:
+        return
+    try:
+        embedded_key = serialization.load_pem_public_key(vm.public_key.encode())
+    except (ValueError, Exception) as e:
+        raise VerificationError(f"invalid embedded public key: {e}")
+    server_key_bytes = server_key.public_bytes(
+        serialization.Encoding.PEM,
+        serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+    embedded_key_bytes = embedded_key.public_bytes(
+        serialization.Encoding.PEM,
+        serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+    if server_key_bytes != embedded_key_bytes:
+        raise VerificationError(
+            "embedded public key does not match server-configured ATTESTATION_VERIFICATION_KEY"
+        )
+
+
 def verify_provenance(filename, sha256, provenance, offline=True):
     """Verify the provenance object is valid for the package.
 
@@ -205,8 +247,10 @@ def verify_provenance(filename, sha256, provenance, offline=True):
                 _enrich_publisher_from_statement(stmt, publisher)
                 if verification_key:
                     _verify_signature(attestation, verification_key)
+                    _verify_embedded_key(attestation, verification_key)
                 else:
                     raise VerificationError(
-                        "Attestation has no Sigstore certificate and no custom "
-                        "verification key is configured (ATTESTATION_VERIFICATION_KEY)"
+                        "Attestation has no Sigstore certificate or no custom "
+                        "verification key configured via ATTESTATION_VERIFICATION_KEY "
+                        "(embedded key in verification_material is optional)"
                     )

pulp_python/tests/functional/api/test_slsa_attestations.py

@@ -67,18 +67,33 @@ def _b64(data: bytes) -> str:
     return base64.b64encode(data).decode()
 
 
-def _make_attestation(statement_bytes, signature_bytes):
+def _make_attestation(statement_bytes, signature_bytes, verification_material=None):
     """Return a single PEP-740 Attestation dict (SLSA flavour)."""
     return {
         "version": 1,
-        "verification_material": None,
+        "verification_material": verification_material,
         "envelope": {
             "statement": _b64(statement_bytes),
             "signature": _b64(signature_bytes),
         },
     }
 
 
+def _read_public_key_pem():
+    """Read the test public key as a PEM string."""
+    with open(TEST_PUBLIC_KEY_PATH) as f:
+        return f.read()
+
+
+def _make_embedded_key_material(public_key_pem):
+    """Build a verification_material dict with an embedded public key."""
+    return {
+        "certificate": None,
+        "public_key": public_key_pem,
+        "transparency_entries": [{"log_index": 0}],
+    }
+
+
 def _make_provenance(attestation):
     """Wrap an attestation into a full PEP-740 Provenance object."""
     return {
@@ -245,3 +260,53 @@ def test_slsa_attestation_via_content_upload(
     assert publisher["builder_id"] == "https://konflux-ci.dev/calunga"
     assert publisher["build_type"] == "https://konflux-ci.dev/PythonWheelBuild@v1"
     assert publisher["kind"] == "konflux-ci.dev"
+
+
+def test_slsa_embedded_key_accepted(
+    python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file
+):
+    """An attestation with an embedded public key matching the server key is accepted."""
+    content = python_content_factory()
+
+    stmt = _build_statement(content.filename, content.sha256)
+    sig = _sign(stmt, test_private_key)
+    vm = _make_embedded_key_material(_read_public_key_pem())
+    att = _make_attestation(stmt, sig, verification_material=vm)
+    prov = _make_provenance(att)
+
+    task = python_bindings.ContentProvenanceApi.create(
+        package=content.pulp_href,
+        file=_provenance_file(prov),
+        verify=True,
+    ).task
+    result = monitor_task(task)
+
+    prov_obj = python_bindings.ContentProvenanceApi.read(result.created_resources[-1])
+    assert prov_obj.package == content.pulp_href
+    stored_vm = prov_obj.provenance["attestation_bundles"][0]["attestations"][0][
+        "verification_material"
+    ]
+    assert stored_vm["public_key"] == _read_public_key_pem()
+
+
+def test_slsa_embedded_key_mismatch_rejected(
+    python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file
+):
+    """An attestation with an embedded key that does not match the server key is rejected."""
+    content = python_content_factory()
+
+    stmt = _build_statement(content.filename, content.sha256)
+    sig = _sign(stmt, test_private_key)
+    wrong_key = "-----BEGIN PUBLIC KEY-----\nMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE\n-----END PUBLIC KEY-----\n"
+    vm = _make_embedded_key_material(wrong_key)
+    att = _make_attestation(stmt, sig, verification_material=vm)
+    prov = _make_provenance(att)
+
+    task = python_bindings.ContentProvenanceApi.create(
+        package=content.pulp_href,
+        file=_provenance_file(prov),
+        verify=True,
+    ).task
+    with pytest.raises(PulpTaskError) as exc_info:
+        monitor_task(task)
+    assert "embedded public key" in exc_info.value.task.error["description"]