Store attestations for PEP740

tests/common/db/attestation.py

@@ -0,0 +1,28 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import hashlib
+
+import factory
+
+from warehouse.attestations.models import ReleaseFileAttestation
+
+from .base import WarehouseFactory
+
+
+class ReleaseAttestationsFactory(WarehouseFactory):
+    class Meta:
+        model = ReleaseFileAttestation
+
+    file = factory.SubFactory("tests.common.db.packaging.FileFactory")
+    attestation_file_sha256_digest = factory.LazyAttribute(
+        lambda o: hashlib.sha256(o.file.filename.encode("utf8")).hexdigest()
+    )

tests/common/db/packaging.py

@@ -34,6 +34,7 @@
 from warehouse.utils import readme
 
 from .accounts import UserFactory
+from .attestation import ReleaseAttestationsFactory
 from .base import WarehouseFactory
 from .observations import ObserverFactory
 
@@ -140,6 +141,12 @@ class Meta:
         )
     )
 
+    attestations = factory.RelatedFactoryList(
+        ReleaseAttestationsFactory,
+        factory_related_name="file",
+        size=1,
+    )
+
 
 class FileEventFactory(WarehouseFactory):
     class Meta:

tests/unit/attestations/__init__.py

@@ -0,0 +1,11 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

tests/unit/attestations/test_core.py

@@ -0,0 +1,107 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import pathlib
+
+from pathlib import Path
+
+import pretend
+
+from pypi_attestations import Attestation, Envelope, VerificationMaterial
+
+import warehouse.packaging
+
+from tests.common.db.packaging import FileFactory
+from warehouse.attestations._core import generate_provenance_file, get_provenance_digest
+from warehouse.events.tags import EventTag
+from warehouse.packaging import ISimpleStorage
+
+from ...common.db.packaging import FileEventFactory
+
+
+def test_get_provenance_digest_succeed(db_request, monkeypatch):
+    file = FileFactory.create()
+    FileEventFactory.create(
+        source=file,
+        tag=EventTag.Project.ReleaseAdd,
+        additional={"publisher_url": "fake-publisher-url"},
+    )
+
+    generate_provenance_file = pretend.call_recorder(
+        lambda request, publisher_url, file_: (Path("fake-path"), "deadbeef")
+    )
+    monkeypatch.setattr(
+        warehouse.attestations._core,
+        "generate_provenance_file",
+        generate_provenance_file,
+    )
+
+    hex_digest = get_provenance_digest(db_request, file)
+
+    assert hex_digest == "deadbeef"
+
+
+def test_get_provenance_digest_fails_no_attestations(db_request, monkeypatch):
+    file = FileFactory.create()
+    monkeypatch.setattr(warehouse.packaging.models.File, "attestations", [])
+
+    provenance_hash = get_provenance_digest(db_request, file)
+    assert provenance_hash is None
+
+
+def test_get_provenance_digest_fails_no_publisher_url(db_request, monkeypatch):
+    file = FileFactory.create()
+
+    provenance_hash = get_provenance_digest(db_request, file)
+    assert provenance_hash is None
+
+
+def test_generate_provenance_file_succeed(db_request, monkeypatch):
+
+    def store_function(path, file_path, *, meta=None):
+        return f"https://files/attestations/{path}.provenance"
+
+    storage_service = pretend.stub(store=pretend.call_recorder(store_function))
+
+    db_request.find_service = pretend.call_recorder(
+        lambda svc, name=None, context=None: {
+            ISimpleStorage: storage_service,
+        }.get(svc)
+    )
+
+    publisher_url = "x-fake-publisher-url"
+    file = FileFactory.create()
+    FileEventFactory.create(
+        source=file,
+        tag=EventTag.Project.ReleaseAdd,
+        additional={"publisher_url": publisher_url},
+    )
+
+    attestation = Attestation(
+        version=1,
+        verification_material=VerificationMaterial(
+            certificate="somebase64string", transparency_entries=[dict()]
+        ),
+        envelope=Envelope(
+            statement="somebase64string",
+            signature="somebase64string",
+        ),
+    )
+
+    read_text = pretend.call_recorder(lambda _: attestation.model_dump_json())
+
+    monkeypatch.setattr(pathlib.Path, "read_text", read_text)
+
+    provenance_file_path, provenance_hash = generate_provenance_file(
+        db_request, publisher_url, file
+    )
+
+    assert provenance_hash is not None

tests/unit/forklift/test_legacy.py

@@ -58,6 +58,7 @@
     Project,
     ProjectMacaroonWarningAssociation,
     Release,
+    ReleaseFileAttestation,
     Role,
 )
 from warehouse.packaging.tasks import sync_file_to_cache, update_bigquery_release_files
@@ -3808,6 +3809,85 @@ def failing_verify(_self, _verifier, _policy, _dist):
         assert resp.status_code == 400
         assert resp.status.startswith(expected_msg)
 
+    def test_upload_succeeds_upload_attestation(
+        self,
+        monkeypatch,
+        pyramid_config,
+        db_request,
+        metrics,
+    ):
+
+        project = ProjectFactory.create()
+        version = "1.0"
+        publisher = GitHubPublisherFactory.create(projects=[project])
+        claims = {
+            "sha": "somesha",
+            "repository": f"{publisher.repository_owner}/{publisher.repository_name}",
+            "workflow": "workflow_name",
+        }
+        identity = PublisherTokenContext(publisher, SignedClaims(claims))
+        db_request.oidc_publisher = identity.publisher
+        db_request.oidc_claims = identity.claims
+
+        db_request.db.add(Classifier(classifier="Environment :: Other Environment"))
+        db_request.db.add(Classifier(classifier="Programming Language :: Python"))
+
+        filename = "{}-{}.tar.gz".format(project.name, "1.0")
+        attestation = Attestation(
+            version=1,
+            verification_material=VerificationMaterial(
+                certificate="somebase64string", transparency_entries=[dict()]
+            ),
+            envelope=Envelope(
+                statement="somebase64string",
+                signature="somebase64string",
+            ),
+        )
+
+        pyramid_config.testing_securitypolicy(identity=identity)
+        db_request.user = None
+        db_request.user_agent = "warehouse-tests/6.6.6"
+        db_request.POST = MultiDict(
+            {
+                "metadata_version": "1.2",
+                "name": project.name,
+                "attestations": f"[{attestation.model_dump_json()}]",
+                "version": version,
+                "summary": "This is my summary!",
+                "filetype": "sdist",
+                "md5_digest": _TAR_GZ_PKG_MD5,
+                "content": pretend.stub(
+                    filename=filename,
+                    file=io.BytesIO(_TAR_GZ_PKG_TESTDATA),
+                    type="application/tar",
+                ),
+            }
+        )
+
+        storage_service = pretend.stub(store=lambda path, filepath, meta: None)
+        db_request.find_service = lambda svc, name=None, context=None: {
+            IFileStorage: storage_service,
+            IMetricsService: metrics,
+        }.get(svc)
+
+        process_attestations = pretend.call_recorder(
+            lambda request, distribution: [attestation]
+        )
+
+        monkeypatch.setattr(legacy, "_process_attestations", process_attestations)
+
+        resp = legacy.file_upload(db_request)
+
+        assert resp.status_code == 200
+
+        attestations_db = (
+            db_request.db.query(ReleaseFileAttestation)
+            .join(ReleaseFileAttestation.file)
+            .filter(File.filename == filename)
+            .all()
+        )
+        assert len(attestations_db) == 1
+
     @pytest.mark.parametrize(
         "version, expected_version",
         [

warehouse/attestations/__init__.py

@@ -0,0 +1,11 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

warehouse/attestations/_core.py

@@ -0,0 +1,105 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import hashlib
+import tempfile
+
+from pathlib import Path
+from typing import Literal
+
+from pydantic import BaseModel, TypeAdapter
+from pypi_attestations import Attestation
+
+from warehouse.packaging import File, ISimpleStorage
+
+
+class Publisher(BaseModel):
+    kind: str
+    """
+    The kind of Trusted Publisher.
+    """
+
+    claims: object
+    """
+    Claims specified by the publisher.
+    """
+
+
+class AttestationBundle(BaseModel):
+    publisher: Publisher
+    """
+    The publisher associated with this set of attestations.
+    """
+
+    attestations: list[Attestation]
+    """
+    The list of attestations included in this bundle.
+    """
+
+
+class Provenance(BaseModel):
+    version: Literal[1]
+    """
+    The provenance object's version, which is always 1.
+    """
+
+    attestation_bundles: list[AttestationBundle]
+    """
+    One or more attestation "bundles".
+    """
+
+
+def get_provenance_digest(request, file: File) -> str | None:
+    if not file.attestations:
+        return None
+
+    publisher_url = file.publisher_url
+    if not publisher_url:
+        return None  # TODO(dm)
+
+    provenance_file_path, provenance_digest = generate_provenance_file(
+        request, publisher_url, file
+    )
+    return provenance_digest
+
+
+def generate_provenance_file(
+    request, publisher_url: str, file: File
+) -> tuple[Path, str]:
+
+    storage = request.find_service(ISimpleStorage)
+    publisher = Publisher(kind=publisher_url, claims={})  # TODO(dm)
+
+    attestation_bundle = AttestationBundle(
+        publisher=publisher,
+        attestations=[
+            TypeAdapter(Attestation).validate_json(
+                Path(release_attestation.attestation_path).read_text()
+            )
+            for release_attestation in file.attestations
+        ],
+    )
+
+    provenance = Provenance(version=1, attestation_bundles=[attestation_bundle])
+
+    provenance_file_path = Path(f"{file.path}.provenance")
+    with tempfile.NamedTemporaryFile() as f:
+        f.write(provenance.model_dump_json().encode("utf-8"))
+        f.flush()
+
+        storage.store(
+            provenance_file_path,
+            f.name,
+        )
+
+        file_digest = hashlib.file_digest(f, "sha256")
+
+    return provenance_file_path, file_digest.hexdigest()