use new methods for building requests

Signed-off-by: Ramon Petgrave <[email protected]>

Author: ramonpetgrave64

sigstore/_internal/rekor/client.py

@@ -18,6 +18,7 @@
 
 from __future__ import annotations
 
+import base64
 import json
 import logging
 from abc import ABC
@@ -26,8 +27,15 @@
 
 import rekor_types
 import requests
+from cryptography.hazmat.primitives import serialization
+from cryptography.x509 import Certificate
 
 from sigstore._internal import USER_AGENT
+from sigstore._internal.rekor import (
+    RekorLogSubmitter,
+)
+from sigstore.dsse import Envelope
+from sigstore.hashes import Hashed
 from sigstore.models import LogEntry
 
 _logger = logging.getLogger(__name__)
@@ -216,7 +224,7 @@ def post(
         return oldest_entry
 
 
-class RekorClient:
+class RekorClient(RekorLogSubmitter):
     """The internal Rekor client"""
 
     def __init__(self, url: str) -> None:
@@ -261,3 +269,63 @@ def log(self) -> RekorLog:
         Returns a `RekorLog` adapter for making requests to a Rekor log.
         """
         return RekorLog(f"{self.url}/log", session=self.session)
+
+    def create_entry(  # type: ignore[override]
+        self, request: rekor_types.Hashedrekord | rekor_types.Dsse
+    ) -> LogEntry:
+        """
+        Submit the request to Rekor.
+        """
+        return self.log.entries.post(request)
+
+    def _build_hashed_rekord_request(  # type: ignore[override]
+        self, hashed_input: Hashed, signature: bytes, certificate: Certificate
+    ) -> rekor_types.Hashedrekord:
+        """
+        Construct a hashed rekord request to submit to Rekor.
+        """
+        return rekor_types.Hashedrekord(
+            spec=rekor_types.hashedrekord.HashedrekordV001Schema(
+                signature=rekor_types.hashedrekord.Signature(
+                    content=base64.b64encode(signature).decode(),
+                    public_key=rekor_types.hashedrekord.PublicKey(
+                        content=base64.b64encode(
+                            certificate.public_bytes(
+                                encoding=serialization.Encoding.PEM
+                            )
+                        ).decode()
+                    ),
+                ),
+                data=rekor_types.hashedrekord.Data(
+                    hash=rekor_types.hashedrekord.Hash(
+                        algorithm=hashed_input._as_hashedrekord_algorithm(),
+                        value=hashed_input.digest.hex(),
+                    )
+                ),
+            ),
+        )
+
+    def _build_dsse_request(  # type: ignore[override]
+        self, envelope: Envelope, certificate: Certificate
+    ) -> rekor_types.Dsse:
+        """
+        Construct a dsse request to submit to Rekor.
+        """
+        return rekor_types.Dsse(
+            spec=rekor_types.dsse.DsseSchema(
+                # NOTE: mypy can't see that this kwarg is correct due to two interacting
+                # behaviors/bugs (one pydantic, one datamodel-codegen):
+                # See: <https://github.com/pydantic/pydantic/discussions/7418#discussioncomment-9024927>
+                # See: <https://github.com/koxudaxi/datamodel-code-generator/issues/1903>
+                proposed_content=rekor_types.dsse.ProposedContent(  # type: ignore[call-arg]
+                    envelope=envelope.to_json(),
+                    verifiers=[
+                        base64.b64encode(
+                            certificate.public_bytes(
+                                encoding=serialization.Encoding.PEM
+                            )
+                        ).decode()
+                    ],
+                ),
+            ),
+        )

sigstore/sign.py

@@ -38,7 +38,6 @@
 
 from __future__ import annotations
 
-import base64
 import logging
 from collections.abc import Iterator
 from contextlib import contextmanager
@@ -47,7 +46,7 @@
 
 import cryptography.x509 as x509
 import rekor_types
-from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.x509.oid import NameOID
 from sigstore_protobuf_specs.dev.sigstore.common.v1 import (
@@ -182,7 +181,7 @@ def _finalize_sign(
         Perform the common "finalizing" steps in a Sigstore signing flow.
         """
         # Submit the proposed entry to the transparency log
-        entry = self._signing_ctx._rekor.log.entries.post(proposed_entry)
+        entry = self._signing_ctx._rekor.create_entry(proposed_entry)
 
         _logger.debug(f"Transparency log entry created with index: {entry.log_index}")
 
@@ -211,26 +210,12 @@ def sign_dsse(
         """
         cert = self._signing_cert()
 
-        # Prepare inputs
-        b64_cert = base64.b64encode(
-            cert.public_bytes(encoding=serialization.Encoding.PEM)
-        )
-
         # Sign the statement, producing a DSSE envelope
         content = dsse._sign(self._private_key, input_)
 
         # Create the proposed DSSE log entry
-        proposed_entry = rekor_types.Dsse(
-            spec=rekor_types.dsse.DsseSchema(
-                # NOTE: mypy can't see that this kwarg is correct due to two interacting
-                # behaviors/bugs (one pydantic, one datamodel-codegen):
-                # See: <https://github.com/pydantic/pydantic/discussions/7418#discussioncomment-9024927>
-                # See: <https://github.com/koxudaxi/datamodel-code-generator/issues/1903>
-                proposed_content=rekor_types.dsse.ProposedContent(  # type: ignore[call-arg]
-                    envelope=content.to_json(),
-                    verifiers=[b64_cert.decode()],
-                ),
-            ),
+        proposed_entry = self._signing_ctx._rekor._build_dsse_request(
+            envelope=content, certificate=cert
         )
 
         return self._finalize_sign(cert, content, proposed_entry)
@@ -255,11 +240,6 @@ def sign_artifact(
 
         cert = self._signing_cert()
 
-        # Prepare inputs
-        b64_cert = base64.b64encode(
-            cert.public_bytes(encoding=serialization.Encoding.PEM)
-        )
-
         # Sign artifact
         hashed_input = sha256_digest(input_)
 
@@ -276,21 +256,8 @@ def sign_artifact(
         )
 
         # Create the proposed hashedrekord entry
-        proposed_entry = rekor_types.Hashedrekord(
-            spec=rekor_types.hashedrekord.HashedrekordV001Schema(
-                signature=rekor_types.hashedrekord.Signature(
-                    content=base64.b64encode(artifact_signature).decode(),
-                    public_key=rekor_types.hashedrekord.PublicKey(
-                        content=b64_cert.decode()
-                    ),
-                ),
-                data=rekor_types.hashedrekord.Data(
-                    hash=rekor_types.hashedrekord.Hash(
-                        algorithm=hashed_input._as_hashedrekord_algorithm(),
-                        value=hashed_input.digest.hex(),
-                    )
-                ),
-            ),
+        proposed_entry = self._signing_ctx._rekor._build_hashed_rekord_request(
+            hashed_input=hashed_input, signature=artifact_signature, certificate=cert
         )
 
         return self._finalize_sign(cert, content, proposed_entry)