wip - domain support

Author: jobselko

.github/workflows/scripts/before_install.sh

@@ -50,7 +50,7 @@ legacy_component_name: "pulp_python"
 component_name: "python"
 component_version: "${COMPONENT_VERSION}"
 pulp_env: {}
-pulp_settings: {"allowed_export_paths": "/tmp", "allowed_import_paths": "/tmp", "api_root": "/pulp/", "attestation_verification_key": "/etc/pki/attestation/test-key.pem", "orphan_protection_time": 0, "pypi_api_hostname": "https://pulp:443"}
+pulp_settings: {"allowed_export_paths": "/tmp", "allowed_import_paths": "/tmp", "api_root": "/pulp/", "attestation_verification_keys": {"default": "/etc/pki/attestation/test-key.pem"}, "orphan_protection_time": 0, "pypi_api_hostname": "https://pulp:443"}
 pulp_scheme: "https"
 image:
   name: "pulp"

CHANGES/1340.feature

@@ -1 +1 @@
-Added custom (non-PEP 740) support for verifying attestations signed with a server-configured public key via `ATTESTATION_VERIFICATION_KEY`.
+Added custom (non-PEP 740) support for verifying attestations signed with a per-domain public key configured via `ATTESTATION_VERIFICATION_KEYS`.

docs/user/guides/attestation.md

@@ -100,27 +100,32 @@ Pulp Python supports verifying attestations signed with a custom public key inst
 certificate. This is useful for environments that sign packages with their own key (e.g. internal
 build systems) rather than using Sigstore's keyless signing.
 
-To enable this, configure the `ATTESTATION_VERIFICATION_KEY` setting with the path to a PEM public
-key file:
+To enable this, configure the `ATTESTATION_VERIFICATION_KEYS` setting, which maps a domain name to
+the path of a PEM public key file. Each domain trusts only its own key. Without domains enabled, all
+content belongs to the `default` domain.
 
 ```
-PULP_ATTESTATION_VERIFICATION_KEY=/path/to/public-key.pem
+PULP_ATTESTATION_VERIFICATION_KEYS=@json {"default": "/path/to/public-key.pem"}
 ```
 
-When this setting is configured, Pulp uses two verification paths that coexist on the same server:
+Pulp uses two verification paths that coexist on the same server:
 
 - **Attestations with a Sigstore certificate** are verified through the standard PEP 740 path,
-  regardless of whether `ATTESTATION_VERIFICATION_KEY` is set.
-- **Attestations without a Sigstore certificate** are verified against the server-configured key using RSA
-  PKCS1v15 signatures.
+  regardless of `ATTESTATION_VERIFICATION_KEYS`.
+- **Attestations without a Sigstore certificate** are verified against the key configured for the
+  current domain using RSA PKCS1v15 signatures. In a domain without a configured key, they are
+  rejected.
 
 This means a single Pulp instance can host packages with Sigstore certificates and
 internally-signed packages with key-based attestations. Each attestation is independently
 routed to the appropriate verification path based on whether it contains a certificate.
 Key-signed attestations must omit `verification_material` or set it to `null`.
+Attestations with a certificate that can't be parsed are rejected.
 
 ### Limitations
 
-- A single key is used across all repositories and domains.
+- Each domain can have only one key.
+- Attestations are verified only on upload. Adding existing content to another repository does not
+  verify them again.
 - The key-based path only supports RSA PKCS1v15 with SHA-256 signatures.
 - Publisher enrichment is supported for SLSA v0.2 provenance statements only.

pulp_python/app/provenance.py

@@ -1,11 +1,11 @@
 import json
-import logging
 from typing import Annotated, Literal, Union, get_args
 from urllib.parse import urlparse
 
-from cryptography.exceptions import InvalidSignature
+from cryptography.exceptions import InvalidSignature, UnsupportedAlgorithm
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import padding as crypto_padding
+from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.x509 import load_der_x509_certificate
 from django.conf import settings
 from pydantic import BaseModel, ConfigDict, Field
@@ -21,7 +21,7 @@
 from sigstore.dsse import Envelope as DSSEEnvelope
 from sigstore.dsse import _pae
 
-log = logging.getLogger(__name__)
+from pulpcore.plugin.util import get_domain
 
 _verification_key_cache = {}
 
@@ -92,28 +92,44 @@ class Provenance(BaseModel):
     attestation_bundles: list[AttestationBundle]
 
 
-def _load_verification_key():
-    """Load the configured attestation verification public key, with caching."""
-    key_path = getattr(settings, "ATTESTATION_VERIFICATION_KEY", None)
+def _load_verification_key(domain_name):
+    """Load the attestation verification public key configured for the domain, with caching."""
+    keys = settings.ATTESTATION_VERIFICATION_KEYS
+    if not isinstance(keys, dict):
+        raise VerificationError(
+            "ATTESTATION_VERIFICATION_KEYS must map domain names to public key paths"
+        )
+    key_path = keys.get(domain_name)
     if not key_path:
         return None
     if key_path not in _verification_key_cache:
-        with open(key_path, "rb") as f:
-            _verification_key_cache[key_path] = serialization.load_pem_public_key(f.read())
+        try:
+            with open(key_path, "rb") as f:
+                key = serialization.load_pem_public_key(f.read())
+        except (OSError, ValueError, UnsupportedAlgorithm) as e:
+            raise VerificationError(
+                f"could not load the verification key for domain '{domain_name}' "
+                f"from {key_path}: {e}"
+            )
+        if not isinstance(key, rsa.RSAPublicKey):
+            raise VerificationError(
+                f"the verification key for domain '{domain_name}' in {key_path} "
+                "must be an RSA public key"
+            )
+        _verification_key_cache[key_path] = key
     return _verification_key_cache[key_path]
 
 
-def _has_valid_certificate(attestation):
-    """Check whether the attestation contains a valid X.509 certificate."""
-    try:
-        vm = attestation.verification_material
-        if vm is None:
-            return False
-        cert_bytes = vm.certificate
-        load_der_x509_certificate(cert_bytes)
-        return True
-    except (ValueError, Exception):
+def _has_certificate(attestation):
+    """Check whether the attestation carries a certificate, rejecting unparseable ones."""
+    vm = attestation.verification_material
+    if vm is None:
         return False
+    try:
+        load_der_x509_certificate(vm.certificate)
+    except ValueError as e:
+        raise VerificationError(f"invalid X.509 certificate in verification_material: {e}")
+    return True
 
 
 def _verify_statement_subject(attestation, dist):
@@ -185,28 +201,30 @@ def verify_provenance(filename, sha256, provenance, offline=True):
 
     Each attestation is routed to one of two coexisting verification paths:
     1. Attestations with a valid Sigstore certificate follow the standard PEP 740 path.
-    2. Attestations without a certificate are verified against the server-configured
-    ATTESTATION_VERIFICATION_KEY using RSA PKCS1v15 signatures, with support for SLSA
-    v0.2 publisher enrichment.
+    2. Attestations without a certificate are verified against the key configured for the
+    current domain in ATTESTATION_VERIFICATION_KEYS using RSA PKCS1v15 signatures, with
+    support for SLSA v0.2 publisher enrichment.
+    Attestations with a certificate that can't be parsed are rejected.
     """
     dist = Distribution(name=filename, digest=sha256)
     for bundle in provenance.attestation_bundles:
         publisher = bundle.publisher
         policy = publisher._as_policy()
         for attestation in bundle.attestations:
-            if _has_valid_certificate(attestation):
+            if _has_certificate(attestation):
                 sig_bundle = attestation.to_bundle()
                 checkpoint = sig_bundle.log_entry._inner.inclusion_proof.checkpoint
                 staging = "sigstage.dev" in checkpoint.envelope
                 attestation.verify(policy, dist, staging=staging, offline=offline)
             else:
-                verification_key = _load_verification_key()
+                domain_name = get_domain().name
+                verification_key = _load_verification_key(domain_name)
                 if verification_key:
                     stmt = _verify_statement_subject(attestation, dist)
                     _enrich_publisher_from_statement(stmt, publisher)
                     _verify_signature(attestation, verification_key)
                 else:
                     raise VerificationError(
-                        "Attestation has no Sigstore certificate and no custom "
-                        "verification key configured via ATTESTATION_VERIFICATION_KEY"
+                        "Attestation has no Sigstore certificate, and "
+                        f"ATTESTATION_VERIFICATION_KEYS has no key for domain '{domain_name}'"
                     )

pulp_python/app/settings.py

@@ -4,7 +4,7 @@
 PYPI_API_HOSTNAME = "https://" + socket.getfqdn()
 PYPI_PATH_PREFIX = "/pypi/"
 
-ATTESTATION_VERIFICATION_KEY = None
+ATTESTATION_VERIFICATION_KEYS = {}
 
 DRF_ACCESS_POLICY = {
     "dynaconf_merge_unique": True,

pulp_python/tests/functional/api/test_key_signed_attestations.py

@@ -4,20 +4,24 @@
 attestations that carry an RSA signature instead of a Sigstore certificate.
 
 A static test keypair is shipped in .ci/assets/keys/ and the matching
-public key is configured as PULP_ATTESTATION_VERIFICATION_KEY so that
+public key is configured for the default domain in PULP_ATTESTATION_VERIFICATION_KEYS so that
 signature verification is fully exercised end-to-end.
 """
 
 import base64
+import hashlib
 import json
 import os
 
 import pytest
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import padding as crypto_padding
+from cryptography.hazmat.primitives.asymmetric import rsa
 
 from pulpcore.tests.functional.utils import PulpTaskError
 
+from pulp_python.tests.functional.constants import PYTHON_EGG_FILENAME, PYTHON_URL
+
 DSSE_PAYLOAD_TYPE = "application/vnd.in-toto+json"
 
 TEST_PRIVATE_KEY_PATH = "/etc/pki/attestation/test-key-private.pem"
@@ -98,13 +102,15 @@ def _make_provenance(attestation):
 
 @pytest.fixture(scope="session")
 def test_private_key(pulp_settings):
-    """Load the CI-generated test attestation signing key."""
+    """Load the static test attestation signing key from .ci/assets/keys/."""
     if not os.path.exists(TEST_PRIVATE_KEY_PATH):
         pytest.skip(
             f"Test attestation private key not found at {TEST_PRIVATE_KEY_PATH} (not running in CI container?)"
         )
-    if pulp_settings.ATTESTATION_VERIFICATION_KEY != TEST_PUBLIC_KEY_PATH:
-        pytest.fail(f"ATTESTATION_VERIFICATION_KEY is not set to {TEST_PUBLIC_KEY_PATH}")
+    if pulp_settings.ATTESTATION_VERIFICATION_KEYS.get("default") != TEST_PUBLIC_KEY_PATH:
+        pytest.fail(
+            f"ATTESTATION_VERIFICATION_KEYS does not map the default domain to {TEST_PUBLIC_KEY_PATH}"
+        )
     with open(TEST_PRIVATE_KEY_PATH, "rb") as f:
         return serialization.load_pem_private_key(f.read(), password=None)
 
@@ -244,3 +250,71 @@ def test_key_signed_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_invalid_certificate_not_verified_with_key(
+    python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file
+):
+    """An attestation with an unparseable certificate is rejected, not verified with the key."""
+    content = python_content_factory()
+
+    stmt = _build_statement(content.filename, content.sha256)
+    att = _make_attestation(stmt, _sign(stmt, test_private_key))
+    att["verification_material"] = {
+        "certificate": _b64(b"not-a-certificate"),
+        "transparency_entries": [{"log_index": 0}],
+    }
+    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 "invalid X.509 certificate" in exc_info.value.task.error["description"]
+
+
+def test_key_signed_attestation_with_other_key_rejected(
+    python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file
+):
+    """An attestation signed with a key other than the domain's configured key is rejected."""
+    content = python_content_factory()
+    other_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+
+    stmt = _build_statement(content.filename, content.sha256)
+    att = _make_attestation(stmt, _sign(stmt, other_key))
+    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 "signature verification failed" in exc_info.value.task.error["description"]
+
+
+def test_key_signed_attestation_rejected_in_domain_without_key(
+    python_bindings, domain_factory, download_python_file, monitor_task, test_private_key
+):
+    """A key configured for the default domain is not trusted in another domain."""
+    domain = domain_factory()
+    python_file = download_python_file(PYTHON_EGG_FILENAME, PYTHON_URL)
+    with open(python_file, "rb") as f:
+        sha256 = hashlib.sha256(f.read()).hexdigest()
+
+    stmt = _build_statement(PYTHON_EGG_FILENAME, sha256)
+    att = _make_attestation(stmt, _sign(stmt, test_private_key))
+    body = {
+        "file": python_file,
+        "relative_path": PYTHON_EGG_FILENAME,
+        "attestations": json.dumps([att]),
+    }
+    task = python_bindings.ContentPackagesApi.create(**body, pulp_domain=domain.name).task
+    with pytest.raises(PulpTaskError) as exc_info:
+        monitor_task(task)
+    err_msg = f"ATTESTATION_VERIFICATION_KEYS has no key for domain '{domain.name}'"
+    assert err_msg in exc_info.value.task.error["description"]

template_config.yml

@@ -47,7 +47,8 @@ pulp_settings:
   allowed_import_paths: "/tmp"
   api_root: "/pulp/"
   orphan_protection_time: 0
-  attestation_verification_key: "/etc/pki/attestation/test-key.pem"
+  attestation_verification_keys:
+    default: "/etc/pki/attestation/test-key.pem"
   pypi_api_hostname: "https://pulp:443"
 pulp_settings_azure:
   MEDIA_ROOT: ""