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

Refactors Inspector so it returns exceptions for invalid data instead of error codes

Browse Source 
Defines InspectionFailed, and exception that it's raised for invalid parameters. The design looks much cleaner now.
This commit is contained in:
Sergi Delgado Segura
2020-03-26 17:10:16 +01:00
parent 4fad6b7b6f
commit b3d67b4ce3
1 changed files with 95 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files

275
teos/inspector.py
View File

@@ -1,11 +1,11 @@
import re import re
import common.cryptographer import common.cryptographer
from common.logger import Logger
from common.constants import LOCATOR_LEN_HEX from common.constants import LOCATOR_LEN_HEX
from common.appointment import Appointment
from teos import errors, LOG_PREFIX from teos import errors, LOG_PREFIX
from common.logger import Logger
from common.appointment import Appointment
logger = Logger(actor="Inspector", log_name_prefix=LOG_PREFIX) logger = Logger(actor="Inspector", log_name_prefix=LOG_PREFIX)
common.cryptographer.logger = Logger(actor="Cryptographer", log_name_prefix=LOG_PREFIX) common.cryptographer.logger = Logger(actor="Cryptographer", log_name_prefix=LOG_PREFIX)
@@ -19,6 +19,14 @@ common.cryptographer.logger = Logger(actor="Cryptographer", log_name_prefix=LOG_
BLOCKS_IN_A_MONTH = 4320 # 4320 = roughly a month in blocks BLOCKS_IN_A_MONTH = 4320 # 4320 = roughly a month in blocks
class InspectionFailed(Exception):
"""Raise this the inspector finds a problem with any of the appointment fields"""
def __init__(self, erno, reason):
self.erno = erno
self.reason = reason
class Inspector: class Inspector:
""" """
The :class:`Inspector` class is in charge of verifying that the appointment data provided by the user is correct. The :class:`Inspector` class is in charge of verifying that the appointment data provided by the user is correct.
@@ -42,43 +50,29 @@ class Inspector:
Returns: Returns:
:obj:`Appointment <teos.appointment.Appointment>` or :obj:`tuple`: An appointment initialized with the :obj:`Appointment <teos.appointment.Appointment>`: An appointment initialized with the
provided data if it is correct. provided data.
Returns a tuple ``(return code, message)`` describing the error otherwise. Raises:
:obj:`InspectionFailed <teos.inspector.InspectionFailed>`: if any of the fields is wrong.
Errors are defined in :mod:`Errors <teos.errors>`.
""" """
if appointment_data is None: if appointment_data is None:
return errors.APPOINTMENT_EMPTY_FIELD, "empty appointment received" raise InspectionFailed(errors.APPOINTMENT_EMPTY_FIELD, "empty appointment received")
elif not isinstance(appointment_data, dict):
raise InspectionFailed(errors.APPOINTMENT_WRONG_FIELD, "wrong appointment format")
block_height = self.block_processor.get_block_count() block_height = self.block_processor.get_block_count()
if block_height is None:
raise InspectionFailed(errors.UNKNOWN_JSON_RPC_EXCEPTION, "Unexpected error occurred")
if block_height is not None: self.check_locator(appointment_data.get("locator"))
rcode, message = self.check_locator(appointment_data.get("locator")) self.check_start_time(appointment_data.get("start_time"), block_height)
self.check_end_time(appointment_data.get("end_time"), appointment_data.get("start_time"), block_height)
self.check_to_self_delay(appointment_data.get("to_self_delay"))
self.check_blob(appointment_data.get("encrypted_blob"))
if rcode == 0: return Appointment.from_dict(appointment_data)
rcode, message = self.check_start_time(appointment_data.get("start_time"), block_height)
if rcode == 0:
rcode, message = self.check_end_time(
appointment_data.get("end_time"), appointment_data.get("start_time"), block_height
)
if rcode == 0:
rcode, message = self.check_to_self_delay(appointment_data.get("to_self_delay"))
if rcode == 0:
rcode, message = self.check_blob(appointment_data.get("encrypted_blob"))
if rcode == 0:
r = Appointment.from_dict(appointment_data)
else:
r = (rcode, message)
else:
# In case of an unknown exception, assign a special rcode and reason.
r = (errors.UNKNOWN_JSON_RPC_EXCEPTION, "Unexpected error occurred")
return r
@staticmethod @staticmethod
def check_locator(locator): def check_locator(locator):
@@ -90,40 +84,23 @@ class Inspector:
Args: Args:
locator (:obj:`str`): the locator to be checked. locator (:obj:`str`): the locator to be checked.
Returns: Raises:
:obj:`tuple`: A tuple (return code, message) as follows: :obj:`InspectionFailed <teos.inspector.InspectionFailed>`: if any of the fields is wrong.
- ``(0, None)`` if the ``locator`` is correct.
- ``!= (0, None)`` otherwise.
The possible return errors are: ``APPOINTMENT_EMPTY_FIELD``, ``APPOINTMENT_WRONG_FIELD_TYPE``,
``APPOINTMENT_WRONG_FIELD_SIZE``, and ``APPOINTMENT_WRONG_FIELD_FORMAT``.
""" """
message = None
rcode = 0
if locator is None: if locator is None:
rcode = errors.APPOINTMENT_EMPTY_FIELD raise InspectionFailed(errors.APPOINTMENT_EMPTY_FIELD, "empty locator received")
message = "empty locator received"
elif type(locator) != str: elif type(locator) != str:
rcode = errors.APPOINTMENT_WRONG_FIELD_TYPE raise InspectionFailed(
message = "wrong locator data type ({})".format(type(locator)) errors.APPOINTMENT_WRONG_FIELD_TYPE, "wrong locator data type ({})".format(type(locator))
)
elif len(locator) != LOCATOR_LEN_HEX: elif len(locator) != LOCATOR_LEN_HEX:
rcode = errors.APPOINTMENT_WRONG_FIELD_SIZE raise InspectionFailed(errors.APPOINTMENT_WRONG_FIELD_SIZE, "wrong locator size ({})".format(len(locator)))
message = "wrong locator size ({})".format(len(locator))
# TODO: #12-check-txid-regexp
elif re.search(r"^[0-9A-Fa-f]+$", locator) is None: elif re.search(r"^[0-9A-Fa-f]+$", locator) is None:
rcode = errors.APPOINTMENT_WRONG_FIELD_FORMAT raise InspectionFailed(errors.APPOINTMENT_WRONG_FIELD_FORMAT, "wrong locator format ({})".format(locator))
message = "wrong locator format ({})".format(locator)
if message is not None:
logger.error(message)
return rcode, message
@staticmethod @staticmethod
def check_start_time(start_time, block_height): def check_start_time(start_time, block_height):
@@ -136,50 +113,35 @@ class Inspector:
start_time (:obj:`int`): the block height at which the tower is requested to start watching for breaches. start_time (:obj:`int`): the block height at which the tower is requested to start watching for breaches.
block_height (:obj:`int`): the chain height. block_height (:obj:`int`): the chain height.
Returns: Raises:
:obj:`tuple`: A tuple (return code, message) as follows: :obj:`InspectionFailed <teos.inspector.InspectionFailed>`: if any of the fields is wrong.
- ``(0, None)`` if the ``start_time`` is correct.
- ``!= (0, None)`` otherwise.
The possible return errors are: ``APPOINTMENT_EMPTY_FIELD``, ``APPOINTMENT_WRONG_FIELD_TYPE``, and
``APPOINTMENT_FIELD_TOO_SMALL``.
""" """
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 # 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). # future will be accepted (even if it's only one block away).
t = type(start_time)
if start_time is None: if start_time is None:
rcode = errors.APPOINTMENT_EMPTY_FIELD raise InspectionFailed(errors.APPOINTMENT_EMPTY_FIELD, "empty start_time received")
message = "empty start_time received"
elif t != int: elif type(start_time) != int:
rcode = errors.APPOINTMENT_WRONG_FIELD_TYPE raise InspectionFailed(
message = "wrong start_time data type ({})".format(t) errors.APPOINTMENT_WRONG_FIELD_TYPE, "wrong start_time data type ({})".format(type(start_time))
)
elif start_time <= block_height: elif start_time < block_height:
rcode = errors.APPOINTMENT_FIELD_TOO_SMALL raise InspectionFailed(errors.APPOINTMENT_FIELD_TOO_SMALL, "start_time is in the past")
if start_time < block_height:
message = "start_time is in the past" elif start_time == block_height:
else: raise InspectionFailed(
message = ( errors.APPOINTMENT_FIELD_TOO_SMALL,
"start_time is too close to current height. " "start_time is too close to current height. Accepted times are: [current_height+1, current_height+6]",
"Accepted times are: [current_height+1, current_height+6]" )
)
elif start_time > block_height + 6: elif start_time > block_height + 6:
rcode = errors.APPOINTMENT_FIELD_TOO_BIG raise InspectionFailed(
message = "start_time is too far in the future. Accepted start times are up to 6 blocks in the future" errors.APPOINTMENT_FIELD_TOO_BIG,
"start_time is too far in the future. Accepted start times are up to 6 blocks in the future",
if message is not None: )
logger.error(message)
return rcode, message
@staticmethod @staticmethod
def check_end_time(end_time, start_time, block_height): def check_end_time(end_time, start_time, block_height):
@@ -193,54 +155,36 @@ class Inspector:
start_time (:obj:`int`): the block height at which the tower is requested to start watching for breaches. start_time (:obj:`int`): the block height at which the tower is requested to start watching for breaches.
block_height (:obj:`int`): the chain height. block_height (:obj:`int`): the chain height.
Returns: Raises:
:obj:`tuple`: A tuple (return code, message) as follows: :obj:`InspectionFailed <teos.inspector.InspectionFailed>`: if any of the fields is wrong.
- ``(0, None)`` if the ``end_time`` is correct.
- ``!= (0, None)`` otherwise.
The possible return errors are: ``APPOINTMENT_EMPTY_FIELD``, ``APPOINTMENT_WRONG_FIELD_TYPE``, and
``APPOINTMENT_FIELD_TOO_SMALL``.
""" """
message = None
rcode = 0
# TODO: What's too close to the current height is not properly defined. Right now any appointment that ends in # 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). # the future will be accepted (even if it's only one block away).
t = type(end_time)
if end_time is None: if end_time is None:
rcode = errors.APPOINTMENT_EMPTY_FIELD raise InspectionFailed(errors.APPOINTMENT_EMPTY_FIELD, "empty end_time received")
message = "empty end_time received"
elif t != int: elif type(end_time) != int:
rcode = errors.APPOINTMENT_WRONG_FIELD_TYPE raise InspectionFailed(
message = "wrong end_time data type ({})".format(t) errors.APPOINTMENT_WRONG_FIELD_TYPE, "wrong end_time data type ({})".format(type(end_time))
)
elif end_time > block_height + BLOCKS_IN_A_MONTH: # 4320 = roughly a month in blocks elif end_time > block_height + BLOCKS_IN_A_MONTH: # 4320 = roughly a month in blocks
rcode = errors.APPOINTMENT_FIELD_TOO_BIG raise InspectionFailed(
message = "end_time should be within the next month (<= current_height + 4320)" errors.APPOINTMENT_FIELD_TOO_BIG, "end_time should be within the next month (<= current_height + 4320)"
)
elif start_time > end_time:
raise InspectionFailed(errors.APPOINTMENT_FIELD_TOO_SMALL, "end_time is smaller than start_time")
elif start_time >= end_time: elif start_time == end_time:
rcode = errors.APPOINTMENT_FIELD_TOO_SMALL raise InspectionFailed(errors.APPOINTMENT_FIELD_TOO_SMALL, "end_time is equal to start_time")
if start_time > end_time:
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 raise InspectionFailed(errors.APPOINTMENT_FIELD_TOO_SMALL, "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: elif block_height == end_time:
logger.error(message) raise InspectionFailed(errors.APPOINTMENT_FIELD_TOO_SMALL, "end_time is too close to current height")
return rcode, message
def check_to_self_delay(self, to_self_delay): def check_to_self_delay(self, to_self_delay):
""" """
@@ -252,46 +196,32 @@ class Inspector:
to_self_delay (:obj:`int`): The ``to_self_delay`` encoded in the ``csv`` of the ``htlc`` that this to_self_delay (:obj:`int`): The ``to_self_delay`` encoded in the ``csv`` of the ``htlc`` that this
appointment is covering. appointment is covering.
Returns: Raises:
:obj:`tuple`: A tuple (return code, message) as follows: :obj:`InspectionFailed <teos.inspector.InspectionFailed>`: if any of the fields is wrong.
- ``(0, None)`` if the ``to_self_delay`` is correct.
- ``!= (0, None)`` otherwise.
The possible return errors are: ``APPOINTMENT_EMPTY_FIELD``, ``APPOINTMENT_WRONG_FIELD_TYPE``, and
``APPOINTMENT_FIELD_TOO_SMALL``.
""" """
message = None
rcode = 0
t = type(to_self_delay)
if to_self_delay is None: if to_self_delay is None:
rcode = errors.APPOINTMENT_EMPTY_FIELD raise InspectionFailed(errors.APPOINTMENT_EMPTY_FIELD, "empty to_self_delay received")
message = "empty to_self_delay received"
elif t != int: elif type(to_self_delay) != int:
rcode = errors.APPOINTMENT_WRONG_FIELD_TYPE raise InspectionFailed(
message = "wrong to_self_delay data type ({})".format(t) errors.APPOINTMENT_WRONG_FIELD_TYPE, "wrong to_self_delay data type ({})".format(type(to_self_delay))
)
elif to_self_delay > pow(2, 32): elif to_self_delay > pow(2, 32):
rcode = errors.APPOINTMENT_FIELD_TOO_BIG raise InspectionFailed(
message = "to_self_delay must fit the transaction nLockTime field ({} > {})".format( errors.APPOINTMENT_FIELD_TOO_BIG,
to_self_delay, pow(2, 32) "to_self_delay must fit the transaction nLockTime field ({} > {})".format(to_self_delay, pow(2, 32)),
) )
elif to_self_delay < self.min_to_self_delay: elif to_self_delay < self.min_to_self_delay:
rcode = errors.APPOINTMENT_FIELD_TOO_SMALL raise InspectionFailed(
message = "to_self_delay too small. The to_self_delay should be at least {} (current: {})".format( errors.APPOINTMENT_FIELD_TOO_SMALL,
self.min_to_self_delay, to_self_delay "to_self_delay too small. The to_self_delay should be at least {} (current: {})".format(
self.min_to_self_delay, to_self_delay
),
) )
if message is not None:
logger.error(message)
return rcode, message
# ToDo: #6-define-checks-encrypted-blob # ToDo: #6-define-checks-encrypted-blob
@staticmethod @staticmethod
def check_blob(encrypted_blob): def check_blob(encrypted_blob):
@@ -301,34 +231,19 @@ class Inspector:
Args: Args:
encrypted_blob (:obj:`str`): the encrypted blob to be checked (hex encoded). encrypted_blob (:obj:`str`): the encrypted blob to be checked (hex encoded).
Returns: Raises:
:obj:`tuple`: A tuple (return code, message) as follows: :obj:`InspectionFailed <teos.inspector.InspectionFailed>`: if any of the fields is wrong.
- ``(0, None)`` if the ``encrypted_blob`` is correct.
- ``!= (0, None)`` otherwise.
The possible return errors are: ``APPOINTMENT_EMPTY_FIELD``, ``APPOINTMENT_WRONG_FIELD_TYPE``, and
``APPOINTMENT_WRONG_FIELD_FORMAT``.
""" """
message = None
rcode = 0
t = type(encrypted_blob)
if encrypted_blob is None: if encrypted_blob is None:
rcode = errors.APPOINTMENT_EMPTY_FIELD raise InspectionFailed(errors.APPOINTMENT_EMPTY_FIELD, "empty encrypted_blob received")
message = "empty encrypted_blob received"
elif t != str: elif type(encrypted_blob) != str:
rcode = errors.APPOINTMENT_WRONG_FIELD_TYPE raise InspectionFailed(
message = "wrong encrypted_blob data type ({})".format(t) errors.APPOINTMENT_WRONG_FIELD_TYPE, "wrong encrypted_blob data type ({})".format(type(encrypted_blob))
)
elif re.search(r"^[0-9A-Fa-f]+$", encrypted_blob) is None: elif re.search(r"^[0-9A-Fa-f]+$", encrypted_blob) is None:
rcode = errors.APPOINTMENT_WRONG_FIELD_FORMAT raise InspectionFailed(
message = "wrong encrypted_blob format ({})".format(encrypted_blob) errors.APPOINTMENT_WRONG_FIELD_FORMAT, "wrong encrypted_blob format ({})".format(encrypted_blob)
)
if message is not None:
logger.error(message)
return rcode, message