legacy: split attestation handling phases (#17067)

* legacy: split attestation handling phases

Only persist attestations once the file is created.

Signed-off-by: William Woodruff <william@trailofbits.com>

* round out coverage

Signed-off-by: William Woodruff <william@trailofbits.com>

* cleanup, remove defunct test

Signed-off-by: William Woodruff <william@trailofbits.com>

---------

Signed-off-by: William Woodruff <william@trailofbits.com>
Co-authored-by: Dustin Ingram <di@users.noreply.github.com>

Author: woodruffw

tests/unit/forklift/test_legacy.py

@@ -1232,6 +1232,7 @@ def storage_service_store(path, file_path, *, meta):
         assert resp.status_code == 200
         assert db_request.find_service.calls == [
             pretend.call(IMetricsService, context=None),
+            pretend.call(IIntegrityService, context=None),
             pretend.call(IFileStorage, name="archive"),
         ]
         assert len(storage_service.store.calls) == 1
@@ -2796,6 +2797,7 @@ def storage_service_store(path, file_path, *, meta):
         assert resp.status_code == 200
         assert db_request.find_service.calls == [
             pretend.call(IMetricsService, context=None),
+            pretend.call(IIntegrityService, context=None),
             pretend.call(IFileStorage, name="archive"),
         ]
         assert storage_service.store.calls == [
@@ -3159,6 +3161,7 @@ def storage_service_store(path, file_path, *, meta):
         assert resp.status_code == 200
         assert db_request.find_service.calls == [
             pretend.call(IMetricsService, context=None),
+            pretend.call(IIntegrityService, context=None),
             pretend.call(IFileStorage, name="archive"),
         ]
         assert storage_service.store.calls == [

warehouse/forklift/legacy.py

@@ -1082,7 +1082,6 @@ def file_upload(request):
         # TODO: Remove sdist zip handling when #12245 is resolved
         # (PEP 625 – Filename of a Source Distribution)
         if form.filetype.data == "sdist" and filename.endswith(".zip"):
-
             # PEP 625: Enforcement on filename extensions. Files ending with
             # .zip will not be permitted.
             send_pep625_extension_email(
@@ -1307,6 +1306,27 @@ def file_upload(request):
                 k: h.hexdigest().lower() for k, h in metadata_file_hashes.items()
             }
 
+        # If the user provided attestations, verify them
+        # We persist these attestations subsequently, only after the
+        # release file is persisted.
+        integrity_service: IIntegrityService = request.find_service(
+            IIntegrityService, context=None
+        )
+        attestations: list[Attestation] = []
+        if "attestations" in request.POST and not request.flags.enabled(
+            AdminFlagValue.DISABLE_PEP740
+        ):
+            try:
+                attestations = integrity_service.parse_attestations(
+                    request,
+                    Distribution(name=filename, digest=file_hashes["sha256"]),
+                )
+            except AttestationUploadError as e:
+                raise _exc_with_message(
+                    HTTPBadRequest,
+                    f"Invalid attestations supplied during upload: {e}",
+                )
+
         # TODO: This should be handled by some sort of database trigger or a
         #       SQLAlchemy hook or the like instead of doing it inline in this
         #       view.
@@ -1375,28 +1395,12 @@ def file_upload(request):
             )
         )
 
-        # If the user provided attestations, verify and store them
-        if "attestations" in request.POST and not request.flags.enabled(
-            AdminFlagValue.DISABLE_PEP740
-        ):
-            integrity_service: IIntegrityService = request.find_service(
-                IIntegrityService, context=None
+        # If we have attestations from above, persist them.
+        if attestations:
+            request.db.add(
+                integrity_service.build_provenance(request, file_, attestations)
             )
 
-            try:
-                attestations: list[Attestation] = integrity_service.parse_attestations(
-                    request,
-                    Distribution(name=filename, digest=file_hashes["sha256"]),
-                )
-                request.db.add(
-                    integrity_service.build_provenance(request, file_, attestations)
-                )
-            except AttestationUploadError as e:
-                raise _exc_with_message(
-                    HTTPBadRequest,
-                    f"Invalid attestations supplied during upload: {e}",
-                )
-
             # Log successful attestation upload
             metrics.increment("warehouse.upload.attestations.ok")