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 cli to avoid multi-type returns (normal return + None). Adds exceptions for errors.

Browse Source
This commit is contained in:
Sergi Delgado Segura
2020-04-03 21:57:38 +02:00
parent f350182012
commit c74f6a49af
4 changed files with 271 additions and 255 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
cli/exceptions.py Normal file
View File

@@ -0,0 +1,22 @@
class InvalidParameter(ValueError):
"""Raised when a command line parameter is invalid (either missing or wrong)"""
def __init__(self, msg, **kwargs):
self.msg = msg
self.kwargs = kwargs
class InvalidKey(Exception):
"""Raised when there is an error loading the keys"""
def __init__(self, msg, **kwargs):
self.reason = msg
self.kwargs = kwargs
class TowerResponseError(Exception):
"""Raised when the tower responds with an error"""
def __init__(self, msg, **kwargs):
self.reason = msg
self.kwargs = kwargs

308
cli/teos_cli.py
View File

@@ -10,8 +10,9 @@ from getopt import getopt, GetoptError
from requests import ConnectTimeout, ConnectionError
from requests.exceptions import MissingSchema, InvalidSchema, InvalidURL
from cli.help import show_usage, help_add_appointment, help_get_appointment, help_register
from cli import DEFAULT_CONF, DATA_DIR, CONF_FILE_NAME, LOG_PREFIX
from cli.exceptions import InvalidKey, InvalidParameter, TowerResponseError
from cli.help import show_usage, help_add_appointment, help_get_appointment, help_register
import common.cryptographer
from common.blob import Blob
@@ -36,26 +37,29 @@ def register(compressed_pk, teos_url):
teos_url (:obj:`str`): the teos base url.
Returns:
:obj:`dict` or :obj:`None`: a dictionary containing the tower response if the registration succeeded. ``None``
otherwise.
:obj:`dict`: a dictionary containing the tower response if the registration succeeded.
Raises:
:obj:`InvalidParameter <cli.exceptions.InvalidParameter>`: if `compressed_pk` is invalid.
:obj:`ConnectionError`: if the client cannot connect to the tower.
:obj:`TowerResponseError <cli.exceptions.TowerResponseError>`: if the tower responded with an error, or the
response was invalid.
"""
if not is_compressed_pk(compressed_pk):
logger.error("The cli public key is not valid")
return None
raise InvalidParameter("The cli public key is not valid")
# Send request to the server.
register_endpoint = "{}/register".format(teos_url)
data = {"public_key": compressed_pk}
logger.info("Registering in the Eye of Satoshi")
server_response = post_request(data, register_endpoint)
if server_response:
response_json = process_post_response(server_response)
return response_json
response = process_post_response(post_request(data, register_endpoint))
return response
def add_appointment(appointment_data, cli_sk, teos_pk, teos_url, appointments_folder_path):
def add_appointment(appointment_data, cli_sk, teos_pk, teos_url):
"""
Manages the add_appointment command.
@@ -67,76 +71,67 @@ def add_appointment(appointment_data, cli_sk, teos_pk, teos_url, appointments_fo
- Send the appointment to the tower
- Wait for the response
- Check the tower's response and signature
- Store the receipt (appointment + signature) on disk
Args:
appointment_data (:obj:`dict`): a dictionary containing the appointment data.
cli_sk (:obj:`PrivateKey`): the client's private key.
teos_pk (:obj:`PublicKey`): the tower's public key.
teos_url (:obj:`str`): the teos base url.
appointments_folder_path (:obj:`str`): the path to the appointments folder.
Returns:
:obj:`bool`: True if the appointment is accepted by the tower and the receipt is properly stored. False if any
error occurs during the process.
:obj:`tuple`: A tuple (`:obj:Appointment <common.appointment.Appointment>`, :obj:`str`) containing the
appointment and the tower's signature.
Raises:
:obj:`InvalidParameter <cli.exceptions.InvalidParameter>`: if `appointment_data` or any of its fields is
invalid.
:obj:`ValueError`: if the appointment cannot be signed.
:obj:`ConnectionError`: if the client cannot connect to the tower.
:obj:`TowerResponseError <cli.exceptions.TowerResponseError>`: if the tower responded with an error, or the
response was invalid.
"""
if appointment_data is None:
logger.error("The provided appointment JSON is empty")
return False
if not is_256b_hex_str(appointment_data.get("tx_id")):
logger.error("The provided txid is not valid")
return False
if not appointment_data:
raise InvalidParameter("The provided appointment JSON is empty")
tx_id = appointment_data.get("tx_id")
tx = appointment_data.get("tx")
if None not in [tx_id, tx]:
appointment_data["locator"] = compute_locator(tx_id)
appointment_data["encrypted_blob"] = Cryptographer.encrypt(Blob(tx), tx_id)
if not is_256b_hex_str(tx_id):
raise InvalidParameter("The provided locator is wrong or missing")
else:
logger.error("Appointment data is missing some fields")
return False
if not tx:
raise InvalidParameter("The provided data is missing the transaction")
appointment_data["locator"] = compute_locator(tx_id)
appointment_data["encrypted_blob"] = Cryptographer.encrypt(Blob(tx), tx_id)
appointment = Appointment.from_dict(appointment_data)
signature = Cryptographer.sign(appointment.serialize(), cli_sk)
if not (appointment and signature):
return False
# FIXME: the cryptographer should return exception we can capture
if not signature:
raise ValueError("The provided appointment cannot be signed")
data = {"appointment": appointment.to_dict(), "signature": signature}
# Send appointment to the server.
add_appointment_endpoint = "{}/add_appointment".format(teos_url)
logger.info("Sending appointment to the Eye of Satoshi")
server_response = post_request(data, add_appointment_endpoint)
if server_response is None:
return False
add_appointment_endpoint = "{}/add_appointment".format(teos_url)
response = process_post_response(post_request(data, add_appointment_endpoint))
response_json = process_post_response(server_response)
if response_json is None:
return False
signature = response_json.get("signature")
signature = response.get("signature")
# Check that the server signed the appointment as it should.
if signature is None:
logger.error("The response does not contain the signature of the appointment")
return False
if not signature:
raise TowerResponseError("The response does not contain the signature of the appointment")
rpk = Cryptographer.recover_pk(appointment.serialize(), signature)
if not Cryptographer.verify_rpk(teos_pk, rpk):
logger.error("The returned appointment's signature is invalid")
return False
raise TowerResponseError("The returned appointment's signature is invalid")
logger.info("Appointment accepted and signed by the Eye of Satoshi")
logger.info("Remaining slots: {}".format(response_json.get("available_slots")))
logger.info("Remaining slots: {}".format(response.get("available_slots")))
# All good, store appointment and signature
return save_appointment_receipt(appointment.to_dict(), signature, appointments_folder_path)
return appointment, signature
def get_appointment(locator, cli_sk, teos_pk, teos_url):
@@ -150,17 +145,20 @@ def get_appointment(locator, cli_sk, teos_pk, teos_url):
teos_url (:obj:`str`): the teos base url.
Returns:
:obj:`dict` or :obj:`None`: a dictionary containing the appointment data if the locator is valid and the tower
responds. ``None`` otherwise.
:obj:`dict`: a dictionary containing the appointment data.
Raises:
:obj:`InvalidParameter <cli.exceptions.InvalidParameter>`: if `appointment_data` or any of its fields is
invalid.
:obj:`ConnectionError`: if the client cannot connect to the tower.
:obj:`TowerResponseError <cli.exceptions.TowerResponseError>`: if the tower responded with an error, or the
response was invalid.
"""
# FIXME: All responses from the tower should be signed. Not using teos_pk atm.
valid_locator = is_locator(locator)
if not valid_locator:
logger.error("The provided locator is not valid", locator=locator)
return None
if not is_locator(locator):
raise InvalidParameter("The provided locator is not valid", locator=locator)
message = "get appointment {}".format(locator)
signature = Cryptographer.sign(message.encode(), cli_sk)
@@ -169,10 +167,9 @@ def get_appointment(locator, cli_sk, teos_pk, teos_url):
# Send request to the server.
get_appointment_endpoint = "{}/get_appointment".format(teos_url)
logger.info("Sending appointment to the Eye of Satoshi")
server_response = post_request(data, get_appointment_endpoint)
response_json = process_post_response(server_response)
response = process_post_response(post_request(data, get_appointment_endpoint))
return response_json
return response
def load_keys(teos_pk_path, cli_sk_path, cli_pk_path):
@@ -185,45 +182,41 @@ def load_keys(teos_pk_path, cli_sk_path, cli_pk_path):
cli_pk_path (:obj:`str`): path to the client public key file.
Returns:
:obj:`tuple` or ``None``: a three-item tuple containing a ``PrivateKey``, a ``PublicKey`` and a ``str``
representing the tower pk, user sk and user compressed pk respectively if all keys can be loaded.
``None`` otherwise.
:obj:`tuple`: a three-item tuple containing a ``PrivateKey``, a ``PublicKey`` and a ``str``
representing the tower pk, user sk and user compressed pk respectively.
Raises:
:obj:`InvalidKey <cli.exceptions.InvalidKey>`: if any of the keys is invalid or cannot be loaded.
"""
if teos_pk_path is None:
logger.error("TEOS's public key file not found. Please check your settings")
return None
if not teos_pk_path:
raise InvalidKey("TEOS's public key file not found. Please check your settings")
if cli_sk_path is None:
logger.error("Client's private key file not found. Please check your settings")
return None
if not cli_sk_path:
raise InvalidKey("Client's private key file not found. Please check your settings")
if cli_pk_path is None:
logger.error("Client's public key file not found. Please check your settings")
return None
if not cli_pk_path:
raise InvalidKey("Client's public key file not found. Please check your settings")
try:
teos_pk_der = Cryptographer.load_key_file(teos_pk_path)
teos_pk = PublicKey(teos_pk_der)
except ValueError:
logger.error("TEOS public key is invalid or cannot be parsed")
return None
raise InvalidKey("TEOS public key is invalid or cannot be parsed")
cli_sk_der = Cryptographer.load_key_file(cli_sk_path)
cli_sk = Cryptographer.load_private_key_der(cli_sk_der)
if cli_sk is None:
logger.error("Client private key is invalid or cannot be parsed")
return None
raise InvalidKey("Client private key is invalid or cannot be parsed")
try:
cli_pk_der = Cryptographer.load_key_file(cli_pk_path)
compressed_cli_pk = Cryptographer.get_compressed_pk(PublicKey(cli_pk_der))
except ValueError:
logger.error("Client public key is invalid or cannot be parsed")
return None
raise InvalidKey("Client public key is invalid or cannot be parsed")
return teos_pk, cli_sk, compressed_cli_pk
@@ -237,26 +230,25 @@ def post_request(data, endpoint):
endpoint (:obj:`str`): the endpoint to send the post request.
Returns:
:obj:`dict` or ``None``: a json-encoded dictionary with the server response if the data can be posted.
``None`` otherwise.
:obj:`dict`: a json-encoded dictionary with the server response if the data can be posted.
Raises:
:obj:`ConnectionError`: if the client cannot connect to the tower.
"""
try:
return requests.post(url=endpoint, json=data, timeout=5)
except ConnectTimeout:
logger.error("Can't connect to the Eye of Satoshi's API. Connection timeout")
message = "Can't connect to the Eye of Satoshi's API. Connection timeout"
except ConnectionError:
logger.error("Can't connect to the Eye of Satoshi's API. Server cannot be reached")
message = "Can't connect to the Eye of Satoshi's API. Server cannot be reached"
except (InvalidSchema, MissingSchema, InvalidURL):
logger.error("Invalid URL. No schema, or invalid schema, found ({})".format(endpoint))
message = "Invalid URL. No schema, or invalid schema, found ({})".format(endpoint)
except requests.exceptions.Timeout:
logger.error("The request timed out")
return None
raise ConnectionError(message)
def process_post_response(response):
@@ -267,27 +259,26 @@ def process_post_response(response):
response (:obj:`requests.models.Response`): a ``Response`` object obtained from the request.
Returns:
:obj:`dict` or :obj:`None`: a dictionary containing the tower's response data if the response type is
``HTTP_OK`` and the response can be properly parsed. ``None`` otherwise.
"""
:obj:`dict`: a dictionary containing the tower's response data if the response type is
``HTTP_OK``.
if not response:
return None
Raises:
:obj:`TowerResponseError <cli.exceptions.TowerResponseError>`: if the tower responded with an error, or the
response was invalid.
"""
try:
response_json = response.json()
except (json.JSONDecodeError, AttributeError):
logger.error(
raise TowerResponseError(
"The server returned a non-JSON response", status_code=response.status_code, reason=response.reason
)
return None
if response.status_code != constants.HTTP_OK:
logger.error(
raise TowerResponseError(
"The server returned an error", status_code=response.status_code, reason=response.reason, data=response_json
)
return None
return response_json
@@ -301,15 +292,18 @@ def parse_add_appointment_args(args):
option and the path to a file containing a json encoded appointment.
Returns:
:obj:`dict` or :obj:`None`: A dictionary containing the appointment data if it can be loaded. ``None``
otherwise.
:obj:`dict`: A dictionary containing the appointment data.
Raises:
:obj:`InvalidParameter <cli.exceptions.InvalidParameter>`: if the appointment data is not JSON encoded.
:obj:`FileNotFoundError`: if -f is passed and the appointment file is not found.
:obj:`IOError`: if -f was passed and the file cannot be read.
"""
use_help = "Use 'help add_appointment' for help of how to use the command"
if not args:
logger.error("No appointment data provided. " + use_help)
return None
raise InvalidParameter("No appointment data provided. " + use_help)
arg_opt = args.pop(0)
@@ -320,22 +314,20 @@ def parse_add_appointment_args(args):
if arg_opt in ["-f", "--file"]:
fin = args.pop(0)
if not os.path.isfile(fin):
logger.error("Can't find file", filename=fin)
return None
raise FileNotFoundError("Cannot find {}".format(fin))
try:
with open(fin) as f:
appointment_data = json.load(f)
except IOError as e:
logger.error("I/O error", errno=e.errno, error=e.strerror)
return None
raise IOError("Cannot read appointment file. {}".format(str(e)))
else:
appointment_data = json.loads(arg_opt)
except json.JSONDecodeError:
logger.error("Non-JSON encoded data provided as appointment. " + use_help)
return None
raise InvalidParameter("Non-JSON encoded data provided as appointment. " + use_help)
return appointment_data
@@ -349,9 +341,6 @@ def save_appointment_receipt(appointment, signature, appointments_folder_path):
signature (:obj:`str`): the signature of the appointment performed by the tower.
appointments_folder_path (:obj:`str`): the path to the appointments folder.
Returns:
:obj:`bool`: True if the appointment if properly saved. False otherwise.
Raises:
IOError: if an error occurs whilst writing the file on disk.
"""
@@ -370,14 +359,12 @@ def save_appointment_receipt(appointment, signature, appointments_folder_path):
with open(filename, "w") as f:
json.dump(data, f)
logger.info("Appointment saved at {}".format(filename))
return True
except IOError as e:
logger.error("There was an error while saving the appointment", error=e)
return False
raise IOError("There was an error while saving the appointment. {}".format(e))
def main(args, command_line_conf):
def main(command, args, command_line_conf):
# Loads config and sets up the data folder and log file
config_loader = ConfigLoader(DATA_DIR, CONF_FILE_NAME, DEFAULT_CONF, command_line_conf)
config = config_loader.build_config()
@@ -391,68 +378,59 @@ def main(args, command_line_conf):
if not teos_url.startswith("http"):
teos_url = "http://" + teos_url
keys = load_keys(config.get("TEOS_PUBLIC_KEY"), config.get("CLI_PRIVATE_KEY"), config.get("CLI_PUBLIC_KEY"))
if keys is not None:
teos_pk, cli_sk, compressed_cli_pk = keys
try:
teos_pk, cli_sk, compressed_cli_pk = load_keys(
config.get("TEOS_PUBLIC_KEY"), config.get("CLI_PRIVATE_KEY"), config.get("CLI_PUBLIC_KEY")
)
try:
if command == "register":
register_data = register(compressed_cli_pk, teos_url)
logger.info("Registration succeeded. Available slots: {}".format(register_data.get("available_slots")))
if command == "add_appointment":
appointment_data = parse_add_appointment_args(args)
appointment, signature = add_appointment(appointment_data, cli_sk, teos_pk, teos_url)
save_appointment_receipt(appointment.to_dict(), signature, config.get("APPOINTMENTS_FOLDER_NAME"))
elif command == "get_appointment":
if not args:
logger.error("No arguments were given")
else:
arg_opt = args.pop(0)
if arg_opt in ["-h", "--help"]:
sys.exit(help_get_appointment())
appointment_data = get_appointment(arg_opt, cli_sk, teos_pk, teos_url)
if appointment_data:
print(appointment_data)
elif command == "help":
if args:
command = args.pop(0)
if command in commands:
if command == "register":
register_data = register(compressed_cli_pk, teos_url)
if register_data:
print(register_data)
if command == "register":
sys.exit(help_register())
if command == "add_appointment":
# Get appointment data from user.
appointment_data = parse_add_appointment_args(args)
add_appointment(
appointment_data, cli_sk, teos_pk, teos_url, config.get("APPOINTMENTS_FOLDER_NAME")
)
if command == "add_appointment":
sys.exit(help_add_appointment())
elif command == "get_appointment":
if not args:
logger.error("No arguments were given")
else:
arg_opt = args.pop(0)
if arg_opt in ["-h", "--help"]:
sys.exit(help_get_appointment())
appointment_data = get_appointment(arg_opt, cli_sk, teos_pk, teos_url)
if appointment_data:
print(appointment_data)
elif command == "help":
if args:
command = args.pop(0)
if command == "register":
sys.exit(help_register())
if command == "add_appointment":
sys.exit(help_add_appointment())
elif command == "get_appointment":
sys.exit(help_get_appointment())
else:
logger.error("Unknown command. Use help to check the list of available commands")
else:
sys.exit(show_usage())
elif command == "get_appointment":
sys.exit(help_get_appointment())
else:
logger.error("Unknown command. Use help to check the list of available commands")
else:
logger.error("No command provided. Use help to check the list of available commands")
sys.exit(show_usage())
except json.JSONDecodeError:
logger.error("Non-JSON encoded appointment passed as parameter")
except (FileNotFoundError, IOError, ConnectionError, ValueError) as e:
logger.error(str(e))
except (InvalidKey, InvalidParameter, TowerResponseError) as e:
logger.error(e.reason, **e.params)
except Exception as e:
logger.error("Unknown error occurred", error=str(e))
if __name__ == "__main__":
@@ -477,7 +455,13 @@ if __name__ == "__main__":
if opt in ["-h", "--help"]:
sys.exit(show_usage())
main(args, command_line_conf)
command = args.pop(0)
if command in commands:
main(command, args, command_line_conf)
elif not command:
logger.error("No command provided. Use help to check the list of available commands")
else:
logger.error("Unknown command. Use help to check the list of available commands")
except GetoptError as e:
logger.error("{}".format(e))

86
test/cli/unit/test_teos_cli.py
View File

@@ -1,19 +1,22 @@
import os
import json
import shutil
import pytest
import responses
from binascii import hexlify
from coincurve import PrivateKey
from requests.exceptions import ConnectionError
from common.blob import Blob
import common.cryptographer
from common.logger import Logger
from common.tools import compute_locator
from common.appointment import Appointment
from common.cryptographer import Cryptographer
from common.blob import Blob
import cli.teos_cli as teos_cli
from cli.exceptions import InvalidParameter, InvalidKey, TowerResponseError
from test.cli.unit.conftest import get_random_value_hex, get_config
common.cryptographer.logger = Logger(actor="Cryptographer", log_name_prefix=teos_cli.LOG_PREFIX)
@@ -84,9 +87,7 @@ def test_add_appointment():
"available_slots": 100,
}
responses.add(responses.POST, add_appointment_endpoint, json=response, status=200)
result = teos_cli.add_appointment(
dummy_appointment_data, dummy_cli_sk, dummy_teos_pk, teos_url, config.get("APPOINTMENTS_FOLDER_NAME")
)
result = teos_cli.add_appointment(dummy_appointment_data, dummy_cli_sk, dummy_teos_pk, teos_url)
assert len(responses.calls) == 1
assert responses.calls[0].request.url == add_appointment_endpoint
@@ -105,13 +106,9 @@ def test_add_appointment_with_invalid_signature(monkeypatch):
}
responses.add(responses.POST, add_appointment_endpoint, json=response, status=200)
result = teos_cli.add_appointment(
dummy_appointment_data, dummy_cli_sk, dummy_teos_pk, teos_url, config.get("APPOINTMENTS_FOLDER_NAME")
)
assert result is False
shutil.rmtree(config.get("APPOINTMENTS_FOLDER_NAME"))
with pytest.raises(TowerResponseError):
teos_cli.add_appointment(dummy_appointment_data, dummy_cli_sk, dummy_teos_pk, teos_url)
@responses.activate
@@ -138,7 +135,8 @@ def test_get_appointment_err():
# Test that get_appointment handles a connection error appropriately.
responses.add(responses.POST, get_appointment_endpoint, body=ConnectionError())
assert not teos_cli.get_appointment(locator, dummy_cli_sk, dummy_teos_pk, teos_url)
with pytest.raises(ConnectionError):
teos_cli.get_appointment(locator, dummy_cli_sk, dummy_teos_pk, teos_url)
def test_load_keys():
@@ -150,7 +148,7 @@ def test_load_keys():
f.write(dummy_cli_sk.to_der())
with open(public_key_file_path, "wb") as f:
f.write(dummy_cli_compressed_pk)
with open(empty_file_path, "wb") as f:
with open(empty_file_path, "wb"):
pass
# Now we can test the function passing the using this files (we'll use the same pk for both)
@@ -158,25 +156,32 @@ def test_load_keys():
assert isinstance(r, tuple)
assert len(r) == 3
# If any param does not match we should get None as result
assert teos_cli.load_keys(None, private_key_file_path, public_key_file_path) is None
assert teos_cli.load_keys(public_key_file_path, None, public_key_file_path) is None
assert teos_cli.load_keys(public_key_file_path, private_key_file_path, None) is None
# If any param does not match the expected, we should get an InvalidKey exception
with pytest.raises(InvalidKey):
teos_cli.load_keys(None, private_key_file_path, public_key_file_path)
with pytest.raises(InvalidKey):
teos_cli.load_keys(public_key_file_path, None, public_key_file_path)
with pytest.raises(InvalidKey):
teos_cli.load_keys(public_key_file_path, private_key_file_path, None)
# The same should happen if we pass a public key where a private should be, for instance
assert teos_cli.load_keys(private_key_file_path, public_key_file_path, private_key_file_path) is None
with pytest.raises(InvalidKey):
teos_cli.load_keys(private_key_file_path, public_key_file_path, private_key_file_path)
# Same if any of the files is empty
assert teos_cli.load_keys(empty_file_path, private_key_file_path, public_key_file_path) is None
assert teos_cli.load_keys(public_key_file_path, empty_file_path, public_key_file_path) is None
assert teos_cli.load_keys(public_key_file_path, private_key_file_path, empty_file_path) is None
with pytest.raises(InvalidKey):
teos_cli.load_keys(empty_file_path, private_key_file_path, public_key_file_path)
with pytest.raises(InvalidKey):
teos_cli.load_keys(public_key_file_path, empty_file_path, public_key_file_path)
with pytest.raises(InvalidKey):
teos_cli.load_keys(public_key_file_path, private_key_file_path, empty_file_path)
# Remove the tmp files
os.remove(private_key_file_path)
os.remove(public_key_file_path)
os.remove(empty_file_path)
# WIP: HERE
@responses.activate
def test_post_request():
response = {
@@ -207,24 +212,23 @@ def test_process_post_response():
# If we modify the response code for a rejection (lets say 404) we should get None
responses.replace(responses.POST, add_appointment_endpoint, json=response, status=404)
r = teos_cli.post_request(json.dumps(dummy_appointment_data), add_appointment_endpoint)
assert teos_cli.process_post_response(r) is None
with pytest.raises(TowerResponseError):
r = teos_cli.post_request(json.dumps(dummy_appointment_data), add_appointment_endpoint)
teos_cli.process_post_response(r)
# The same should happen if the response is not in json
# The same should happen if the response is not in json independently of the return type
responses.replace(responses.POST, add_appointment_endpoint, status=404)
r = teos_cli.post_request(json.dumps(dummy_appointment_data), add_appointment_endpoint)
assert teos_cli.process_post_response(r) is None
with pytest.raises(TowerResponseError):
r = teos_cli.post_request(json.dumps(dummy_appointment_data), add_appointment_endpoint)
teos_cli.process_post_response(r)
responses.replace(responses.POST, add_appointment_endpoint, status=200)
with pytest.raises(TowerResponseError):
r = teos_cli.post_request(json.dumps(dummy_appointment_data), add_appointment_endpoint)
teos_cli.process_post_response(r)
def test_parse_add_appointment_args():
# If no args are passed, function should fail.
appt_data = teos_cli.parse_add_appointment_args(None)
assert not appt_data
# If file doesn't exist, function should fail.
appt_data = teos_cli.parse_add_appointment_args(["-f", "nonexistent_file"])
assert not appt_data
# If file exists and has data in it, function should work.
with open("appt_test_file", "w") as f:
json.dump(dummy_appointment_data, f)
@@ -232,12 +236,22 @@ def test_parse_add_appointment_args():
appt_data = teos_cli.parse_add_appointment_args(["-f", "appt_test_file"])
assert appt_data
os.remove("appt_test_file")
# If appointment json is passed in, function should work.
appt_data = teos_cli.parse_add_appointment_args([json.dumps(dummy_appointment_data)])
assert appt_data
os.remove("appt_test_file")
def test_parse_add_appointment_args_wrong():
# If no args are passed, function should fail.
with pytest.raises(InvalidParameter):
teos_cli.parse_add_appointment_args(None)
# If file doesn't exist, function should fail.
with pytest.raises(FileNotFoundError):
teos_cli.parse_add_appointment_args(["-f", "nonexistent_file"])
def test_save_appointment_receipt(monkeypatch):
appointments_folder = "test_appointments_receipts"

110
test/teos/e2e/test_basic_e2e.py
View File

@@ -1,8 +1,10 @@
import pytest
from time import sleep
from riemann.tx import Tx
from binascii import hexlify
from coincurve import PrivateKey
from cli.exceptions import TowerResponseError
from cli import teos_cli, DATA_DIR, DEFAULT_CONF, CONF_FILE_NAME
import common.cryptographer
@@ -48,9 +50,7 @@ def get_appointment_info(locator, sk=cli_sk):
def add_appointment(appointment_data, sk=cli_sk):
return teos_cli.add_appointment(
appointment_data, sk, teos_pk, teos_base_endpoint, cli_config.get("APPOINTMENTS_FOLDER_NAME")
)
return teos_cli.add_appointment(appointment_data, sk, teos_pk, teos_base_endpoint)
def test_commands_non_registered(bitcoin_cli, create_txs):
@@ -61,10 +61,12 @@ def test_commands_non_registered(bitcoin_cli, create_txs):
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
assert add_appointment(appointment_data) is False
with pytest.raises(TowerResponseError):
assert add_appointment(appointment_data)
# Get appointment
assert get_appointment_info(appointment_data.get("locator")) is None
with pytest.raises(TowerResponseError):
assert get_appointment_info(appointment_data.get("locator"))
def test_commands_registered(bitcoin_cli, create_txs):
@@ -76,37 +78,39 @@ def test_commands_registered(bitcoin_cli, create_txs):
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
assert add_appointment(appointment_data) is True
appointment, available_slots = add_appointment(appointment_data)
assert isinstance(appointment, Appointment) and isinstance(available_slots, str)
# Get appointment
r = get_appointment_info(appointment_data.get("locator"))
assert r.get("locator") == appointment_data.get("locator")
assert r.get("appointment").get("locator") == appointment_data.get("locator")
assert r.get("appointment").get("encrypted_blob") == appointment_data.get("encrypted_blob")
assert r.get("appointment").get("start_time") == appointment_data.get("start_time")
assert r.get("appointment").get("end_time") == appointment_data.get("end_time")
assert r.get("locator") == appointment.locator
assert r.get("appointment") == appointment.to_dict()
def test_appointment_life_cycle(bitcoin_cli, create_txs):
# First of all we need to register
# FIXME: requires register command in the cli
teos_cli.register(compressed_cli_pk, teos_base_endpoint)
# After that we can build an appointment and send it to the tower
commitment_tx, penalty_tx = create_txs
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
locator = compute_locator(commitment_tx_id)
appointment, available_slots = add_appointment(appointment_data)
assert add_appointment(appointment_data) is True
# Get the information from the tower to check that it matches
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "being_watched"
assert appointment_info.get("locator") == locator
assert appointment_info.get("appointment") == appointment.to_dict()
# Trigger a breach and check again
new_addr = bitcoin_cli.getnewaddress()
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "dispute_responded"
assert appointment_info.get("locator") == locator
# It can be also checked by ensuring that the penalty transaction made it to the network
penalty_tx_id = bitcoin_cli.decoderawtransaction(penalty_tx).get("txid")
@@ -120,13 +124,12 @@ def test_appointment_life_cycle(bitcoin_cli, create_txs):
assert False
# Now let's mine some blocks so the appointment reaches its end.
# Since we are running all the nodes remotely data may take more time than normal, and some confirmations may be
# missed, so we generate more than enough confirmations and add some delays.
for _ in range(int(1.5 * END_TIME_DELTA)):
sleep(1)
for _ in range(END_TIME_DELTA):
bitcoin_cli.generatetoaddress(1, new_addr)
assert get_appointment_info(locator) is None
# The appointment is no longer in the tower
with pytest.raises(TowerResponseError):
get_appointment_info(locator)
def test_appointment_malformed_penalty(bitcoin_cli, create_txs):
@@ -142,18 +145,24 @@ def test_appointment_malformed_penalty(bitcoin_cli, create_txs):
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, mod_penalty_tx.hex())
locator = compute_locator(commitment_tx_id)
assert add_appointment(appointment_data) is True
appointment, _ = add_appointment(appointment_data)
# Get the information from the tower to check that it matches
appointment_info = get_appointment_info(locator)
assert appointment_info.get("status") == "being_watched"
assert appointment_info.get("locator") == locator
assert appointment_info.get("appointment") == appointment.to_dict()
# Broadcast the commitment transaction and mine a block
new_addr = bitcoin_cli.getnewaddress()
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
# The appointment should have been removed since the penalty_tx was malformed.
sleep(1)
assert get_appointment_info(locator) is None
with pytest.raises(TowerResponseError):
get_appointment_info(locator)
def test_appointment_wrong_key(bitcoin_cli, create_txs):
def test_appointment_wrong_decryption_key(bitcoin_cli, create_txs):
# This tests an appointment encrypted with a key that has not been derived from the same source as the locator.
# Therefore the tower won't be able to decrypt the blob once the appointment is triggered.
commitment_tx, penalty_tx = create_txs
@@ -176,7 +185,6 @@ def test_appointment_wrong_key(bitcoin_cli, create_txs):
# Check that the server has accepted the appointment
signature = response_json.get("signature")
assert signature is not None
rpk = Cryptographer.recover_pk(appointment.serialize(), signature)
assert Cryptographer.verify_rpk(teos_pk, rpk) is True
assert response_json.get("locator") == appointment.locator
@@ -186,14 +194,14 @@ def test_appointment_wrong_key(bitcoin_cli, create_txs):
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
# The appointment should have been removed since the decryption failed.
sleep(1)
assert get_appointment_info(appointment.locator) is None
with pytest.raises(TowerResponseError):
get_appointment_info(appointment.locator)
def test_two_identical_appointments(bitcoin_cli, create_txs):
# Tests sending two identical appointments to the tower.
# This tests sending an appointment with two valid transaction with the same locator.
# If they come from the same user, the last one will be kept
# If they come from the same user, the last one will be kept.
commitment_tx, penalty_tx = create_txs
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
@@ -201,18 +209,16 @@ def test_two_identical_appointments(bitcoin_cli, create_txs):
locator = compute_locator(commitment_tx_id)
# Send the appointment twice
assert add_appointment(appointment_data) is True
assert add_appointment(appointment_data) is True
add_appointment(appointment_data)
add_appointment(appointment_data)
# Broadcast the commitment transaction and mine a block
new_addr = bitcoin_cli.getnewaddress()
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
# The last appointment should have made it to the Responder
sleep(1)
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "dispute_responded"
assert appointment_info.get("appointment").get("penalty_rawtx") == penalty_tx
@@ -266,7 +272,7 @@ def test_two_identical_appointments(bitcoin_cli, create_txs):
def test_two_appointment_same_locator_different_penalty_different_users(bitcoin_cli, create_txs):
# This tests sending an appointment with two valid transaction with the same locator.
# This tests sending an appointment with two valid transaction with the same locator fro different users
commitment_tx, penalty_tx1 = create_txs
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
@@ -279,28 +285,24 @@ def test_two_appointment_same_locator_different_penalty_different_users(bitcoin_
appointment2_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx2)
locator = compute_locator(commitment_tx_id)
# tmp keys from a different user
# tmp keys for a different user
tmp_sk = PrivateKey()
tmp_compressed_pk = hexlify(tmp_sk.public_key.format(compressed=True)).decode("utf-8")
teos_cli.register(tmp_compressed_pk, teos_base_endpoint)
assert add_appointment(appointment1_data) is True
assert add_appointment(appointment2_data, sk=tmp_sk) is True
appointment, _ = add_appointment(appointment1_data)
appointment_2, _ = add_appointment(appointment2_data, sk=tmp_sk)
# Broadcast the commitment transaction and mine a block
new_addr = bitcoin_cli.getnewaddress()
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
# One of the transactions must have made it to the Responder while the other must have been dropped for
# double-spending
sleep(1)
appointment_info = get_appointment_info(locator)
appointment2_info = get_appointment_info(locator, sk=tmp_sk)
# One of the two request must be None, while the other must be valid
assert (appointment_info is None and appointment2_info is not None) or (
appointment2_info is None and appointment_info is not None
)
# double-spending. That means that one of the responses from the tower should fail
appointment_info = None
with pytest.raises(TowerResponseError):
appointment_info = get_appointment_info(locator)
appointment2_info = get_appointment_info(locator, sk=tmp_sk)
if appointment_info is None:
appointment_info = appointment2_info
@@ -308,6 +310,7 @@ def test_two_appointment_same_locator_different_penalty_different_users(bitcoin_
assert appointment_info.get("status") == "dispute_responded"
assert appointment_info.get("locator") == appointment1_data.get("locator")
assert appointment_info.get("appointment").get("penalty_tx") == appointment1_data.get("penalty_tx")
def test_appointment_shutdown_teos_trigger_back_online(create_txs, bitcoin_cli):
@@ -320,7 +323,7 @@ def test_appointment_shutdown_teos_trigger_back_online(create_txs, bitcoin_cli):
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
locator = compute_locator(commitment_tx_id)
assert add_appointment(appointment_data) is True
appointment, _ = add_appointment(appointment_data)
# Restart teos
teosd_process.terminate()
@@ -329,21 +332,17 @@ def test_appointment_shutdown_teos_trigger_back_online(create_txs, bitcoin_cli):
assert teos_pid != teosd_process.pid
# Check that the appointment is still in the Watcher
sleep(1)
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "being_watched"
assert appointment_info.get("appointment") == appointment.to_dict()
# Trigger appointment after restart
new_addr = bitcoin_cli.getnewaddress()
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
# The appointment should have been moved to the Responder
sleep(1)
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "dispute_responded"
@@ -357,12 +356,12 @@ def test_appointment_shutdown_teos_trigger_while_offline(create_txs, bitcoin_cli
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
locator = compute_locator(commitment_tx_id)
assert add_appointment(appointment_data) is True
appointment, _ = add_appointment(appointment_data)
# Check that the appointment is still in the Watcher
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "being_watched"
assert appointment_info.get("appointment") == appointment.to_dict()
# Shutdown and trigger
teosd_process.terminate()
@@ -374,10 +373,7 @@ def test_appointment_shutdown_teos_trigger_while_offline(create_txs, bitcoin_cli
assert teos_pid != teosd_process.pid
# The appointment should have been moved to the Responder
sleep(1)
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "dispute_responded"
teosd_process.terminate()