aljaz/python-teos
1
0
Fork 0
mirror of https://github.com/aljazceru/python-teos.git synced 2025-12-17 22:24:23 +01:00
Code Issues Packages Projects Releases Wiki Activity

Adds Inspector unittests. Fixes some related bugs.

Browse Source
This commit is contained in:
Sergi Delgado Segura
2019-10-07 15:18:00 +01:00
parent 067efcca73
commit 4be506412e
4 changed files with 270 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
pisa/block_processor.py
View File

@@ -30,6 +30,18 @@ class BlockProcessor:
return block_hash
@staticmethod
def get_block_count():
block_count = None
try:
block_count = bitcoin_cli.getblockcount()
except JSONRPCException as e:
logging.error("[BlockProcessor] couldn't get block block count. Error code {}".format(e))
return block_count
@staticmethod
def get_potential_matches(txids, locator_uuid_map):
potential_locators = {sha256(binascii.unhexlify(txid)).hexdigest(): txid for txid in txids}

44
pisa/inspector.py
View File

@@ -2,9 +2,14 @@ import re
from pisa import errors
import pisa.conf as conf
from pisa import logging, bitcoin_cli
from pisa import logging
from pisa.appointment import Appointment
from pisa.utils.auth_proxy import JSONRPCException
from pisa.block_processor import BlockProcessor
# FIXME: The inspector logs the wrong messages sent form the users. A possible attack surface would be to send a really
# long field that, even if not accepted by PISA, would be stored in the logs. This is a possible DoS surface
# since pisa would store any kind of message (no matter the length). Solution: truncate the length of the fields
# stored + blacklist if multiple wrong requests are received.
class Inspector:
@@ -17,10 +22,11 @@ class Inspector:
cipher = data.get('cipher')
hash_function = data.get('hash_function')
try:
block_height = bitcoin_cli.getblockcount()
block_height = BlockProcessor.get_block_count()
if block_height is not None:
rcode, message = self.check_locator(locator)
if rcode == 0:
rcode, message = self.check_start_time(start_time, block_height)
if rcode == 0:
@@ -39,9 +45,7 @@ class Inspector:
else:
r = (rcode, message)
except JSONRPCException as e:
logging.error("[Inspector] JSONRPCException. Error code {}".format(e))
else:
# In case of an unknown exception, assign a special rcode and reason.
r = (errors.UNKNOWN_JSON_RPC_EXCEPTION, "Unexpected error occurred")
@@ -76,6 +80,9 @@ class Inspector:
message = None
rcode = 0
# TODO: What's too close to the current height is not properly defined. Right now any appointment that is in the
# future will be accepted (even if it's only one block away).
t = type(start_time)
if start_time is None:
@@ -89,7 +96,7 @@ class Inspector:
if start_time < block_height:
message = "start_time is in the past"
else:
message = "start_time too close to current height"
message = "start_time is too close to current height"
if message is not None:
logging.error("[Inspector] {}".format(message))
@@ -101,6 +108,9 @@ class Inspector:
message = None
rcode = 0
# TODO: What's too close to the current height is not properly defined. Right now any appointment that ends in
# the future will be accepted (even if it's only one block away).
t = type(end_time)
if end_time is None:
@@ -115,9 +125,12 @@ class Inspector:
message = "end_time is smaller than start_time"
else:
message = "end_time is equal to start_time"
elif block_height > end_time:
elif block_height >= end_time:
rcode = errors.APPOINTMENT_FIELD_TOO_SMALL
message = 'end_time is in the past'
if block_height > end_time:
message = 'end_time is in the past'
else:
message = 'end_time is too close to current height'
if message is not None:
logging.error("[Inspector] {}".format(message))
@@ -161,10 +174,9 @@ class Inspector:
elif t != str:
rcode = errors.APPOINTMENT_WRONG_FIELD_TYPE
message = "wrong encrypted_blob data type ({})".format(t)
elif encrypted_blob == '':
# ToDo: #6 We may want to define this to be at least as long as one block of the cipher we are using
rcode = errors.APPOINTMENT_WRONG_FIELD
message = "wrong encrypted_blob"
elif re.search(r'^[0-9A-Fa-f]+$', encrypted_blob) is None:
rcode = errors.APPOINTMENT_WRONG_FIELD_FORMAT
message = "wrong encrypted_blob format ({})".format(encrypted_blob)
if message is not None:
logging.error("[Inspector] {}".format(message))
@@ -184,7 +196,7 @@ class Inspector:
elif t != str:
rcode = errors.APPOINTMENT_WRONG_FIELD_TYPE
message = "wrong cipher data type ({})".format(t)
elif cipher not in conf.SUPPORTED_CIPHERS:
elif cipher.upper() not in conf.SUPPORTED_CIPHERS:
rcode = errors.APPOINTMENT_CIPHER_NOT_SUPPORTED
message = "cipher not supported: {}".format(cipher)
@@ -206,7 +218,7 @@ class Inspector:
elif t != str:
rcode = errors.APPOINTMENT_WRONG_FIELD_TYPE
message = "wrong hash_function data type ({})".format(t)
elif hash_function not in conf.SUPPORTED_HASH_FUNCTIONS:
elif hash_function.upper() not in conf.SUPPORTED_HASH_FUNCTIONS:
rcode = errors.APPOINTMENT_HASH_FUNCTION_NOT_SUPPORTED
message = "hash_function not supported {}".format(hash_function)

130
test/appointment_tests.py
View File

@@ -1,130 +0,0 @@
import pisa.conf as conf
from pisa.inspector import Inspector
from pisa.appointment import Appointment
from pisa import errors, logging, bitcoin_cli
from pisa.utils.auth_proxy import JSONRPCException
appointment = {"locator": None, "start_time": None, "end_time": None, "dispute_delta": None,
"encrypted_blob": None, "cipher": None, "hash_function": None}
try:
block_height = bitcoin_cli.getblockcount()
except JSONRPCException as e:
logging.error("[Inspector] JSONRPCException. Error code {}".format(e))
locators = [None, 0, 'A' * 31, "A" * 63 + "_"]
start_times = [None, 0, '', 15.0, block_height - 10]
end_times = [None, 0, '', 26.123, block_height - 11]
dispute_deltas = [None, 0, '', 1.2, -3, 30]
encrypted_blobs = [None, 0, '']
ciphers = [None, 0, '', 'foo']
hash_functions = [None, 0, '', 'foo']
locators_rets = [errors.APPOINTMENT_EMPTY_FIELD, errors.APPOINTMENT_WRONG_FIELD_TYPE,
errors.APPOINTMENT_WRONG_FIELD_SIZE, errors.APPOINTMENT_WRONG_FIELD_FORMAT]
start_time_rets = [errors.APPOINTMENT_EMPTY_FIELD, errors.APPOINTMENT_FIELD_TOO_SMALL,
errors.APPOINTMENT_WRONG_FIELD_TYPE, errors.APPOINTMENT_WRONG_FIELD_TYPE,
errors.APPOINTMENT_FIELD_TOO_SMALL]
end_time_rets = [errors.APPOINTMENT_EMPTY_FIELD, errors.APPOINTMENT_FIELD_TOO_SMALL,
errors.APPOINTMENT_WRONG_FIELD_TYPE, errors.APPOINTMENT_WRONG_FIELD_TYPE,
errors.APPOINTMENT_FIELD_TOO_SMALL]
dispute_delta_rets = [errors.APPOINTMENT_EMPTY_FIELD, errors.APPOINTMENT_FIELD_TOO_SMALL,
errors.APPOINTMENT_WRONG_FIELD_TYPE, errors.APPOINTMENT_WRONG_FIELD_TYPE,
errors.APPOINTMENT_FIELD_TOO_SMALL]
encrypted_blob_rets = [errors.APPOINTMENT_EMPTY_FIELD, errors.APPOINTMENT_WRONG_FIELD_TYPE,
errors.APPOINTMENT_WRONG_FIELD]
cipher_rets = [errors.APPOINTMENT_EMPTY_FIELD, errors.APPOINTMENT_WRONG_FIELD_TYPE,
errors.APPOINTMENT_CIPHER_NOT_SUPPORTED, errors.APPOINTMENT_CIPHER_NOT_SUPPORTED]
hash_function_rets = [errors.APPOINTMENT_EMPTY_FIELD, errors.APPOINTMENT_WRONG_FIELD_TYPE,
errors.APPOINTMENT_HASH_FUNCTION_NOT_SUPPORTED, errors.APPOINTMENT_HASH_FUNCTION_NOT_SUPPORTED]
inspector = Inspector()
print("Locator tests\n")
for locator, ret in zip(locators, locators_rets):
appointment["locator"] = locator
r = inspector.inspect(appointment)
assert r[0] == ret
print(r)
# Set locator to a 'valid' one
appointment['locator'] = 'A' * 64
print("\nStart time tests\n")
for start_time, ret in zip(start_times, start_time_rets):
appointment["start_time"] = start_time
r = inspector.inspect(appointment)
assert r[0] == ret
print(r)
# Setting the start time to some time in the future
appointment['start_time'] = block_height + 10
print("\nEnd time tests\n")
for end_time, ret in zip(end_times, end_time_rets):
appointment["end_time"] = end_time
r = inspector.inspect(appointment)
assert r[0] == ret
print(r)
# Setting the end time to something consistent with start time
appointment['end_time'] = block_height + 30
print("\nDelta tests\n")
for dispute_delta, ret in zip(dispute_deltas, dispute_delta_rets):
appointment["dispute_delta"] = dispute_delta
r = inspector.inspect(appointment)
assert r[0] == ret
print(r)
# Setting the a proper dispute delta
appointment['dispute_delta'] = appointment['end_time'] - appointment['start_time']
print("\nEncrypted blob tests\n")
for encrypted_blob, ret in zip(encrypted_blobs, encrypted_blob_rets):
appointment["encrypted_blob"] = encrypted_blob
r = inspector.inspect(appointment)
assert r[0] == ret
print(r)
# Setting the encrypted blob to something that may pass
appointment['encrypted_blob'] = 'A' * 32
print("\nCipher tests\n")
for cipher, ret in zip(ciphers, cipher_rets):
appointment["cipher"] = cipher
r = inspector.inspect(appointment)
assert r[0] == ret
print(r)
# Setting the cipher to the only supported one for now
appointment['cipher'] = conf.SUPPORTED_CIPHERS[0]
print("\nHash function tests\n")
for hash_function, ret in zip(hash_functions, hash_function_rets):
appointment["hash_function"] = hash_function
r = inspector.inspect(appointment)
assert r[0] == ret
print(r)
# Setting the cipher to the only supported one for now
appointment['hash_function'] = conf.SUPPORTED_HASH_FUNCTIONS[0]
r = inspector.inspect(appointment)
assert type(r) == Appointment
print("\nAll tests passed!")

230
test/unit/test_inspector.py Normal file
View File

@@ -0,0 +1,230 @@
from os import urandom
from pisa import logging
from pisa.errors import *
from pisa.inspector import Inspector
from pisa.appointment import Appointment
from pisa.block_processor import BlockProcessor
from pisa.conf import MIN_DISPUTE_DELTA, SUPPORTED_CIPHERS, SUPPORTED_HASH_FUNCTIONS
inspector = Inspector()
APPOINTMENT_OK = (0, None)
NO_HEX_STINGS = ["R" * 64, urandom(31).hex() + "PP", "$"*64, " "*64]
WRONG_TYPES = [[], '', urandom(32).hex(), 3.2, 2.0, (), object, {}, " "*32, object()]
WRONG_TYPES_NO_STR = [[], urandom(32), 3.2, 2.0, (), object, {}, object()]
def test_check_locator():
# Right appointment type, size and format
locator = urandom(32).hex()
assert(Inspector.check_locator(locator) == APPOINTMENT_OK)
# Wrong size (too big)
locator = urandom(33).hex()
assert(Inspector.check_locator(locator)[0] == APPOINTMENT_WRONG_FIELD_SIZE)
# Wrong size (too small)
locator = urandom(31).hex()
assert(Inspector.check_locator(locator)[0] == APPOINTMENT_WRONG_FIELD_SIZE)
# Empty
locator = None
assert (Inspector.check_locator(locator)[0] == APPOINTMENT_EMPTY_FIELD)
# Wrong type (several types tested, it should do for anything that is not a string)
locators = [[], -1, 3.2, 0, 4, (), object, {}, object()]
for locator in locators:
assert (Inspector.check_locator(locator)[0] == APPOINTMENT_WRONG_FIELD_TYPE)
# Wrong format (no hex)
locators = NO_HEX_STINGS
for locator in locators:
assert (Inspector.check_locator(locator)[0] == APPOINTMENT_WRONG_FIELD_FORMAT)
def test_check_start_time():
# Time is defined in block height
current_time = 100
# Right format and right value (start time in the future)
start_time = 101
assert (Inspector.check_start_time(start_time, current_time) == APPOINTMENT_OK)
# Start time too small (either same block or block in the past)
start_times = [100, 99, 98, -1]
for start_time in start_times:
assert (Inspector.check_start_time(start_time, current_time)[0] == APPOINTMENT_FIELD_TOO_SMALL)
# Empty field
start_time = None
assert (Inspector.check_start_time(start_time, current_time)[0] == APPOINTMENT_EMPTY_FIELD)
# Wrong data type
start_times = WRONG_TYPES
for start_time in start_times:
assert (Inspector.check_start_time(start_time, current_time)[0] == APPOINTMENT_WRONG_FIELD_TYPE)
def test_check_end_time():
# Time is defined in block height
current_time = 100
start_time = 120
# Right format and right value (start time before end and end in the future)
end_time = 121
assert (Inspector.check_end_time(end_time, start_time, current_time) == APPOINTMENT_OK)
# End time too small (start time after end time)
end_times = [120, 119, 118, -1]
for end_time in end_times:
assert (Inspector.check_end_time(end_time, start_time, current_time)[0] == APPOINTMENT_FIELD_TOO_SMALL)
# End time too small (either same height as current block or in the past)
current_time = 130
end_times = [130, 129, 128, -1]
for end_time in end_times:
assert (Inspector.check_end_time(end_time, start_time, current_time)[0] == APPOINTMENT_FIELD_TOO_SMALL)
# Empty field
end_time = None
assert (Inspector.check_end_time(end_time, start_time, current_time)[0] == APPOINTMENT_EMPTY_FIELD)
# Wrong data type
end_times = WRONG_TYPES
for end_time in end_times:
assert (Inspector.check_end_time(end_time, start_time, current_time)[0] == APPOINTMENT_WRONG_FIELD_TYPE)
def test_check_delta():
# Right value, right format
deltas = [MIN_DISPUTE_DELTA, MIN_DISPUTE_DELTA+1, MIN_DISPUTE_DELTA+1000]
for delta in deltas:
assert (Inspector.check_delta(delta) == APPOINTMENT_OK)
# Delta too small
deltas = [MIN_DISPUTE_DELTA-1, MIN_DISPUTE_DELTA-2, 0, -1, -1000]
for delta in deltas:
assert (Inspector.check_delta(delta)[0] == APPOINTMENT_FIELD_TOO_SMALL)
# Empty field
delta = None
assert(Inspector.check_delta(delta)[0] == APPOINTMENT_EMPTY_FIELD)
# Wrong data type
deltas = WRONG_TYPES
for delta in deltas:
assert (Inspector.check_delta(delta)[0] == APPOINTMENT_WRONG_FIELD_TYPE)
def test_check_blob():
# Right format and length
encrypted_blob = urandom(120).hex()
assert(Inspector.check_blob(encrypted_blob) == APPOINTMENT_OK)
# # Wrong content
# # FIXME: There is not proper defined format for this yet. It should be restricted by size at least, and check it
# # is multiple of the block size defined by the encryption function.
# Wrong type
encrypted_blobs = WRONG_TYPES_NO_STR
for encrypted_blob in encrypted_blobs:
assert (Inspector.check_blob(encrypted_blob)[0] == APPOINTMENT_WRONG_FIELD_TYPE)
# Empty field
encrypted_blob = None
assert (Inspector.check_blob(encrypted_blob)[0] == APPOINTMENT_EMPTY_FIELD)
# Wrong format (no hex)
encrypted_blobs = NO_HEX_STINGS
for encrypted_blob in encrypted_blobs:
assert (Inspector.check_blob(encrypted_blob)[0] == APPOINTMENT_WRONG_FIELD_FORMAT)
def test_check_cipher():
# Right format and content (any case combination should be accepted)
for cipher in SUPPORTED_CIPHERS:
cipher_cases = [cipher, cipher.lower(), cipher.capitalize()]
for case in cipher_cases:
assert(Inspector.check_cipher(case) == APPOINTMENT_OK)
# Wrong type
ciphers = WRONG_TYPES_NO_STR
for cipher in ciphers:
assert(Inspector.check_cipher(cipher)[0] == APPOINTMENT_WRONG_FIELD_TYPE)
# Wrong value
ciphers = NO_HEX_STINGS
for cipher in ciphers:
assert(Inspector.check_cipher(cipher)[0] == APPOINTMENT_CIPHER_NOT_SUPPORTED)
# Empty field
cipher = None
assert (Inspector.check_cipher(cipher)[0] == APPOINTMENT_EMPTY_FIELD)
def test_check_hash_function():
# Right format and content (any case combination should be accepted)
for hash_function in SUPPORTED_HASH_FUNCTIONS:
hash_function_cases = [hash_function, hash_function.lower(), hash_function.capitalize()]
for case in hash_function_cases:
assert (Inspector.check_hash_function(case) == APPOINTMENT_OK)
# Wrong type
hash_functions = WRONG_TYPES_NO_STR
for hash_function in hash_functions:
assert (Inspector.check_hash_function(hash_function)[0] == APPOINTMENT_WRONG_FIELD_TYPE)
# Wrong value
hash_functions = NO_HEX_STINGS
for hash_function in hash_functions:
assert (Inspector.check_hash_function(hash_function)[0] == APPOINTMENT_HASH_FUNCTION_NOT_SUPPORTED)
# Empty field
hash_function = None
assert (Inspector.check_hash_function(hash_function)[0] == APPOINTMENT_EMPTY_FIELD)
def test_inspect():
# Running this required bitcoind to be running (or mocked) since the block height is queried by inspect.
# At this point every single check function has been already tested, let's test inspect with an invalid and a valid
# appointments.
# Invalid appointment, every field is empty
appointment_data = dict()
appointment = inspector.inspect(appointment_data)
assert (type(appointment) == tuple and appointment[0] != 0)
# Valid appointment
locator = urandom(32).hex()
start_time = BlockProcessor.get_block_count() + 5
end_time = start_time + 20
dispute_delta = MIN_DISPUTE_DELTA
encrypted_blob = urandom(64).hex()
cipher = SUPPORTED_CIPHERS[0]
hash_function = SUPPORTED_HASH_FUNCTIONS[0]
appointment_data = {"locator": locator, "start_time": start_time, "end_time": end_time,
"dispute_delta": dispute_delta, "encrypted_blob": encrypted_blob, "cipher": cipher,
"hash_function": hash_function}
appointment = inspector.inspect(appointment_data)
assert(type(appointment) == Appointment and appointment.locator == locator and appointment.start_time == start_time
and appointment.end_time == end_time and appointment.dispute_delta == dispute_delta and
appointment.encrypted_blob.data == encrypted_blob and appointment.cipher == cipher and
appointment.hash_function == hash_function)
logging.getLogger().disabled = True
test_check_locator()
test_check_start_time()
test_check_end_time()
test_check_delta()
test_check_blob()
test_check_cipher()
test_check_hash_function()
test_inspect()