In the first post we got the board set up and read one block from the chain. This time, we'll tackle the somewhat daunting task of constructing, signing, and broadcasting a transaction to the chain.
Signing
This may be the most technical part of the whole process. There's a lot of scary stuff like cryptographic functions and fiddling with bits. The good news is that all this stuff is available in open-source libraries and old tutorial posts.
Here is some minimal code needed to sign a custom_json. There's room to add other operations if we need them down the line. To keep things clean, this will be saved as hivesign.py separate from the main script.
# hivesign.py
import datetime
from hashlib import sha256
import rapidjson as json
import secp256k1prp as secp256k1
import varint
# simplified for this demo
ops_dict = {'custom_json':18}
def fix_custom_json(op):
if not isinstance(op['json'], str):
op['json'] = json.dumps(op['json'])
return op
def serialize_custom_json(op):
b = b''
b += varint.encode(ops_dict['custom_json'])
b += varint.encode(len(op['required_auths']))
if op['required_auths']:
b += varint.encode(len(op['required_auths'][0]))
b += op['required_auths'][0].encode()
b += varint.encode(len(op['required_posting_auths']))
if op['required_posting_auths']:
b += varint.encode(len(op['required_posting_auths'][0]))
b += op['required_posting_auths'][0].encode()
b += varint.encode(len(op['id']))
b += op['id'].encode()
b += varint.encode(len(op['json']))
b += op['json'].encode()
return b
def serialize_op(op_type, op):
if (op_type == 'custom_json'):
op = fix_custom_json(op)
return serialize_custom_json(op)
def serialize_trx(trx):
b = b''
b += trx['ref_block_num'].to_bytes(2, 'little')
b += trx['ref_block_prefix'].to_bytes(4, 'little')
ts = int(datetime.datetime.fromisoformat(trx['expiration']+'+00:00').timestamp())
b += ts.to_bytes(4, 'little')
b += varint.encode(len(trx['operations']))
for op in trx['operations']:
b += serialize_op(*op)
b += int(0).to_bytes(1, 'little')
return b
def derive_digest(b):
chain_id = 'beeab0de00000000000000000000000000000000000000000000000000000000'
m = bytes.fromhex(chain_id) + b
return sha256(m).digest()
def is_canonical(sig):
return (
not (sig[0] & 0x80)
and not (sig[0] == 0 and not (sig[1] & 0x80))
and not (sig[32] & 0x80)
and not (sig[32] == 0 and not (sig[33] & 0x80))
)
def get_signature(p, digest):
i = 0
ndata = secp256k1.ffi.new('const int *ndata')
ndata[0] = 0
while True:
ndata[0] += 1
privkey = secp256k1.PrivateKey(p, raw=True)
sig = secp256k1.ffi.new('secp256k1_ecdsa_recoverable_signature *')
signed = secp256k1.lib.secp256k1_ecdsa_sign_recoverable(
privkey.ctx,
sig,
digest,
privkey.private_key,
secp256k1.ffi.NULL,
ndata
)
assert signed == 1
signature, i = privkey.ecdsa_recoverable_serialize(sig)
if is_canonical(signature):
i += 4
i += 27
break
result = i.to_bytes(1, 'little')
result += signature
return result
def sign_trx(trx, key):
b = serialize_trx(trx)
trx_id = sha256(b).hexdigest()[:40]
d = derive_digest(b)
sig = get_signature(key, d)
trx['signatures'].append(sig.hex())
return [trx, trx_id]
Construction and Broadcasting
Now we're ready to use the sign_trx function from hivesign.py in the main script. We also made a nice do_condenser function to easily use anything from that API, which will give us lots of options in the future. This can be its own module too, when the main script gets a little busier.
import datetime
import requests
from hivesign import sign_trx
class Dummy(): pass
script = Dummy()
script._id = 1
script.node = 'https://api.deathwing.me'
script.account = 'orange-pi'
def get_props(expiration=60):
result = do_condenser('get_dynamic_global_properties')
props = {}
props['head_block_num'] = result['head_block_number']
props['ref_block_num'] = props['head_block_num'] & 0xFFFF
props['ref_block_prefix'] = int.from_bytes(bytes.fromhex(result['head_block_id'][8:16]), 'little')
e = datetime.datetime.fromisoformat(result['time']) + datetime.timedelta(seconds=expiration)
props['expiration'] = e.isoformat()
return props
def contruct_trx(operations, key):
props = get_props()
trx = {
'expiration':props['expiration'],
'ref_block_num':props['ref_block_num'],
'ref_block_prefix':props['ref_block_prefix'],
'operations':operations,
'extensions':[],
'signatures':[]
}
return sign_trx(trx, key)
def do_condenser(method, params=[]):
if not isinstance(params, list):
params = [params]
data = {'jsonrpc':'2.0', 'method':f'condenser_api.{method}', 'params':params, 'id':script._id}
with script.session.post(script.node, json=data) as r:
script._id += 1
return r.json().get('result')
def broadcast(operations, key):
trx, trx_id = contruct_trx(operations, key)
do_condenser('broadcast_transaction', trx)
return [trx, trx_id]
def main():
custom = {
'id':'orange-pi-test',
'json':{
'message':f'hello from {script.account}',
'numbers':456789
},
'required_auths':[],
'required_posting_auths':[script.account]
}
with requests.Session() as script.session:
trx, trx_id = broadcast([['custom_json', custom]], get_key(script.account))
print(f'trx_id: {trx_id}')
print(f'trx: {trx}')
if __name__ == '__main__':
main()
And... It works! The transaction is on the blockchain for anyone to see.
Next Time
Now we're ready to stream blocks, read transactions, and react accordingly. I'm still open to ideas for what this thing should actually do.
Thanks for reading!