From 6c902122d5baf48341226994501938fd43e2fcfd Mon Sep 17 00:00:00 2001 From: Jason Hang Date: Wed, 23 Sep 2026 01:47:58 +0100 Subject: [PATCH] Support multiple transparency log entries Signed-off-by: Jason Hang --- CHANGELOG.md | 7 + README.md | 14 +- sigstore/_cli.py | 18 +- sigstore/models.py | 126 ++++++----- sigstore/verify/verifier.py | 339 ++++++++++++++++++------------ test/unit/internal/test_trust.py | 5 +- test/unit/test_models.py | 35 ++- test/unit/verify/test_verifier.py | 156 +++++++++++++- 8 files changed, 503 insertions(+), 197 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7992807df..bfc3cb178 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 224af561a..559734cf4 100644 --- a/README.md +++ b/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) diff --git a/sigstore/_cli.py b/sigstore/_cli.py index 7573b6436..cfb03772f 100644 --- a/sigstore/_cli.py +++ b/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(): diff --git a/sigstore/models.py b/sigstore/models.py index 2237b772e..18047bd6a 100644 --- a/sigstore/models.py +++ b/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.""" diff --git a/sigstore/verify/verifier.py b/sigstore/verify/verifier.py index 7dccb6670..a9cb31d8c 100644 --- a/sigstore/verify/verifier.py +++ b/sigstore/verify/verifier.py @@ -57,11 +57,16 @@ verify_sct, ) from sigstore._internal.timestamp import TimestampSource, TimestampVerificationResult -from sigstore._internal.trust import KeyringPurpose +from sigstore._internal.trust import Keyring, KeyringPurpose, RekorKeyring from sigstore._utils import base64_encode_pem_cert, sha256_digest from sigstore.errors import CertValidationError, VerificationError from sigstore.hashes import Hashed -from sigstore.models import Bundle, ClientTrustConfig, TrustedRoot +from sigstore.models import ( + Bundle, + ClientTrustConfig, + TransparencyLogEntry, + TrustedRoot, +) from sigstore.verify.policy import VerificationPolicy _logger = logging.getLogger(__name__) @@ -80,15 +85,22 @@ class Verifier: The primary API for verification operations. """ - def __init__(self, *, trusted_root: TrustedRoot): + def __init__(self, *, trusted_root: TrustedRoot, tlog_threshold: int = 1): """ Create a new `Verifier`. `trusted_root` is the `TrustedRoot` object containing the root of trust for the verification process. + + `tlog_threshold` is the minimum number of trusted transparency log + operators required for verification. It defaults to 1. """ + if tlog_threshold < 1: + raise ValueError("transparency log threshold must be at least 1") + self._fulcio_certificate_chain = trusted_root.get_fulcio_certs() self._trusted_root = trusted_root + self._tlog_threshold = tlog_threshold # this is an ugly hack needed for verifying "detached" materials # In reality we should be choosing the rekor instance based on the logid @@ -101,31 +113,31 @@ def __init__(self, *, trusted_root: TrustedRoot): self._rekor = RekorClient(url) @classmethod - def production(cls, *, offline: bool = False) -> Verifier: + def production(cls, *, offline: bool = False, tlog_threshold: int = 1) -> Verifier: """ Return a `Verifier` instance configured against Sigstore's production-level services. - `offline` controls the Trusted Root refresh behavior: if `True`, - the verifier uses the Trusted Root in the local TUF cache. If `False`, - a TUF repository refresh is attempted. + `offline` controls Trusted Root refresh behavior. + `tlog_threshold` controls the minimum transparency log threshold. """ config = ClientTrustConfig.production(offline=offline) return cls( trusted_root=config.trusted_root, + tlog_threshold=tlog_threshold, ) @classmethod - def staging(cls, *, offline: bool = False) -> Verifier: + def staging(cls, *, offline: bool = False, tlog_threshold: int = 1) -> Verifier: """ Return a `Verifier` instance configured against Sigstore's staging-level services. - `offline` controls the Trusted Root refresh behavior: if `True`, - the verifier uses the Trusted Root in the local TUF cache. If `False`, - a TUF repository refresh is attempted. + `offline` controls Trusted Root refresh behavior. + `tlog_threshold` controls the minimum transparency log threshold. """ config = ClientTrustConfig.staging(offline=offline) return cls( trusted_root=config.trusted_root, + tlog_threshold=tlog_threshold, ) def _verify_signed_timestamp( @@ -207,18 +219,165 @@ def _verify_timestamp_authority( return verified_timestamps - def _establish_time(self, bundle: Bundle) -> list[TimestampVerificationResult]: + def _verify_tlog_entry_body( + self, + bundle: Bundle, + entry: TransparencyLogEntry, + hashed_input: Hashed | None, + ) -> None: + """Verify that a transparency log entry matches the bundle contents.""" + if bundle._dsse_envelope is not None: + kind = entry._inner.kind_version.kind + version = entry._inner.kind_version.version + + if kind == "hashedrekord" and version == "0.0.2": + _validate_hashedrekord_v002_dsse_entry_body(bundle, entry) + elif kind == "dsse" and version == "0.0.1": + _validate_dsse_v001_entry_body(bundle, entry) + else: + raise VerificationError( + f"Unsupported DSSE log entry type: {kind}/{version}" + ) + return + + if hashed_input is None: + raise VerificationError( + "missing artifact digest for log entry verification" + ) + + if entry._inner.kind_version.kind != "hashedrekord": + raise VerificationError( + f"Expected entry type hashedrekord, got {entry._inner.kind_version.kind}" + ) + + version = entry._inner.kind_version.version + if version == "0.0.2": + _validate_hashedrekord_v002_entry_body(bundle, hashed_input, entry) + elif version == "0.0.1": + _validate_hashedrekord_v001_entry_body(bundle, hashed_input, entry) + else: + raise VerificationError(f"Unsupported hashedrekord version {version}") + + def _verify_tlog_entries( + self, + bundle: Bundle, + hashed_input: Hashed | None = None, + ) -> list[TimestampVerificationResult]: """ - Establish the time for bundle verification. + Verify the bundle's transparency log entries and enforce the configured + threshold. - This method uses timestamps from two possible sources: - 1. RFC3161 signed timestamps from a Timestamping Authority (TSA) - 2. Transparency Log timestamps + A log entry contributes to the threshold only after its transparency log + proof and its consistency with the signed bundle contents have both been + verified. """ - verified_timestamps = [] + trusted_tlogs = self._trusted_root._rekor_tlogs(KeyringPurpose.VERIFY) + + seen_entries: set[tuple[bytes, int]] = set() + verified_entries = 0 + verified_operators: set[str] = set() + verified_timestamps: list[TimestampVerificationResult] = [] + + for entry in bundle._log_entries: + entry_identity = ( + bytes(entry._inner.log_id.key_id), + entry._inner.log_index, + ) + if entry_identity in seen_entries: + raise VerificationError("duplicate transparency log entry") + seen_entries.add(entry_identity) + + # Prefer a trusted log whose configured log ID matches the bundle + # entry. Rekor v2 log IDs are not necessarily the same as the key ID + # computed from the log's public key, so matching must use the + # TransparencyLogInstance metadata rather than Keyring's key IDs. + # + # If no usable trusted log has a matching log ID, preserve the + # existing Keyring behavior by treating the bundle log ID as only a + # hint and trying all successfully loaded trusted Rekor keys. + candidate_keyrings = [] + exact_candidate_keyrings = [] + + for tlog in trusted_tlogs: + keyring = RekorKeyring(Keyring([tlog.public_key])) + if not keyring._keyring: + continue + + candidate = (tlog, keyring) + candidate_keyrings.append(candidate) + + if tlog.log_id.key_id == entry._inner.log_id.key_id: + exact_candidate_keyrings.append(candidate) + + candidates = exact_candidate_keyrings or candidate_keyrings + + verified_tlogs = [] + for tlog, keyring in candidates: + try: + entry._verify(keyring) + except VerificationError: + continue + verified_tlogs.append(tlog) + + if not verified_tlogs: + continue + + # The log proof alone is insufficient: the entry must describe the + # artifact or DSSE envelope that is actually being verified. + self._verify_tlog_entry_body(bundle, entry, hashed_input) + + if self._tlog_threshold == 1: + verified_entries += 1 + else: + if any(not tlog.operator for tlog in verified_tlogs): + raise VerificationError( + "operator metadata is required for transparency log " + "thresholds greater than 1" + ) + + operators = {tlog.operator for tlog in verified_tlogs if tlog.operator} + if len(operators) != 1: + raise VerificationError( + "transparency log entry matches multiple operators" + ) + + verified_operators.update(operators) + + timestamp = entry._inner.integrated_time + if timestamp and entry._inner.inclusion_promise: + kv = entry._inner.kind_version + if not (kv.kind in ["dsse", "hashedrekord"] and kv.version == "0.0.1"): + raise VerificationError( + "Integrated time only supported for " + "dsse/hashedrekord 0.0.1 types" + ) + + verified_timestamps.append( + TimestampVerificationResult( + source=TimestampSource.TRANSPARENCY_SERVICE, + time=datetime.fromtimestamp(timestamp, tz=timezone.utc), + ) + ) + + verified_count = ( + verified_entries if self._tlog_threshold == 1 else len(verified_operators) + ) + if verified_count < self._tlog_threshold: + raise VerificationError( + "transparency log threshold not met: " + f"{verified_count} < {self._tlog_threshold}" + ) + + return verified_timestamps + + def _establish_time( + self, + bundle: Bundle, + tlog_timestamps: list[TimestampVerificationResult], + ) -> list[TimestampVerificationResult]: + """Establish verified signing times for bundle verification.""" + verified_timestamps = list(tlog_timestamps) - # If a timestamp from the timestamping service is available, the Verifier MUST - # perform path validation using the timestamp from the Timestamping Service. if bundle.verification_material.timestamp_verification_data: if not self._trusted_root.get_timestamp_authorities(): msg = ( @@ -230,26 +389,6 @@ def _establish_time(self, bundle: Bundle) -> list[TimestampVerificationResult]: timestamp_from_tsa = self._verify_timestamp_authority(bundle) verified_timestamps.extend(timestamp_from_tsa) - # If a timestamp from the Transparency Service is available, the Verifier MUST - # perform path validation using the timestamp from the Transparency Service. - # NOTE: We only include this timestamp if it's accompanied by an inclusion - # promise that cryptographically binds it. We verify the inclusion promise - # itself later, as part of log entry verification. - if ( - timestamp := bundle.log_entry._inner.integrated_time - ) and bundle.log_entry._inner.inclusion_promise: - kv = bundle.log_entry._inner.kind_version - if not (kv.kind in ["dsse", "hashedrekord"] and kv.version == "0.0.1"): - raise VerificationError( - "Integrated time only supported for dsse/hashedrekord 0.0.1 types" - ) - - verified_timestamps.append( - TimestampVerificationResult( - source=TimestampSource.TRANSPARENCY_SERVICE, - time=datetime.fromtimestamp(timestamp, tz=timezone.utc), - ) - ) return verified_timestamps def _verify_chain_at_time( @@ -286,7 +425,10 @@ def _verify_chain_at_time( ) def _verify_common_signing_cert( - self, bundle: Bundle, policy: VerificationPolicy + self, + bundle: Bundle, + policy: VerificationPolicy, + tlog_timestamps: list[TimestampVerificationResult], ) -> None: """ Performs the signing certificate verification steps that are shared between @@ -295,47 +437,33 @@ def _verify_common_signing_cert( Raises `VerificationError` on all failures. """ - # In order to verify an artifact, we need to achieve the following: - # - # 0. Establish a time for the signature. - # 1. Verify that the signing certificate chains to the root of trust - # and is valid at the time of signing. - # 2. Verify the signing certificate's SCT. - # 3. Verify that the signing certificate conforms to the Sigstore - # X.509 profile as well as the passed-in `VerificationPolicy`. - # 4. Verify the inclusion proof and signed checkpoint for the log - # entry. - # 5. Verify the inclusion promise for the log entry, if present. - # 6. Verify the timely insertion of the log entry against the validity - # period for the signing certificate. - # 7. Verify the signature and input against the signing certificate's - # public key. - # 8. Verify the transparency log entry's consistency against the other - # materials, to prevent variants of CVE-2022-36056. + # Transparency log evidence and bundle-entry consistency are verified + # before this method. This ensures that Rekor integrated time is trusted + # only after the supporting log evidence has been authenticated. # - # This method performs steps (0) through (6) above. Its caller - # MUST perform steps (7) and (8) separately, since they vary based on - # the kind of verification being performed (i.e. hashedrekord, DSSE, etc.) + # This method validates the signing certificate against the established + # verified times, checks its SCT and verification policy, and enforces + # its validity period. cert = bundle.signing_certificate - # (0): Establishing a Time for the Signature + # Establish verified signing times. # First, establish verified times for the signature. This is required to # validate the certificate chain, so this step comes first. # These include TSA timestamps and (in the case of rekor v1 entries) # rekor log integrated time. - verified_timestamps = self._establish_time(bundle) + verified_timestamps = self._establish_time(bundle, tlog_timestamps) if len(verified_timestamps) < VERIFIED_TIME_THRESHOLD: raise VerificationError("not enough sources of verified time") - # (1): verify that the signing certificate is signed by the root + # Verify that the signing certificate is signed by the root # certificate and that the signing certificate was valid at the # time of signing. chain: list[Certificate] = [] for vts in verified_timestamps: chain = self._verify_chain_at_time(cert, vts) - # (2): verify the signing certificate's SCT. + # Verify the signing certificate's SCT. try: verify_sct( cert, @@ -345,7 +473,7 @@ def _verify_common_signing_cert( except VerificationError as e: raise VerificationError(f"failed to verify SCT on signing certificate: {e}") - # (3): verify the signing certificate against the Sigstore + # Verify the signing certificate against the Sigstore # X.509 profile and verify against the given `VerificationPolicy`. usage_ext = cert.extensions.get_extension_for_class(KeyUsage) if not usage_ext.value.digital_signature: @@ -359,16 +487,7 @@ def _verify_common_signing_cert( _logger.debug("Successfully verified signing certificate validity...") - # (4): verify the inclusion proof and signed checkpoint for the - # log entry. - # (5): verify the inclusion promise for the log entry, if present. - entry = bundle.log_entry - try: - entry._verify(self._trusted_root.rekor_keyring(KeyringPurpose.VERIFY)) - except VerificationError as exc: - raise VerificationError(f"invalid log entry: {exc}") - - # (6): verify our established times (timestamps or the log integration time) are + # Verify our established times (timestamps or log integration time) are # within signing certificate validity period. for vts in verified_timestamps: if not ( @@ -403,43 +522,22 @@ def verify_dsse( manner. """ - # (1) through (6) are performed by `_verify_common_signing_cert`. - self._verify_common_signing_cert(bundle, policy) - - # (7): verify the bundle's signature and DSSE envelope against the - # signing certificate's public key. envelope = bundle._dsse_envelope if envelope is None: raise VerificationError( "cannot perform DSSE verification on a bundle without a DSSE envelope" ) + tlog_timestamps = self._verify_tlog_entries(bundle) + self._verify_common_signing_cert(bundle, policy, tlog_timestamps) + + # Verify the bundle's signature and DSSE envelope against the signing + # certificate's public key. + signing_key = bundle.signing_certificate.public_key() signing_key = cast(ec.EllipticCurvePublicKey, signing_key) dsse._verify(signing_key, envelope) - # (8): verify the consistency of the log entry's body against - # the other bundle materials. - # Rekor v2 records DSSE envelopes as hashedrekord/0.0.2 entries whose - # digest covers PAE(payloadType, payload) and whose signature.content - # equals envelope.signatures[0].sig (rekor-v2-spec §6.1.4). Rekor v1 - # used a dsse/0.0.1 entry, which is slightly weaker than the - # hashedrekord consistency check: dsse entries record an envelope - # hash that we *cannot* verify (the envelope is uncanonicalized JSON), - # so we manually pick apart the entry body and verify the parts we - # can (payload hash and signature list). - entry = bundle.log_entry - kind = entry._inner.kind_version.kind - version = entry._inner.kind_version.version - if kind == "hashedrekord" and version == "0.0.2": - _validate_hashedrekord_v002_dsse_entry_body(bundle) - elif kind == "dsse" and version == "0.0.1": - _validate_dsse_v001_entry_body(bundle) - else: - raise VerificationError( - f"Unsupported DSSE log entry type: {kind}/{version}" - ) - return (envelope._inner.payload_type, envelope._inner.payload) def verify_artifact( @@ -461,14 +559,14 @@ def verify_artifact( On failure, this method raises `VerificationError`. """ - # (1) through (6) are performed by `_verify_common_signing_cert`. - self._verify_common_signing_cert(bundle, policy) - hashed_input = sha256_digest(input_) bundle_signature = bundle._inner.message_signature if bundle_signature is None: raise VerificationError("Missing bundle message signature") + tlog_timestamps = self._verify_tlog_entries(bundle, hashed_input) + self._verify_common_signing_cert(bundle, policy, tlog_timestamps) + # signature is verified over input digest, but if the bundle documents the digest we still # want to ensure it matches the input digest: if ( @@ -491,29 +589,11 @@ def verify_artifact( _logger.debug("Successfully verified signature...") - # (8): verify the consistency of the log entry's body against - # the other bundle materials (and input being verified). - entry = bundle.log_entry - if entry._inner.kind_version.kind != "hashedrekord": - raise VerificationError( - f"Expected entry type hashedrekord, got {entry._inner.kind_version.kind}" - ) - - if entry._inner.kind_version.version == "0.0.2": - _validate_hashedrekord_v002_entry_body(bundle, hashed_input) - elif entry._inner.kind_version.version == "0.0.1": - _validate_hashedrekord_v001_entry_body(bundle, hashed_input) - else: - raise VerificationError( - f"Unsupported hashedrekord version {entry._inner.kind_version.version}" - ) - -def _validate_dsse_v001_entry_body(bundle: Bundle) -> None: +def _validate_dsse_v001_entry_body(bundle: Bundle, entry: TransparencyLogEntry) -> None: """ Validate the Entry body for dsse v001. """ - entry = bundle.log_entry envelope = bundle._dsse_envelope if envelope is None: raise VerificationError( @@ -550,12 +630,11 @@ def _validate_dsse_v001_entry_body(bundle: Bundle) -> None: def _validate_hashedrekord_v001_entry_body( - bundle: Bundle, hashed_input: Hashed + bundle: Bundle, hashed_input: Hashed, entry: TransparencyLogEntry ) -> None: """ Validate the Entry body for hashedrekord v001. """ - entry = bundle.log_entry expected_body = _hashedrekord_from_parts( bundle.signing_certificate, bundle._inner.message_signature.signature, # type: ignore[union-attr] @@ -570,7 +649,9 @@ def _validate_hashedrekord_v001_entry_body( ) -def _validate_hashedrekord_v002_dsse_entry_body(bundle: Bundle) -> None: +def _validate_hashedrekord_v002_dsse_entry_body( + bundle: Bundle, entry: TransparencyLogEntry +) -> None: """ Validate Entry body for a Rekor v2 DSSE envelope encoded as a hashedrekord/0.0.2 entry (rekor-v2-spec §6.1.4). @@ -582,7 +663,6 @@ def _validate_hashedrekord_v002_dsse_entry_body(bundle: Bundle) -> None: - signature.content = envelope.signatures[0].sig. - signature.verifier = the bundle's signing certificate. """ - entry = bundle.log_entry envelope = bundle._dsse_envelope if envelope is None: raise VerificationError( @@ -621,12 +701,11 @@ def _validate_hashedrekord_v002_dsse_entry_body(bundle: Bundle) -> None: def _validate_hashedrekord_v002_entry_body( - bundle: Bundle, hashed_input: Hashed + bundle: Bundle, hashed_input: Hashed, entry: TransparencyLogEntry ) -> None: """ Validate Entry body for hashedrekord v002. """ - entry = bundle.log_entry if bundle._inner.message_signature is None: raise VerificationError( "invalid hashedrekord log entry: missing message signature" diff --git a/test/unit/internal/test_trust.py b/test/unit/internal/test_trust.py index 4340ee007..551ceee04 100644 --- a/test/unit/internal/test_trust.py +++ b/test/unit/internal/test_trust.py @@ -223,7 +223,10 @@ def test_good(self, asset, file): assert len(root._inner.ctlogs) == 2 assert len(root._inner.timestamp_authorities) == 1 - # only one of the two rekor keys is actually supported + # only usable transparency log instances contribute verifying keys + rekor_tlogs = root._rekor_tlogs(KeyringPurpose.VERIFY) + assert len(rekor_tlogs) == 1 + assert rekor_tlogs[0] is root._inner.tlogs[0] assert len(root.rekor_keyring(KeyringPurpose.VERIFY)._keyring) == 1 assert len(root.ct_keyring(KeyringPurpose.VERIFY)._keyring) == 2 assert root.get_fulcio_certs() is not None diff --git a/test/unit/test_models.py b/test/unit/test_models.py index 0285be8f8..a8c27dc95 100644 --- a/test/unit/test_models.py +++ b/test/unit/test_models.py @@ -138,10 +138,43 @@ def test_invalid_empty_cert_chain(self, signing_bundle): def test_invalid_no_log_entry(self, signing_bundle): with pytest.raises( - InvalidBundle, match="expected exactly one log entry in bundle" + InvalidBundle, match="expected at least one log entry in bundle" ): signing_bundle("bundle_no_log_entry.txt") + def test_multiple_log_entries(self, signing_bundle): + _, bundle = signing_bundle("bundle.txt") + raw = json.loads(bundle.to_json()) + entries = raw["verificationMaterial"]["tlogEntries"] + entries.append(entries[0]) + + bundle = Bundle.from_json(json.dumps(raw)) + + assert len(bundle._log_entries) == 2 + assert bundle.log_entry == bundle._log_entries[0] + + def test_too_many_log_entries(self, signing_bundle): + _, bundle = signing_bundle("bundle.txt") + raw = json.loads(bundle.to_json()) + entry = raw["verificationMaterial"]["tlogEntries"][0] + raw["verificationMaterial"]["tlogEntries"] = [entry] * 33 + + with pytest.raises(InvalidBundle, match="expected at most 32 log entries"): + Bundle.from_json(json.dumps(raw)) + + def test_invalid_additional_log_entry(self, signing_bundle): + _, bundle = signing_bundle("bundle.txt") + raw = json.loads(bundle.to_json()) + entries = raw["verificationMaterial"]["tlogEntries"] + entries.append(json.loads(json.dumps(entries[0]))) + entries[1]["kindVersion"]["version"] = "0.0.3" + + with pytest.raises( + InvalidBundle, + match="Expected log entry version 0.0.1 - 0.0.2", + ): + Bundle.from_json(json.dumps(raw)) + def test_verification_materials_offline_no_checkpoint(self, signing_bundle): with pytest.raises( InvalidBundle, match="entry must contain inclusion proof, with checkpoint" diff --git a/test/unit/verify/test_verifier.py b/test/unit/verify/test_verifier.py index de4de1a3c..c2e43b2e7 100644 --- a/test/unit/verify/test_verifier.py +++ b/test/unit/verify/test_verifier.py @@ -13,6 +13,8 @@ # limitations under the License. +import base64 +import copy import hashlib import json import logging @@ -23,10 +25,11 @@ import rfc3161_client from sigstore_models.trustroot import v1 as trustroot_v1 -from sigstore._internal.trust import CertificateAuthority +from sigstore._internal.trust import CertificateAuthority, KeyringPurpose +from sigstore._utils import sha256_digest from sigstore.dsse import StatementBuilder, Subject from sigstore.errors import CertValidationError, VerificationError -from sigstore.models import Bundle, TrustedRoot +from sigstore.models import Bundle, TransparencyLogEntry, TrustedRoot from sigstore.verify import policy from sigstore.verify.verifier import Verifier @@ -143,10 +146,155 @@ def test_verifier_bundle_offline(signing_bundle, null_policy, filename): verifier.verify_artifact(file.read_bytes(), bundle, null_policy) +def test_verifier_tlog_threshold_one_accepts_one_valid_entry( + signing_bundle, null_policy +): + file, bundle = signing_bundle("bundle.txt") + raw = json.loads(bundle.to_json()) + + extra = copy.deepcopy(raw["verificationMaterial"]["tlogEntries"][0]) + extra["logId"]["keyId"] = base64.b64encode(b"\x01" * 32).decode() + raw["verificationMaterial"]["tlogEntries"].append(extra) + + bundle = Bundle.from_json(json.dumps(raw)) + verifier = Verifier.staging(offline=True) + + verifier.verify_artifact(file.read_bytes(), bundle, null_policy) + + +def test_verifier_rejects_duplicate_tlog_entries(signing_bundle, null_policy): + file, bundle = signing_bundle("bundle.txt") + raw = json.loads(bundle.to_json()) + + raw["verificationMaterial"]["tlogEntries"].append( + copy.deepcopy(raw["verificationMaterial"]["tlogEntries"][0]) + ) + + bundle = Bundle.from_json(json.dumps(raw)) + verifier = Verifier.staging(offline=True) + + with pytest.raises( + VerificationError, + match="duplicate transparency log entry", + ): + verifier.verify_artifact(file.read_bytes(), bundle, null_policy) + + +def test_verifier_tlog_threshold_requires_operator_metadata( + signing_bundle, null_policy +): + file, bundle = signing_bundle("bundle.txt") + verifier = Verifier.staging(offline=True, tlog_threshold=2) + + with pytest.raises( + VerificationError, + match="operator metadata is required", + ): + verifier.verify_artifact(file.read_bytes(), bundle, null_policy) + + +def test_verifier_rejects_invalid_tlog_threshold(): + with pytest.raises( + ValueError, + match="transparency log threshold must be at least 1", + ): + Verifier.staging(offline=True, tlog_threshold=0) + + +def _add_synthetic_second_tlog_entry( + bundle: Bundle, + log_id: bytes, +) -> TransparencyLogEntry: + second_inner = copy.deepcopy(bundle._log_entries[0]._inner) + second_inner.log_id.key_id = log_id + second_inner.log_index = type(second_inner.log_index)( + int(second_inner.log_index) + 1 + ) + + second = TransparencyLogEntry(second_inner) + bundle._log_entries.append(second) + return second + + +def _configure_two_tlog_operators( + verifier: Verifier, + bundle: Bundle, + *, + same_operator: bool, +) -> None: + first_entry_id = bundle._log_entries[0]._inner.log_id.key_id + trusted_tlogs = verifier._trusted_root._rekor_tlogs(KeyringPurpose.VERIFY) + + first_tlog = copy.deepcopy( + next(tlog for tlog in trusted_tlogs if tlog.log_id.key_id == first_entry_id) + ) + second_tlog = copy.deepcopy( + next(tlog for tlog in trusted_tlogs if tlog.log_id.key_id != first_entry_id) + ) + + first_tlog.operator = "operator-a.example" + second_tlog.operator = ( + "operator-a.example" if same_operator else "operator-b.example" + ) + + verifier._trusted_root._inner.tlogs = [first_tlog, second_tlog] + _add_synthetic_second_tlog_entry(bundle, second_tlog.log_id.key_id) + + +def test_verifier_tlog_threshold_counts_distinct_operators( + signing_bundle, null_policy, monkeypatch +): + file, bundle = signing_bundle("bundle.txt") + verifier = Verifier.staging(offline=True, tlog_threshold=2) + _configure_two_tlog_operators( + verifier, + bundle, + same_operator=False, + ) + + # Quorum counting is the behavior under test here. The normal cryptographic + # verification path is covered by the existing verifier tests and by the + # threshold-1 multi-entry test above. + monkeypatch.setattr( + TransparencyLogEntry, + "_verify", + lambda self, keyring: None, + ) + + verifier.verify_artifact(file.read_bytes(), bundle, null_policy) + + +def test_verifier_tlog_threshold_counts_operator_once( + signing_bundle, null_policy, monkeypatch +): + file, bundle = signing_bundle("bundle.txt") + verifier = Verifier.staging(offline=True, tlog_threshold=2) + _configure_two_tlog_operators( + verifier, + bundle, + same_operator=True, + ) + + monkeypatch.setattr( + TransparencyLogEntry, + "_verify", + lambda self, keyring: None, + ) + + with pytest.raises( + VerificationError, + match=r"transparency log threshold not met: 1 < 2", + ): + verifier.verify_artifact(file.read_bytes(), bundle, null_policy) + + def test_verifier_certificate_chain_rejects_invalid_time(signing_bundle): - _, bundle = signing_bundle("bundle.txt") + file, bundle = signing_bundle("bundle.txt") verifier = Verifier.staging(offline=True) - timestamp = verifier._establish_time(bundle)[0] + tlog_timestamps = verifier._verify_tlog_entries( + bundle, sha256_digest(file.read_bytes()) + ) + timestamp = verifier._establish_time(bundle, tlog_timestamps)[0] timestamp.time = datetime(2000, 1, 1, tzinfo=timezone.utc) with pytest.raises(