Merkle Tree Implementation

Complete Python implementation of AirChain's Merkle tree verification.

Installation

pip install pycardano requests

Complete Implementation

import hashlib
import json
from typing import List, Tuple

class MerkleTree:
    def __init__(self, data: List[dict]):
        self.leaves = [self._hash_data(d) for d in data]
        self.tree = self._build_tree()

    def _hash_data(self, data: dict) -> str:
        json_str = json.dumps(data, sort_keys=True)
        return hashlib.sha256(json_str.encode()).hexdigest()

    def _hash_pair(self, left: str, right: str) -> str:
        combined = left + right
        return hashlib.sha256(combined.encode()).hexdigest()

    def _build_tree(self) -> List[List[str]]:
        tree = [self.leaves]
        current_level = self.leaves

        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 left
                next_level.append(self._hash_pair(left, right))
            tree.append(next_level)
            current_level = next_level

        return tree

    def get_root(self) -> str:
        return self.tree[-1][0]

    def get_proof(self, index: int) -> Tuple[List[str], List[int]]:
        proof = []
        positions = []

        for level in self.tree[:-1]:
            if index % 2 == 0:
                sibling_index = index + 1
                positions.append(1)  # Right
            else:
                sibling_index = index - 1
                positions.append(0)  # Left

            if sibling_index < len(level):
                proof.append(level[sibling_index])
            else:
                proof.append(level[index])

            index = index // 2

        return proof, positions

    def verify_proof(self, data: dict, proof: List[str], positions: List[int]) -> bool:
        current_hash = self._hash_data(data)

        for sibling_hash, position in zip(proof, positions):
            if position == 0:  # Sibling is on left
                current_hash = self._hash_pair(sibling_hash, current_hash)
            else:  # Sibling is on right
                current_hash = self._hash_pair(current_hash, sibling_hash)

        return current_hash == self.get_root()


# Example Usage
if __name__ == "__main__":
    # Sample sensor readings
    readings = [
        {"sensor_id": "NGR-LOS-YAB-001", "timestamp": "2025-11-29T14:00:00Z", "pm25": 45.3, "aqi": 112},
        {"sensor_id": "NGR-LOS-YAB-002", "timestamp": "2025-11-29T14:00:00Z", "pm25": 38.7, "aqi": 95},
        {"sensor_id": "NGR-LOS-LEK-001", "timestamp": "2025-11-29T14:00:00Z", "pm25": 52.1, "aqi": 128},
        {"sensor_id": "NGR-LOS-IKJ-001", "timestamp": "2025-11-29T14:00:00Z", "pm25": 61.4, "aqi": 145},
    ]

    # Build Merkle tree
    tree = MerkleTree(readings)
    root = tree.get_root()
    print(f"Merkle Root: {root}")

    # Generate proof for first reading
    proof, positions = tree.get_proof(0)
    print(f"\nProof for reading 0:")
    print(f"Sibling hashes: {proof}")
    print(f"Positions: {positions}")

    # Verify the proof
    is_valid = tree.verify_proof(readings[0], proof, positions)
    print(f"\nVerification result: {is_valid}")

    # Try to verify with tampered data
    tampered_reading = readings[0].copy()
    tampered_reading["pm25"] = 999.9
    is_valid_tampered = tree.verify_proof(tampered_reading, proof, positions)
    print(f"Tampered data verification: {is_valid_tampered}")

Using with AirChain API

import requests

API_KEY = "your_api_key"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Get verification proof from API
response = requests.get(
    "https://api.airchain.ng/v1/readings/NGR-LOS-YAB-001/verify?timestamp=2025-11-29T14:00:00Z",
    headers=headers
)

verification = response.json()
merkle_root = verification["merkle_proof"]["root"]
proof_path = verification["merkle_proof"]["path"]

# Verify against Cardano blockchain
cardano_tx = verification["cardano"]["tx_hash"]
print(f"Verified on Cardano: {cardano_tx}")

Next Steps

• Blockchain Verification Guide
• Python SDK
• Contributing