Change API

Signed-off-by: laurentsimon <laurentsimon@google.com>

Author: laurentsimon

sigstore/_utils.py

@@ -23,13 +23,13 @@
 import sys
 from typing import IO, NewType, Union
 
-from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import ec, rsa
-from cryptography.hazmat.primitives.asymmetric.utils import Prehashed
 from cryptography.x509 import Certificate, ExtensionNotFound, Version
 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):
@@ -141,26 +141,13 @@ def key_id(key: PublicKey) -> KeyID:
 
     return KeyID(hashlib.sha256(public_bytes).digest())
 
-def hazmat_digest_to_bundle(algo: str) -> HashAlgorithm:
-    lookup = {hashes.SHA256().name: HashAlgorithm.SHA2_256}
-    if algo in lookup:
-        return HashAlgorithm(lookup[algo])
-    return ValueError(f"unknown digest algorithm {algo}")
-
-def get_digest(
-        input_: IO[bytes],
-        algorithm_: Prehashed = None,
-    ) -> (bytes, Prehashed):
-    if algorithm_ is None:
-        return sha256_streaming(input_), Prehashed(hashes.SHA256())
-
-    if isinstance(algorithm_, Prehashed):
-        # Check we have a 256-bit digest size for compatibility with secp256r1.
-        if algorithm_.digest_size != 32:
-            return ValueError(f"invalid digest size ({algorithm_.digest_size}), expected 32")
-        return input_.getvalue(), algorithm_
-
-    raise ValueError("invalid arguments")
+def get_digest(input_: IO[bytes] | sigstore_hashes.Hashed) -> sigstore_hashes.Hashed:
+    if isinstance(input_, sigstore_hashes.Hashed):
+        return input_
+
+    # NOTE: Not able to check for the type `TypeError: Subscripted generics cannot be used with class and instance checks`
+    # when calling isinstance(_input, IO[bytes])
+    return sigstore_hashes.Hashed(digest=sha256_streaming(input_), algorithm=HashAlgorithm.SHA2_256)
 
 def sha256_streaming(io: IO[bytes]) -> bytes:
     """

sigstore/hashes.py

@@ -0,0 +1,50 @@
+
+# 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.
+
+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 DIRSHA256_P1(hashes.HashAlgorithm):
+    name = "dirsha256-p1"
+    digest_size = 32
+    block_size = -1
+
+class Hashed(BaseModel):
+    """
+    Represents hashed value.
+    """
+
+    algorithm: HashAlgorithm
+    """
+    The digest algorithm uses to compute the digest.
+    """
+
+    digest: bytes
+    """
+    The digest representing the hash value.
+    """
+
+    def as_prehashed(self) -> Prehashed:
+        return Prehashed(self.hazmat_algorithm())
+
+    def hazmat_algorithm(self) -> hashes.HashAlgorithm:
+        if self.algorithm == HashAlgorithm.SHA2_256:
+            return hashes.SHA256()
+        if self.algorithm == HashAlgorithm.HASH_ALGORITHM_UNSPECIFIED:
+            return DIRSHA256_P1()
+        raise ValueError(f"unknown hash algorithm: {self.algorithm}")

sigstore/sign.py

@@ -49,14 +49,14 @@
 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 pydantic import BaseModel
 from sigstore_protobuf_specs.dev.sigstore.bundle.v1 import (
     Bundle,
     VerificationMaterial,
 )
 from sigstore_protobuf_specs.dev.sigstore.common.v1 import (
+    HashAlgorithm,
     HashOutput,
     LogId,
     MessageSignature,
@@ -71,6 +71,7 @@
     TransparencyLogEntry,
 )
 
+from sigstore import hashes as sigstore_hashes
 from sigstore._internal.fulcio import (
     ExpiredCertificate,
     FulcioCertificateSigningResponse,
@@ -79,7 +80,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 B64Str, HexStr, PEMCert, get_digest, hazmat_digest_to_bundle
+from sigstore._utils import B64Str, HexStr, PEMCert, get_digest
 from sigstore.oidc import ExpiredIdentity, IdentityToken
 from sigstore.transparency import LogEntry
 
@@ -173,11 +174,10 @@ def _signing_cert(
 
     def sign(
         self,
-        input_: IO[bytes],
-        algorithm_: Prehashed = None,
+        input_: IO[bytes] | sigstore_hashes.Hashed,
     ) -> SigningResult:
         """Public API for signing blobs"""
-        input_digest, algorithm = get_digest(input_, algorithm_)
+        hashed_input = get_digest(input_)
         private_key = self._private_key
 
         if not self._identity_token.in_validity_period():
@@ -201,7 +201,7 @@ def sign(
 
         # Sign artifact
         artifact_signature = private_key.sign(
-            input_digest, ec.ECDSA(algorithm)
+            hashed_input.digest, ec.ECDSA(hashed_input.as_prehashed())
         )
         b64_artifact_signature = B64Str(base64.b64encode(artifact_signature).decode())
 
@@ -223,8 +223,8 @@ def sign(
                 ),
                 data=rekor_types.hashedrekord.Data(
                     hash=rekor_types.hashedrekord.Hash(
-                        algorithm=algorithm._algorithm.name,
-                        value=input_digest.hex(),
+                        algorithm=hashed_input.hazmat_algorithm().name,
+                        value=hashed_input.digest.hex(),
                     )
                 ),
             ),
@@ -234,8 +234,8 @@ def sign(
         logger.debug(f"Transparency log entry created with index: {entry.log_index}")
 
         return SigningResult(
-            digest_algorithm=hazmat_digest_to_bundle(algorithm._algorithm.name),
-            input_digest=HexStr(input_digest.hex()),
+            digest_algorithm=hashed_input.algorithm,
+            input_digest=HexStr(hashed_input.digest.hex()),
             cert_pem=PEMCert(
                 cert.public_bytes(encoding=serialization.Encoding.PEM).decode()
             ),
@@ -314,16 +314,11 @@ class SigningResult(BaseModel):
     Represents the artifacts of a signing operation.
     """
 
-    digest_algorithm: str
+    digest_algorithm: HashAlgorithm
     """
     The digest algorithm to use for the hash.
     """
 
-    input_digest: HexStr
-    """
-    The hex-encoded SHA256 digest of the input that was signed for.
-    """
-
     cert_pem: PEMCert
     """
     The PEM-encoded public half of the certificate used for signing.

sigstore/verify/models.py

@@ -26,7 +26,6 @@
 from typing import IO
 
 import rekor_types
-from cryptography.hazmat.primitives.asymmetric.utils import Prehashed
 from cryptography.hazmat.primitives.serialization import Encoding
 from cryptography.x509 import (
     Certificate,
@@ -54,6 +53,7 @@
     TransparencyLogEntry,
 )
 
+from sigstore import hashes as sigstore_hashes
 from sigstore._internal.rekor import RekorClient
 from sigstore._utils import (
     B64Str,
@@ -62,7 +62,6 @@
     cert_is_leaf,
     cert_is_root_ca,
     get_digest,
-    hazmat_digest_to_bundle,
 )
 from sigstore.errors import Error
 from sigstore.transparency import LogEntry, LogInclusionProof
@@ -180,17 +179,11 @@ class VerificationMaterials:
     Represents the materials needed to perform a Sigstore verification.
     """
 
-    digest_algorithm: Prehashed
+    hashed_input: sigstore_hashes.Hashed
     """
-    The digest algorithm to use for the hash.
+    The hash of the verification input.
     """
 
-    input_digest: bytes
-    """
-    The 'digest_algorithm' hash of the verification input, as raw bytes.
-    """
-
-
     certificate: Certificate
     """
     The certificate that attests to and contains the public signing key.
@@ -234,12 +227,11 @@ class VerificationMaterials:
     def __init__(
         self,
         *,
-        input_: IO[bytes],
+        input_: IO[bytes] | sigstore_hashes.Hashed,
         cert_pem: PEMCert,
         signature: bytes,
         offline: bool = False,
         rekor_entry: LogEntry | None,
-        algorithm_: Prehashed = None
     ):
         """
         Create a new `VerificationMaterials` from the given materials.
@@ -254,7 +246,7 @@ def __init__(
         Effect: `input_` is consumed as part of construction.
         """
 
-        self.input_digest, self.digest_algorithm = get_digest(input_, algorithm_)
+        self.hashed_input = get_digest(input_)
         self.certificate = load_pem_x509_certificate(cert_pem.encode())
         self.signature = signature
 
@@ -426,8 +418,8 @@ def rekor_entry(self, client: RekorClient) -> LogEntry:
                 ),
                 data=rekor_types.hashedrekord.Data(
                     hash=rekor_types.hashedrekord.Hash(
-                        algorithm=self.digest_algorithm._algorithm.name,
-                        value=self.input_digest.hex(),
+                        algorithm=self.hashed_input.hazmat_algorithm().name,
+                        value=self.hashed_input.digest.hex(),
                     ),
                 ),
             ),
@@ -520,8 +512,8 @@ def to_bundle(self) -> Bundle:
             ),
             message_signature=MessageSignature(
                 message_digest=HashOutput(
-                    algorithm=hazmat_digest_to_bundle(self.digest_algorithm._algorithm.name),
-                    digest=self.input_digest,
+                    algorithm=self.hashed_input.algorithm,
+                    digest=self.hashed_input.digest,
                 ),
                 signature=self.signature,
             ),

sigstore/verify/verifier.py

@@ -223,8 +223,8 @@ def verify(
             signing_key = cast(ec.EllipticCurvePublicKey, signing_key)
             signing_key.verify(
                 materials.signature,
-                materials.input_digest,
-                ec.ECDSA(materials.digest_algorithm),
+                materials.hashed_input.digest,
+                ec.ECDSA(materials.hashed_input.as_prehashed()),
             )
         except InvalidSignature:
             return VerificationFailure(reason="Signature is invalid for input")
@@ -239,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(