Allow multiple attestations per distribution (#17134)

* attestations: allow multiple attestations per dist

Signed-off-by: Facundo Tuesca <[email protected]>

* docs: document multiple attestations per file upload

Signed-off-by: Facundo Tuesca <[email protected]>

* attestations: improve error messages

Signed-off-by: Facundo Tuesca <[email protected]>

---------

Signed-off-by: Facundo Tuesca <[email protected]>

Author: facutuesca

docs/user/attestations/index.md

@@ -32,6 +32,11 @@ Currently, PyPI allows the following attestation predicates:
 * [SLSA Provenance]
 * [PyPI Publish]
 
+Each file can be uploaded along its attestations. Currently PyPI supports two
+attestations per file: one for each of the allowed predicates. Uploads with more
+than two attestations per file, or with attestations with repeated predicates will
+be rejected.
+
 [in-toto Attestation Framework]: https://github.com/in-toto/attestation/blob/main/spec/README.md
 
 [PEP 740]: https://peps.python.org/pep-0740/

tests/unit/attestations/test_services.py

@@ -10,6 +10,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import json
+import re
 
 import pretend
 import pytest
@@ -132,28 +133,74 @@ def test_parse_attestations_fails_malformed_attestation(self, metrics, db_reques
             in metrics.increment.calls
         )
 
-    def test_parse_attestations_fails_multiple_attestations(
+    def test_parse_attestations_fails_multiple_attestations_exceeds_limit(
         self, metrics, db_request, dummy_attestation
     ):
         integrity_service = services.IntegrityService(
             metrics=metrics,
             session=db_request.db,
         )
 
+        max_attestations = len(services.SUPPORTED_ATTESTATION_TYPES)
+
         db_request.oidc_publisher = pretend.stub(attestation_identity=pretend.stub())
         db_request.POST["attestations"] = TypeAdapter(list[Attestation]).dump_json(
-            [dummy_attestation, dummy_attestation]
+            [dummy_attestation] * (max_attestations + 1)
         )
         with pytest.raises(
-            AttestationUploadError, match="Only a single attestation per file"
+            AttestationUploadError,
+            match=f"A maximum of {max_attestations} attestations per file are "
+            f"supported",
         ):
             integrity_service.parse_attestations(
                 db_request,
                 pretend.stub(),
             )
 
         assert (
-            pretend.call("warehouse.upload.attestations.failed_multiple_attestations")
+            pretend.call(
+                "warehouse.upload.attestations.failed_limit_multiple_attestations"
+            )
+            in metrics.increment.calls
+        )
+
+    def test_parse_attestations_fails_multiple_attestations_same_predicate(
+        self, metrics, monkeypatch, db_request, dummy_attestation
+    ):
+        integrity_service = services.IntegrityService(
+            metrics=metrics,
+            session=db_request.db,
+        )
+        max_attestations = len(services.SUPPORTED_ATTESTATION_TYPES)
+        db_request.oidc_publisher = pretend.stub(
+            attestation_identity=pretend.stub(),
+        )
+        db_request.oidc_claims = {"sha": "somesha"}
+        db_request.POST["attestations"] = TypeAdapter(list[Attestation]).dump_json(
+            [dummy_attestation] * max_attestations
+        )
+
+        monkeypatch.setattr(Verifier, "production", lambda: pretend.stub())
+        monkeypatch.setattr(
+            Attestation, "verify", lambda *args: (AttestationType.PYPI_PUBLISH_V1, {})
+        )
+
+        with pytest.raises(
+            AttestationUploadError,
+            match=re.escape(
+                "Multiple attestations for the same file with the same predicate "
+                "type (https://docs.pypi.org/attestations/publish/v1) are not supported"
+            ),
+        ):
+            integrity_service.parse_attestations(
+                db_request,
+                pretend.stub(),
+            )
+
+        assert (
+            pretend.call(
+                "warehouse.upload.attestations.failed_duplicate_predicate_type"
+            )
             in metrics.increment.calls
         )
 

warehouse/attestations/services.py

@@ -81,13 +81,14 @@ def _extract_attestations_from_request(request: Request) -> list[Attestation]:
             "Malformed attestations: an empty attestation set is not permitted"
         )
 
-    # This is a temporary constraint; multiple attestations per file will
-    # be supported in the future.
-    if len(attestations) > 1:
-        metrics.increment("warehouse.upload.attestations.failed_multiple_attestations")
-
+    # We currently allow at most one attestation per predicate type
+    max_attestations = len(SUPPORTED_ATTESTATION_TYPES)
+    if len(attestations) > max_attestations:
+        metrics.increment(
+            "warehouse.upload.attestations.failed_limit_multiple_attestations"
+        )
         raise AttestationUploadError(
-            "Only a single attestation per file is supported",
+            f"A maximum of {max_attestations} attestations per file are supported",
         )
 
     return attestations
@@ -158,9 +159,11 @@ def parse_attestations(
         # Sanity-checked above.
         expected_identity = request.oidc_publisher.attestation_identity
 
+        seen_predicate_types: set[AttestationType] = set()
+
         for attestation_model in attestations:
             try:
-                predicate_type, _ = attestation_model.verify(
+                predicate_type_str, _ = attestation_model.verify(
                     expected_identity,
                     distribution,
                 )
@@ -182,13 +185,26 @@ def parse_attestations(
                     f"Unknown error while trying to verify included attestations: {e}",
                 )
 
-            if predicate_type not in SUPPORTED_ATTESTATION_TYPES:
+            if predicate_type_str not in SUPPORTED_ATTESTATION_TYPES:
                 self.metrics.increment(
                     "warehouse.upload.attestations.failed_unsupported_predicate_type"
                 )
                 raise AttestationUploadError(
-                    f"Attestation with unsupported predicate type: {predicate_type}",
+                    f"Attestation with unsupported predicate type: "
+                    f"{predicate_type_str}",
                 )
+            predicate_type = AttestationType(predicate_type_str)
+
+            if predicate_type in seen_predicate_types:
+                self.metrics.increment(
+                    "warehouse.upload.attestations.failed_duplicate_predicate_type"
+                )
+                raise AttestationUploadError(
+                    f"Multiple attestations for the same file with the same "
+                    f"predicate type ({predicate_type.value}) are not supported",
+                )
+
+            seen_predicate_types.add(predicate_type)
 
         return attestations