feat: Enable signing with pre-computed hash (#860)

Added prehashing support to Signer.sign(): caller can now precompute the hash and provide
a Hashed object instead of plain input bytes.

API changes:
* Changed sign.Signer.sign() backwards-compatibly
* Added hashes module

The algorithm given to rekor remains unchanged (but capability of providing other
algorithms in the future now exists).

Signed-off-by: laurentsimon <laurentsimon@google.com>
Signed-off-by: William Woodruff <william@trailofbits.com>
Co-authored-by: William Woodruff <william@trailofbits.com>

Author: laurentsimon

CHANGELOG.md

@@ -10,11 +10,14 @@ All versions prior to 0.9.0 are untracked.
 
 ### Added
 
+* API: `Signer.sign()` can now take a `Hashed` as an input,
+  performing a signature on a pre-computed hash value
+  ([#860](https://github.com/sigstore/sigstore-python/pull/860))
+
 * API: `Signer.sign()` can now take an in-toto `Statement` as an input,
   producing a DSSE-formatted signature rather than a "bare" signature
   ([#804](https://github.com/sigstore/sigstore-python/pull/804))
 
-
 * API: `SigningResult.content` has been added, representing either the
   `hashedrekord` entry's message signature or the `dsse` entry's envelope
   ([#804](https://github.com/sigstore/sigstore-python/pull/804))

sigstore/_utils.py

@@ -32,7 +32,9 @@
     load_der_x509_certificate,
 )
 from cryptography.x509.oid import ExtendedKeyUsageOID, ExtensionOID
+from sigstore_protobuf_specs.dev.sigstore.common.v1 import HashAlgorithm
 
+from sigstore import hashes as sigstore_hashes
 from sigstore.errors import Error
 
 if sys.version_info < (3, 11):
@@ -158,6 +160,19 @@ def key_id(key: PublicKey) -> KeyID:
     return KeyID(hashlib.sha256(public_bytes).digest())
 
 
+def get_digest(input_: IO[bytes] | sigstore_hashes.Hashed) -> sigstore_hashes.Hashed:
+    """
+    Compute the SHA256 digest of an input stream or, if given a `Hashed`,
+    return it directly.
+    """
+    if isinstance(input_, sigstore_hashes.Hashed):
+        return input_
+
+    return sigstore_hashes.Hashed(
+        digest=sha256_streaming(input_), algorithm=HashAlgorithm.SHA2_256
+    )
+
+
 def sha256_streaming(io: IO[bytes]) -> bytes:
     """
     Compute the SHA256 of a stream.

sigstore/hashes.py

@@ -0,0 +1,55 @@
+# Copyright 2023 The Sigstore Authors
+#
+# 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.
+
+"""
+Hashing APIs.
+"""
+
+import rekor_types
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric.utils import Prehashed
+from pydantic import BaseModel
+from sigstore_protobuf_specs.dev.sigstore.common.v1 import HashAlgorithm
+
+
+class Hashed(BaseModel):
+    """
+    Represents a hashed value.
+    """
+
+    algorithm: HashAlgorithm
+    """
+    The digest algorithm uses to compute the digest.
+    """
+
+    digest: bytes
+    """
+    The digest representing the hash value.
+    """
+
+    def _as_hashedrekord_algorithm(self) -> rekor_types.hashedrekord.Algorithm:
+        """
+        Returns an appropriate `hashedrekord.Algorithm` for this `Hashed`.
+        """
+        if self.algorithm == HashAlgorithm.SHA2_256:
+            return rekor_types.hashedrekord.Algorithm.SHA256
+        raise ValueError(f"unknown hash algorithm: {self.algorithm}")
+
+    def _as_prehashed(self) -> Prehashed:
+        """
+        Returns an appropriate Cryptography `Prehashed` for this `Hashed`.
+        """
+        if self.algorithm == HashAlgorithm.SHA2_256:
+            return Prehashed(hashes.SHA256())
+        raise ValueError(f"unknown hash algorithm: {self.algorithm}")

sigstore/sign.py

@@ -49,15 +49,13 @@
 import rekor_types
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import ec
-from cryptography.hazmat.primitives.asymmetric.utils import Prehashed
 from cryptography.x509.oid import NameOID
 from in_toto_attestation.v1.statement import Statement
 from sigstore_protobuf_specs.dev.sigstore.bundle.v1 import (
     Bundle,
     VerificationMaterial,
 )
 from sigstore_protobuf_specs.dev.sigstore.common.v1 import (
-    HashAlgorithm,
     HashOutput,
     LogId,
     MessageSignature,
@@ -73,6 +71,7 @@
 )
 from sigstore_protobuf_specs.io.intoto import Envelope
 
+from sigstore import hashes as sigstore_hashes
 from sigstore._internal import dsse
 from sigstore._internal.fulcio import (
     ExpiredCertificate,
@@ -82,7 +81,7 @@
 from sigstore._internal.rekor.client import RekorClient
 from sigstore._internal.sct import verify_sct
 from sigstore._internal.trustroot import TrustedRoot
-from sigstore._utils import PEMCert, sha256_streaming
+from sigstore._utils import PEMCert, get_digest
 from sigstore.oidc import ExpiredIdentity, IdentityToken
 from sigstore.transparency import LogEntry
 
@@ -176,7 +175,7 @@ def _signing_cert(
 
     def sign(
         self,
-        input_: IO[bytes] | Statement,
+        input_: IO[bytes] | Statement | sigstore_hashes.Hashed,
     ) -> Bundle:
         """Public API for signing blobs"""
         private_key = self._private_key
@@ -219,16 +218,16 @@ def sign(
                 ),
             )
         else:
-            input_digest = sha256_streaming(input_)
+            hashed_input = get_digest(input_)
 
             artifact_signature = private_key.sign(
-                input_digest, ec.ECDSA(Prehashed(hashes.SHA256()))
+                hashed_input.digest, ec.ECDSA(hashed_input._as_prehashed())
             )
 
             content = MessageSignature(
                 message_digest=HashOutput(
-                    algorithm=HashAlgorithm.SHA2_256,
-                    digest=input_digest,
+                    algorithm=hashed_input.algorithm,
+                    digest=hashed_input.digest,
                 ),
                 signature=artifact_signature,
             )
@@ -244,8 +243,8 @@ def sign(
                     ),
                     data=rekor_types.hashedrekord.Data(
                         hash=rekor_types.hashedrekord.Hash(
-                            algorithm=rekor_types.hashedrekord.Algorithm.SHA256,
-                            value=input_digest.hex(),
+                            algorithm=hashed_input._as_hashedrekord_algorithm(),
+                            value=hashed_input.digest.hex(),
                         )
                     ),
                 ),

sigstore/verify/models.py

@@ -38,7 +38,6 @@
     VerificationMaterial,
 )
 from sigstore_protobuf_specs.dev.sigstore.common.v1 import (
-    HashAlgorithm,
     HashOutput,
     LogId,
     MessageSignature,
@@ -54,14 +53,15 @@
     TransparencyLogEntry,
 )
 
+from sigstore import hashes as sigstore_hashes
 from sigstore._internal.rekor import RekorClient
 from sigstore._utils import (
     B64Str,
     PEMCert,
     base64_encode_pem_cert,
     cert_is_leaf,
     cert_is_root_ca,
-    sha256_streaming,
+    get_digest,
 )
 from sigstore.errors import Error
 from sigstore.transparency import LogEntry, LogInclusionProof
@@ -179,9 +179,9 @@ class VerificationMaterials:
     Represents the materials needed to perform a Sigstore verification.
     """
 
-    input_digest: bytes
+    hashed_input: sigstore_hashes.Hashed
     """
-    The SHA256 hash of the verification input, as raw bytes.
+    The hash of the verification input.
     """
 
     certificate: Certificate
@@ -227,7 +227,7 @@ class VerificationMaterials:
     def __init__(
         self,
         *,
-        input_: IO[bytes],
+        input_: IO[bytes] | sigstore_hashes.Hashed,
         cert_pem: PEMCert,
         signature: bytes,
         offline: bool = False,
@@ -246,7 +246,7 @@ def __init__(
         Effect: `input_` is consumed as part of construction.
         """
 
-        self.input_digest = sha256_streaming(input_)
+        self.hashed_input = get_digest(input_)
         self.certificate = load_pem_x509_certificate(cert_pem.encode())
         self.signature = signature
 
@@ -416,8 +416,8 @@ def rekor_entry(self, client: RekorClient) -> LogEntry:
                 ),
                 data=rekor_types.hashedrekord.Data(
                     hash=rekor_types.hashedrekord.Hash(
-                        algorithm=rekor_types.hashedrekord.Algorithm.SHA256,
-                        value=self.input_digest.hex(),
+                        algorithm=self.hashed_input._as_hashedrekord_algorithm(),
+                        value=self.hashed_input.digest.hex(),
                     ),
                 ),
             ),
@@ -510,8 +510,8 @@ def to_bundle(self) -> Bundle:
             ),
             message_signature=MessageSignature(
                 message_digest=HashOutput(
-                    algorithm=HashAlgorithm.SHA2_256,
-                    digest=self.input_digest,
+                    algorithm=self.hashed_input.algorithm,
+                    digest=self.hashed_input.digest,
                 ),
                 signature=self.signature,
             ),

sigstore/verify/verifier.py

@@ -24,9 +24,7 @@
 from typing import List, cast
 
 from cryptography.exceptions import InvalidSignature
-from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import ec
-from cryptography.hazmat.primitives.asymmetric.utils import Prehashed
 from cryptography.x509 import Certificate, ExtendedKeyUsage, KeyUsage
 from cryptography.x509.oid import ExtendedKeyUsageOID
 from OpenSSL.crypto import (
@@ -225,8 +223,8 @@ def verify(
             signing_key = cast(ec.EllipticCurvePublicKey, signing_key)
             signing_key.verify(
                 materials.signature,
-                materials.input_digest,
-                ec.ECDSA(Prehashed(hashes.SHA256())),
+                materials.hashed_input.digest,
+                ec.ECDSA(materials.hashed_input._as_prehashed()),
             )
         except InvalidSignature:
             return VerificationFailure(reason="Signature is invalid for input")
@@ -241,7 +239,7 @@ def verify(
         except RekorEntryMissingError:
             return LogEntryMissing(
                 signature=B64Str(base64.b64encode(materials.signature).decode()),
-                artifact_hash=HexStr(materials.input_digest.hex()),
+                artifact_hash=HexStr(materials.hashed_input.digest.hex()),
             )
         except InvalidRekorEntryError:
             return VerificationFailure(

test/unit/conftest.py

@@ -40,6 +40,7 @@
 from sigstore.sign import SigningContext
 from sigstore.verify import VerificationMaterials
 from sigstore.verify.policy import VerificationSuccess
+from sigstore.verify.verifier import Verifier
 
 _ASSETS = (Path(__file__).parent / "assets").resolve()
 assert _ASSETS.is_dir()
@@ -50,7 +51,9 @@
 
 def _has_oidc_id():
     # If there are tokens manually defined for us in the environment, use them.
-    if os.getenv("SIGSTORE_IDENTITY_TOKEN_production") is not None:
+    if os.getenv("SIGSTORE_IDENTITY_TOKEN_production") or os.getenv(
+        "SIGSTORE_IDENTITY_TOKEN_staging"
+    ):
         return True
 
     try:
@@ -240,7 +243,7 @@ def tuf_dirs(monkeypatch, tmp_path):
     ],
     ids=["production", "staging"],
 )
-def id_config(request) -> tuple[SigningContext, IdentityToken]:
+def signer_and_ident(request) -> tuple[type[SigningContext], type[IdentityToken]]:
     env, signer = request.param
     # Detect env variable for local interactive tests.
     token = os.getenv(f"SIGSTORE_IDENTITY_TOKEN_{env}")
@@ -251,6 +254,20 @@ def id_config(request) -> tuple[SigningContext, IdentityToken]:
     return signer, IdentityToken(token)
 
 
+@pytest.fixture
+def staging() -> tuple[type[SigningContext], type[Verifier], IdentityToken]:
+    signer = SigningContext.staging
+    verifier = Verifier.staging
+
+    # Detect env variable for local interactive tests.
+    token = os.getenv("SIGSTORE_IDENTITY_TOKEN_staging")
+    if not token:
+        # If the variable is not defined, try getting an ambient token.
+        token = detect_credential(_DEFAULT_AUDIENCE)
+
+    return signer, verifier, IdentityToken(token)
+
+
 @pytest.fixture
 def dummy_jwt():
     def _dummy_jwt(claims: dict):