Merkle Tree em Python Puro: A Árvore Que Detecta Se Um Único Byte Foi Alterado em 10 GB de Dados (Sem Conferir Tudo)
Você tem 10 GB de logs. Ou 50.000 arquivos de backup. Ou um dataset crítico que precisa provar que não foi adulterado. Conferir byte por byte? Esquece. O Merkle Tree resolve isso com uma única hash de 64 caracteres — e te diz exatamente onde está a diferença se algo mudou.
O Problema Que Ninguém Te Conta
Quando você precisa verificar se um arquivo grande foi corrompido ou alterado, a solução óbvia é calcular o hash do arquivo inteiro. SHA-256, pronto. Mas e quando você tem milhares de arquivos? E quando só um bloco de 4 KB foi alterado em 10 GB de dados? Recalcular tudo é burrice.
O Git resolve isso de forma elegante: ele não armazena só o hash do commit. Ele constrói uma árvore de hashes onde cada nó pai é o hash dos filhos. Se um único byte muda em qualquer arquivo, a hash raiz muda. E você pode encontrar qual bloco mudou em O(log n) — sem recalcular tudo.
Isso é um Merkle Tree. E hoje você vai implementar um do zero.
O Que É Um Merkle Tree (Sem Diagrama Chato)
Pense assim:
1. Você quebra seus dados em blocos (ex: 4 KB cada)
2. Calcula o hash de cada bloco → essas são as folhas
3. Agrupa as folhas em pares e calcula o hash de cada par → nós internos
4. Repete até sobrar um único hash → a raiz (root hash)
Se qualquer bloco muda, todos os hashes acima dele mudam. A raiz final é uma “impressão digital” de todos os dados. E a mágica: você pode provar que um bloco específico pertence ao conjunto original mostrando apenas os hashes do “caminho” até a raiz — isso se chama Merkle Proof.
Implementação Completa: Merkle Tree em Python Puro
Sem bibliotecas externas. Só hashlib da stdlib.
import hashlib
from typing import List, Optional, Tuple
from dataclasses import dataclass
@dataclass
class MerkleNode:
hash: str
left: Optional['MerkleNode'] = None
right: Optional['MerkleNode'] = None
@property
def is_leaf(self) -> bool:
return self.left is None and self.right is None
class MerkleTree:
def __init__(self, data_blocks: List[bytes]):
if not data_blocks:
raise ValueError("Precisa de pelo menos um bloco de dados")
self.blocks = data_blocks
self.leaves = [self._hash(block) for block in data_blocks]
self.root = self._build_tree(self.leaves)
def _hash(self, data: bytes) -> str:
"""SHA-256 de qualquer coisa."""
return hashlib.sha256(data).hexdigest()
def _hash_pair(self, left: str, right: str) -> str:
"""Hash da concatenação de dois hashes (ordenados)."""
# Ordenar garante que hash(a,b) == hash(b,a) não aconteça
combined = left + right if left <= right else right + left
return self._hash(combined.encode())
def _build_tree(self, hashes: List[str]) -> MerkleNode:
"""Constrói a árvore bottom-up."""
if len(hashes) == 1:
return MerkleNode(hash=hashes[0])
nodes = [MerkleNode(hash=h) for h in hashes]
while len(nodes) > 1:
next_level = []
for i in range(0, len(nodes), 2):
left = nodes[i]
# Se ímpar, duplica o último nó
right = nodes[i + 1] if i + 1 < len(nodes) else nodes[i]
parent_hash = self._hash_pair(left.hash, right.hash)
parent = MerkleNode(hash=parent_hash, left=left, right=right)
next_level.append(parent)
nodes = next_level
return nodes[0]
def get_proof(self, index: int) -> List[Tuple[str, str]]:
"""
Retorna a Merkle Proof para o bloco no índice dado.
Cada elemento é (hash, posição) onde posição é 'left' ou 'right'.
"""
if index < 0 or index >= len(self.leaves):
raise IndexError(f"Índice {index} fora do range [0, {len(self.leaves)})")
proof = []
current_level = self.leaves
current_index = index
while len(current_level) > 1:
next_level = []
for i in range(0, len(current_level), 2):
left = current_level[i]
right = current_level[i + 1] if i + 1 < len(current_level) else current_level[i]
# Se estamos no par atual, adiciona o irmão à proof
if i == current_index - (current_index % 2):
if current_index % 2 == 0:
# Sou o left, meu irmão é right
proof.append((right, 'right'))
else:
# Sou o right, meu irmão é left
proof.append((left, 'left'))
parent_hash = self._hash_pair(left, right)
next_level.append(parent_hash)
current_level = next_level
current_index = current_index // 2
return proof
def verify_proof(self, block: bytes, proof: List[Tuple[str, str]], root: str) -> bool:
"""Verifica se um bloco pertence à árvore usando apenas a proof."""
current_hash = self._hash(block)
for sibling_hash, position in proof:
if position == 'left':
current_hash = self._hash_pair(sibling_hash, current_hash)
else:
current_hash = self._hash_pair(current_hash, sibling_hash)
return current_hash == rootCaso de Uso Real: Verificador de Integridade de Backups
Agora a parte que importa. Vamos usar o Merkle Tree para verificar se um backup de 10 GB não foi corrompido — sem recalcular tudo.
import os
from pathlib import Path
class BackupIntegrityChecker:
def __init__(self, chunk_size: int = 4096): # 4 KB chunks
self.chunk_size = chunk_size
def chunk_file(self, filepath: str) -> List[bytes]:
"""Lê um arquivo em chunks de tamanho fixo."""
chunks = []
with open(filepath, 'rb') as f:
while True:
chunk = f.read(self.chunk_size)
if not chunk:
break
chunks.append(chunk)
return chunks
def create_manifest(self, filepath: str) -> dict:
"""
Cria um manifest com a root hash e todas as leaf hashes.
Salve isso em lugar seguro.
"""
chunks = self.chunk_file(filepath)
tree = MerkleTree(chunks)
return {
'filepath': filepath,
'root_hash': tree.root.hash,
'leaf_hashes': tree.leaves,
'chunk_size': self.chunk_size,
'total_chunks': len(chunks),
'file_size': os.path.getsize(filepath)
}
def verify_file(self, filepath: str, manifest: dict) -> dict:
"""
Verifica se o arquivo ainda corresponde ao manifest.
Retorna quais chunks estão corrompidos (se houver).
"""
chunks = self.chunk_file(filepath)
if len(chunks) != manifest['total_chunks']:
return {
'valid': False,
'error': f"Tamanho mudou: {len(chunks)} chunks vs {manifest['total_chunks']} esperado"
}
# Verifica a root hash primeiro (rápido)
tree = MerkleTree(chunks)
if tree.root.hash == manifest['root_hash']:
return {'valid': True, 'corrupted_chunks': []}
# Se falhou, encontra quais chunks específicos mudaram
corrupted = []
for i, (chunk, expected_hash) in enumerate(zip(chunks, manifest['leaf_hashes'])):
actual_hash = tree._hash(chunk)
if actual_hash != expected_hash:
corrupted.append({
'chunk_index': i,
'byte_offset': i * self.chunk_size,
'expected_hash': expected_hash,
'actual_hash': actual_hash
})
return {
'valid': False,
'corrupted_chunks': corrupted,
'total_corrupted': len(corrupted)
}
# Exemplo de uso
checker = BackupIntegrityChecker(chunk_size=8192) # 8 KB chunks
# Primeira vez: cria o manifest
manifest = checker.create_manifest('/backup/database_2026_07_14.sql')
print(f"Root hash: {manifest['root_hash']}")
print(f"Total chunks: {manifest['total_chunks']}")
# Salva o manifest (JSON, banco, onde quiser)
import json
with open('/backup/database_2026_07_14.manifest.json', 'w') as f:
json.dump(manifest, f, indent=2)
# Depois: verifica integridade
with open('/backup/database_2026_07_14.manifest.json', 'r') as f:
saved_manifest = json.load(f)
result = checker.verify_file('/backup/database_2026_07_14.sql', saved_manifest)
if result['valid']:
print("✅ Arquivo íntegro")
else:
print(f"❌ {result['total_corrupted']} chunks corrompidos:")
for chunk in result['corrupted_chunks']:
offset_mb = chunk['byte_offset'] / (1024 * 1024)
print(f" Chunk {chunk['chunk_index']} (offset: {offset_mb:.2f} MB)")Por Que Isso É Melhor Que SHA-256 Simples?
| Abordagem | Tempo para verificar 10 GB | Detecta **onde** mudou? |
|———–|—————————|————————|
| SHA-256 do arquivo inteiro | ~30 segundos | ❌ Não |
| SHA-256 de cada chunk (sem árvore) | ~30 segundos + O(n) comparações | ✅ Sim, mas lento |
| **Merkle Tree** | ~30 segundos (primeira vez) + O(log n) depois | ✅ Sim, rápido |
A mágica está na Merkle Proof: você pode provar que um bloco específico pertence ao conjunto mostrando apenas ~20 hashes (para 1 milhão de blocos), não todos.
🧠 O Perrengue do Olivetto
Semana passada eu tinha um script que fazia backup de 8 GB de logs toda madrugada. Depois copiava pra um NAS via rsync. Um dia o rsync disse “done” mas o arquivo no NAS estava com 3 bytes corrompidos no meio — provavelmente erro de rede que o TCP não pegou.
Demorei 2 dias pra perceber que os logs estavam inconsistentes. Se eu tivesse guardado um manifest com Merkle Tree, teria detectado o chunk corrompido em segundos — e sabido exatamente quais 4 KB retransmitir, não os 8 GB inteiros.
A lição: integridade não é binária. Não basta saber “mudou ou não”. Você precisa saber onde mudou pra consertar rápido.
Onde Mais Isso É Usado?
– Git: cada commit tem uma Merkle Tree dos arquivos
– IPFS: arquivos são endereçados pela root hash
– Bitcoin/Ethereum: blocos usam Merkle Trees pra provar transações
– Btrfs/ZFS: filesystems com checksum por bloco (mesma ideia)
– S3/Glacier: alguns serviços usam pra verificação de integridade
CTA: O Que Você Vai Verificar Hoje?
Agora você tem um verificador de integridade que:
– Detecta corrupção em qualquer arquivo
– Te diz exatamente onde está o problema
– Usa O(log n) espaço pra proofs
– Roda em Python puro, sem dependências
Pergunta: qual arquivo crítico do seu sistema você não verifica hoje? Aquele backup que você “tem certeza” que está OK? Roda esse código nele agora. Se der match, dorme tranquilo. Se não der, você acabou de se livrar de um bug que ia te acordar às 3 da manhã.
Testa aí e me conta nos comentários: qual foi o arquivo mais suspeito que você verificou?
