upload: add attestations to PackageFile

tests/test_package.py

@@ -11,6 +11,7 @@
 # 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 json
 import string
 
 import pretend
@@ -114,6 +115,40 @@ def test_package_signed_name_is_correct():
     assert package.signed_filename == (filename + ".asc")
 
 
+def test_package_add_attestations(tmp_path):
+    package = package_file.PackageFile.from_filename(helpers.WHEEL_FIXTURE, None)
+
+    assert package.attestations is None
+
+    attestations = []
+    for i in range(3):
+        path = tmp_path / f"fake.{i}.attestation"
+        path.write_text(json.dumps({"fake": f"attestation {i}"}))
+        attestations.append(str(path))
+
+    package.add_attestations(attestations)
+
+    assert package.attestations == [
+        {"fake": "attestation 0"},
+        {"fake": "attestation 1"},
+        {"fake": "attestation 2"},
+    ]
+
+
+def test_package_add_attestations_invalid_json(tmp_path):
+    package = package_file.PackageFile.from_filename(helpers.WHEEL_FIXTURE, None)
+
+    assert package.attestations is None
+
+    attestation = tmp_path / "fake.publish.attestation"
+    attestation.write_text("this is not valid JSON")
+
+    with pytest.raises(
+        exceptions.InvalidDistribution, match="invalid JSON in attestation"
+    ):
+        package.add_attestations([attestation])
+
+
 @pytest.mark.parametrize(
     "pkg_name,expected_name",
     [
@@ -185,7 +220,8 @@ def test_metadata_dictionary_keys():
 
 
 @pytest.mark.parametrize("gpg_signature", [(None), (pretend.stub())])
-def test_metadata_dictionary_values(gpg_signature):
+@pytest.mark.parametrize("attestation", [(None), ({"fake": "attestation"})])
+def test_metadata_dictionary_values(gpg_signature, attestation):
     """Pass values from pkginfo.Distribution through to dictionary."""
     meta = pretend.stub(
         name="whatever",
@@ -226,6 +262,8 @@ def test_metadata_dictionary_values(gpg_signature):
         filetype=pretend.stub(),
     )
     package.gpg_signature = gpg_signature
+    if attestation:
+        package.attestations = [attestation]
 
     result = package.metadata_dictionary()
 
@@ -277,6 +315,12 @@ def test_metadata_dictionary_values(gpg_signature):
     # GPG signature
     assert result.get("gpg_signature") == gpg_signature
 
+    # Attestations
+    if attestation:
+        assert result["attestations"] == json.dumps(package.attestations)
+    else:
+        assert "attestations" not in result
+
 
 TWINE_1_5_0_WHEEL_HEXDIGEST = package_file.Hexdigest(
     "1919f967e990bee7413e2a4bc35fd5d1",

tests/test_upload.py

@@ -69,10 +69,11 @@ def test_make_package_pre_signed_dist(upload_settings, caplog):
     upload_settings.sign = True
     upload_settings.verbose = True
 
-    package = upload._make_package(filename, signatures, upload_settings)
+    package = upload._make_package(filename, signatures, [], upload_settings)
 
     assert package.filename == filename
     assert package.gpg_signature is not None
+    assert package.attestations is None
 
     assert caplog.messages == [
         f"{filename} ({expected_size})",
@@ -94,7 +95,7 @@ def stub_sign(package, *_):
 
     monkeypatch.setattr(package_file.PackageFile, "sign", stub_sign)
 
-    package = upload._make_package(filename, signatures, upload_settings)
+    package = upload._make_package(filename, signatures, [], upload_settings)
 
     assert package.filename == filename
     assert package.gpg_signature is not None
@@ -105,6 +106,16 @@ def stub_sign(package, *_):
     ]
 
 
+def test_make_package_attestations_flagged_but_missing(upload_settings):
+    """Fail when the user requests attestations but does not supply any attestations."""
+    upload_settings.attestations = True
+
+    with pytest.raises(
+        exceptions.InvalidDistribution, match="Upload with attestations requested"
+    ):
+        upload._make_package(helpers.NEW_WHEEL_FIXTURE, {}, [], upload_settings)
+
+
 def test_split_inputs():
     """Split inputs into dists, signatures, and attestations."""
     inputs = [

twine/commands/upload.py

@@ -73,9 +73,16 @@ def skip_upload(
 
 
 def _make_package(
-    filename: str, signatures: Dict[str, str], upload_settings: settings.Settings
+    filename: str,
+    signatures: Dict[str, str],
+    attestations: List[str],
+    upload_settings: settings.Settings,
 ) -> package_file.PackageFile:
-    """Create and sign a package, based off of filename, signatures and settings."""
+    """Create and sign a package, based off of filename, signatures, and settings.
+
+    Additionally, any supplied attestations are attached to the package when
+    the settings indicate to do so.
+    """
     package = package_file.PackageFile.from_filename(filename, upload_settings.comment)
 
     signed_name = package.signed_basefilename
@@ -84,6 +91,17 @@ def _make_package(
     elif upload_settings.sign:
         package.sign(upload_settings.sign_with, upload_settings.identity)
 
+    # Attestations are only attached if explicitly requested with `--attestations`.
+    if upload_settings.attestations:
+        # Passing `--attestations` without any actual attestations present
+        # indicates user confusion, so we fail rather than silently allowing it.
+        if not attestations:
+            raise exceptions.InvalidDistribution(
+                "Upload with attestations requested, but "
+                f"{filename} has no associated attestations"
+            )
+        package.add_attestations(attestations)
+
     file_size = utils.get_file_size(package.filename)
     logger.info(f"{package.filename} ({file_size})")
     if package.gpg_signature:
@@ -154,14 +172,17 @@ def upload(upload_settings: settings.Settings, dists: List[str]) -> None:
     """
     dists = commands._find_dists(dists)
     # Determine if the user has passed in pre-signed distributions or any attestations.
-    uploads, signatures, _ = _split_inputs(dists)
+    uploads, signatures, attestations_by_dist = _split_inputs(dists)
 
     upload_settings.check_repository_url()
     repository_url = cast(str, upload_settings.repository_config["repository"])
     print(f"Uploading distributions to {repository_url}")
 
     packages_to_upload = [
-        _make_package(filename, signatures, upload_settings) for filename in uploads
+        _make_package(
+            filename, signatures, attestations_by_dist[filename], upload_settings
+        )
+        for filename in uploads
     ]
 
     if any(p.gpg_signature for p in packages_to_upload):

twine/package.py

@@ -13,11 +13,12 @@
 # limitations under the License.
 import hashlib
 import io
+import json
 import logging
 import os
 import re
 import subprocess
-from typing import Dict, NamedTuple, Optional, Sequence, Tuple, Union, cast
+from typing import Any, Dict, List, NamedTuple, Optional, Sequence, Tuple, Union, cast
 
 import importlib_metadata
 import pkginfo
@@ -78,6 +79,7 @@ def __init__(
         self.signed_filename = self.filename + ".asc"
         self.signed_basefilename = self.basefilename + ".asc"
         self.gpg_signature: Optional[Tuple[str, bytes]] = None
+        self.attestations: Optional[List[Dict[Any, str]]] = None
 
         hasher = HashManager(filename)
         hasher.hash()
@@ -186,6 +188,9 @@ def metadata_dictionary(self) -> Dict[str, MetadataValue]:
         if self.gpg_signature is not None:
             data["gpg_signature"] = self.gpg_signature
 
+        if self.attestations is not None:
+            data["attestations"] = json.dumps(self.attestations)
+
         # FIPS disables MD5 and Blake2, making the digest values None. Some package
         # repositories don't allow null values, so this only sends non-null values.
         # See also: https://github.com/pypa/twine/issues/775
@@ -197,6 +202,19 @@ def metadata_dictionary(self) -> Dict[str, MetadataValue]:
 
         return data
 
+    def add_attestations(self, attestations: List[str]) -> None:
+        loaded_attestations = []
+        for attestation in attestations:
+            with open(attestation, "rb") as att:
+                try:
+                    loaded_attestations.append(json.load(att))
+                except json.JSONDecodeError:
+                    raise exceptions.InvalidDistribution(
+                        f"invalid JSON in attestation: {attestation}"
+                    )
+
+        self.attestations = loaded_attestations
+
     def add_gpg_signature(
         self, signature_filepath: str, signature_filename: str
     ) -> None:

twine/repository.py

@@ -25,7 +25,7 @@
 import twine
 from twine import package as package_file
 
-KEYWORDS_TO_NOT_FLATTEN = {"gpg_signature", "content"}
+KEYWORDS_TO_NOT_FLATTEN = {"gpg_signature", "attestations", "content"}
 
 LEGACY_PYPI = "https://pypi.python.org/"
 LEGACY_TEST_PYPI = "https://testpypi.python.org/"