Building a Bitcoin P2TR Tx by Hand

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import hashlib
import schnorr_lib

def convertbits(data, frombits, tobits, pad=True):
    """General power-of-2 base conversion."""
    # From https://github.com/sipa/bech32/blob/master/ref/python/segwit_addr.py
    acc = 0
    bits = 0
    ret = []
    maxv = (1 << tobits) - 1
    max_acc = (1 << (frombits + tobits - 1)) - 1
    for value in data:
        if value < 0 or (value >> frombits):
            return None
        acc = ((acc << frombits) | value) & max_acc
        bits += frombits
        while bits >= tobits:
            bits -= tobits
            ret.append((acc >> bits) & maxv)
    if pad:
        if bits:
            ret.append((acc << (tobits - bits)) & maxv)
    elif bits >= frombits or ((acc << (tobits - bits)) & maxv):
        return None
    return ret


def bech32_decode(bech):
    """Validate a Bech32/Bech32m string, and determine HRP and data."""
    # From https://github.com/sipa/bech32/blob/master/ref/python/segwit_addr.py
    CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
    if ((any(ord(x) < 33 or ord(x) > 126 for x in bech)) or
            (bech.lower() != bech and bech.upper() != bech)):
        return (None, None)
    bech = bech.lower()
    pos = bech.rfind('1')
    if pos < 1 or pos + 7 > len(bech) or len(bech) > 90:
        return (None, None)
    if not all(x in CHARSET for x in bech[pos + 1:]):
        return (None, None)
    hrp = bech[:pos]
    data = [CHARSET.find(x) for x in bech[pos + 1:]]
    return (hrp, data[:-6])


def p2tr_address_to_scriptpubkey(address: str) -> bytes:
    """Convert p2tr address into scriptpubkey (locking script)"""
    if not any(address.startswith(item) for item in ["tb1p", "bc1p", "bcrt1p"]):
        raise Exception("invalid p2tr address")
    hrp, data = bech32_decode(address)
    if hrp is None or data is None:
        raise Exception("bech32_decode failed")
    decoded = convertbits(data[1:], 5, 8, False)
    if decoded is None:
        raise Exception("convertbits failed")
    binary_str = ""
    for number in decoded:
        binary_str += bin(number)[2:].zfill(8)  # Convert the integer to binary and pad with zeros to 8 digits
    hex_str = hex(int(binary_str, 2))[2:]  # Convert binary to hexadecimal and remove the prefix '0x'
    return bytes.fromhex("5120" + hex_str)


def sha256(data: bytes) -> bytes:
    """One round of SHA256"""
    return hashlib.sha256(data).digest()


def reverse_byte_order(input: bytes) -> bytes:
    """Reverse byte order"""
    return input[::-1]


def varint_len(data: bytes) -> bytes:
    """returns the length of the input as a variable integer"""
    l = len(data)
    if l < int('fd', 16):
        varint = l.to_bytes(1, byteorder="little", signed=False)
    elif l < int('ffff', 16):
        varint = bytes.fromhex("fd") + l.to_bytes(2, byteorder="little", signed=False)
    else:
        raise Exception("This function only handles up to 0xffff bytes")
    return varint


# Private keys
# Please make sure the private keys is related to utxos_to_spend
private_keys = [
    "d5d3a71a06e45f9c215f764612b209cc41d086651a0aa62f78551408cc64ec2b",  # tb1pevpm69vf0my4ge9m0ne9m8s056mma2uxpw9u9m76cug2vcsmu6ask9u37q
    "9ab7cdd3bfc955da43de564fcebb9b43fbb51d3e31f1b0cfc7875f06d22cd157",  # tb1plmq7t40y68ve96pppv6mqe78kxyuhaq2hl4anxym5jaywcz00xeqthwx6l
]

utxos_to_spend = [
    {
        # https://api.blockcypher.com/v1/btc/test3/txs/01b748006933d1827c44cf510635fc1ba02eeee940fb877cb92495bc5a01db77
        "txid": "01b748006933d1827c44cf510635fc1ba02eeee940fb877cb92495bc5a01db77",
        "vout": 1,
        "value": 3000,
        "scriptpubkey": "5120cb03bd15897ec95464bb7cf25d9e0fa6b7beab860b8bc2efdac710a6621be6bb"  # i.e. locking script
    },
    {
        # https://api.blockcypher.com/v1/btc/test3/txs/3eec92fec94a12ca67e245311a45dafbe25c972c120c351870b9453ceb60e82b
        "txid": "3eec92fec94a12ca67e245311a45dafbe25c972c120c351870b9453ceb60e82b",
        "vout": 0,
        "value": 99800,
        "scriptpubkey": "5120fec1e5d5e4d1d992e8210b35b067c7b189cbf40abfebd9989ba4ba47604f79b2"  # i.e. locking script
    }
]

tx_outputs = [
    {
        "value": 102520,
        "scriptpubkey_address": "tb1pwxpw8xkd2w9e0d38kvlrzqpnszhnaqwuvcpf0l2nlc3t5fhujgzs58jxsz"  # receive address
    }
]

# data for the tx I want to create
version: bytes = (1).to_bytes(4, byteorder="little", signed=False)  # version 1, 01000000
sequence: bytes = 0xfffffffd.to_bytes(4, byteorder="little", signed=False)  # enables replace-by-fee and absolute lock-time but disables relative lock-time
lock_time: bytes = bytes.fromhex("00000000")
hash_type: bytes = bytes.fromhex("00")  # SIGHASH_DEFAULT

# ----------
# CONSTRUCTING SIGHASH x INPUT (taproot)
# ----------
# sha_prevouts (32) = SHA256(serialization of all input outpoints)
prevouts: bytes = b''
for utxo_to_spend in utxos_to_spend:
    prevouts += reverse_byte_order(bytes.fromhex(utxo_to_spend["txid"]))
    prevouts += utxo_to_spend["vout"].to_bytes(4, byteorder="little", signed=False)
sha_prevouts = sha256(prevouts)

# sha_amounts (32): the SHA256 of the serialization of all spent output amounts
amounts: bytes = b''
for utxo_to_spend in utxos_to_spend:
    amounts += utxo_to_spend["value"].to_bytes(8, byteorder="little", signed=False)
sha_amounts = sha256(amounts)

# sha_scriptpubkeys (32): the SHA256 of all spent outputs' scriptPubKeys, serialized as script inside CTxOut
scriptpubkeys: bytes = b''
for utxo_to_spend in utxos_to_spend:
    if not (utxo_to_spend["scriptpubkey"].startswith("5120") and len(utxo_to_spend["scriptpubkey"]) == 68):
        raise Exception("invalid scriptpubkey, only p2tr is supported")
    scriptpubkey: bytes = bytes.fromhex(utxo_to_spend["scriptpubkey"])
    scriptpubkeys += varint_len(scriptpubkey)
    scriptpubkeys += scriptpubkey
sha_scriptpubkeys: bytes = sha256(scriptpubkeys)

# sha_sequences (32): the SHA256 of the serialization of all input nSequence.
sequences: bytes = b''
for utxo_to_spend in utxos_to_spend:
    sequences += sequence
sha_sequences: bytes = sha256(sequences)

# sha_outputs (32): the SHA256 of the serialization of all outputs in CTxOut format.
outputs: bytes = b''
for tx_output in tx_outputs:
    scriptpubkey: bytes = p2tr_address_to_scriptpubkey(tx_output["scriptpubkey_address"])
    outputs += tx_output["value"].to_bytes(8, byteorder="little", signed=False)
    outputs += varint_len(scriptpubkey)
    outputs += scriptpubkey
sha_outputs: bytes = sha256(outputs)

# spend_type (1): equal to (ext_flag * 2) + annex_present, where annex_present is 0 if no annex is present,
# or 1 otherwise (the original witness stack has two or more witness elements,
# and the first byte of the last element is 0x50)
spend_type = bytes.fromhex("00")

preimage_hashes = []
for i, _ in enumerate(utxos_to_spend):
    # input_index (4): index of this input in the transaction input vector. Index of the first input is 0
    input_index: bytes = i.to_bytes(4, byteorder="little", signed=False)

    # https://github.com/bitcoin/bips/blob/master/bip-0341.mediawiki
    sig_to_hash = b'\x00' + hash_type + version + lock_time \
                  + sha_prevouts + sha_amounts + sha_scriptpubkeys + sha_sequences \
                  + sha_outputs + spend_type + input_index

    print("sig_to_hash =", sig_to_hash.hex())

    tag_hash = sha256("TapSighash".encode())
    preimage_hash = sha256(tag_hash + tag_hash + sig_to_hash)
    print("preimage_hash =", preimage_hash.hex())
    preimage_hashes.append(preimage_hash)


# ----------
# SIGNING
# ----------
assert (len(preimage_hashes) == len(private_keys))
signatures = []
for i, private_key in enumerate(private_keys):
    sig = schnorr_lib.schnorr_sign(preimage_hashes[i], private_key)
    signatures.append(sig)


# ----------
# CONSTRUCTING WITNESS
# ----------
witnesses: bytes = b''
for witness_sig in signatures:
    witness_count = (1).to_bytes(1, byteorder="little", signed=False)  # Witness Count 1 means P2TR (Key Path)
    witness_sig_len: bytes = varint_len(witness_sig)
    witnesses += witness_count
    witnesses += witness_sig_len
    witnesses += witness_sig


# ----------
# TX READY
# ----------
marker = bytes.fromhex("00")
flag = bytes.fromhex("01")
txin_count: bytes = len(utxos_to_spend).to_bytes(1, byteorder="little", signed=False)
txout_count: bytes = len(tx_outputs).to_bytes(1, byteorder="little", signed=False)

txins = b''
for utxo_to_spend in utxos_to_spend:
    txins += reverse_byte_order(bytes.fromhex(utxo_to_spend["txid"]))
    txins += utxo_to_spend["vout"].to_bytes(4, byteorder="little", signed=False)
    txins += bytes.fromhex("00")
    txins += sequence

# witness serialization format for tx: see https://github.com/bitcoin/bips/blob/master/bip-0144.mediawiki
tx = version \
     + marker + flag \
     + txin_count + txins \
     + txout_count + outputs \
     + witnesses \
     + lock_time

# Since the signature data is different every time, tx hex may be different every time
raw_tx = tx.hex()  # this tx can be broadcast
print("raw_tx =", raw_tx)  # 0100000000010277db015abc9524b97c87fb40e9ee2ea01bfc350651cf447c82d133690048b7010100000000fdffffff2be860eb3c45b97018350c122c975ce2fbda451a3145e267ca124ac9fe92ec3e0000000000fdffffff0178900100000000002251207182e39acd538b97b627b33e31003380af3e81dc660297fd53fe22ba26fc920501404a43ea0ee1ffef5ddd894cc271c6dce2265f1fe02f066d4f7fdb0c6e18eda325c9e0a6eef6ace0ef2380cc57c655f175cd0ad180a8c58e788756d682c8e21d5c01403bfe16498e4712951bb1659c21d7be53e63c28df59d19f76065e412f67adf15eba76fce9b595342f4949572135c8502f8586cf7e6d2c9ddb9c9bdfc43ead9d3800000000

# result tx_id:
# https://mempool.space/testnet/tx/0195fce44d57453a0cdf8ef1e36ed022994f59278a9e915503e04e9a11f06892

from typing import Tuple, Optional
from binascii import unhexlify
import hashlib
import os

# From: https://github.com/BitPolito/schnorr-sig/blob/master/schnorr_lib.py

# Elliptic curve parameters
p = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEFFFFFC2F
n = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
G = (0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798,
     0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8)

# Points are tuples of X and Y coordinates
# the point at infinity is represented by the None keyword
Point = Tuple[int, int]


# Get bytes from an int
def bytes_from_int(a: int) -> bytes:
    return a.to_bytes(32, byteorder="big")


# Get bytes from a point
def bytes_from_point(P: Point) -> bytes:
    return bytes_from_int(x(P))


# Get an int from bytes
def int_from_bytes(b: bytes) -> int:
    return int.from_bytes(b, byteorder="big")


# Get an int from hex
def int_from_hex(a: hex) -> int:
    return int.from_bytes(unhexlify(a), byteorder="big")


# Get x coordinate from a point
def x(P: Point) -> int:
    return P[0]


# Get y coordinate from a point
def y(P: Point) -> int:
    return P[1]


# Point addition
def point_add(P1: Optional[Point], P2: Optional[Point]) -> Optional[Point]:
    if P1 is None:
        return P2
    if P2 is None:
        return P1
    if (x(P1) == x(P2)) and (y(P1) != y(P2)):
        return None
    if P1 == P2:
        lam = (3 * x(P1) * x(P1) * pow(2 * y(P1), p - 2, p)) % p
    else:
        lam = ((y(P2) - y(P1)) * pow(x(P2) - x(P1), p - 2, p)) % p
    x3 = (lam * lam - x(P1) - x(P2)) % p
    return x3, (lam * (x(P1) - x3) - y(P1)) % p


# Point multiplication
def point_mul(P: Optional[Point], d: int) -> Optional[Point]:
    R = None
    for i in range(256):
        if (d >> i) & 1:
            R = point_add(R, P)
        P = point_add(P, P)
    return R


# Note:
# This implementation can be sped up by storing the midstate
# after hashing tag_hash instead of rehashing it all the time
# Get the hash digest of (tag_hashed || tag_hashed || message)
def tagged_hash(tag: str, msg: bytes) -> bytes:
    tag_hash = hashlib.sha256(tag.encode()).digest()
    return hashlib.sha256(tag_hash + tag_hash + msg).digest()


# Check if a point is at infinity
def is_infinity(P: Optional[Point]) -> bool:
    return P is None


# Get xor of bytes
def xor_bytes(b0: bytes, b1: bytes) -> bytes:
    return bytes(a ^ b for (a, b) in zip(b0, b1))


# Get a point from bytes
def lift_x_square_y(b: bytes) -> Optional[Point]:
    x = int_from_bytes(b)
    if x >= p:
        return None
    y_sq = (pow(x, 3, p) + 7) % p
    y = pow(y_sq, (p + 1) // 4, p)
    if pow(y, 2, p) != y_sq:
        return None
    return x, y


def lift_x_even_y(b: bytes) -> Optional[Point]:
    P = lift_x_square_y(b)
    if P is None:
        return None
    else:
        return x(P), y(P) if y(P) % 2 == 0 else p - y(P)


# Check if an int is square
def is_square(a: int) -> bool:
    return int(pow(a, (p - 1) // 2, p)) == 1


# Check if a point has even y coordinate
def has_even_y(P: Point) -> bool:
    return y(P) % 2 == 0


# Generate auxiliary random of 32 bytes
def get_aux_rand() -> bytes:
    return os.urandom(32)


# Extract R_x int value from signature
def get_int_R_from_sig(sig: bytes) -> int:
    return int_from_bytes(sig[0:32])


# Extract s int value from signature
def get_int_s_from_sig(sig: bytes) -> int:
    return int_from_bytes(sig[32:64])


# Generate Schnorr signature
def schnorr_sign(msg: bytes, privateKey: str) -> bytes:
    if len(msg) != 32:
        raise ValueError('The message must be a 32-byte array.')
    d0 = int_from_hex(privateKey)
    if not (1 <= d0 <= n - 1):
        raise ValueError(
            'The secret key must be an integer in the range 1..n-1.')
    P = point_mul(G, d0)
    assert P is not None
    d = d0 if has_even_y(P) else n - d0
    t = xor_bytes(bytes_from_int(d), tagged_hash("BIP0340/aux", get_aux_rand()))
    k0 = int_from_bytes(tagged_hash("BIP0340/nonce", t + bytes_from_point(P) + msg)) % n
    if k0 == 0:
        raise RuntimeError('Failure. This happens only with negligible probability.')
    R = point_mul(G, k0)
    assert R is not None
    k = n - k0 if not has_even_y(R) else k0
    e = int_from_bytes(tagged_hash("BIP0340/challenge", bytes_from_point(R) + bytes_from_point(P) + msg)) % n
    sig = bytes_from_point(R) + bytes_from_int((k + e * d) % n)

    if not schnorr_verify(msg, bytes_from_point(P), sig):
        raise RuntimeError('The created signature does not pass verification.')
    return sig


# Verify Schnorr signature
def schnorr_verify(msg: bytes, pubkey: bytes, sig: bytes) -> bool:
    if len(msg) != 32:
        raise ValueError('The message must be a 32-byte array.')
    if len(pubkey) != 32:
        raise ValueError('The public key must be a 32-byte array.')
    if len(sig) != 64:
        raise ValueError('The signature must be a 64-byte array.')
    P = lift_x_even_y(pubkey)
    r = get_int_R_from_sig(sig)
    s = get_int_s_from_sig(sig)
    if (P is None) or (r >= p) or (s >= n):
        return False
    e = int_from_bytes(tagged_hash("BIP0340/challenge", sig[0:32] + pubkey + msg)) % n
    R = point_add(point_mul(G, s), point_mul(P, n - e))
    if (R is None) or (not has_even_y(R)):
        # print("Please, recompute the sign. R is None or has even y")
        return False
    if x(R) != r:
        # print("There's something wrong")
        return False
    return True