Merkle Trees: From Zero to Hero - A Complete Guide to Cryptographic Data Structures

import hashlib

def sha256(data):
    """Compute SHA-256 hash of data"""
    return hashlib.sha256(data.encode()).hexdigest()

# Example
print(sha256("Hello, World!"))
# Output: dffd6021bb2bd5b0af676290809ec3a53191dd81c7f70a4b28688a362182986f

import hashlib

def hash_data(data):
    """Hash a piece of data using SHA-256"""
    return hashlib.sha256(data.encode()).hexdigest()

def hash_pair(left, right):
    """Hash two hashes together"""
    return hashlib.sha256((left + right).encode()).hexdigest()

# Our data
transactions = ["TxA", "TxB", "TxC", "TxD"]

# Step 1: Create leaf hashes
leaves = [hash_data(tx) for tx in transactions]
print("Leaves:", leaves)

# Step 2: First level
hab = hash_pair(leaves[0], leaves[1])
hcd = hash_pair(leaves[2], leaves[3])
print(f"HAB: {hab}")
print(f"HCD: {hcd}")

# Step 3: Root
root = hash_pair(hab, hcd)
print(f"Root: {root}")

class MerkleTree:
    def __init__(self, data):
        self.leaves = [hash_data(d) for d in data]
        self.tree = self.build_tree(self.leaves)
        self.root = self.tree[0]
    
    def build_tree(self, leaves):
        """Build tree bottom-up"""
        if len(leaves) == 1:
            return leaves
        
        # Ensure even number of leaves
        if len(leaves) % 2 != 0:
            leaves.append(leaves[-1])
        
        # Build next level
        next_level = []
        for i in range(0, len(leaves), 2):
            parent = hash_pair(leaves[i], leaves[i+1])
            next_level.append(parent)
        
        return self.build_tree(next_level) + leaves
    
    def get_proof(self, index):
        """Get Merkle proof for leaf at index"""
        proof = []
        current_index = index
        current_level_size = len(self.leaves)
        
        # Start from leaves, go up to root
        level_start = len(self.tree) - len(self.leaves)
        
        while current_level_size > 1:
            # Get sibling
            if current_index % 2 == 0:
                sibling_index = current_index + 1
            else:
                sibling_index = current_index - 1
            
            if sibling_index < current_level_size:
                sibling = self.tree[level_start + sibling_index]
                proof.append({
                    'hash': sibling,
                    'position': 'right' if current_index % 2 == 0 else 'left'
                })
            
            # Move to parent level
            current_index = current_index // 2
            current_level_size = (current_level_size + 1) // 2
            level_start -= current_level_size
        
        return proof
    
    def verify_proof(self, data, index, proof):
        """Verify a Merkle proof"""
        current_hash = hash_data(data)
        
        for p in proof:
            if p['position'] == 'right':
                current_hash = hash_pair(current_hash, p['hash'])
            else:
                current_hash = hash_pair(p['hash'], current_hash)
        
        return current_hash == self.root

# Example usage
transactions = ["TxA", "TxB", "TxC", "TxD"]
tree = MerkleTree(transactions)

# Get proof for TxC (index 2)
proof = tree.get_proof(2)
print(f"Proof for TxC: {proof}")

# Verify the proof
is_valid = tree.verify_proof("TxC", 2, proof)
print(f"Proof valid: {is_valid}")

# Try with wrong data
is_valid_fake = tree.verify_proof("TxX", 2, proof)
print(f"Fake proof valid: {is_valid_fake}")

import hashlib
from typing import List, Optional, Dict, Any

class MerkleNode:
    """Represents a node in the Merkle tree"""
    def __init__(self, hash_value: str, left=None, right=None, data=None):
        self.hash = hash_value
        self.left = left
        self.right = right
        self.data = data  # Only for leaf nodes
    
    def is_leaf(self):
        return self.left is None and self.right is None

class MerkleTree:
    """
    A complete implementation of a Merkle tree with proof generation
    and verification capabilities.
    """
    
    def __init__(self, data_blocks: List[str]):
        """
        Initialize Merkle tree from data blocks.
        
        Args:
            data_blocks: List of data items to include in tree
        """
        if not data_blocks:
            raise ValueError("Cannot create Merkle tree from empty data")
        
        self.data_blocks = data_blocks
        self.leaves = self._create_leaf_nodes(data_blocks)
        self.root = self._build_tree(self.leaves)
    
    @staticmethod
    def _hash(data: str) -> str:
        """Compute SHA-256 hash of data"""
        return hashlib.sha256(data.encode('utf-8')).hexdigest()
    
    def _create_leaf_nodes(self, data_blocks: List[str]) -> List[MerkleNode]:
        """Create leaf nodes from data blocks"""
        return [
            MerkleNode(self._hash(data), data=data) 
            for data in data_blocks
        ]
    
    def _build_tree(self, nodes: List[MerkleNode]) -> MerkleNode:
        """
        Recursively build tree from leaf nodes up to root.
        
        Args:
            nodes: Current level of nodes
            
        Returns:
            Root node of tree
        """
        if len(nodes) == 1:
            return nodes[0]
        
        # Handle odd number of nodes by duplicating last
        if len(nodes) % 2 != 0:
            nodes.append(nodes[-1])
        
        # Build parent level
        parent_nodes = []
        for i in range(0, len(nodes), 2):
            left = nodes[i]
            right = nodes[i + 1]
            parent_hash = self._hash(left.hash + right.hash)
            parent = MerkleNode(parent_hash, left=left, right=right)
            parent_nodes.append(parent)
        
        return self._build_tree(parent_nodes)
    
    def get_root_hash(self) -> str:
        """Get the Merkle root hash"""
        return self.root.hash
    
    def generate_proof(self, index: int) -> List[Dict[str, Any]]:
        """
        Generate Merkle proof for data block at given index.
        
        Args:
            index: Index of data block in original list
            
        Returns:
            List of proof elements (hash and position)
        """
        if index < 0 or index >= len(self.data_blocks):
            raise IndexError(f"Index {index} out of range")
        
        proof = []
        current_node = self.leaves[index]
        nodes_at_level = self.leaves[:]
        
        # Traverse up the tree
        while len(nodes_at_level) > 1:
            # Ensure even number
            if len(nodes_at_level) % 2 != 0:
                nodes_at_level.append(nodes_at_level[-1])
            
            # Find current node's index and sibling
            current_index = nodes_at_level.index(current_node)
            sibling_index = current_index + 1 if current_index % 2 == 0 else current_index - 1
            sibling = nodes_at_level[sibling_index]
            
            proof.append({
                'hash': sibling.hash,
                'position': 'right' if current_index % 2 == 0 else 'left'
            })
            
            # Move to parent level
            parent_nodes = []
            for i in range(0, len(nodes_at_level), 2):
                left = nodes_at_level[i]
                right = nodes_at_level[i + 1]
                parent_hash = self._hash(left.hash + right.hash)
                parent = MerkleNode(parent_hash, left=left, right=right)
                parent_nodes.append(parent)
            
            # Find which parent corresponds to current node
            parent_index = current_index // 2
            current_node = parent_nodes[parent_index]
            nodes_at_level = parent_nodes
        
        return proof
    
    def verify_proof(self, data: str, index: int, proof: List[Dict[str, Any]]) -> bool:
        """
        Verify a Merkle proof.
        
        Args:
            data: The data block to verify
            index: Original index of data block
            proof: Merkle proof from generate_proof()
            
        Returns:
            True if proof is valid, False otherwise
        """
        current_hash = self._hash(data)
        
        for proof_element in proof:
            sibling_hash = proof_element['hash']
            position = proof_element['position']
            
            if position == 'right':
                current_hash = self._hash(current_hash + sibling_hash)
            else:
                current_hash = self._hash(sibling_hash + current_hash)
        
        return current_hash == self.root.hash
    
    def print_tree(self, node: Optional[MerkleNode] = None, level: int = 0):
        """Print visual representation of tree"""
        if node is None:
            node = self.root
        
        indent = "  " * level
        if node.is_leaf():
            print(f"{indent}Leaf: {node.hash[:8]}... (data: {node.data})")
        else:
            print(f"{indent}Node: {node.hash[:8]}...")
            if node.left:
                self.print_tree(node.left, level + 1)
            if node.right:
                self.print_tree(node.right, level + 1)

# Example usage and testing
def main():
    # Create sample data
    transactions = [
        "Alice pays Bob 10 BTC",
        "Charlie pays Dave 5 BTC",
        "Eve pays Frank 3 BTC",
        "Grace pays Henry 7 BTC"
    ]
    
    print("Building Merkle Tree...")
    tree = MerkleTree(transactions)
    
    print(f"\nMerkle Root: {tree.get_root_hash()}\n")
    
    print("Tree Structure:")
    tree.print_tree()
    
    # Generate and verify proof
    print("\n" + "="*60)
    print("MERKLE PROOF DEMONSTRATION")
    print("="*60)
    
    test_index = 2
    test_data = transactions[test_index]
    
    print(f"\nGenerating proof for: '{test_data}'")
    proof = tree.generate_proof(test_index)
    
    print(f"\nProof (requires {len(proof)} hashes):")
    for i, p in enumerate(proof):
        print(f"  {i+1}. Hash: {p['hash'][:16]}... (position: {p['position']})")
    
    # Verify correct data
    is_valid = tree.verify_proof(test_data, test_index, proof)
    print(f"\nVerification with correct data: {is_valid}")
    
    # Verify tampered data
    tampered_data = "Alice pays Bob 100 BTC"
    is_valid_tampered = tree.verify_proof(tampered_data, test_index, proof)
    print(f"Verification with tampered data: {is_valid_tampered}")
    
    print("\n" + "="*60)
    print("EFFICIENCY DEMONSTRATION")
    print("="*60)
    
    # Show efficiency for different sizes
    sizes = [10, 100, 1000, 10000]
    for size in sizes:
        test_data = [f"Transaction {i}" for i in range(size)]
        test_tree = MerkleTree(test_data)
        test_proof = test_tree.generate_proof(0)
        print(f"\nDataset size: {size:,} items")
        print(f"Proof size: {len(test_proof)} hashes")
        print(f"Efficiency: {len(test_proof)/size*100:.2f}% of full data")

if __name__ == "__main__":
    main()

class BitcoinSPVClient:
    def __init__(self):
        self.block_headers = {}  # Only store headers
    
    def add_block_header(self, block_height, header):
        """Store block header (80 bytes)"""
        self.block_headers[block_height] = {
            'merkle_root': header['merkle_root'],
            'timestamp': header['timestamp'],
            'difficulty': header['difficulty']
        }
    
    def verify_transaction(self, tx, block_height, merkle_proof):
        """Verify transaction was included in block"""
        header = self.block_headers.get(block_height)
        if not header:
            return False
        
        # Reconstruct root from proof
        computed_root = self._compute_root(tx, merkle_proof)
        return computed_root == header['merkle_root']
    
    def _compute_root(self, tx, proof):
        current = self._hash(tx)
        for sibling, position in proof:
            if position == 'left':
                current = self._hash(sibling + current)
            else:
                current = self._hash(current + sibling)
        return current

from concurrent.futures import ThreadPoolExecutor

def build_level_parallel(nodes, max_workers=4):
    """Build one level of tree in parallel"""
    if len(nodes) % 2 != 0:
        nodes.append(nodes[-1])
    
    def hash_pair_at(i):
        left = nodes[i]
        right = nodes[i + 1]
        return MerkleNode(hash_pair(left.hash, right.hash), left, right)
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        indices = range(0, len(nodes), 2)
        parent_nodes = list(executor.map(hash_pair_at, indices))
    
    return parent_nodes

class OptimizedMerkleTree:
    def __init__(self, data):
        self.leaves = [hash_data(d) for d in data]
        self.levels = self._build_with_cache(self.leaves)
        self.root = self.levels[0][0]
    
    def _build_with_cache(self, leaves):
        """Build tree and cache all levels"""
        levels = [leaves]
        current = leaves
        
        while len(current) > 1:
            if len(current) % 2 != 0:
                current.append(current[-1])
            
            next_level = []
            for i in range(0, len(current), 2):
                parent = hash_pair(current[i], current[i+1])
                next_level.append(parent)
            
            levels.insert(0, next_level)
            current = next_level
        
        return levels
    
    def get_proof_optimized(self, index):
        """O(log n) proof generation using cached levels"""
        proof = []
        current_index = index
        
        # Start from leaf level
        for level in range(len(self.levels)-1, 0, -1):
            # Get sibling
            if current_index % 2 == 0:
                sibling_index = current_index + 1
            else:
                sibling_index = current_index - 1
            
            if sibling_index < len(self.levels[level]):
                sibling = self.levels[level][sibling_index]
                proof.append({
                    'hash': sibling,
                    'position': 'right' if current_index % 2 == 0 else 'left'
                })
            
            # Move to parent level
            current_index = current_index // 2
        
        return proof

def batch_verify(tree, proofs):
    """Verify multiple proofs in parallel"""
    results = []
    
    with ThreadPoolExecutor() as executor:
        futures = []
        for proof in proofs:
            future = executor.submit(
                tree.verify_proof, 
                proof['data'], 
                proof['index'], 
                proof['proof']
            )
            futures.append(future)
        
        for future in futures:
            results.append(future.result())
    
    return results

class StreamingMerkleTree:
    def __init__(self, hash_func=hash_data):
        self.hash_func = hash_func
        self.buffer = []
        self.root = None
        self.levels = {}
    
    def add(self, data):
        """Add data to tree with limited memory"""
        self.buffer.append(self.hash_func(data))
        
        # Process complete pairs
        while len(self.buffer) >= 2:
            # Process pairs
            new_buffer = []
            for i in range(0, len(self.buffer) - 1, 2):
                parent = self.hash_func(self.buffer[i] + self.buffer[i+1])
                new_buffer.append(parent)
            
            # Handle odd number
            if len(self.buffer) % 2 != 0:
                new_buffer.append(self.buffer[-1])
            
            self.buffer = new_buffer
    
    def finalize(self):
        """Finalize tree and return root"""
        while len(self.buffer) > 1:
            if len(self.buffer) % 2 != 0:
                self.buffer.append(self.buffer[-1])
            
            new_buffer = []
            for i in range(0, len(self.buffer), 2):
                parent = self.hash_func(self.buffer[i] + self.buffer[i+1])
                new_buffer.append(parent)
            
            self.buffer = new_buffer
        
        self.root = self.buffer[0] if self.buffer else None
        return self.root

def compress_proof(proof):
    """Compress proof using bitmask for positions"""
    # Extract hashes and positions
    hashes = [p['hash'] for p in proof]
    positions = [p['position'] for p in proof]
    
    # Convert positions to bitmask
    bitmask = 0
    for i, pos in enumerate(positions):
        if pos == 'left':
            bitmask |= (1 << i)
    
    # Return compressed format
    return {
        'hashes': hashes,
        'bitmask': bitmask
    }

def decompress_proof(compressed):
    """Decompress proof from bitmask format"""
    hashes = compressed['hashes']
    bitmask = compressed['bitmask']
    
    proof = []
    for i, hash_val in enumerate(hashes):
        position = 'left' if (bitmask & (1 << i)) else 'right'
        proof.append({
            'hash': hash_val,
            'position': position
        })
    
    return proof