Root hash signature verification v2

sigstore/_internal/merkle.py

@@ -26,7 +26,9 @@
 import struct
 from typing import List, Tuple
 
-from sigstore._utils import HexStr
+from sigstore._internal.rekor import RekorClient, SignedCheckpoint
+from sigstore._internal.rekor.checkpoint import InvalidSignedNote
+from sigstore._utils import HexStr, KeyID
 from sigstore.transparency import LogEntry
 
 
@@ -133,3 +135,38 @@ def verify_merkle_inclusion(entry: LogEntry) -> None:
             f"Inclusion proof contains invalid root hash: expected {inclusion_proof}, calculated "
             f"{calc_hash}"
         )
+
+
+def verify_checkpoint(client: RekorClient, entry: LogEntry) -> None:
+    """
+    Verify the inclusion proof's checkpoint.
+    """
+
+    inclusion_proof = entry.inclusion_proof
+    if inclusion_proof is None:
+        raise InvalidInclusionProofError("Rekor entry has no inclusion proof")
+    if inclusion_proof.checkpoint is None:
+        return
+
+    # verififcaiton occurs in two stages:
+    # 1) verify the signature on the checkpoint
+    # 2) verify the root hash in the checkpoint matches the root hash from the inclusion proof.
+
+    try:
+        signed_checkpoint = SignedCheckpoint.from_text(inclusion_proof.checkpoint)
+    except ValueError as e:
+        raise InvalidInclusionProofError("Failed to parse checkpoint") from e
+
+    try:
+        signed_checkpoint.signed_note.verify(client, KeyID(bytes.fromhex(entry.log_id)))
+    except InvalidSignedNote as e:
+        raise InvalidInclusionProofError("Failed to verify checkpoint signature") from e
+
+    checkpoint_hash = signed_checkpoint.checkpoint.log_hash
+    root_hash = inclusion_proof.root_hash
+
+    if checkpoint_hash != root_hash:
+        raise InvalidInclusionProofError(
+            "Inclusion proof contains invalid root hash signature: ",
+            f"expected {str(checkpoint_hash)} got {str(root_hash)}",
+        )

sigstore/_internal/rekor/__init__.py

@@ -16,6 +16,7 @@
 APIs for interacting with Rekor.
 """
 
+from .checkpoint import SignedCheckpoint
 from .client import RekorClient
 
-__all__ = ["RekorClient"]
+__all__ = ["RekorClient", "SignedCheckpoint"]

sigstore/_internal/rekor/checkpoint.py

@@ -0,0 +1,202 @@
+# 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.
+
+"""
+Rekor Checkpoint machinery.
+"""
+
+from __future__ import annotations
+
+import base64
+import re
+import struct
+from dataclasses import dataclass
+from typing import List
+
+from pydantic import BaseModel, Field, StrictStr
+
+from sigstore._internal.keyring import KeyringSignatureError
+from sigstore._internal.rekor.client import RekorClient
+from sigstore._utils import KeyID
+
+
+@dataclass(frozen=True)
+class RekorSignature:
+    """
+    Represents a `RekorSignature` containing:
+    - the name of the signature, e.g. "rekor.sigstage.dev"
+    - the signature hash
+    - the base64 signature
+    """
+
+    # FIXME(jl): this does not feel like a de novo definition...
+    # does this exist already in sigstore-python (or its depenencices)?
+    name: str
+    sig_hash: bytes
+    signature: bytes
+
+
+class LogCheckpoint(BaseModel):
+    """
+    Represents a Rekor `LogCheckpoint` containing:
+    - an origin, e.g. "rekor.sigstage.dev - 8050909264565447525"
+    - the size of the log,
+    - the hash of the log,
+    - and any ancillary contants, e.g. "Timestamp: 1679349379012118479"
+    see: https://github.com/transparency-dev/formats/blob/main/log/README.md
+    """
+
+    origin: StrictStr
+    log_size: int
+    log_hash: StrictStr
+    other_content: List[str]
+
+    @classmethod
+    def from_text(cls, text: str) -> LogCheckpoint:
+        """
+        Serialize from the text header ("note") of a SignedNote.
+        """
+
+        lines = text.strip().split("\n")
+        if len(lines) < 4:
+            raise ValueError("Malformed LogCheckpoint: too few items in header!")
+
+        origin = lines[0]
+        if len(origin) == 0:
+            raise ValueError("Malformed LogCheckpoint: empty origin!")
+
+        log_size = int(lines[1])
+        root_hash = base64.b64decode(lines[2]).hex()
+
+        return LogCheckpoint(
+            origin=origin,
+            log_size=log_size,
+            log_hash=root_hash,
+            other_content=lines[3:],
+        )
+
+    @classmethod
+    def to_text(self) -> str:
+        """
+        Serialize a `LogCheckpoint` into text format.
+        See class definition for a prose description of the format.
+        """
+        return "\n".join(
+            [
+                self.origin,
+                str(self.log_size),
+                self.log_hash,
+            ]
+            + self.other_content
+        )
+
+
+class InvalidSignedNote(Exception):
+    """
+    Raised during SignedNote verification if invalid in some way.
+    """
+
+    pass
+
+
+@dataclass(frozen=True)
+class SignedNote:
+    """
+    Represents a signed `Note` containing a note and its corresponding list of signatures.
+    """
+
+    note: StrictStr = Field(..., alias="note")
+    signatures: list[RekorSignature] = Field(..., alias="signatures")
+
+    @classmethod
+    def from_text(cls, text: str) -> SignedNote:
+        """
+        Serialize from a bundled text 'note'.
+
+        A note contains:
+        - a name, a string associated with the signer,
+        - a separator blank line,
+        - and signature(s), each signature takes the form
+            `\u2014 NAME SIGNATURE\n`
+          (where \u2014 == em dash).
+
+        An adaptation of the Rekor's `UnmarshalText`:
+        https://github.com/sigstore/rekor/blob/4b1fa6661cc6dfbc844b4c6ed9b1f44e7c5ae1c0/pkg/util/signed_note.go#L141
+        """
+
+        separator: str = "\n\n"
+        if text.count(separator) != 1:
+            raise ValueError(
+                "Note must contain one blank line, deliniating the text from the signature block"
+            )
+        split = text.index(separator)
+
+        header: str = text[: split + 1]
+        data: str = text[split + len(separator) :]
+
+        if len(data) == 0:
+            raise ValueError("Malformed Note: must contain at least one signature!")
+        if data[-1] != "\n":
+            raise ValueError("Malformed Note: data section must end with newline!")
+
+        sig_parser = re.compile(r"\u2014 (\S+) (\S+)\n")
+        signatures: list[RekorSignature] = []
+        for name, signature in re.findall(sig_parser, data):
+            signature_bytes: bytes = base64.b64decode(signature)
+            if len(signature_bytes) < 5:
+                raise ValueError("Malformed Note: signature contains too few bytes")
+
+            signature = RekorSignature(
+                name=name,
+                # FIXME(jl): In Go, construct a big-endian UInt32 from 4 bytes. Is this equivalent?
+                sig_hash=struct.unpack(">4s", signature_bytes[0:4])[0],
+                signature=base64.b64encode(signature_bytes[4:]),
+            )
+            signatures.append(signature)
+
+        return cls(note=header, signatures=signatures)
+
+    def verify(self, client: RekorClient, key_id: KeyID) -> None:
+        note = str.encode(self.note)
+
+        for sig in self.signatures:
+            if sig.sig_hash != key_id[:4]:
+                raise InvalidSignedNote("sig_hash hint does not match expected key_id")
+
+            try:
+                client._rekor_keyring.verify(
+                    key_id=key_id, signature=base64.b64decode(sig.signature), data=note
+                )
+            except KeyringSignatureError as sig_err:
+                raise InvalidSignedNote("invalid signature") from sig_err
+
+
+@dataclass(frozen=True)
+class SignedCheckpoint:
+    """
+    Represents a *signed* `Checkpoint`: a `LogCheckpoint` and its corresponding *signed* `Note`.
+    """
+
+    signed_note: SignedNote
+    checkpoint: LogCheckpoint
+
+    @classmethod
+    def from_text(cls, text: str) -> SignedCheckpoint:
+        """
+        Create a new `SignedCheckpoint` from the text representation.
+        """
+
+        signed_note = SignedNote.from_text(text)
+        checkpoint = LogCheckpoint.from_text(signed_note.note)
+        return cls(signed_note=signed_note, checkpoint=checkpoint)

sigstore/sign.py

@@ -60,6 +60,7 @@
     X509CertificateChain,
 )
 from sigstore_protobuf_specs.dev.sigstore.rekor.v1 import (
+    Checkpoint,
     InclusionPromise,
     InclusionProof,
     KindVersion,
@@ -238,6 +239,12 @@ def _to_bundle(self) -> Bundle:
                 hashes=[
                     bytes.fromhex(h) for h in self.log_entry.inclusion_proof.hashes
                 ],
+                checkpoint=Checkpoint(
+                    envelope=self.log_entry.inclusion_proof.checkpoint or ""
+                )
+                # FIXME(jl): checkpoint should be a serialzied field here.
+                # this causes tests depending on the commited `bundle.txt.sigstore` to fail;
+                # the bundle doesn't contain the `"checkpoint"` field.
             )
 
         tlog_entry = TransparencyLogEntry(

sigstore/transparency.py

@@ -90,7 +90,6 @@ def _from_response(cls, dict_: dict[str, Any]) -> LogEntry:
             raise ValueError("Received multiple entries in response")
 
         uuid, entry = entries[0]
-
         return LogEntry(
             uuid=uuid,
             body=entry["body"],
@@ -125,10 +124,11 @@ class LogInclusionProof(BaseModel):
     Represents an inclusion proof for a transparency log entry.
     """
 
+    checkpoint: Optional[StrictStr] = Field(..., alias="checkpoint")
+    hashes: List[StrictStr] = Field(..., alias="hashes")
     log_index: StrictInt = Field(..., alias="logIndex")
     root_hash: StrictStr = Field(..., alias="rootHash")
     tree_size: StrictInt = Field(..., alias="treeSize")
-    hashes: List[StrictStr] = Field(..., alias="hashes")
 
     class Config:
         allow_population_by_field_name = True

sigstore/verify/models.py

@@ -239,12 +239,15 @@ def from_bundle(
                 f"expected exactly one log entry, got {len(tlog_entries)}"
             )
         tlog_entry = tlog_entries[0]
+        inclusion_proof = tlog_entry.inclusion_proof
+        checkpoint = inclusion_proof.checkpoint
 
         inclusion_proof = LogInclusionProof(
-            log_index=tlog_entry.inclusion_proof.log_index,
-            root_hash=tlog_entry.inclusion_proof.root_hash.hex(),
-            tree_size=tlog_entry.inclusion_proof.tree_size,
-            hashes=[h.hex() for h in tlog_entry.inclusion_proof.hashes],
+            checkpoint=checkpoint.envelope if checkpoint.envelope != "" else None,
+            hashes=[h.hex() for h in inclusion_proof.hashes],
+            log_index=inclusion_proof.log_index,
+            root_hash=inclusion_proof.root_hash.hex(),
+            tree_size=inclusion_proof.tree_size,
         )
         entry = LogEntry(
             uuid=None,

sigstore/verify/verifier.py

@@ -38,6 +38,7 @@
 
 from sigstore._internal.merkle import (
     InvalidInclusionProofError,
+    verify_checkpoint,
     verify_merkle_inclusion,
 )
 from sigstore._internal.rekor.client import RekorClient
@@ -254,6 +255,12 @@ def verify(
                 return VerificationFailure(
                     reason=f"invalid Rekor inclusion proof: {exc}"
                 )
+
+            try:
+                verify_checkpoint(self._rekor, entry)
+            except InvalidInclusionProofError as exc:
+                return VerificationFailure(reason=f"invalid Rekor root hash: {exc}")
+
         else:
             logger.debug(
                 "offline verification requested: skipping Merkle inclusion proof"

test/unit/assets/bundle.txt.crt

@@ -0,0 +1,19 @@
+-----BEGIN CERTIFICATE-----
+MIIC5zCCAmygAwIBAgIUJ3vpewdf6e91rgjqCqagstF4qn8wCgYIKoZIzj0EAwMw
+NzEVMBMGA1UEChMMc2lnc3RvcmUuZGV2MR4wHAYDVQQDExVzaWdzdG9yZS1pbnRl
+cm1lZGlhdGUwHhcNMjMwNDI2MDAyMTA4WhcNMjMwNDI2MDAzMTA4WjAAMHYwEAYH
+KoZIzj0CAQYFK4EEACIDYgAE2sd6+lOBcn5MXtnbwca7zcwpprl7GUZiKTO9IWpA
+UfVTtx+BXGHQCRwsFy/d7dLlf4hurIqhzMD5yaC2kcU9/8c9G55JyBXF8Dx5SQm9
+y2rPWFIdm29Ql9A3I3yyEFyPo4IBbjCCAWowDgYDVR0PAQH/BAQDAgeAMBMGA1Ud
+JQQMMAoGCCsGAQUFBwMDMB0GA1UdDgQWBBTlaUfjpiXGhBP3hOCW0JJZDSPxgzAf
+BgNVHSMEGDAWgBRxhjCmFHxib/n31vQFGn9f/+tvrDAYBgNVHREBAf8EDjAMgQph
+QHRueS50b3duMCwGCisGAQQBg78wAQEEHmh0dHBzOi8vZ2l0aHViLmNvbS9sb2dp
+bi9vYXV0aDAuBgorBgEEAYO/MAEIBCAMHmh0dHBzOi8vZ2l0aHViLmNvbS9sb2dp
+bi9vYXV0aDCBigYKKwYBBAHWeQIEAgR8BHoAeAB2ACswvNxoiMni4dgmKV50H0g5
+MZYC8pwzy15DQP6yrIZ6AAABh7rveBsAAAQDAEcwRQIhAKOZPMN9Q9qO1HXigHBP
+t+Ic16yy2Zgv2KQ23i5WLj16AiAzrFpuayGXdoK+hYePl9dEeXjG/vB2jK/E3sEs
+IrXtETAKBggqhkjOPQQDAwNpADBmAjEAgmhg80mI/Scr0isBnD5FYXZ8WxA8tnBB
+Pmdf4aNGForGazGXaFQVPXgBVPv+YGI/AjEA0QzPC5dHD/WWXW2GbEC4dpwFk8OG
+RkiExMOy/+CqabbVg+/lx1N9VGBTlUTft45d
+-----END CERTIFICATE-----
+

test/unit/assets/bundle.txt.sig

@@ -0,0 +1 @@
+MGUCMQCOOJqTY6XWgB64izK2WVP07b0SG9M5WPCwKhfTPwMvtsgUi8KeRGwQkvvLYbKHdqUCMEbOXFG0NMqEQxWVb6rmGnexdADuGf6Jl8qAC8tn67p3QfVoXzMvFA61PzxwVwvb8g==