Support multiple transparency log entries

CHANGELOG.md

@@ -8,6 +8,13 @@ All versions prior to 0.9.0 are untracked.
 
 ## [Unreleased]
 
+### Added
+
+* Verification now supports bundles with multiple transparency log entries
+  and configurable transparency log thresholds through
+  `Verifier(tlog_threshold=...)` and `--tlog-threshold`
+  ([#1821](https://github.com/sigstore/sigstore-python/issues/1821)).
+
 ### Fixed
 
 * Parsing a malformed in-toto statement now includes the underlying validation

README.md

@@ -225,8 +225,8 @@ Output options:
 ```
 usage: sigstore verify identity [-h] [-v] [--certificate FILE]
                                 [--signature FILE] [--bundle FILE] [--offline]
-                                --cert-identity IDENTITY --cert-oidc-issuer
-                                URL
+                                [--tlog-threshold N] --cert-identity IDENTITY
+                                --cert-oidc-issuer URL
                                 FILE_OR_DIGEST [FILE_OR_DIGEST ...]
 
 options:
@@ -248,6 +248,8 @@ Verification inputs:
 Verification options:
   --offline             Perform offline verification; requires a Sigstore
                         bundle (default: False)
+  --tlog-threshold N    Require verification by at least N transparency log
+                        operators (default: 1)
   --cert-identity IDENTITY
                         The identity to check for in the certificate's Subject
                         Alternative Name (default: None)
@@ -263,9 +265,9 @@ Verification options:
 ```
 usage: sigstore verify github [-h] [-v] [--certificate FILE]
                               [--signature FILE] [--bundle FILE] [--offline]
-                              [--cert-identity IDENTITY] [--trigger EVENT]
-                              [--sha SHA] [--name NAME] [--repository REPO]
-                              [--ref REF]
+                              [--tlog-threshold N] [--cert-identity IDENTITY]
+                              [--trigger EVENT] [--sha SHA] [--name NAME]
+                              [--repository REPO] [--ref REF]
                               FILE_OR_DIGEST [FILE_OR_DIGEST ...]
 
 options:
@@ -287,6 +289,8 @@ Verification inputs:
 Verification options:
   --offline             Perform offline verification; requires a Sigstore
                         bundle (default: False)
+  --tlog-threshold N    Require verification by at least N transparency log
+                        operators (default: 1)
   --cert-identity IDENTITY
                         The identity to check for in the certificate's Subject
                         Alternative Name (default: None)

sigstore/_cli.py

@@ -187,12 +187,25 @@ def file_or_digest(arg: str) -> Hashed | Path:
 
 
 def _add_shared_verification_options(group: argparse._ArgumentGroup) -> None:
+    def positive_int(value: str) -> int:
+        threshold = int(value)
+        if threshold < 1:
+            raise argparse.ArgumentTypeError("must be at least 1")
+        return threshold
+
     group.add_argument(
         "--offline",
         action="store_true",
         default=_boolify_env("SIGSTORE_OFFLINE"),
         help="Perform offline verification; requires a Sigstore bundle",
     )
+    group.add_argument(
+        "--tlog-threshold",
+        type=positive_int,
+        default=1,
+        metavar="N",
+        help="Require verification by at least N transparency log operators",
+    )
 
 
 def _add_shared_oidc_options(
@@ -1090,7 +1103,10 @@ def _collect_verification_state(
             )
 
     trust_config = _get_trust_config(args)
-    verifier = Verifier(trusted_root=trust_config.trusted_root)
+    verifier = Verifier(
+        trusted_root=trust_config.trusted_root,
+        tlog_threshold=args.tlog_threshold,
+    )
 
     all_materials = []
     for file_or_hashed, materials in input_map.items():

sigstore/models.py

@@ -74,6 +74,10 @@
 
 _logger = logging.getLogger(__name__)
 
+# Bound attacker-controlled transparency log entries to limit verification work.
+# This matches sigstore-go's current limit.
+_MAX_ALLOWED_TLOG_ENTRIES = 32
+
 
 class TransparencyLogEntry:
     """
@@ -438,60 +442,55 @@ def _verify(self) -> None:
 
         self._signing_certificate = leaf_cert
 
-        # Extract the log entry. For the time being, we expect
-        # bundles to only contain a single log entry.
+        # Extract and validate the transparency log entries.
         tlog_entries = self._inner.verification_material.tlog_entries
-        if len(tlog_entries) != 1:
-            raise InvalidBundle("expected exactly one log entry in bundle")
-        tlog_entry = tlog_entries[0]
-
-        if tlog_entry.kind_version.version not in ["0.0.1", "0.0.2"]:
-            raise IncompatibleEntry(
-                f"Expected log entry version 0.0.1 - 0.0.2, got {tlog_entry.kind_version.version}"
+        if not tlog_entries:
+            raise InvalidBundle("expected at least one log entry in bundle")
+        if len(tlog_entries) > _MAX_ALLOWED_TLOG_ENTRIES:
+            raise InvalidBundle(
+                f"expected at most {_MAX_ALLOWED_TLOG_ENTRIES} log entries in bundle"
             )
 
-        # Handling of inclusion promises and proofs varies between bundle
-        # format versions:
-        #
-        # * For 0.1, an inclusion promise is required; the client
-        #   MUST verify the inclusion promise.
-        #   The inclusion proof is NOT required. If provided, it might NOT
-        #   contain a checkpoint; in this case, we ignore it (since it's
-        #   useless without one).
-        #
-        # * For 0.2+, an inclusion proof is required; the client MUST
-        #   verify the inclusion proof. The inclusion prof MUST contain
-        #   a checkpoint.
-        #
-        #   The inclusion promise is NOT required if another source of signed
-        #   time (such as a signed timestamp) is present. If no other source
-        #   of signed time is present, then the inclusion promise MUST be
-        #   present.
-        #
-        # Before all of this, we require that the inclusion proof be present
-        # (when constructing the LogEntry).
-        log_entry = TransparencyLogEntry(tlog_entry)
-
-        if media_type == Bundle.BundleType.BUNDLE_0_1:
-            if not log_entry._inner.inclusion_promise:
-                raise InvalidBundle("bundle must contain an inclusion promise")
-            if not log_entry._inner.inclusion_proof.checkpoint:
-                _logger.debug(
-                    "0.1 bundle contains inclusion proof without checkpoint; ignoring"
+        log_entries: list[TransparencyLogEntry] = []
+        for tlog_entry in tlog_entries:
+            if tlog_entry.kind_version.version not in ["0.0.1", "0.0.2"]:
+                raise IncompatibleEntry(
+                    "Expected log entry version 0.0.1 - 0.0.2, "
+                    f"got {tlog_entry.kind_version.version}"
                 )
-        else:
-            if not log_entry._inner.inclusion_proof.checkpoint:
-                raise InvalidBundle("expected checkpoint in inclusion proof")
 
-            if (
-                not log_entry._inner.inclusion_promise
-                and not self.verification_material.timestamp_verification_data
-            ):
-                raise InvalidBundle(
-                    "bundle must contain an inclusion promise or signed timestamp(s)"
-                )
+            # Handling of inclusion promises and proofs varies between bundle
+            # format versions:
+            #
+            # * For 0.1, an inclusion promise is required; the client
+            #   MUST verify the inclusion promise.
+            # * For 0.2+, an inclusion proof is required; the client MUST
+            #   verify the inclusion proof. An inclusion promise is optional
+            #   when another signed source of time is present.
+            log_entry = TransparencyLogEntry(tlog_entry)
+
+            if media_type == Bundle.BundleType.BUNDLE_0_1:
+                if not log_entry._inner.inclusion_promise:
+                    raise InvalidBundle("bundle must contain an inclusion promise")
+                if not log_entry._inner.inclusion_proof.checkpoint:
+                    _logger.debug(
+                        "0.1 bundle contains inclusion proof without checkpoint; ignoring"
+                    )
+            else:
+                if not log_entry._inner.inclusion_proof.checkpoint:
+                    raise InvalidBundle("expected checkpoint in inclusion proof")
+
+                if (
+                    not log_entry._inner.inclusion_promise
+                    and not self.verification_material.timestamp_verification_data
+                ):
+                    raise InvalidBundle(
+                        "bundle must contain an inclusion promise or signed timestamp(s)"
+                    )
+
+            log_entries.append(log_entry)
 
-        self._log_entry = log_entry
+        self._log_entries = log_entries
 
     @property
     def signing_certificate(self) -> Certificate:
@@ -501,10 +500,12 @@ def signing_certificate(self) -> Certificate:
     @property
     def log_entry(self) -> TransparencyLogEntry:
         """
-        Returns the bundle's log entry, containing an inclusion proof
-        (with checkpoint) and an inclusion promise (if the latter is present).
+        Returns the bundle's first transparency log entry.
+
+        This property is retained for compatibility with callers that expect
+        single-entry bundles.
         """
-        return self._log_entry
+        return self._log_entries[0]
 
     @property
     def _dsse_envelope(self) -> dsse.Envelope | None:
@@ -836,12 +837,11 @@ def from_file(
         inner = trustroot_v1.TrustedRoot.from_json(Path(path).read_bytes())
         return cls(inner)
 
-    def _get_tlog_keys(
+    def _get_usable_tlogs(
         self, tlogs: list[trustroot_v1.TransparencyLogInstance], purpose: KeyringPurpose
-    ) -> Iterable[common_v1.PublicKey]:
+    ) -> Iterable[trustroot_v1.TransparencyLogInstance]:
         """
-        Yields an iterator of public keys for transparency log instances that
-        are suitable for `purpose`.
+        Yields transparency log instances that are suitable for `purpose`.
         """
         allow_expired = purpose is KeyringPurpose.VERIFY
         for tlog in tlogs:
@@ -850,8 +850,24 @@ def _get_tlog_keys(
             ):
                 continue
 
+            yield tlog
+
+    def _get_tlog_keys(
+        self, tlogs: list[trustroot_v1.TransparencyLogInstance], purpose: KeyringPurpose
+    ) -> Iterable[common_v1.PublicKey]:
+        """
+        Yields public keys for transparency log instances that are suitable
+        for `purpose`.
+        """
+        for tlog in self._get_usable_tlogs(tlogs, purpose):
             yield tlog.public_key
 
+    def _rekor_tlogs(
+        self, purpose: KeyringPurpose
+    ) -> list[trustroot_v1.TransparencyLogInstance]:
+        """Return usable Rekor transparency log instances."""
+        return list(self._get_usable_tlogs(self._inner.tlogs, purpose))
+
     def rekor_keyring(self, purpose: KeyringPurpose) -> RekorKeyring:
         """Return keyring with keys for Rekor."""