This is a code-intensive chapter. In the previous two chapters, we have examined the theory of cryptocurrency transactions and seen how the merkle root can be used to verify that the transactions in a block have not been tampered with. In this chapter, we are going to put this theoretical knowledge into practice by implementing Python code for processing Helium transactions.
Transaction Processing Code Walkthrough
The canonical transaction processing workflow is as follows. An entity creates a transaction and then places the transaction on the Helium peer-to-peer network for processing. The transaction is then picked up by Helium mining nodes. These nodes are engaged in the activity of processing transactions and mining blocks in order to reap rewards. In the case of Bitcoin, the reward is the issuance of some new bitcoins and the transaction fees which are attached to the transactions in the block that is successfully mined by the node. A transaction may have a zero transaction fee attached to it. The quantum of the transaction fee is within the sole discretion of the entity transferring value. A miner can elect to include a transaction in a block that it is mining or decline such an inclusion. Therefore, different miners will typically be mining different blocks at the same time. If a miner succeeds in adding a block to the blockchain, we say that the miner has mined the block. The transaction processing workflow is identical in Helium.
When a miner receives a block, the very first thing that it will do is to verify the integrity of the transaction. Verification means that the transaction is examined to ascertain that it contains valid values. For example, the transaction locktime parameter will be invalid if it is a negative number. An important aspect of verification is to ascertain that the transferor of value is spending values that it owns. This chapter addresses transaction validation. If a transaction is valid, it can become part of a block that a miner intends to mine.
After a miner has mined a block, it adds the block to its local blockchain. The miner then broadcasts the newly mined block on the Helium network. Other miners who receive this block will then add it to their local blockchain.
Let us suppose that a miner mines a block with an invalid transaction. It may maliciously add this block to its local copy of the blockchain. But other miners will decline to add the block to their local blockchains, because they will ascertain that the block contains an invalid transaction.
It is worth noting that miners on a Helium network may have local blockchains that differ from each other. But due to the distributed consensus mechanism, the network will always converge to a dominant blockchain. We will discuss this in a subsequent chapter. Because of the manner in which distributed consensus works, trust is not needed in the network. We do not care that a mining node is acting mala fides. This is Satoshi Nakamoto’s breakthrough insight.
Alright, let us start with the code walkthrough. Copy the program code that follows into a file called tx.py and save it the transaction directory.
You will notice that we have typically wrapped function bodies in this module within exception blocks.
In dynamically typed languages such as Python, the debugging process consists primarily of detecting and fixing type errors at runtime. This burden rises exponentially as program size increases. This factor makes interpreted languages with dynamic typing unsuitable for large-scale program development. In contrast, Go and C++ are very strictly typed. Go and C++ programs will simply refuse to compile if there is a type error in the program. In Go, once the program has compiled, debugging runtime errors is typically very fast (probably O(n)).
The workhorse function in the tx module is validate_transaction
; it verifies the integrity of a transaction. This function performs the following tests:
- 1.
Verify that required transaction attributes are present.
- 2.
The locktime is greater than or equal to zero.
- 3.
The locktime value is less than the maximum specified in the hconfig module.
- 4.
The version number is valid.
- 5.
The transaction ID has a valid format.
- 6.
The vin list has positive length if the transaction is not in the genesis block or if the transaction is not a coinbase transaction.
- 7.
For genesis block transactions or a coinbase transaction, the vin list is empty.
- 8.
The vin list is not greater than the maximum allowable length specified in the hconfig module.
- 9.
The vin list has valid format.
- 10.
The vout list has positive length.
- 11.
The vout list is not greater than the maximum allowable length specified in the hconfig module.
- 12.
The vout list elements have valid format.
- 13.
The transaction implements a p2pkhash script.
- 14.
The reference to a previous transaction those output is consumed is valid.
- 15.
The total value spent is less than or equal to the total spendable value.
- 16.
All of the spent values in a transaction are greater than zero.
- 17.
The transaction inputs can be unlocked. That is, an entity only consumes inputs that it owns.
- 18.
A coinbase transaction value is not consumed before its locktime has expired.
In the tx module, the prevtx_value function references an unspent value in a previous transaction. This function is mocked in the unit tests since we have not implemented a LevelDB database to get unspent transaction amounts. We will look at Helium databases in the next chapter. The transaction_fee function computes the transaction fee for the transaction.
The unlock_transaction_fragment function determines whether values in a previous transaction can be unlocked for use in the present transaction. This function implements the p2pkhash stack processor that was discussed two chapters previously:
"""
tx.py: The tx module defines the structure of Helium transactions and implements
basic transaction validation operations.
"""
import hconfig
import hblockchain as hchain
import json
import rcrypt
import secrets
import pdb
import logging
"""
log debugging messages to the file debug.log
"""
logging.basicConfig(filename="debug.log",filemode="w", \
format='%(asctime)s:%(levelname)s:%(message)s',level=logging.DEBUG)
"""
Each transaction is modelled as a dictionary:
{
"transactionid": <string>
"version": <integer>
"locktime": <integer>
"vin": <list<dictionary>>
"vout": <list<<dictionary>>
}
transactionid is the id of the the present transaction.
vin is a list of spendable inputs that are owned by the entity spending them.
Each spendable input comes from the vout list of a previous transaction.
Each vin is a list and each element is a dictionary object with the
following structure:
vin element:
{
"txid": <string>
"vout_index": <int>
"ScriptSig": <list<string>>
}
txid is the transaction id of a previous transaction. vout_index is an index
into this previous transaction's vout list.
vout is a list and each element in the vout list is a dictionary with the
following structure:
vout element:
{
"value": <integer>,
"ScriptPubKey": <list<string>>
}
"""
def create_transaction(transaction: 'dictionary', zero_inputs: 'boolean'=False) -> 'bool':
"""
creates a transaction. Receives a transaction object and a predicate.
zero_inputs is True if the transaction is in the genesis block or if the transaction
is a coinbase transaction, otherwise zero_inputs is False.
Returns False if the transaction parameters are invalid. Otherwise returns True
"""
if validate_transaction(transaction, zero_inputs) == False: return False
return True
def validate_transaction(trans: "dictionary", zero_inputs: "boolean"=False) -> "bool":
"""
verifies that a transaction has valid values.
receives a transaction and a predicate.
zero_inputs is True if the transaction is in the genesis block or if the transaction
is a coinbase transaction, otherwise zero_inputs is False.
The following transaction validation tests are performed:
(1) The required attributes are present.
(2) The locktime is greater than or equal to zero.
(3) The locktime is less than a prescribed value.
(4) The version number is valid.
(5) The transaction ID has a valid format.
(6) The vin list has positive length for transactions that are not in the genesis block and are not coinbase transactions.
(7) The vin list is empty for coinbase transactions.
(8) The genesis block does not have any inputs
(9) The vin list is not greater than the maximum allowable length.
(10) The vin list has valid format.
(11) The vout list has positive length.
(12) The vout list is not greater than the maximum allowable length.
(13) The vout list elements have valid format.
(14) The The vout list implements a p2pkhash script.
(15) The reference to a previous transaction ID is valid.
(16) The total value spent is less than or equal to the total spendable value.
(17) The spent values are greater than zero.
(18) The index values in the vin array reference valid elements in the
vout array of the previous transaction.
(19) The transaction inputs can be spent
"""
try:
if type(trans) != dict:
raise(ValueError("not dict type"))
# Verify that required attributes are present
if trans.get("transactionid") == None: return False
if 'version' not in trans: return False
if trans['locktime'] == None: return False
if trans['vin'] == None: return False
if trans['vout'] == None: return False
# validate the format of the transaction id
if rcrypt.validate_SHA256_hash(trans['transactionid']) == False:
raise(ValueError("not SHA-256 hash error"))
# validate the transaction version and locktime values
if trans['version'] != hconfig.conf["VERSION_NO"]:
raise(ValueError("improper version no"))
if trans['locktime'] < 0:
raise(ValueError("invalid locktime"))
# genesis block or a coinbase transaction does not have any inputs
if zero_inputs == True and len(trans["vin"]) > 0:
raise(ValueError("genesis block cannot have inputs"))
# validate the vin elements
# there are no vin inputs for a genesis block transaction
spendable_fragments = []
for vin_element in trans['vin']:
if validate_vin(vin_element) == False: return False
tx_key = vin_element['txid'] + '_' + str(vin_element['vout_index'])
spendable_fragment = prevtx_value(tx_key)
if spendable_fragment == False:
raise(ValueError("invalid spendable input for transaction"))
else:
spendable_fragments.append(spendable_fragment)
# validate the transaction's vout list
if len(trans['vout']) > hconfig.conf['MAX_OUTPUTS'] or len(trans['vout']) <= 0:
raise(ValueError("vout list length error"))
for vout_element in trans['vout']:
if validate_vout(vout_element) == False: return False
# validate the transaction fee
if zero_inputs == False:
if (transaction_fee(trans, spendable_fragments)) == False: return False
# test that the transaction inputs are unlocked
ctr = 0
for vin in trans['vin']:
if unlock_transaction_fragment(vin, spendable_fragments[ctr]) == False:
raise(ValueError("failed to unlock transaction"))
ctr += 1
except Exception as err:
logging.debug('validate_transaction: exception: ' + str(err))
return False
return True
def validate_vin(vin_element: 'dictionary') -> bool:
"""
tests whether a vin element has valid values
returns True if the vin is valid, False otherwise
"""
try:
if vin_element['vout_index'] < 0: return False
if len(vin_element['ScriptSig']) != 2: return False
if len(vin_element['ScriptSig'][0]) == 0: return False
if len(vin_element['ScriptSig'][1]) == 0: return False
if rcrypt.validate_SHA256_hash(vin_element['txid']) == False:
raise(ValueError("txid is invalid"))
except Exception as err:
print("validate_vin exception" + str(err))
logging.debug('validate_vin: exception: ' + str(err))
return False
return True
def validate_vout(vout_element: 'dictionary') -> bool:
"""
tests whether a vout element has valid values
returns True if the vout is valid, False otherwise
"""
try:
if vout_element['value'] <= 0:
raise(ValueError("value is <= 0"))
# validate the p2pkhash script
if len(vout_element["ScriptPubKey"]) != 5:
raise(ValueError("ScriptPubKey length error"))
if vout_element["ScriptPubKey"][0] != '<DUP>':
raise(ValueError("ScriptPubKey <DUP> error"))
if vout_element["ScriptPubKey"][1] != '<HASH-160>':
raise(ValueError("ScriptPubKey <HASH-160> error"))
if vout_element["ScriptPubKey"][3] != '<EQ-VERIFY>':
raise(ValueError("ScriptPubKey <EQ-VERIFY> error"))
if vout_element["ScriptPubKey"][4] != '<CHECK-SIG>':
raise(ValueError("ScriptPubKey <CHECK-SIG> error"))
if len(vout_element["ScriptPubKey"][2]) == 0:
raise(ValueError("ScriptPubKey RIPEMD-160 hash error"))
except Exception as err:
print("validate_vout exception" + str(err))
logging.debug('validate_vout: exception: ' + str(err))
return False
return True
def prevtx_value(txkey: "string") -> 'bool':
"""
examines the chainstate database and returns the transaction fragment:
{
"pkkhash": <string>,
"value": <int>,
"spent": <bool>,
"tx_chain": <string>
}
receives a transaction fragment key:
"previous txid" + "_" + str(prevtx_vout_index)
Returns False if the transaction if does not exist, or if the
value has been spent or if the value is not a positive integer
otherwise returns the previous transaction fragment data
"""
return "mocked value"
def transaction_fee(trans: 'dictionary', value_list: 'list<dictionary>' ) -> 'integer or bool':
"""
calculates the transaction fee for a transaction.
Receives a transaction and a list of chainstate transaction fragments
Returns the transaction fee (in helium_cents) or False if there is an error.
There is no transaction fee for the genesis block or coinbase transactions.
"""
try:
spendable_value = 0
spent_value = 0
for val in value_list:
spendable_value += val["value"]
for vout in trans['vout']:
spent_value += vout['value']
if spendable_value <= 0:
raise(ValueError("spendable value <= 0 "))
if spent_value <= 0:
raise(ValueError("spent value <= 0 "))
if spendable_value < spent_value:
raise(ValueError("spendable value < spent value"))
except Exception as err:
logging.debug('transaction_fee: exception: ' + str(err))
return False
return spendable_value - spent_value
def unlock_transaction_fragment(vinel: "dictionary", fragment: "dictionary") -> "boolean":
"""
unlocks a previous transaction fragment using the p2pkhash script.
Executes the script and returns False if the transaction is not unlocked
Receives: the consuming vin and the previous transaction fragment consumed
by the vin element.
"""
try:
execution_stack = []
result_stack = []
# make the execution stack for a p2pkhash script
# since we only have one type of script in Helium
# we can use hard-coded values
execution_stack.append('SIG')
execution_stack.append('PUBKEY')
execution_stack.append('<DUP>')
execution_stack.append('<HASH_160>')
execution_stack.append('HASH-160')
execution_stack.append('<EQ-VERIFY>')
execution_stack.append('<CHECK_SIG>')
# Run the p2pkhash execution stack
result_stack.insert(0, vinel['ScriptSig'][0])
result_stack.insert(0, vinel['ScriptSig'][1])
result_stack.insert(0, vinel['ScriptSig'][1])
hash_160 = rcrypt.make_SHA256_hash(vinel['ScriptSig'][1])
hash_160 = rcrypt.make_RIPEMD160_hash(hash_160)
result_stack.insert(0, hash_160)
result_stack.insert(0, fragment["pkhash"])
# do the EQ_VERIFY operation
tmp1 = result_stack.pop(0)
tmp2 = result_stack.pop(0)
# test for RIPEMD-160 hash match
if tmp1 != tmp2:
raise(ValueError("public key match failure"))
# test for a signature match
ret = rcrypt.verify_signature(vinel['ScriptSig'][1], \
vinel['ScriptSig'][1], vinel['ScriptSig'][0])
if ret == False:
raise(ValueError("signature match failure"))
except Exception as err:
logging.debug('unlock_transaction_fragment: exception: ' + str(err))
return False
return True
Transaction Processing Unit Tests
Copy the following code into a file named test_tx.py and copy this file into the unit_tests directory
. Traverse to the unit_tests directory and run the tests:1
$(virtual) pytest test_tx.py -s
You will note that at the outset we create a symbolic transaction as well as a symbolic previous transaction that is used by the unit tests to create a pair of transactions with randomized values. The randomize_list function provides randomized access to vin and vout list elements for our unit tests. This module makes extensive use of pytest mocks, so if you are not conversant with this technology, you will want to review Appendix 3. The last two unit tests are noteworthy for your consideration; they test the unlocking of previous transaction outputs.
After executing the command line, you should see 35 unit tests passing:
"""
test_tx.py: implements unit tests for the transactions module
"""
import rcrypt
import hconfig
import hblockchain as bchain
import tx
import pytest
import pdb
import secrets
import random
previous_tx = None
prev_tx_keys = []
num_prev_tx_outputs = 0
##################################
# Synthetic Previous Transaction
##################################
def make_synthetic_previous_transaction(num_outputs):
"""
makes synthetic previous transaction with randomized values
we abstract away the inputs to the previous transaction.
"""
transaction = {}
transaction['version'] = 1
transaction['transactionid'] = rcrypt.make_uuid()
transaction['locktime'] = 0
transaction['vin'] = []
transaction['vout'] = []
def make_synthetic_vout():
"""
make a synthetic vout object.
returns a vout dictionary item
"""
key_pair = rcrypt.make_ecc_keys()
global prev_tx_keys
prev_tx_keys.append([key_pair[0],key_pair[1]])
vout = {}
vout['value'] = secrets.randbelow(1000000) + 1000 # helium cents
tmp = []
tmp.append('<DUP>')
tmp.append('<HASH-160>')
tmp.append(key_pair[1])
tmp.append('<EQ-VERIFY>')
tmp.append('<CHECK-SIG>')
vout['ScriptPubKey'] = tmp
return vout
ctr = 0
while ctr < num_outputs:
ret = make_synthetic_vout()
transaction['vout'].append(ret)
ctr += 1
return transaction
##################################
# Synthetic Transaction
##################################
def make_synthetic_transaction(num_outputs):
"""
makes a synthetic transaction with randomized values
"""
transaction = {}
transaction['version'] = 1
transaction['transactionid'] = rcrypt.make_uuid()
transaction['locktime'] = 0
transaction['vin'] = []
transaction['vout'] = []
ctr = 0
while ctr < num_outputs:
vout = make_synthetic_vout()
transaction['vout'].append(vout)
ctr += 1
# randomize how many of the previous transaction outputs
# are consumed and the order in which they are consumed
num_inputs = secrets.randbelow(num_prev_tx_outputs) + 1
indices = list(range(num_inputs))
random.shuffle(indices)
for ctr in indices:
vin = make_synthetic_vin(previous_tx["transactionid"], ctr)
transaction['vin'].append(vin)
ctr += 1
return transaction
#############################
# make a synthetic vin list
#############################
def make_synthetic_vin(prev_txid, prev_tx_vout_index):
"""
makes a vin object.
receives a previous transaction id and an index into the vout list.
returns the vin element
"""
vin = {}
vin['txid'] = prev_txid
vin['vout_index'] = prev_tx_vout_index
vin['ScriptSig'] = []
global prev_tx_keys
vin['ScriptSig'].append(rcrypt.sign_message(prev_tx_keys[prev_tx_vout_index][0],
prev_tx_keys[prev_tx_vout_index][1]))
vin['ScriptSig'].append(prev_tx_keys[prev_tx_vout_index][1])
return vin
#############################
# make a synthetic vout list
#############################
def make_synthetic_vout():
"""
make a randomized vout element
receives a public key
returns the vout dict element
"""
vout = {}
vout['value'] = 1 # helium cents
tmp = []
tmp.append('<DUP>')
tmp.append('<HASH-160>')
tmp.append(secrets.token_hex(64))
tmp.append('<EQ-VERIFY>')
tmp.append('<CHECK-SIG>')
vout['ScriptPubKey'] = tmp
return vout
#############################
# randomize access to a list
#############################
def randomize_list(lst: "list"):
"""
randomly shuffles the indexes into a list
returns the randomized index list
"""
ctr = len(lst)
p = 0
index = []
while p < ctr: index.append(p);p += 1
random.shuffle(index)
return index
########################################
# Make a synthetic previous transaction
########################################
num_prev_tx_outputs = secrets.randbelow(3) + 1
previous_tx = make_synthetic_previous_transaction(num_prev_tx_outputs)
"""
test the transactions
"""
def test_type_transaction():
"""
test transaction type
"""
assert tx.validate_transaction([]) == False
def test_missing_transactionid():
"""
test missing transaction id
"""
trns = make_synthetic_transaction(1)
del trns["transactionid"]
assert tx.validate_transaction(trns) == False
def test_bad_transactionid():
"""
test transaction id with an invalid length
"""
trns = make_synthetic_transaction(1)
# replace the last char with an invalid char
trns["transactionid"] = trns["transactionid"][:-1] + "z"
assert tx.validate_transaction(trns) == False
def test_zero_version_number():
"""
test zero version no
"""
trns = make_synthetic_transaction(1)
trns["version"] = 0
assert tx.validate_transaction(trns) == False
def test_negative_locktime():
"""
test for a negative locktime
"""
trns = make_synthetic_transaction(1)
trns["locktime"] = -1
assert tx.validate_transaction(trns) == False
def test_invalid_vout_length():
"""
test that the vout list length is not greater than
the maximum allowed by the Helium config
"""
trn = make_synthetic_transaction(11)
assert tx.validate_transaction(trn) == False
def test_no_prev_tx_reference():
"""
test that the transaction's vin element has a
reference to a previous transaction.
"""
txn = make_synthetic_transaction(1)
indices = randomize_list(txn["vin"])
for ctr in list(range(len(indices))):
txn['vin'][indices[ctr]]['txid'] = ""
assert tx.validate_transaction(txn) == False
def test_empty_scriptSig():
"""
test that ScriptPubKey element is not empty
"""
txn = make_synthetic_transaction(8)
indices = randomize_list(txn['vin'])
for ctr in list(range(len(indices))):
txn['vin'][indices[ctr]]['ScriptSig'] = []
vin = txn['vin'][indices[ctr]]
assert tx.validate_vin(vin) == False
def test_scriptsig_length_error():
"""
test if the ScriptSig list has invalid length
"""
txn = make_synthetic_transaction(2)
indices = randomize_list(txn['vin'])
keys = rcrypt.make_ecc_keys()
for ctr in list(range(len(indices))):
txn["vin"][indices[ctr]]['ScriptSig'].append(keys[1])
assert tx.validate_vin(txn['vin'][indices[ctr]]) == False
def test_empty_scriptSig_sig():
"""
test that ScriptSig signature field is not empty
"""
txn = make_synthetic_transaction(2)
indices = randomize_list(txn['vin'])
for ctr in list(range(len(indices))):
txn['vin'][indices[ctr]]['ScriptSig'][0] = ''
vin = txn['vin'][indices[ctr]]
assert tx.validate_vin(vin) == False
def test_empty_scriptSig_pubkey():
"""
test that ScriptSig public key field is not empty
"""
txn = make_synthetic_transaction(2)
indices = randomize_list(txn['vin'])
for ctr in list(range(len(indices))):
txn['vin'][indices[ctr]]['ScriptSig'][1] = ''
assert tx.validate_transaction(txn) == False
def test_scriptpubkey_length_error():
"""
test that ScriptPubKey list has a valid length
"""
txn = make_synthetic_transaction(4)
indices = randomize_list(txn['vout'])
for ctr in list(range(len(indices))):
element = txn['vout'][indices[ctr]]['ScriptPubKey'][1]
txn['vout'][indices[ctr]]['ScriptPubKey'].append(element)
assert tx.validate_transaction(txn) == False
@pytest.mark.parametrize("index, value, ret", [
(0, 'DUP', False),
(0, '<DUP>', True),
(1, '<HASH-160',False),
(1, '<HASH-160<',False),
(1, '<HASH-160>',True),
(1, '<HASH160>', False),
(3, '<VERIFY>', False),
(3, '<EQ-VERIFY>', True),
(4, '<CHECKSIG>', False),
(4, '<CHECK-SIG>', True),
])
def test_bad_scriptpubkey_script(monkeypatch, index, value, ret):
"""
test if ScriptPubKey has an invalid element
"""
monkeypatch.setattr(tx, "transaction_fee", 5)
monkeypatch.setattr(tx, "unlock_transaction_fragment", True)
txn = make_synthetic_transaction(7)
indices = randomize_list(txn['vout'])
txn['vout'][indices[0]]['ScriptPubKey'][index] = value
assert tx.validate_vout(txn['vout'][indices[0]]) == ret
def test_pubkeyhash():
"""
test a public key hash in scriptpubkey for a
valid RIPEMD-160 format
"""
txn = make_synthetic_transaction(5)
indices = randomize_list(txn['vout'])
for ctr in list(range(len(indices))):
val = txn['vout'][indices[ctr]]['ScriptPubKey'][2]
val = rcrypt.make_RIPEMD160_hash(rcrypt.make_SHA256_hash(val))
assert rcrypt.validate_RIPEMD160_hash(val) == True
def test_invalid_txid():
"""
test the format of a previous transaction
"""
txn = make_synthetic_transaction(2)
id = rcrypt.make_uuid()
id = "j" + id[1:]
indices = randomize_list(txn['vin'])
ctr = secrets.randbelow(len(indices))
txn['vin'][indices[ctr]]['txid'] = id
assert tx.validate_transaction(txn) == False
def test_invalid_txid_length():
"""
test a previous transaction id for length
"""
txn = make_synthetic_transaction(2)
id = rcrypt.make_uuid()
id = "p" + id[0:]
indices = randomize_list(txn['vin'])
ctr = secrets.randbelow(len(indices))
txn['vin'][indices[ctr]]['txid'] = id
assert tx.validate_transaction(txn) == False
def test_bad_transaction_inputs(monkeypatch):
"""
tests that the inputs received are valid.
fails if the previous transaction does not exist,
or previous value has been spent, or is negative
"""
monkeypatch.setattr(tx, 'prevtx_value', lambda x: False)
assert tx.prevtx_value("stubbed function") == False
def test_transaction_inputs(monkeypatch):
"""
tests that the inputs received are valid
fails if the previous transaction does not exist,
or previous value has been spent, or is negative
"""
txn = make_synthetic_transaction(2)
monkeypatch.setattr(tx, 'prevtx_value', lambda x:
[{
"txid_vout": txn['vin'][0]['txid'] + '_' + str(txn['vin'][0]['vout_index']),
"value": 210,
"pkhash": previous_tx['vout'][0]['ScriptPubKey'][2],
"block": 100,
"spent": False
},
{
"txid_vout": txn['vin'][0]['txid'] + '_' + str(txn['vin'][0]['vout_index']),
"value": 38,
"pkhash": previous_tx['vout'][0]['ScriptPubKey'][2],
"block": 29,
"spent": False
}]
)
assert bool(tx.prevtx_value(txn)) == True
def test_zero_output_value(monkeypatch):
"""
test a zero transaction output value
"""
txn = make_synthetic_transaction(4)
indices = randomize_list(txn['vout'])
ctr = secrets.randbelow(len(indices))
monkeypatch.setitem(txn['vout'][indices[ctr]], 'value', 0 )
assert tx.validate_transaction(txn) == False
def test_negative_output_value(monkeypatch):
"""
test a negative transaction output value
"""
txn = make_synthetic_transaction(4)
indices = randomize_list(txn['vout'])
ctr = secrets.randbelow(len(indices))
monkeypatch.setitem(txn['vout'][indices[ctr]], 'value', -456712)
assert tx.validate_transaction(txn) == False
def test_transaction_fee(monkeypatch):
"""
test various transaction fees
"""
txn = make_synthetic_transaction(2)
txn['vout'][0]["value"] = 10
txn['vout'][1]["value"] = 203
value_list = [{
"txid_vout": txn['vin'][0]['txid'] + '_' + str(txn['vin'][0]['vout_index']),
"value": 210,
"pkhash": previous_tx['vout'][0]['ScriptPubKey'][2],
"block": 100,
"spent": False
},
{
"txid_vout": txn['vin'][0]['txid'] + '_' + str(txn['vin'][0]['vout_index']),
"value": 38,
"pkhash": previous_tx['vout'][0]['ScriptPubKey'][2],
"block": 29,
"spent": False
}
]
assert tx.transaction_fee(txn, value_list) != False
def test_negative_transaction_fee(monkeypatch):
"""
test a negative transaction fee
"""
txn = make_synthetic_transaction(3)
monkeypatch.setitem(txn['vout'][0], 'value', 10000)
value_list = [{
"txid_vout": txn['vin'][0]['txid'] + '_' + str(txn['vin'][0]['vout_index']),
"value": 210,
"pkhash": previous_tx['vout'][0]['ScriptPubKey'][2],
"block": 100,
"spent": False
},
{
"txid_vout": txn['vin'][0]['txid'] + '_' + str(txn['vin'][0]['vout_index']),
"value": 38,
"pkhash": previous_tx['vout'][0]['ScriptPubKey'][2],
"block": 29,
"spent": False
}
]
assert tx.transaction_fee(txn, value_list) == False
def test_unlock_bad_hash(monkeypatch):
"""
test unlocking previous transaction output with
bad RIPEMD-160 hash p2pkhash value
"""
global prev_tx_keys
prev_tx_keys.clear()
txn1 = make_synthetic_previous_transaction(4)
txn2 = make_synthetic_transaction(2)
# make a transaction fragment where the first vin element
# of txn2 consumes the value of the first vout element of txn1
# synthetic consuming vin element in tx2
vin = {}
vin['txid'] = txn1["transactionid"]
vin['vout_index'] = 0
vin['ScriptSig'] = {}
signature = rcrypt.sign_message(prev_tx_keys[1][0], prev_tx_keys[1][1])
pubkey = prev_tx_keys[1][1]
sig = []
sig.append(signature)
sig.append(pubkey + "corrupted")
vin['ScriptSig'] = sig
# use the wrong public key hash in txn2
key_pair = rcrypt.make_ecc_keys()
ripemd_hash = rcrypt.make_RIPEMD160_hash(rcrypt.make_SHA256_hash(key_pair[1]))
fragment = {
"value": 210,
"pkhash": ripemd_hash,
"spent": False,
"tx_chain": txn2["transactionid"] + "_" + "0",
"checksum": rcrypt.make_SHA256_hash(txn1["transactionid"])
}
assert tx.unlock_transaction_fragment(vin, fragment) == False
def test_unlock_bad_signature(monkeypatch):
"""
test unlocking a transaction with a bad signature
"""
global prev_tx_keys
prev_tx_keys.clear()
txn1 = make_synthetic_previous_transaction(4)
txn2 = make_synthetic_transaction(2)
# make a transaction fragment where the first vin element
# of txn2 consumes the value of the first vout element of txn1
# synthetic consuming vin element in tx2
vin = {}
vin['txid'] = txn1["transactionid"]
vin['vout_index'] = 0
vin['ScriptSig'] = {}
# use wrong private key to sign
key_pair = rcrypt.make_ecc_keys()
signature = rcrypt.sign_message(key_pair[0], prev_tx_keys[1][1])
pubkey = prev_tx_keys[1][1]
sig = []
sig.append(signature)
sig.append(pubkey + "corrupted")
vin['ScriptSig'] = sig
# public key hash in txn2
ripemd_hash = rcrypt.make_RIPEMD160_hash(rcrypt.make_SHA256_hash(prev_tx_keys[1][1]))
fragment = {
"value": 210,
"pkhash": ripemd_hash,
"spent": False,
"tx_chain": txn2["transactionid"] + "_" + "0",
"checksum": rcrypt.make_SHA256_hash(txn1["transactionid"])
}
assert tx.unlock_transaction_fragment(vin, fragment) == False
def test_unlock_bad_pubkey():
"""
test unlocking a transaction with a bad public key
"""
global prev_tx_keys
prev_tx_keys.clear()
txn1 = make_synthetic_previous_transaction(4)
txn2 = make_synthetic_transaction(2)
# make a transaction fragment where the first vin element
# of txn2 consumes the value of the first vout element of txn1
# synthetic consuming vin element in tx2
vin = {}
vin['txid'] = txn1["transactionid"]
vin['vout_index'] = 0
vin['ScriptSig'] = {}
signature = rcrypt.sign_message(prev_tx_keys[1][0], prev_tx_keys[1][1])
pubkey = prev_tx_keys[1][1]
sig = []
sig.append(signature)
sig.append(pubkey + "corrupted")
vin['ScriptSig'] = sig
# public key hash in txn2
ripemd_hash = rcrypt.make_RIPEMD160_hash(rcrypt.make_SHA256_hash(prev_tx_keys[1][1]))
fragment = {
"value": 210,
"pkhash": ripemd_hash,
"spent": False,
"tx_chain": txn2["transactionid"] + "_" + "0",
"checksum": rcrypt.make_SHA256_hash(txn1["transactionid"])
}
assert tx.unlock_transaction_fragment(vin, fragment) == False
def test_unlock_good():
"""
test unlocking a transaction with a good private-public key pair
"""
global prev_tx_keys
prev_tx_keys.clear()
txn1 = make_synthetic_previous_transaction(4)
txn2 = make_synthetic_transaction(2)
# make a transaction fragment where the first vin element
# of txn2 consumes the value of the first vout element of txn1
# synthetic consuming vin element in tx2
vin = {}
vin['txid'] = txn1["transactionid"]
vin['vout_index'] = 0
vin['ScriptSig'] = {}
signature = rcrypt.sign_message(prev_tx_keys[1][0], prev_tx_keys[1][1])
pubkey = prev_tx_keys[1][1]
sig = []
sig.append(signature)
sig.append(pubkey)
vin['ScriptSig'] = sig
# public key hash in txn2
ripemd_hash = rcrypt.make_RIPEMD160_hash(rcrypt.make_SHA256_hash(prev_tx_keys[1][1]))
fragment = {
"value": 210,
"pkhash": ripemd_hash,
"spent": False,
"tx_chain": txn2["transactionid"] + "_" + "0",
"checksum": rcrypt.make_SHA256_hash(txn1["transactionid"])
}
assert tx.unlock_transaction_fragment(vin, fragment) == True
Update the Helium Blockchain Module
Now that we have a functional transaction module; we update the hblockchain module
to validate transactions in a block. Ensure that you import the tx module into hblockchain. Remove the add_block function from the hblockchain module and add the following variant in lieu thereof:
def add_block(block: "dictionary") -> "bool":
"""
add_block: adds a block to the blockchain. Receives a block.
The block attributes are checked for validity and each transaction
in the block is tested for validity. If there are no errors, the block
is written to a file as a sequence of raw bytes. Then the block is added
to the blockchain.
returns True if the block is added to the blockchain and False otherwise.
"""
try:
# validate the received block parameters
if validate_block(block) == False:
raise(ValueError("block validation error"))
# validate the transactions in the block
for trx in block['tx']:
# first transaction in the block is a coinbase transaction
if block["height"] == 0 or block['tx'][0] == trx: zero_inputs = True
else: zero_inputs = False
if tx.validate_transaction(trx, zero_inputs) == False:
raise(ValueError("transaction validation error"))
# serialize the block to a file
if (serialize_block(block) == False):
raise(ValueError("serialize block error"))
# add the block to the blockchain in memory
blockchain.append(block)
except Exception as err:
print(str(err))
logging.debug('add_block: exception: ' + str(err))
return False
return True
This validates all of the transactions in a block before the block is added to the blockchain.
Conclusion
In this chapter, we have written program code to validate Helium transactions. We have also written unit tests for our transaction program code. Lastly, we have refined the Helium blockchain code to verify transactions before a block is added to the blockchain.
You will have noticed that we have not implemented any code to retrieve unspent transaction values from previous transactions. This feature was mocked in our unit tests. The next chapter addresses this matter.
We will implement two databases for Helium. In the same manner as Bitcoin, we are going to implement a LevelDB database to verify and get unspent transaction amounts. We will also implement a second LevelDB database that lets us quickly determine the block in which a particular transaction is located.