<filename>code/merkle_root.py
from base64 import b64decode, b64encode
from contextlib import redirect_stderr
from dataclasses_json import dataclass_json
from dataclasses import dataclass
import hashlib
import io
import json
import logging
import re

from pgpy import PGPKey, PGPMessage
import requests


KEYBASE_MERKLE_ROOT_URL = 'https://keybase.io/_/api/1.0/merkle/root.json'
TEMPLATE_MERKLE_URL = KEYBASE_MERKLE_ROOT_URL + "?seqno={seqno}"
KEYBASE_KID = '<KEY>'
logger = logging.getLogger(__name__)


class VerificationError(Exception):
    pass


@dataclass_json
@dataclass
class MerkleRoot:
    # if the format of anything in here changes, make sure to
    # bump the version of StampedMerkleRoot
    seqno: int = 0
    ctime_string: str = "1970-01-01T00:00:00.000Z"
    root_hash: str = 128 * "0"
    b64stamped: str = 128 * "0"  # hash of validated sig over merkle root payload
    stable_url: str = TEMPLATE_MERKLE_URL.format(seqno=0)

    @property
    def data_to_stamp(self) -> bytes:
        return b64decode(self.b64stamped)


# fetch from the keybase API and verify a bunch of things
def fetch_keybase_merkle_root() -> MerkleRoot:
    # fetch the current merkle root from the keybase api
    resp = requests.get(KEYBASE_MERKLE_ROOT_URL)
    full_kb_merkle_root = resp.json()
    seqno = full_kb_merkle_root['seqno']
    stable_url = TEMPLATE_MERKLE_URL.format(seqno=seqno)
    root_hash = full_kb_merkle_root['hash']

    # extract the signature and signed payload into a PGP message for verification
    raw_pgp_sig_msg = full_kb_merkle_root['sigs'][KEYBASE_KID]['sig']

    # the message itself is too big to include with OTS data in a chat message
    # so let's take the hash of it and use that instead. this is kind of a
    # bummer because it's almost OK to use the whole thing.
    hash_of_raw_pgp_sig = hashlib.sha512(raw_pgp_sig_msg.encode()).digest()
    b64stamped = b64encode(hash_of_raw_pgp_sig).decode('utf-8')

    # this function will raise an exception if PGP verification fails
    signed_payload = _verify_keybase_signature(raw_pgp_sig_msg)

    # as a sanity check, ensure that the signed payload matches
    # the raw json payload which is also present in the keybase
    # API response
    claimed_payload = json.loads(full_kb_merkle_root['payload_json'])
    if signed_payload != claimed_payload:
        logger.error(f"signed_payload: {signed_payload}")
        logger.error(f"api response payload: {claimed_payload}")
        raise VerificationError(f"signed payload doesn't match API response payload at {stable_url}")

    # and most importantly, that the root hash that was signed is the same as the one that's
    # at the top level of the API response body.
    signed_root_hash = signed_payload['body']['root']
    if (signed_root_hash != root_hash or len(signed_root_hash) == 0):
        logger.error(f"signed_payload: {signed_payload}")
        raise VerificationError(f"keybase signed a different root hash ({signed_root_hash}) from the one in the payload {root_hash}")

    return MerkleRoot(
        seqno=seqno,
        root_hash=root_hash,
        ctime_string=full_kb_merkle_root['ctime_string'],
        b64stamped=b64stamped,
        stable_url=stable_url,
    )


def _verify_keybase_signature(raw_pgp_sig_msg):
    # load the raw pgp message
    pgp_msg = PGPMessage.from_blob(raw_pgp_sig_msg)
    # load keybase's claimed public key
    # see: https://keybase.io/docs/server_security/our_merkle_key
    kb_public_key, _ = PGPKey.from_blob(KEYBASE_PGP_VERIFICATION_KEY)

    # verify it: https://pgpy.readthedocs.io/en/latest/examples.html#verifying-things
    f = io.StringIO()
    with redirect_stderr(f):
        # suppress unnecessary stdout
        verification_result = kb_public_key.verify(pgp_msg)
    if not verification_result:
        raise VerificationError("API response did not verify with Keybase's public key")

    good_signatures = list(verification_result.good_signatures)
    if len(good_signatures) != 1:
        logger.error(f"good_signatures = {good_signatures}")
        raise VerificationError(f"Expected 1 valid signature, got {len(good_signatures)} from {specific_url}")

    return json.loads(good_signatures[0].subject)



KEYBASE_PGP_VERIFICATION_KEY = """
-----BEGIN PGP PUBLIC KEY BLOCK-----
Version: GnuPG/MacGPG2 v2.0.22 (Darwin)
Comment: GPGTools - https://gpgtools.org

<KEY>
-----END PGP PUBLIC KEY BLOCK-----
"""
<filename>code/merkle_root.py
from base64 import b64decode, b64encode
from contextlib import redirect_stderr
from dataclasses_json import dataclass_json
from dataclasses import dataclass
import hashlib
import io
import json
import logging
import re

from pgpy import PGPKey, PGPMessage
import requests


KEYBASE_MERKLE_ROOT_URL = 'https://keybase.io/_/api/1.0/merkle/root.json'
TEMPLATE_MERKLE_URL = KEYBASE_MERKLE_ROOT_URL + "?seqno={seqno}"
KEYBASE_KID = '<KEY>'
logger = logging.getLogger(__name__)


class VerificationError(Exception):
    pass


@dataclass_json
@dataclass
class MerkleRoot:
    # if the format of anything in here changes, make sure to
    # bump the version of StampedMerkleRoot
    seqno: int = 0
    ctime_string: str = "1970-01-01T00:00:00.000Z"
    root_hash: str = 128 * "0"
    b64stamped: str = 128 * "0"  # hash of validated sig over merkle root payload
    stable_url: str = TEMPLATE_MERKLE_URL.format(seqno=0)

    @property
    def data_to_stamp(self) -> bytes:
        return b64decode(self.b64stamped)


# fetch from the keybase API and verify a bunch of things
def fetch_keybase_merkle_root() -> MerkleRoot:
    # fetch the current merkle root from the keybase api
    resp = requests.get(KEYBASE_MERKLE_ROOT_URL)
    full_kb_merkle_root = resp.json()
    seqno = full_kb_merkle_root['seqno']
    stable_url = TEMPLATE_MERKLE_URL.format(seqno=seqno)
    root_hash = full_kb_merkle_root['hash']

    # extract the signature and signed payload into a PGP message for verification
    raw_pgp_sig_msg = full_kb_merkle_root['sigs'][KEYBASE_KID]['sig']

    # the message itself is too big to include with OTS data in a chat message
    # so let's take the hash of it and use that instead. this is kind of a
    # bummer because it's almost OK to use the whole thing.
    hash_of_raw_pgp_sig = hashlib.sha512(raw_pgp_sig_msg.encode()).digest()
    b64stamped = b64encode(hash_of_raw_pgp_sig).decode('utf-8')

    # this function will raise an exception if PGP verification fails
    signed_payload = _verify_keybase_signature(raw_pgp_sig_msg)

    # as a sanity check, ensure that the signed payload matches
    # the raw json payload which is also present in the keybase
    # API response
    claimed_payload = json.loads(full_kb_merkle_root['payload_json'])
    if signed_payload != claimed_payload:
        logger.error(f"signed_payload: {signed_payload}")
        logger.error(f"api response payload: {claimed_payload}")
        raise VerificationError(f"signed payload doesn't match API response payload at {stable_url}")

    # and most importantly, that the root hash that was signed is the same as the one that's
    # at the top level of the API response body.
    signed_root_hash = signed_payload['body']['root']
    if (signed_root_hash != root_hash or len(signed_root_hash) == 0):
        logger.error(f"signed_payload: {signed_payload}")
        raise VerificationError(f"keybase signed a different root hash ({signed_root_hash}) from the one in the payload {root_hash}")

    return MerkleRoot(
        seqno=seqno,
        root_hash=root_hash,
        ctime_string=full_kb_merkle_root['ctime_string'],
        b64stamped=b64stamped,
        stable_url=stable_url,
    )


def _verify_keybase_signature(raw_pgp_sig_msg):
    # load the raw pgp message
    pgp_msg = PGPMessage.from_blob(raw_pgp_sig_msg)
    # load keybase's claimed public key
    # see: https://keybase.io/docs/server_security/our_merkle_key
    kb_public_key, _ = PGPKey.from_blob(KEYBASE_PGP_VERIFICATION_KEY)

    # verify it: https://pgpy.readthedocs.io/en/latest/examples.html#verifying-things
    f = io.StringIO()
    with redirect_stderr(f):
        # suppress unnecessary stdout
        verification_result = kb_public_key.verify(pgp_msg)
    if not verification_result:
        raise VerificationError("API response did not verify with Keybase's public key")

    good_signatures = list(verification_result.good_signatures)
    if len(good_signatures) != 1:
        logger.error(f"good_signatures = {good_signatures}")
        raise VerificationError(f"Expected 1 valid signature, got {len(good_signatures)} from {specific_url}")

    return json.loads(good_signatures[0].subject)



KEYBASE_PGP_VERIFICATION_KEY = """
-----BEGIN PGP PUBLIC KEY BLOCK-----
Version: GnuPG/MacGPG2 v2.0.22 (Darwin)
Comment: GPGTools - https://gpgtools.org

<KEY>
-----END PGP PUBLIC KEY BLOCK-----
"""
# if the format of anything in here changes, make sure to # bump the version of StampedMerkleRoot # hash of validated sig over merkle root payload # fetch from the keybase API and verify a bunch of things # fetch the current merkle root from the keybase api # extract the signature and signed payload into a PGP message for verification # the message itself is too big to include with OTS data in a chat message # so let's take the hash of it and use that instead. this is kind of a # bummer because it's almost OK to use the whole thing. # this function will raise an exception if PGP verification fails # as a sanity check, ensure that the signed payload matches # the raw json payload which is also present in the keybase # API response # and most importantly, that the root hash that was signed is the same as the one that's # at the top level of the API response body. # load the raw pgp message # load keybase's claimed public key # see: https://keybase.io/docs/server_security/our_merkle_key # verify it: https://pgpy.readthedocs.io/en/latest/examples.html#verifying-things # suppress unnecessary stdout -----BEGIN PGP PUBLIC KEY BLOCK----- Version: GnuPG/MacGPG2 v2.0.22 (Darwin) Comment: GPGTools - https://gpgtools.org  <KEY> -----END PGP PUBLIC KEY BLOCK-----