Merkle Path Verification

When verifying BEEF structures, it's necessary to ensure that all transactions are well-anchored: this is to say, that they come from inputs in the honest chain. The SDK doesn't ship with a headers client, but this guide shows an example of how to use it with Pulse: a popular client suitable for a wide range of use-cases.

Pre-requisites

As stated in the README, you will need to be running a Pulse instance. Get it up and running, and configure a level of authentication appropriate for your use-case:

docker pull bsvb/block-headers-service
docker run bsvb/block-headers-service:latest

Building our Client

The SDK's ChainTracker interface defines the required structure for our implementation, as follows:

class ChainTracker(ABC):
    """
    The Chain Tracker is responsible for verifying the validity of a given Merkle root
    for a specific block height within the blockchain.

    Chain Trackers ensure the integrity of the blockchain by
    validating new headers against the chain's history. They use accumulated
    proof-of-work and protocol adherence as metrics to assess the legitimacy of blocks.
    """

    @abstractmethod
    async def is_valid_root_for_height(self, root: str, height: int) -> bool:
        """
        Verify the validity of a Merkle root for a given block height.

        :param root: The Merkle root to verify.
        :param height: The block height to verify against.
        :return: A boolean indicating if the Merkle root is valid for the specified block height.
        """
        pass

Given an array of merkle roots and corresponding block heights, we return a boolean indicating whether they're all valid.

We can plug in the Block Header Service API with appropriate HTTP handling logic as follows:

class WhatsOnChainTracker(ChainTracker):
    def __init__(
            self,
            network: str = "main",
            api_key: Optional[str] = None,
            http_client: Optional[HttpClient] = None,
    ):
        self.network = network
        self.URL = f"https://api.whatsonchain.com/v1/bsv/{network}"
        self.http_client = (
            http_client if http_client else default_http_client()
        )
        self.api_key = api_key

    async def is_valid_root_for_height(self, root: str, height: int) -> bool:
        request_options = {"method": "GET", "headers": self.get_headers()}

        response = await self.http_client.fetch(
            f"{self.URL}/block/{height}/header", request_options
        )
        if response.ok:
            merkleroot = response.json()["data"]["merkleroot"]
            return merkleroot == root
        elif response.status_code == 404:
            return False
        else:
            raise Exception(
                f"Failed to verify merkleroot for height {height} because of an error: {response.json()}"
            )

    def get_headers(self) -> Dict[str, str]:
        headers = {
            "Accept": "application/json",
        }
        if self.api_key:
            headers["Authorization"] = self.api_key
        return headers

Now, we can use our WhatsOnChainTracker as a ChainTracker when calling the Transaction object's .verify() method. You can see an example in the BEEF verification guide.

This provides the ability to ensure that a transaction is well-anchored.