cli: allow DSSE verification

sigstore/_cli.py

@@ -26,7 +26,7 @@
 from cryptography.x509 import load_pem_x509_certificate
 from rich.logging import RichHandler
 
-from sigstore import __version__
+from sigstore import __version__, dsse
 from sigstore._internal.fulcio.client import (
     DEFAULT_FULCIO_URL,
     ExpiredCertificate,
@@ -813,13 +813,9 @@ def _verify_identity(args: argparse.Namespace) -> None:
         )
 
         try:
-            verifier.verify_artifact(
-                input_=hashed,
-                bundle=bundle,
-                policy=policy_,
-            )
+            _verify_common(verifier, hashed, bundle, policy_)
             print(f"OK: {file}")
-        except VerificationError as exc:
+        except Error as exc:
             _logger.error(f"FAIL: {file}")
             exc.log_and_exit(_logger, args.verbose >= 1)
 
@@ -851,13 +847,48 @@ def _verify_github(args: argparse.Namespace) -> None:
     verifier, materials = _collect_verification_state(args)
     for file, hashed, bundle in materials:
         try:
-            verifier.verify_artifact(input_=hashed, bundle=bundle, policy=policy_)
+            _verify_common(verifier, hashed, bundle, policy_)
             print(f"OK: {file}")
-        except VerificationError as exc:
+        except Error as exc:
             _logger.error(f"FAIL: {file}")
             exc.log_and_exit(_logger, args.verbose >= 1)
 
 
+def _verify_common(
+    verifier: Verifier,
+    hashed: Hashed,
+    bundle: Bundle,
+    policy_: policy.VerificationPolicy,
+) -> None:
+    """
+    Common verification handling.
+
+    This dispatches to either artifact or DSSE verification, depending on
+    `bundle`'s inner type.
+    """
+
+    # If the bundle specifies a DSSE envelope, perform DSSE verification
+    # and assert that the inner payload is an in-toto statement bound
+    # to a subject matching the input's digest.
+    if bundle._dsse_envelope:
+        type_, payload = verifier.verify_dsse(bundle=bundle, policy=policy_)
+        if type_ != dsse.Envelope._TYPE:
+            raise VerificationError(f"expected JSON payload for DSSE, got {type_}")
+
+        stmt = dsse.Statement(payload)
+        if not stmt._matches_digest(hashed):
+            raise VerificationError(
+                f"in-toto statement has no subject for digest {hashed.digest.hex()}"
+            )
+    else:
+        verifier.verify_artifact(
+            input_=hashed,
+            bundle=bundle,
+            policy=policy_,
+        )
+    pass
+
+
 def _get_identity(args: argparse.Namespace) -> Optional[IdentityToken]:
     token = None
     if not args.oidc_disable_ambient_providers:

sigstore/dsse.py

@@ -25,10 +25,12 @@
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import ec
 from pydantic import BaseModel, ConfigDict, Field, RootModel, StrictStr, ValidationError
+from sigstore_protobuf_specs.dev.sigstore.common.v1 import HashAlgorithm
 from sigstore_protobuf_specs.io.intoto import Envelope as _Envelope
 from sigstore_protobuf_specs.io.intoto import Signature
 
-from sigstore.errors import VerificationError
+from sigstore.errors import Error, VerificationError
+from sigstore.hashes import Hashed
 
 _logger = logging.getLogger(__name__)
 
@@ -97,9 +99,28 @@ def __init__(self, contents: bytes) -> None:
         """
         self._contents = contents
         try:
-            self._statement = _Statement.model_validate_json(contents)
+            self._inner = _Statement.model_validate_json(contents)
         except ValidationError:
-            raise ValueError("malformed in-toto statement")
+            raise Error("malformed in-toto statement")
+
+    def _matches_digest(self, digest: Hashed) -> bool:
+        """
+        Returns a boolean indicating whether this in-toto Statement contains a subject
+        matching the given digest. The subject's name is **not** checked.
+
+        No digests other than SHA256 are currently supported.
+        """
+        if digest.algorithm != HashAlgorithm.SHA2_256:
+            raise VerificationError(f"unexpected digest algorithm: {digest.algorithm}")
+
+        for sub in self._inner.subjects:
+            sub_digest = sub.digest.root.get("sha256")
+            if sub_digest is None:
+                continue
+            if sub_digest == digest.digest.hex():
+                return True
+
+        return False
 
     def _pae(self) -> bytes:
         """
@@ -160,7 +181,7 @@ def build(self) -> Statement:
                 predicate=self._predicate,
             )
         except ValidationError as e:
-            raise ValueError(f"invalid statement: {e}")
+            raise Error(f"invalid statement: {e}")
 
         return Statement(stmt.model_dump_json(by_alias=True).encode())