Offline Rekor bundle generation and verification

README.md

@@ -146,8 +146,9 @@ Verifying:
 <!-- @begin-sigstore-verify-help@ -->
 ```
 usage: sigstore verify [-h] [--certificate FILE] [--signature FILE]
-                       [--cert-email EMAIL] [--cert-oidc-issuer URL]
-                       [--staging] [--rekor-url URL]
+                       [--bundle FILE] [--cert-email EMAIL]
+                       [--cert-oidc-issuer URL] [--offline] [--staging]
+                       [--rekor-url URL]
                        FILE [FILE ...]
 
 positional arguments:
@@ -162,13 +163,17 @@ Verification inputs:
                         used with multiple inputs (default: None)
   --signature FILE      The signature to verify against; not used with
                         multiple inputs (default: None)
+  --bundle FILE         The offline Rekor bundle to verify with; not used with
+                        multiple inputs (default: None)
 
 Extended verification options:
   --cert-email EMAIL    The email address to check for in the certificate's
                         Subject Alternative Name (default: None)
   --cert-oidc-issuer URL
                         The OIDC issuer URL to check for in the certificate's
                         OIDC issuer extension (default: None)
+  --offline             Perform offline Rekor verification using a bundle;
+                        implied by --bundle (default: False)
 
 Sigstore instance options:
   --staging             Use sigstore's staging instances, instead of the

sigstore/_cli.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import argparse
+import json
 import logging
 import os
 import sys
@@ -33,7 +34,11 @@
     STAGING_OAUTH_ISSUER,
     get_identity_token,
 )
-from sigstore._internal.rekor.client import DEFAULT_REKOR_URL, RekorClient
+from sigstore._internal.rekor.client import (
+    DEFAULT_REKOR_URL,
+    RekorClient,
+    RekorEntry,
+)
 from sigstore._sign import Signer
 from sigstore._verify import (
     CertificateVerificationFailure,
@@ -245,6 +250,13 @@ def _parser() -> argparse.ArgumentParser:
         default=os.getenv("SIGSTORE_SIGNATURE"),
         help="The signature to verify against; not used with multiple inputs",
     )
+    input_options.add_argument(
+        "--bundle",
+        metavar="FILE",
+        type=Path,
+        default=os.getenv("SIGSTORE_BUNDLE"),
+        help="The offline Rekor bundle to verify with; not used with multiple inputs",
+    )
 
     verification_options = verify.add_argument_group("Extended verification options")
     verification_options.add_argument(
@@ -261,6 +273,12 @@ def _parser() -> argparse.ArgumentParser:
         default=os.getenv("SIGSTORE_CERT_OIDC_ISSUER"),
         help="The OIDC issuer URL to check for in the certificate's OIDC issuer extension",
     )
+    verification_options.add_argument(
+        "--offline",
+        action="store_true",
+        default=_boolify_env("SIGSTORE_OFFLINE"),
+        help="Perform offline Rekor verification using a bundle; implied by --bundle",
+    )
 
     instance_options = verify.add_argument_group("Sigstore instance options")
     _add_shared_instance_options(instance_options)
@@ -403,10 +421,10 @@ def _sign(args: argparse.Namespace) -> None:
 
 
 def _verify(args: argparse.Namespace) -> None:
-    # Fail if `--certificate` or `--signature` is specified and we have more than one input.
-    if (args.certificate or args.signature) and len(args.files) > 1:
+    # Fail if --certificate, --signature, or --bundle is specified and we have more than one input.
+    if (args.certificate or args.signature or args.bundle) and len(args.files) > 1:
         args._parser.error(
-            "--certificate and --signature can only be used with a single input file"
+            "--certificate, --signature, and --bundle can only be used with a single input file"
         )
 
     # The converse of `sign`: we build up an expected input map and check
@@ -416,24 +434,31 @@ def _verify(args: argparse.Namespace) -> None:
         if not file.is_file():
             args._parser.error(f"Input must be a file: {file}")
 
-        sig, cert = args.signature, args.certificate
+        sig, cert, bundle = args.signature, args.certificate, args.bundle
         if sig is None:
             sig = file.parent / f"{file.name}.sig"
         if cert is None:
             cert = file.parent / f"{file.name}.crt"
+        if bundle is None:
+            bundle = file.parent / f"{file.name}.bundle"
 
         missing = []
         if not sig.is_file():
             missing.append(str(sig))
         if not cert.is_file():
             missing.append(str(cert))
+        if not bundle.is_file() and args.offline:
+            # NOTE: We only produce errors on missing bundle files
+            # if the user has explicitly requested offline-only verification.
+            # Otherwise, we fall back on online verification.
+            missing.append(str(bundle))
 
         if missing:
             args._parser.error(
                 f"Missing verification materials for {(file)}: {', '.join(missing)}"
             )
 
-        input_map[file] = {"cert": cert, "sig": sig}
+        input_map[file] = {"cert": cert, "sig": sig, "bundle": bundle}
 
     if args.staging:
         logger.debug("verify: staging instances requested")
@@ -456,6 +481,12 @@ def _verify(args: argparse.Namespace) -> None:
         logger.debug(f"Using signature from: {inputs['sig']}")
         signature = inputs["sig"].read_bytes().rstrip()
 
+        entry: Optional[RekorEntry] = None
+        if inputs["bundle"].is_file():
+            logger.debug(f"Using offline Rekor bundle from: {inputs['bundle']}")
+            bundle = json.loads(inputs["bundle"].read_text())
+            entry = RekorEntry.from_bundle(bundle)
+
         logger.debug(f"Verifying contents from: {file}")
 
         result = verifier.verify(
@@ -464,6 +495,7 @@ def _verify(args: argparse.Namespace) -> None:
             signature=signature,
             expected_cert_email=args.cert_email,
             expected_cert_oidc_issuer=args.cert_oidc_issuer,
+            offline_rekor_entry=entry,
         )
 
         if result:

sigstore/_internal/merkle.py

@@ -26,7 +26,7 @@
 import struct
 from typing import List, Tuple
 
-from sigstore._internal.rekor import RekorEntry, RekorInclusionProof
+from sigstore._internal.rekor import RekorEntry
 
 
 class InvalidInclusionProofError(Exception):
@@ -91,10 +91,11 @@ def _hash_leaf(leaf: bytes) -> bytes:
     return hashlib.sha256(data).digest()
 
 
-def verify_merkle_inclusion(
-    inclusion_proof: RekorInclusionProof, entry: RekorEntry
-) -> None:
+def verify_merkle_inclusion(entry: RekorEntry) -> None:
     """Verify the Merkle Inclusion Proof for a given Rekor entry"""
+    inclusion_proof = entry.inclusion_proof
+    if inclusion_proof is None:
+        raise InvalidInclusionProofError("Rekor entry has no inclusion proof")
 
     # Figure out which subset of hashes corresponds to the inner and border nodes.
     inner, border = _decomp_inclusion_proof(

sigstore/_internal/rekor/client.py

@@ -29,6 +29,7 @@
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import ec, rsa
 from pydantic import BaseModel, Field, StrictInt, StrictStr, validator
+from securesystemslib.formats import encode_canonical
 
 logger = logging.getLogger(__name__)
 
@@ -48,13 +49,47 @@
 
 @dataclass(frozen=True)
 class RekorEntry:
-    uuid: str
+    uuid: Optional[str]
+    """
+    This entry's unique ID in the Rekor instance it was retrieved from.
+
+    For sharded Rekor deployments, IDs are unique per-shard.
+
+    Not present for `RekorEntry` instances loaded from offline bundles.
+    """
+
     body: str
+    """
+    The base64-encoded body of the Rekor entry.
+    """
+
     integrated_time: int
+    """
+    The UNIX time at which this entry was integrated into the Rekor log.
+    """
+
     log_id: str
+    """
+    The log's ID (as the SHA256 hash of the DER-encoded public key for the log
+    at the time of entry inclusion).
+    """
+
     log_index: int
-    verification: dict
-    raw_data: dict
+    """
+    The index of this entry within the log.
+    """
+
+    inclusion_proof: Optional[RekorInclusionProof]
+    """
+    An optional inclusion proof for this log entry.
+
+    Only present for entries retrieved from online logs.
+    """
+
+    signed_entry_timestamp: str
+    """
+    The base64-encoded Signed Entry Timestamp (SET) for this log entry.
+    """
 
     @classmethod
     def from_response(cls, dict_: Dict[str, Any]) -> RekorEntry:
@@ -71,10 +106,48 @@ def from_response(cls, dict_: Dict[str, Any]) -> RekorEntry:
             integrated_time=entry["integratedTime"],
             log_id=entry["logID"],
             log_index=entry["logIndex"],
-            verification=entry["verification"],
-            raw_data=entry,
+            inclusion_proof=RekorInclusionProof.parse_obj(
+                entry["verification"]["inclusionProof"]
+            ),
+            signed_entry_timestamp=entry["verification"]["signedEntryTimestamp"],
         )
 
+    @classmethod
+    def from_bundle(cls, dict_: Dict[str, Any]) -> RekorEntry:
+        """
+        Creates a `RekorEntry` from an offline Rekor bundle.
+
+        See: <https://github.com/sigstore/cosign/blob/main/specs/SIGNATURE_SPEC.md#properties>
+        """
+
+        payload = dict_["payload"]
+        return cls(
+            uuid=None,
+            body=payload["body"],
+            integrated_time=payload["body"],
+            log_id=payload["body"],
+            log_index=payload["body"],
+            inclusion_proof=None,
+            signed_entry_timestamp=dict_["SignedEntryTimestamp"],
+        )
+
+    def encode_canonical(self) -> bytes:
+        """
+        Returns a base64-encoded, canonicalized JSON (RFC 8785) representation
+        of the Rekor log entry.
+
+        This encoded representation is suitable for verification against
+        the Signed Entry Timestamp.
+        """
+        payload = {
+            "body": self.body,
+            "integratedTime": self.integrated_time,
+            "logID": self.log_id,
+            "logIndex": self.log_index,
+        }
+
+        return encode_canonical(payload).encode()  # type: ignore
+
 
 @dataclass(frozen=True)
 class RekorLogInfo:

sigstore/_internal/set.py

@@ -21,7 +21,6 @@
 import cryptography.hazmat.primitives.asymmetric.ec as ec
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives import hashes
-from securesystemslib.formats import encode_canonical
 
 from sigstore._internal.rekor import RekorClient, RekorEntry
 
@@ -34,27 +33,15 @@ def verify_set(client: RekorClient, entry: RekorEntry) -> None:
     """
     Verify the Signed Entry Timestamp for a given Rekor `entry` using the given `client`.
     """
-
-    # Put together the payload
-    #
-    # This involves removing any non-required fields (verification and attestation) and then
-    # canonicalizing the remaining JSON in accordance with IETF's RFC 8785.
-    raw_data = entry.raw_data.copy()
-    raw_data.pop("verification", None)
-    raw_data.pop("attestation", None)
-    canon_data: bytes = encode_canonical(raw_data).encode()
-
     # Decode the SET field
-    signed_entry_ts: bytes = base64.b64decode(
-        entry.verification["signedEntryTimestamp"].encode()
-    )
+    signed_entry_ts: bytes = base64.b64decode(entry.signed_entry_timestamp)
 
     # Validate the SET
     try:
         client._pubkey.verify(
             signature=signed_entry_ts,
-            data=canon_data,
+            data=entry.encode_canonical(),
             signature_algorithm=ec.ECDSA(hashes.SHA256()),
         )
     except InvalidSignature as inval_sig:
-        raise InvalidSetError from inval_sig
+        raise InvalidSetError("invalid signature") from inval_sig