Implement verification of Merkle inclusion proof (#38)

* sigstore: Initial implementation of verify command

* _verify, _cli: Check the cert for the signer email

* _verify: Add proper checks

* setup: Add PyOpenSSL as a dependency

* _verify: Remove redundant newline

* Fix lint

* Fix comment formatting

* Return None for consistency

* Use click files instead of paths and minor fixes

* treewide: embed key material, refactor to accomodate

* gitignore: only ignore junk in the root

* sigstore: add fulcio root

* Format data correctly for Rekor endpoint

* Fix Rekor API calls in verification

* Fix type checking

* Use the updated Fulcio root key

* Use CTFE key as a file properly and fix type hints

* Add more progress feedback

* Use pydantic and remove custom `from_dict` helper

* Add type hints to verify API

* Add pydantic to setup

* merkle, _verify: Initial attempt at inclusion proof verification

* merkle: Get root calculation working

* Cleanup

* Verify all matching entries

* merkle: Documentation

* Formatting

Co-authored-by: William Woodruff <[email protected]>

Author: tetsuo-cpp

sigstore/_internal/merkle.py

@@ -1,9 +1,115 @@
 """
 Utilities for verifying proof-of-inclusion within Rekor's Merkle Tree.
+
+This code is based off Google's Trillian Merkle Tree implementation which Cosign uses to validate
+Rekor entries.
+
+The data format for the Merkle tree nodes is described in IETF's RFC 6962.
 """
 
-from sigstore._internal.rekor import RekorInclusionProof
+import base64
+import hashlib
+import struct
+from typing import List, Tuple
 
+from sigstore._internal.rekor import RekorEntry, RekorInclusionProof
 
-def verify_merkle_inclusion(inclusion_proof: RekorInclusionProof) -> None:
+
+class InvalidInclusionProofError(Exception):
     pass
+
+
+LEAF_HASH_PREFIX = 0
+NODE_HASH_PREFIX = 1
+
+
+def _decomp_inclusion_proof(index: int, size: int) -> Tuple[int, int]:
+    """
+    Breaks down inclusion proof for a leaf at the specified |index| in a tree of the specified
+    |size| into 2 components. The splitting point between them is where paths to leaves |index| and
+    |size-1| diverge.
+
+    Returns lengths of the bottom and upper proof parts correspondingly. The sum of the two
+    determines the correct length of the inclusion proof.
+    """
+
+    inner = (index ^ (size - 1)).bit_length()
+    border = bin(index >> inner).count("1")
+    return inner, border
+
+
+def _chain_inner(seed: bytes, hashes: List[str], log_index: int) -> bytes:
+    """
+    Computes a subtree hash for a node on or below the tree's right border. Assumes |proof| hashes
+    are ordered from lower levels to upper, and |seed| is the initial subtree/leaf hash on the path
+    located at the specified |index| on its level.
+    """
+
+    for i in range(len(hashes)):
+        h = bytes.fromhex(hashes[i])
+        if (log_index >> i) & 1 == 0:
+            seed = _hash_children(seed, h)
+        else:
+            seed = _hash_children(h, seed)
+    return seed
+
+
+def _chain_border_right(seed: bytes, hashes: List[str]) -> bytes:
+    """
+    Chains proof hashes along tree borders. This differs from inner chaining because |proof|
+    contains only left-side subtree hashes.
+    """
+
+    for h in hashes:
+        seed = _hash_children(bytes.fromhex(h), seed)
+    return seed
+
+
+def _hash_children(lhs: bytes, rhs: bytes) -> bytes:
+    pattern = f"B{len(lhs)}s{len(rhs)}s"
+    data = struct.pack(pattern, NODE_HASH_PREFIX, lhs, rhs)
+    return hashlib.sha256(data).digest()
+
+
+def _hash_leaf(leaf: bytes) -> bytes:
+    pattern = f"B{len(leaf)}s"
+    data = struct.pack(pattern, LEAF_HASH_PREFIX, leaf)
+    return hashlib.sha256(data).digest()
+
+
+def verify_merkle_inclusion(
+    inclusion_proof: RekorInclusionProof, entry: RekorEntry
+) -> None:
+    """Verify the Merkle Inclusion Proof for a given Rekor entry"""
+
+    # Figure out which subset of hashes corresponds to the inner and border nodes.
+    inner, border = _decomp_inclusion_proof(
+        inclusion_proof.log_index, inclusion_proof.tree_size
+    )
+
+    # Check against the number of hashes.
+    if len(inclusion_proof.hashes) != (inner + border):
+        raise InvalidInclusionProofError(
+            f"Inclusion proof has wrong size: expected {inner + border}, got "
+            f"{len(inclusion_proof.hashes)}"
+        )
+
+    # The new entry's hash isn't included in the inclusion proof so we should calculate this
+    # ourselves.
+    leaf_hash: bytes = _hash_leaf(base64.b64decode(entry.body))
+
+    # Now chain the hashes belonging to the inner and border portions. We should expect the
+    # calculated hash to match the root hash.
+    intermediate_result: bytes = _chain_inner(
+        leaf_hash, inclusion_proof.hashes[:inner], inclusion_proof.log_index
+    )
+
+    calc_hash: str = _chain_border_right(
+        intermediate_result, inclusion_proof.hashes[inner:]
+    ).hex()
+
+    if calc_hash != inclusion_proof.root_hash:
+        raise InvalidInclusionProofError(
+            f"Inclusion proof contains invalid root hash: expected {inclusion_proof}, calculated "
+            f"{calc_hash}"
+        )

sigstore/_internal/rekor/_client.py

@@ -11,7 +11,7 @@
 from urllib.parse import urljoin
 
 import requests
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, validator
 
 DEFAULT_REKOR_URL = "https://rekor.sigstore.dev/api/v1/"
 
@@ -52,6 +52,27 @@ class RekorInclusionProof(BaseModel):
     tree_size: int = Field(..., alias="treeSize")
     hashes: List[str] = Field(..., alias="hashes")
 
+    @validator("log_index")
+    def log_index_positive(cls, v):
+        if v < 0:
+            raise ValueError(f"Inclusion proof has invalid log index: {v} < 0")
+        return v
+
+    @validator("tree_size")
+    def tree_size_positive(cls, v):
+        if v < 0:
+            raise ValueError(f"Inclusion proof has invalid tree size: {v} < 0")
+        return v
+
+    @validator("tree_size")
+    def log_index_within_tree_size(cls, v, values, **kwargs):
+        if v <= values["log_index"]:
+            raise ValueError(
+                "Inclusion proof has log index greater than or equal to tree size: "
+                f"{v} <= {values['log_index']}"
+            )
+        return v
+
 
 class RekorClientError(Exception):
     pass

sigstore/_verify.py

@@ -19,7 +19,10 @@
 from cryptography.x509.oid import ExtendedKeyUsageOID
 from OpenSSL.crypto import X509, X509Store, X509StoreContext
 
-from sigstore._internal.merkle import verify_merkle_inclusion
+from sigstore._internal.merkle import (
+    InvalidInclusionProofError,
+    verify_merkle_inclusion,
+)
 from sigstore._internal.rekor import (
     RekorClient,
     RekorEntry,
@@ -139,7 +142,13 @@ def verify(
         inclusion_proof = RekorInclusionProof.parse_obj(
             entry.verification.get("inclusionProof")
         )
-        verify_merkle_inclusion(inclusion_proof)
+        try:
+            verify_merkle_inclusion(inclusion_proof, entry)
+        except InvalidInclusionProofError as inval_inclusion_proof:
+            output(
+                f"Failed to validate Rekor entry's inclusion proof: {inval_inclusion_proof}"
+            )
+            continue
 
         # 5) Verify the Signed Entry Timestamp (SET) supplied by Rekor for this artifact
         try: