Skip to content

Merkle Proof Formats for Weber

Notice: This document is a work-in-progress for researchers and implementers of the Weber protocol.

Table of contents

Introduction

Merkle proofs are a fundamental component of the Weber protocol, enabling efficient verification of data inclusion and state transitions. This document outlines the various Merkle proof formats used in Weber, their applications, and implementation details.

Weber's proof system builds upon established Merkle tree techniques while introducing optimizations for: - Reduced proof sizes for bandwidth-constrained environments - Efficient on-chain verification for cross-chain applications - Support for complex state transitions in the Weber consensus protocol - Light client synchronization with minimal computational overhead

The proof formats described in this document are used throughout the Weber protocol, from basic transaction verification to complex state transition proofs in the beacon chain.

Helper functions

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def get_power_of_two_ceil(x: int) -> int:
    """
    Get the power of 2 for given input, or the closest higher power of 2 if the input is not a power of 2.
    Commonly used for "how many nodes do I need for a bottom tree layer fitting x elements?"
    Example: 0->1, 1->1, 2->2, 3->4, 4->4, 5->8, 6->8, 7->8, 8->8, 9->16.
    """
    if x <= 1:
        return 1
    elif x == 2:
        return 2
    else:
        return 2 * get_power_of_two_ceil((x + 1) // 2)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
def get_power_of_two_floor(x: int) -> int:
    """
    Get the power of 2 for given input, or the closest lower power of 2 if the input is not a power of 2.
    The zero case is a placeholder and not used for math with generalized indices.
    Commonly used for "what power of two makes up the root bit of the generalized index?"
    Example: 0->1, 1->1, 2->2, 3->2, 4->4, 5->4, 6->4, 7->4, 8->8, 9->8
    """
    if x <= 1:
        return 1
    if x == 2:
        return x
    else:
        return 2 * get_power_of_two_floor(x // 2)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def count_trailing_zeros(x: int) -> int:
    """
    Count the number of trailing zero bits in an integer.
    Used in Weber's optimized proof path computation.
    """
    if x == 0:
        return 0
    count = 0
    while (x & 1) == 0:
        count += 1
        x >>= 1
    return count

Generalized Merkle tree index

In a binary Merkle tree, we define a "generalized index" of a node as 2**depth + index. Visually, this looks as follows:

1
2
3
4
    1
 2     3
4 5   6 7
   ...

Note that the generalized index has the convenient property that the two children of node k are 2k and 2k+1, and also that it equals the position of a node in the linear representation of the Merkle tree that's computed by this function:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
def merkle_tree(leaves: Sequence[Bytes32]) -> Sequence[Bytes32]:
    """
    Return an array representing the tree nodes by generalized index: 
    [0, 1, 2, 3, 4, 5, 6, 7], where each layer is a power of 2. The 0 index is ignored. The 1 index is the root.
    The result will be twice the size as the padded bottom layer for the input leaves.
    """
    bottom_length = get_power_of_two_ceil(len(leaves))
    o = [Bytes32()] * bottom_length + list(leaves) + [Bytes32()] * (bottom_length - len(leaves))
    for i in range(bottom_length - 1, 0, -1):
        o[i] = hash(o[i * 2] + o[i * 2 + 1])
    return o

We define a custom type GeneralizedIndex as a Python integer type in this document. It can be represented as a Bitvector/Bitlist object as well.

We will define Merkle proofs in terms of generalized indices.

SSZ object to index

We can describe the hash tree of any SSZ object, rooted in hash_tree_root(object), as a binary Merkle tree whose depth may vary. For example, an object {x: bytes32, y: List[uint64]} would look as follows:

1
2
3
4
5
6
7
8
     root
    /    \
   x    y_root
        /    \
y_data_root  len(y)
    / \
   /\ /\
  .......

We can now define a concept of a "path", a way of describing a function that takes as input an SSZ object and outputs some specific (possibly deeply nested) member. For example, foo -> foo.x is a path, as are foo -> len(foo.y) and foo -> foo.y[5].w. We'll describe paths as lists, which can have two representations. In "human-readable form", they are ["x"], ["y", "__len__"] and ["y", 5, "w"] respectively. In "encoded form", they are lists of uint64 values, in these cases (assuming the fields of foo in order are x then y, and w is the first field of y[i]) [0], [1, 2**64-1], [1, 5, 0]. We define SSZVariableName as the member variable name string, i.e., a path is presented as a sequence of integers and SSZVariableName.

1
2
3
4
5
6
7
8
def item_length(typ: SSZType) -> int:
    """
    Return the number of bytes in a basic type, or 32 (a full hash) for compound types.
    """
    if issubclass(typ, BasicValue):
        return typ.byte_len
    else:
        return 32

1
2
3
4
5
6
7
def get_elem_type(typ: Union[BaseBytes, BaseList, Container],
                  index_or_variable_name: Union[int, SSZVariableName]) -> SSZType:
    """
    Return the type of the element of an object of the given type with the given index
    or member variable name (eg. `7` for `x[7]`, `"foo"` for `x.foo`)
    """
    return typ.get_fields()[index_or_variable_name] if issubclass(typ, Container) else typ.elem_type

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
def chunk_count(typ: SSZType) -> int:
    """
    Return the number of hashes needed to represent the top-level elements in the given type
    (eg. `x.foo` or `x[7]` but not `x[7].bar` or `x.foo.baz`). In all cases except lists/vectors
    of basic types, this is simply the number of top-level elements, as each element gets one
    hash. For lists/vectors of basic types, it is often fewer because multiple basic elements
    can be packed into one 32-byte chunk.
    """
    # typ.length describes the limit for list types, or the length for vector types.
    if issubclass(typ, BasicValue):
        return 1
    elif issubclass(typ, Bits):
        return (typ.length + 255) // 256
    elif issubclass(typ, Elements):
        return (typ.length * item_length(typ.elem_type) + 31) // 32
    elif issubclass(typ, Container):
        return len(typ.get_fields())
    else:
        raise Exception(f"Type not supported: {typ}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
def get_item_position(typ: SSZType, index_or_variable_name: Union[int, SSZVariableName]) -> Tuple[int, int, int]:
    """
    Return three variables:
        (i) the index of the chunk in which the given element of the item is represented;
        (ii) the starting byte position within the chunk;
        (iii) the ending byte position within the chunk.
    For example: for a 6-item list of uint64 values, index=2 will return (0, 16, 24), index=5 will return (1, 8, 16)
    """
    if issubclass(typ, Elements):
        index = int(index_or_variable_name)
        start = index * item_length(typ.elem_type)
        return start // 32, start % 32, start % 32 + item_length(typ.elem_type)
    elif issubclass(typ, Container):
        variable_name = index_or_variable_name
        return typ.get_field_names().index(variable_name), 0, item_length(get_elem_type(typ, variable_name))
    else:
        raise Exception("Only lists/vectors/containers supported")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
def get_generalized_index(typ: SSZType, *path: PyUnion[int, SSZVariableName]) -> GeneralizedIndex:
    """
    Converts a path (eg. `[7, "foo", 3]` for `x[7].foo[3]`, `[12, "bar", "__len__"]` for
    `len(x[12].bar)`) into the generalized index representing its position in the Merkle tree.
    """
    root = GeneralizedIndex(1)
    for p in path:
        assert not issubclass(typ, BasicValue)  # If we descend to a basic type, the path cannot continue further
        if p == '__len__':
            typ = uint64
            assert issubclass(typ, (List, ByteList))
            root = GeneralizedIndex(root * 2 + 1)
        else:
            pos, _, _ = get_item_position(typ, p)
            base_index = (GeneralizedIndex(2) if issubclass(typ, (List, ByteList)) else GeneralizedIndex(1))
            root = GeneralizedIndex(root * base_index * get_power_of_two_ceil(chunk_count(typ)) + pos)
            typ = get_elem_type(typ, p)
    return root

Helpers for generalized indices

Usage note: functions outside this section should manipulate generalized indices using only functions inside this section. This is to make it easier for developers to implement generalized indices with underlying representations other than bigints.

concat_generalized_indices

1
2
3
4
5
6
7
8
9
def concat_generalized_indices(*indices: GeneralizedIndex) -> GeneralizedIndex:
    """
    Given generalized indices i1 for A -> B, i2 for B -> C .... i_n for Y -> Z, returns
    the generalized index for A -> Z.
    """
    o = GeneralizedIndex(1)
    for i in indices:
        o = GeneralizedIndex(o * get_power_of_two_floor(i) + (i - get_power_of_two_floor(i)))
    return o

get_generalized_index_length

1
2
3
4
5
def get_generalized_index_length(index: GeneralizedIndex) -> int:
    """
    Return the length of a path represented by a generalized index.
    """
    return int(log2(index))

get_generalized_index_bit

1
2
3
4
5
def get_generalized_index_bit(index: GeneralizedIndex, position: int) -> bool:
    """
    Return the given bit of a generalized index.
    """
    return (index & (1 << position)) > 0

generalized_index_sibling

1
2
def generalized_index_sibling(index: GeneralizedIndex) -> GeneralizedIndex:
    return GeneralizedIndex(index ^ 1)

generalized_index_child

1
2
def generalized_index_child(index: GeneralizedIndex, right_side: bool) -> GeneralizedIndex:
    return GeneralizedIndex(index * 2 + right_side)

generalized_index_parent

1
2
def generalized_index_parent(index: GeneralizedIndex) -> GeneralizedIndex:
    return GeneralizedIndex(index // 2)

generalized_index_ancestor

1
2
3
4
5
6
7
8
def generalized_index_ancestor(index: GeneralizedIndex, generations: int) -> GeneralizedIndex:
    """
    Weber-specific function to find an ancestor node n generations up.
    More efficient than calling generalized_index_parent repeatedly.
    """
    if generations == 0:
        return index
    return GeneralizedIndex(index >> generations)

Merkle multiproofs

We define a Merkle multiproof as a minimal subset of nodes in a Merkle tree needed to fully authenticate that a set of nodes actually are part of a Merkle tree with some specified root, at a particular set of generalized indices. For example, here is the Merkle multiproof for positions 0, 1, 6 in an 8-node Merkle tree (i.e. generalized indices 8, 9, 14):

1
2
3
4
       .
   .       .
 .   *   *   .
x x . . . . x *

. are unused nodes, * are used nodes, x are the values we are trying to prove. Notice how despite being a multiproof for 3 values, it requires only 3 auxiliary nodes, only one node more than would be required to prove a single value. Normally the efficiency gains are not quite that extreme, but the savings relative to individual Merkle proofs are still significant. As a rule of thumb, a multiproof for k nodes at the same level of an n-node tree has size k * (n/k + log(n/k)).

First, we provide a method for computing the generalized indices of the auxiliary tree nodes that a proof of a given set of generalized indices will require:

1
2
3
4
5
6
7
8
9
def get_branch_indices(tree_index: GeneralizedIndex) -> Sequence[GeneralizedIndex]:
    """
    Get the generalized indices of the sister chunks along the path from the chunk with the
    given tree index to the root.
    """
    o = [generalized_index_sibling(tree_index)]
    while o[-1] > 1:
        o.append(generalized_index_sibling(generalized_index_parent(o[-1])))
    return o[:-1]

1
2
3
4
5
6
7
8
9
def get_path_indices(tree_index: GeneralizedIndex) -> Sequence[GeneralizedIndex]:
    """
    Get the generalized indices of the chunks along the path from the chunk with the
    given tree index to the root.
    """
    o = [tree_index]
    while o[-1] > 1:
        o.append(generalized_index_parent(o[-1]))
    return o[:-1]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
def get_helper_indices(indices: Sequence[GeneralizedIndex]) -> Sequence[GeneralizedIndex]:
    """
    Get the generalized indices of all "extra" chunks in the tree needed to prove the chunks with the given
    generalized indices. Note that the decreasing order is chosen deliberately to ensure equivalence to the
    order of hashes in a regular single-item Merkle proof in the single-item case.
    """
    all_helper_indices: Set[GeneralizedIndex] = set()
    all_path_indices: Set[GeneralizedIndex] = set()
    for index in indices:
        all_helper_indices = all_helper_indices.union(set(get_branch_indices(index)))
        all_path_indices = all_path_indices.union(set(get_path_indices(index)))

    return sorted(all_helper_indices.difference(all_path_indices), reverse=True)

Now we provide the Merkle proof verification functions. First, for single item proofs:

1
2
3
4
5
6
7
8
def calculate_merkle_root(leaf: Bytes32, proof: Sequence[Bytes32], index: GeneralizedIndex) -> Root:
    assert len(proof) == get_generalized_index_length(index)
    for i, h in enumerate(proof):
        if get_generalized_index_bit(index, i):
            leaf = hash(h + leaf)
        else:
            leaf = hash(leaf + h)
    return leaf

1
2
def verify_merkle_proof(leaf: Bytes32, proof: Sequence[Bytes32], index: GeneralizedIndex, root: Root) -> bool:
    return calculate_merkle_root(leaf, proof, index) == root

Now for multi-item proofs:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def calculate_multi_merkle_root(leaves: Sequence[Bytes32],
                                proof: Sequence[Bytes32],
                                indices: Sequence[GeneralizedIndex]) -> Root:
    assert len(leaves) == len(indices)
    helper_indices = get_helper_indices(indices)
    assert len(proof) == len(helper_indices)
    objects = {
        **{index: node for index, node in zip(indices, leaves)},
        **{index: node for index, node in zip(helper_indices, proof)}
    }
    keys = sorted(objects.keys(), reverse=True)
    pos = 0
    while pos < len(keys):
        k = keys[pos]
        if k in objects and k ^ 1 in objects and k // 2 not in objects:
            objects[GeneralizedIndex(k // 2)] = hash(
                objects[GeneralizedIndex((k | 1) ^ 1)] +
                objects[GeneralizedIndex(k | 1)]
            )
            keys.append(GeneralizedIndex(k // 2))
        pos += 1
    return objects[GeneralizedIndex(1)]

1
2
3
4
5
def verify_merkle_multiproof(leaves: Sequence[Bytes32],
                             proof: Sequence[Bytes32],
                             indices: Sequence[GeneralizedIndex],
                             root: Root) -> bool:
    return calculate_multi_merkle_root(leaves, proof, indices) == root

Note that the single-item proof is a special case of a multi-item proof; a valid single-item proof verifies correctly when put into the multi-item verification function (making the natural trivial changes to input arguments, index -> [index] and leaf -> [leaf]). Note also that calculate_merkle_root and calculate_multi_merkle_root can be used independently to compute the new Merkle root of a proof with leaves updated.

Weber optimized proof formats

Weber introduces several optimizations to reduce proof sizes and computational overhead for various use cases.

Compressed multiproofs

Weber introduces an advanced compression technique for Merkle multiproofs that significantly reduces proof size compared to standard approaches.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def compress_proof(proof: Sequence[Bytes32], indices: Sequence[GeneralizedIndex]) -> Tuple[Sequence[Bytes32], bytes]:
    """
    Compress a multiproof by identifying common subtrees and structural patterns.
    Returns the compressed proof elements and a compression descriptor.

    In Weber, certain common proof patterns can be compressed with specialized encoding.
    """
    # Sort indices to identify adjacent leaves
    sorted_indices = sorted(indices)

    # Find common subtrees (siblings that can be omitted)
    common_parents = set()
    for i in range(len(sorted_indices) - 1):
        if generalized_index_parent(sorted_indices[i]) == generalized_index_parent(sorted_indices[i + 1]):
            common_parents.add(generalized_index_parent(sorted_indices[i]))

    # Create compression descriptor
    # Format: 1 byte version + N bytes describing the compression pattern
    descriptor = bytearray([1])  # Version 1 of compression algorithm

    # Encode compression pattern (which nodes can be reconstructed)
    omitted_indices = []
    compressed_proof = []
    original_helper_indices = get_helper_indices(indices)

    for i, node in enumerate(proof):
        index = original_helper_indices[i]
        # Check if this node can be omitted (part of a common subtree)
        if generalized_index_parent(index) in common_parents:
            omitted_indices.append(i)
            descriptor.append(1)  # Mark as omitted
        else:
            compressed_proof.append(node)
            descriptor.append(0)  # Mark as included

    # Add metadata about compression ratio to descriptor
    compression_ratio = len(proof) / max(1, len(compressed_proof))
    descriptor.extend(struct.pack("f", compression_ratio))

    return compressed_proof, bytes(descriptor)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def decompress_proof(compressed_proof: Sequence[Bytes32], 
                     descriptor: bytes, 
                     indices: Sequence[GeneralizedIndex]) -> Sequence[Bytes32]:
    """
    Reconstruct a full proof from its compressed form.
    """
    # Parse descriptor
    version = descriptor[0]
    if version != 1:
        raise ValueError(f"Unsupported compression version: {version}")

    # Extract compression pattern
    pattern = descriptor[1:-4]  # Skip version byte and compression ratio

    # Calculate original helper indices
    helper_indices = get_helper_indices(indices)

    # Reconstruct full proof
    full_proof = []
    compressed_idx = 0

    for i, included in enumerate(pattern):
        if i >= len(helper_indices):
            break

        if included == 0:  # Node is included in compressed proof
            full_proof.append(compressed_proof[compressed_idx])
            compressed_idx += 1
        else:  # Node needs to be reconstructed
            # Find sibling information
            idx = helper_indices[i]
            parent_idx = generalized_index_parent(idx)

            # This is a simplified reconstruction - actual implementation would
            # need to reconstruct based on the known leaves and other proof elements
            # using the multiproof algorithm backward
            reconstructed_hash = bytes([0] * 32)  # Placeholder
            full_proof.append(reconstructed_hash)

    return full_proof

Incremental verification

Weber supports incremental proof verification with optimal efficiency for blockchain state updates:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
def update_merkle_proof(original_leaf: Bytes32, 
                        new_leaf: Bytes32, 
                        proof: Sequence[Bytes32], 
                        index: GeneralizedIndex) -> Root:
    """
    Calculate the new Merkle root when a leaf changes, without recomputing the entire tree.
    Particularly useful for state transition verification in Weber.
    """
    # The basic approach simply recalculates the path with the new leaf
    return calculate_merkle_root(new_leaf, proof, index)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def batch_update_merkle_proof(original_leaves: Sequence[Bytes32],
                             new_leaves: Sequence[Bytes32],
                             proof: Sequence[Bytes32],
                             indices: Sequence[GeneralizedIndex]) -> Root:
    """
    Weber-specific optimization for updating multiple leaves simultaneously.
    This is more efficient than updating leaves one by one for related state changes.
    """
    assert len(original_leaves) == len(new_leaves) == len(indices)

    # First check if we can use the optimized batch update algorithm
    # (when leaves are close together in the tree)
    can_optimize = True
    base_depth = None

    for idx in indices:
        depth = get_generalized_index_length(idx)
        if base_depth is None:
            base_depth = depth
        elif depth != base_depth:
            can_optimize = False
            break

    if can_optimize and max(indices) - min(indices) < 32:
        # Use optimized batch update for adjacent or nearby leaves
        helper_indices = get_helper_indices(indices)

        # Build partial tree containing only the affected leaves and proof nodes
        objects = {
            **{index: node for index, node in zip(indices, new_leaves)},
            **{index: node for index, node in zip(helper_indices, proof)}
        }

        # Recalculate only the affected part of the tree
        keys = sorted(objects.keys(), reverse=True)
        pos = 0
        while pos < len(keys):
            k = keys[pos]
            if k in objects and k ^ 1 in objects and k // 2 not in objects:
                objects[GeneralizedIndex(k // 2)] = hash(
                    objects[GeneralizedIndex((k | 1) ^ 1)] +
                    objects[GeneralizedIndex(k | 1)]
                )
                keys.append(GeneralizedIndex(k // 2))
            pos += 1

        return objects[GeneralizedIndex(1)]
    else:
        # Fall back to standard multiproof calculation
        return calculate_multi_merkle_root(new_leaves, proof, indices)

State transition proofs

Weber introduces specialized state transition proofs optimized for its PoS consensus mechanism:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class StateTransitionProof:
    """
    A specialized proof format for validating state transitions in Weber.

    Fields:
    - pre_state_indices: Indices of state elements before transition
    - post_state_values: Values of state elements after transition
    - proof_elements: Auxiliary nodes needed for verification
    - transition_metadata: Additional data specific to the transition type
    """
    pre_state_indices: List[GeneralizedIndex]
    pre_state_values: List[Bytes32]
    post_state_values: List[Bytes32]
    proof_elements: List[Bytes32]
    transition_metadata: TransitionMetadata

class TransitionMetadata:
    """
    Metadata about a state transition to assist in verification.

    Fields:
    - transition_type: Type of state transition (block processing, epoch, etc.)
    - validator_indices: Indices of validators involved in this transition
    - validator_signatures: Signatures authorizing the transition
    """
    transition_type: int
    validator_indices: List[ValidatorIndex]
    validator_signatures: List[BLSSignature]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def verify_state_transition(pre_state_root: Root,
                           post_state_root: Root,
                           transition_proof: StateTransitionProof) -> bool:
    """
    Verify that a state transition from pre_state to post_state is valid.

    1. Validate that pre_state elements match the pre_state_root using the proof
    2. Apply the state transition rules based on transition_type
    3. Verify that the resulting state matches post_state_root
    """
    # Step 1: Verify pre-state values are correct
    if not verify_merkle_multiproof(
        transition_proof.pre_state_values,
        transition_proof.proof_elements,
        transition_proof.pre_state_indices,
        pre_state_root
    ):
        return False

    # Step 2: Apply transition rules based on transition type
    expected_post_values = []

    if transition_proof.transition_metadata.transition_type == BLOCK_PROCESSING:
        # Apply block processing rules
        for i, pre_value in enumerate(transition_proof.pre_state_values):
            # This is simplified - actual implementation would apply the full state transition rules
            # based on the specific part of the state being updated
            if i in transition_proof.transition_metadata.validator_indices:
                # Apply validator-specific updates
                expected_post_values.append(apply_validator_update(pre_value, transition_proof))
            else:
                # Apply standard updates
                expected_post_values.append(apply_standard_update(pre_value, transition_proof))

    elif transition_proof.transition_metadata.transition_type == EPOCH_PROCESSING:
        # Apply epoch processing rules
        expected_post_values = apply_epoch_transition(
            transition_proof.pre_state_values,
            transition_proof.transition_metadata
        )

    else:
        return False  # Unknown transition type

    # Step 3: Verify post-state matches expectations
    if expected_post_values != transition_proof.post_state_values:
        return False

    # Step 4: Verify post-state root matches
    calculated_post_root = calculate_multi_merkle_root(
        transition_proof.post_state_values,
        transition_proof.proof_elements,
        transition_proof.pre_state_indices  # Same indices, different values
    )

    return calculated_post_root == post_state_root

Cross-chain verification

Weber implements efficient cross-chain verification with compact Merkle proof formats:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class BridgeProof:
    """
    Compact proof format optimized for cross-chain verification.

    Fields:
    - block_height: Height of the block containing the data
    - validator_signatures: Aggregated BLS signatures from validators
    - compressed_proof: Minimized Merkle proof using Weber's compression
    - metadata: Cross-chain specific verification data
    """
    block_height: uint64
    validator_signatures: BLSSignature
    compressed_proof: List[Bytes32]
    compressed_descriptor: bytes
    metadata: BridgeMetadata

class BridgeMetadata:
    """
    Additional data needed for cross-chain verification.

    Fields:
    - target_chain_id: Identifier of the destination chain
    - source_chain_id: Identifier of Weber (the source chain)
    - gas_optimization_flags: Flags to optimize verification on different VMs
    """
    target_chain_id: uint64
    source_chain_id: uint64
    gas_optimization_flags: uint32

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def generate_bridge_proof(block_root: Root, 
                         data_index: GeneralizedIndex,
                         witness_data: Bytes32) -> BridgeProof:
    """
    Generate a compact proof suitable for verification on external chains.
    BridgeProof is optimized for gas efficiency when verified in EVM-compatible chains.
    """
    # Get the standard Merkle proof first
    standard_proof = generate_merkle_proof(block_root, data_index)

    # Compress the proof for cross-chain efficiency
    compressed_proof, descriptor = compress_proof([standard_proof], [data_index])

    # Get validator signatures attesting to this block
    validator_signatures = get_validator_signatures_for_block(block_root)

    # Create bridge-specific metadata
    metadata = BridgeMetadata(
        target_chain_id=determine_target_chain(),
        source_chain_id=WEBER_CHAIN_ID,
        gas_optimization_flags=select_gas_optimizations(determine_target_chain())
    )

    return BridgeProof(
        block_height=get_block_height(block_root),
        validator_signatures=aggregate_signatures(validator_signatures),
        compressed_proof=compressed_proof,
        compressed_descriptor=descriptor,
        metadata=metadata
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def verify_bridge_proof_on_evm(proof: BridgeProof, expected_root: bytes32, data_index: uint256) -> bool:
    """
    Solidity-compatible verification function for Weber bridge proofs.
    This is the implementation that would run on an EVM-compatible chain.

    Note: This is pseudocode representing the equivalent Solidity implementation.
    """
    # 1. Verify the validator signatures (using precompiles for BLS verification)
    if not verify_aggregate_signature(
        proof.validator_signatures,
        construct_signing_root(proof.block_height),
        get_weber_validators()
    ):
        return false

    # 2. Decompress the proof (optimized for EVM gas usage)
    full_proof = evm_decompress_proof(
        proof.compressed_proof, 
        proof.compressed_descriptor,
        data_index
    )

    # 3. Verify the Merkle proof (gas-optimized version for EVM)
    leaf_node = get_leaf_from_proof(full_proof)
    calculated_root = evm_calculate_merkle_root(
        leaf_node,
        full_proof,
        data_index
    )

    # 4. Check against expected root
    return calculated_root == expected_root

Security considerations

When implementing Merkle proof verification in Weber, consider the following security aspects:

  1. Proof Size Limit: Weber enforces a maximum proof size of 1MB to prevent DoS attacks. The verification algorithm checks proof size before processing.

  2. Validated Types: All proven data is typed using SSZ schemas, ensuring proper serialization/deserialization and preventing type confusion attacks.

  3. Second Preimage Resistance: Weber uses SHA-256 for Merkle tree hashing, providing strong second preimage resistance. The hash function selection meets NIST security standards and is resistant to length extension attacks.

  4. Gas Optimization: For on-chain verification, Weber's proofs include additional metadata enabling gas-efficient verification on EVM chains. This includes: - Batched hash verification - Proof compression to minimize calldata costs - Precomputation of common verification patterns

  5. Long-Range Attack Resistance: Weber's proof system tracks validator set changes and includes checkpoint signatures in long-range proofs, making it resistant to long-range attacks. This is particularly important for cross-chain proofs where verifiers may not follow the source chain closely.

  6. Front-running Protection: Weber's cross-chain proofs include destination-specific binding data to prevent malicious front-running or replay of proofs across different recipient contracts.

Implementation guidelines

When implementing Weber's Merkle proof system:

  1. Constant-time Verification: Use constant-time comparison for root verification to prevent timing side-channel attacks. This is especially important for implementations running in shared environments.

  2. Proof Caching: Implement a two-level caching strategy: - L1 cache: Recently verified proof results (root + indices + leaf values → boolean result) - L2 cache: Recently calculated intermediate nodes (saves recalculation for similar proofs)

  3. Resource-constrained Environments: For IoT or mobile devices: - Implement progressive proof verification allowing partial computation with memory checkpoints - Support chunk-based proof ingestion rather than requiring the entire proof at once - Optimize for binary size by combining similar tree operations

  4. Smart Contract Optimization: When verifying Weber proofs in smart contracts: - Store frequently used proof elements in contract storage to amortize gas costs - Use assembly for tree hashing operations to reduce gas consumption - Implement batched verification for multiple related proofs

  5. Indexing Efficiency: Weber nodes should: - Maintain a sparse Merkle trie of recent state roots for O(log n) proof generation - Index "hot" state fields that are frequently proven or verified - Use specialized databases optimized for tree traversal patterns

  6. Testing Requirements: Implementation must pass: - Standard test vectors provided in the Weber test suite - Fuzz testing with random trees and indices - Hostile proof tests with malformed or malicious proof elements - Performance benchmarks against reference implementation