mirror of
https://github.com/aljazceru/python-teos.git
synced 2025-12-18 06:34:19 +01:00
Renames pisa_cli to wt_cli
This commit is contained in:
Sergi Delgado Segura
@@ -11,9 +11,9 @@ conf_fields = {
|
||||
"DATA_FOLDER": {"value": conf.DATA_FOLDER, "type": str},
|
||||
"CLIENT_LOG_FILE": {"value": conf.CLIENT_LOG_FILE, "type": str, "path": True},
|
||||
"APPOINTMENTS_FOLDER_NAME": {"value": conf.APPOINTMENTS_FOLDER_NAME, "type": str, "path": True},
|
||||
"CLI_PUBLIC_KEY": {"value": conf.CLI_PUBLIC_KEY, "type": str, "path": True},
|
||||
"CLI_PRIVATE_KEY": {"value": conf.CLI_PRIVATE_KEY, "type": str, "path": True},
|
||||
"PISA_PUBLIC_KEY": {"value": conf.PISA_PUBLIC_KEY, "type": str, "path": True},
|
||||
# "CLI_PUBLIC_KEY": {"value": conf.CLI_PUBLIC_KEY, "type": str, "path": True},
|
||||
# "CLI_PRIVATE_KEY": {"value": conf.CLI_PRIVATE_KEY, "type": str, "path": True},
|
||||
# "PISA_PUBLIC_KEY": {"value": conf.PISA_PUBLIC_KEY, "type": str, "path": True},
|
||||
}
|
||||
|
||||
# Expand user (~) if found and check fields are correct
|
||||
|
||||
@@ -7,7 +7,3 @@ DATA_FOLDER = "~/.pisa_btc/"
|
||||
|
||||
CLIENT_LOG_FILE = "cli.log"
|
||||
APPOINTMENTS_FOLDER_NAME = "appointment_receipts"
|
||||
|
||||
CLI_PUBLIC_KEY = "cli_pk.der"
|
||||
CLI_PRIVATE_KEY = "cli_sk.der"
|
||||
PISA_PUBLIC_KEY = "pisa_pk.der"
|
||||
|
||||
@@ -22,65 +22,49 @@ from common.tools import check_sha256_hex_format, check_locator_format, compute_
|
||||
logger = Logger(actor="Client", log_name_prefix=LOG_PREFIX)
|
||||
|
||||
|
||||
# FIXME: TESTING ENDPOINT, WON'T BE THERE IN PRODUCTION
|
||||
def generate_dummy_appointment():
|
||||
get_block_count_end_point = "http://{}:{}/get_block_count".format(pisa_api_server, pisa_api_port)
|
||||
r = requests.get(url=get_block_count_end_point, timeout=5)
|
||||
|
||||
current_height = r.json().get("block_count")
|
||||
|
||||
dummy_appointment_data = {
|
||||
"tx": os.urandom(192).hex(),
|
||||
"tx_id": os.urandom(32).hex(),
|
||||
"start_time": current_height + 5,
|
||||
"end_time": current_height + 10,
|
||||
"to_self_delay": 20,
|
||||
}
|
||||
|
||||
logger.info(
|
||||
"Generating dummy appointment data:" "\n\n" + json.dumps(dummy_appointment_data, indent=4, sort_keys=True)
|
||||
)
|
||||
|
||||
json.dump(dummy_appointment_data, open("dummy_appointment_data.json", "w"))
|
||||
|
||||
logger.info("\nData stored in dummy_appointment_data.json")
|
||||
# FIXME: creating a simpler load_keys for the alpha. Client keys will not be necessary. PISA key is hardcoded.
|
||||
# def load_keys(pisa_pk_path, cli_sk_path, cli_pk_path):
|
||||
# """
|
||||
# Loads all the keys required so sign, send, and verify the appointment.
|
||||
#
|
||||
# Args:
|
||||
# pisa_pk_path (:obj:`str`): path to the PISA public key file.
|
||||
# cli_sk_path (:obj:`str`): path to the client private key file.
|
||||
# cli_pk_path (:obj:`str`): path to the client public key file.
|
||||
#
|
||||
# Returns:
|
||||
# :obj:`tuple` or ``None``: a three item tuple containing a pisa_pk object, cli_sk object and the cli_sk_der
|
||||
# encoded key if all keys can be loaded. ``None`` otherwise.
|
||||
# """
|
||||
#
|
||||
# pisa_pk_der = Cryptographer.load_key_file(pisa_pk_path)
|
||||
# pisa_pk = Cryptographer.load_public_key_der(pisa_pk_der)
|
||||
#
|
||||
# if pisa_pk is None:
|
||||
# logger.error("PISA's public key file not found. Please check your settings")
|
||||
# return None
|
||||
#
|
||||
# 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's private key file not found. Please check your settings")
|
||||
# return None
|
||||
#
|
||||
# cli_pk_der = Cryptographer.load_key_file(cli_pk_path)
|
||||
#
|
||||
# if cli_pk_der is None:
|
||||
# logger.error("Client's public key file not found. Please check your settings")
|
||||
# return None
|
||||
#
|
||||
# return pisa_pk, cli_sk, cli_pk_der
|
||||
|
||||
|
||||
def load_keys(pisa_pk_path, cli_sk_path, cli_pk_path):
|
||||
"""
|
||||
Loads all the keys required so sign, send, and verify the appointment.
|
||||
def load_keys():
|
||||
PISA_PUBLIC_KEY = "3056301006072a8648ce3d020106052b8104000a0342000430053e39c53b8bcb43354a4ed886b8082af1d1e8fc14956e60ad0592bfdfab511b7e309f6ac83b7495462196692e145bf7b1a321e96ec8fc4d678719c77342da"
|
||||
pisa_pk = Cryptographer.load_public_key_der(binascii.unhexlify(PISA_PUBLIC_KEY))
|
||||
|
||||
Args:
|
||||
pisa_pk_path (:obj:`str`): path to the PISA public key file.
|
||||
cli_sk_path (:obj:`str`): path to the client private key file.
|
||||
cli_pk_path (:obj:`str`): path to the client public key file.
|
||||
|
||||
Returns:
|
||||
:obj:`tuple` or ``None``: a three item tuple containing a pisa_pk object, cli_sk object and the cli_sk_der
|
||||
encoded key if all keys can be loaded. ``None`` otherwise.
|
||||
"""
|
||||
|
||||
pisa_pk_der = Cryptographer.load_key_file(pisa_pk_path)
|
||||
pisa_pk = Cryptographer.load_public_key_der(pisa_pk_der)
|
||||
|
||||
if pisa_pk is None:
|
||||
logger.error("PISA's public key file not found. Please check your settings")
|
||||
return None
|
||||
|
||||
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's private key file not found. Please check your settings")
|
||||
return None
|
||||
|
||||
cli_pk_der = Cryptographer.load_key_file(cli_pk_path)
|
||||
|
||||
if cli_pk_der is None:
|
||||
logger.error("Client's public key file not found. Please check your settings")
|
||||
return None
|
||||
|
||||
return pisa_pk, cli_sk, cli_pk_der
|
||||
return pisa_pk
|
||||
|
||||
|
||||
def add_appointment(args):
|
||||
@@ -109,16 +93,20 @@ def add_appointment(args):
|
||||
: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.
|
||||
"""
|
||||
# FIXME: creating a simpler load_keys for the alpha. Client keys will not be necessary. PISA key is hardcoded.
|
||||
# pisa_pk, cli_sk, cli_pk_der = load_keys(
|
||||
# config.get("PISA_PUBLIC_KEY"), config.get("CLI_PRIVATE_KEY"), config.get("CLI_PUBLIC_KEY")
|
||||
# )
|
||||
#
|
||||
# try:
|
||||
# hex_pk_der = binascii.hexlify(cli_pk_der)
|
||||
#
|
||||
# except binascii.Error as e:
|
||||
# logger.error("Could not successfully encode public key as hex", error=str(e))
|
||||
# return False
|
||||
pisa_pk = load_keys()
|
||||
|
||||
pisa_pk, cli_sk, cli_pk_der = load_keys(
|
||||
config.get("PISA_PUBLIC_KEY"), config.get("CLI_PRIVATE_KEY"), config.get("CLI_PUBLIC_KEY")
|
||||
)
|
||||
|
||||
try:
|
||||
hex_pk_der = binascii.hexlify(cli_pk_der)
|
||||
|
||||
except binascii.Error as e:
|
||||
logger.error("Could not successfully encode public key as hex", error=str(e))
|
||||
if pisa_pk is None:
|
||||
return False
|
||||
|
||||
# Get appointment data from user.
|
||||
@@ -146,15 +134,21 @@ def add_appointment(args):
|
||||
return False
|
||||
|
||||
appointment = Appointment.from_dict(appointment_data)
|
||||
signature = Cryptographer.sign(appointment.serialize(), cli_sk)
|
||||
|
||||
if not (appointment and signature):
|
||||
return False
|
||||
|
||||
data = {"appointment": appointment.to_dict(), "signature": signature, "public_key": hex_pk_der.decode("utf-8")}
|
||||
# FIXME: getting rid of the client-side signature for the alpha. A proper authentication is required.
|
||||
# signature = Cryptographer.sign(appointment.serialize(), cli_sk)
|
||||
#
|
||||
# if not (appointment and signature):
|
||||
# return False
|
||||
#
|
||||
# data = {"appointment": appointment.to_dict(), "signature": signature, "public_key": hex_pk_der.decode("utf-8")}
|
||||
data = {"appointment": appointment.to_dict()}
|
||||
|
||||
# Send appointment to the server.
|
||||
server_response = post_appointment(data)
|
||||
if server_response is None:
|
||||
return False
|
||||
|
||||
response_json = process_post_appointment_response(server_response)
|
||||
|
||||
if response_json is None:
|
||||
@@ -316,6 +310,7 @@ def save_appointment_receipt(appointment, signature):
|
||||
try:
|
||||
with open(filename, "w") as f:
|
||||
json.dump(data, f)
|
||||
logger.info("Appointment saved at {}".format(filename))
|
||||
return True
|
||||
|
||||
except IOError as e:
|
||||
@@ -378,7 +373,6 @@ if __name__ == "__main__":
|
||||
pisa_api_server = config.get("DEFAULT_PISA_API_SERVER")
|
||||
pisa_api_port = config.get("DEFAULT_PISA_API_PORT")
|
||||
commands = ["add_appointment", "get_appointment", "help"]
|
||||
testing_commands = ["generate_dummy_appointment"]
|
||||
|
||||
try:
|
||||
opts, args = getopt(argv[1:], "s:p:h", ["server", "port", "help"])
|
||||
@@ -432,11 +426,6 @@ if __name__ == "__main__":
|
||||
else:
|
||||
sys.exit(show_usage())
|
||||
|
||||
# FIXME: testing command, not for production
|
||||
elif command in testing_commands:
|
||||
if command == "generate_dummy_appointment":
|
||||
generate_dummy_appointment()
|
||||
|
||||
else:
|
||||
logger.error("Unknown command. Use help to check the list of available commands")
|
||||
|
||||
@@ -58,8 +58,8 @@ class Inspector:
|
||||
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:
|
||||
rcode, message = self.check_appointment_signature(appointment_data, signature, public_key)
|
||||
# if rcode == 0:
|
||||
# rcode, message = self.check_appointment_signature(appointment_data, signature, public_key)
|
||||
|
||||
if rcode == 0:
|
||||
r = Appointment.from_dict(appointment_data)
|
||||
@@ -336,6 +336,11 @@ class Inspector:
|
||||
rcode = errors.APPOINTMENT_EMPTY_FIELD
|
||||
message = "empty signature received"
|
||||
|
||||
elif pk_der is None:
|
||||
rcode = errors.APPOINTMENT_EMPTY_FIELD
|
||||
message = "empty public key received"
|
||||
|
||||
else:
|
||||
pk = Cryptographer.load_public_key_der(unhexlify(pk_der))
|
||||
valid_sig = Cryptographer.verify(Appointment.from_dict(appointment_data).serialize(), signature, pk)
|
||||
|
||||
|
||||
@@ -14,7 +14,7 @@ from common.appointment import Appointment
|
||||
from common.cryptographer import Cryptographer
|
||||
|
||||
from apps.cli.blob import Blob
|
||||
import apps.cli.pisa_cli as pisa_cli
|
||||
import apps.cli.wt_cli as wt_cli
|
||||
from test.apps.cli.unit.conftest import get_random_value_hex
|
||||
|
||||
# dummy keys for the tests
|
||||
@@ -33,11 +33,11 @@ dummy_pk_der = dummy_pk.public_bytes(
|
||||
|
||||
|
||||
# Replace the key in the module with a key we control for the tests
|
||||
pisa_cli.pisa_public_key = dummy_pk
|
||||
wt_cli.pisa_public_key = dummy_pk
|
||||
# Replace endpoint with dummy one
|
||||
pisa_cli.pisa_api_server = "dummy.com"
|
||||
pisa_cli.pisa_api_port = 12345
|
||||
pisa_endpoint = "http://{}:{}/".format(pisa_cli.pisa_api_server, pisa_cli.pisa_api_port)
|
||||
wt_cli.pisa_api_server = "dummy.com"
|
||||
wt_cli.pisa_api_port = 12345
|
||||
pisa_endpoint = "http://{}:{}/".format(wt_cli.pisa_api_server, wt_cli.pisa_api_port)
|
||||
|
||||
dummy_appointment_request = {
|
||||
"tx": get_random_value_hex(192),
|
||||
@@ -62,7 +62,8 @@ dummy_appointment = Appointment.from_dict(dummy_appointment_full)
|
||||
|
||||
|
||||
def load_dummy_keys(*args):
|
||||
return dummy_pk, dummy_sk, dummy_pk_der
|
||||
# return dummy_pk, dummy_sk, dummy_pk_der
|
||||
return dummy_pk
|
||||
|
||||
|
||||
def get_dummy_pisa_pk_der(*args):
|
||||
@@ -81,30 +82,30 @@ def get_bad_signature(*args):
|
||||
return Cryptographer.sign(dummy_appointment.serialize(), another_sk)
|
||||
|
||||
|
||||
def test_load_keys():
|
||||
# Let's first create a private key and public key files
|
||||
private_key_file_path = "sk_test_file"
|
||||
public_key_file_path = "pk_test_file"
|
||||
with open(private_key_file_path, "wb") as f:
|
||||
f.write(dummy_sk_der)
|
||||
with open(public_key_file_path, "wb") as f:
|
||||
f.write(dummy_pk_der)
|
||||
|
||||
# Now we can test the function passing the using this files (we'll use the same pk for both)
|
||||
r = pisa_cli.load_keys(public_key_file_path, private_key_file_path, public_key_file_path)
|
||||
assert isinstance(r, tuple)
|
||||
assert len(r) == 3
|
||||
|
||||
# If any param does not match we should get None as result
|
||||
assert pisa_cli.load_keys(None, private_key_file_path, public_key_file_path) is None
|
||||
assert pisa_cli.load_keys(public_key_file_path, None, public_key_file_path) is None
|
||||
assert pisa_cli.load_keys(public_key_file_path, private_key_file_path, None) is None
|
||||
|
||||
# The same should happen if we pass a public key where a private should be, for instance
|
||||
assert pisa_cli.load_keys(private_key_file_path, public_key_file_path, private_key_file_path) is None
|
||||
|
||||
os.remove(private_key_file_path)
|
||||
os.remove(public_key_file_path)
|
||||
# def test_load_keys():
|
||||
# # Let's first create a private key and public key files
|
||||
# private_key_file_path = "sk_test_file"
|
||||
# public_key_file_path = "pk_test_file"
|
||||
# with open(private_key_file_path, "wb") as f:
|
||||
# f.write(dummy_sk_der)
|
||||
# with open(public_key_file_path, "wb") as f:
|
||||
# f.write(dummy_pk_der)
|
||||
#
|
||||
# # Now we can test the function passing the using this files (we'll use the same pk for both)
|
||||
# r = wt_cli.load_keys(public_key_file_path, private_key_file_path, public_key_file_path)
|
||||
# assert isinstance(r, tuple)
|
||||
# assert len(r) == 3
|
||||
#
|
||||
# # If any param does not match we should get None as result
|
||||
# assert wt_cli.load_keys(None, private_key_file_path, public_key_file_path) is None
|
||||
# assert wt_cli.load_keys(public_key_file_path, None, public_key_file_path) is None
|
||||
# assert wt_cli.load_keys(public_key_file_path, private_key_file_path, None) is None
|
||||
#
|
||||
# # The same should happen if we pass a public key where a private should be, for instance
|
||||
# assert wt_cli.load_keys(private_key_file_path, public_key_file_path, private_key_file_path) is None
|
||||
#
|
||||
# os.remove(private_key_file_path)
|
||||
# os.remove(public_key_file_path)
|
||||
|
||||
|
||||
# TODO: 90-add-more-add-appointment-tests
|
||||
@@ -112,11 +113,11 @@ def test_load_keys():
|
||||
def test_add_appointment(monkeypatch):
|
||||
# Simulate a request to add_appointment for dummy_appointment, make sure that the right endpoint is requested
|
||||
# and the return value is True
|
||||
monkeypatch.setattr(pisa_cli, "load_keys", load_dummy_keys)
|
||||
monkeypatch.setattr(wt_cli, "load_keys", load_dummy_keys)
|
||||
|
||||
response = {"locator": dummy_appointment.locator, "signature": get_dummy_signature()}
|
||||
responses.add(responses.POST, pisa_endpoint, json=response, status=200)
|
||||
result = pisa_cli.add_appointment([json.dumps(dummy_appointment_request)])
|
||||
result = wt_cli.add_appointment([json.dumps(dummy_appointment_request)])
|
||||
|
||||
assert len(responses.calls) == 1
|
||||
assert responses.calls[0].request.url == pisa_endpoint
|
||||
@@ -129,7 +130,7 @@ def test_add_appointment_with_invalid_signature(monkeypatch):
|
||||
# make sure that the right endpoint is requested, but the return value is False
|
||||
|
||||
# Make sure the test uses the bad dummy signature
|
||||
monkeypatch.setattr(pisa_cli, "load_keys", load_dummy_keys)
|
||||
monkeypatch.setattr(wt_cli, "load_keys", load_dummy_keys)
|
||||
|
||||
response = {
|
||||
"locator": dummy_appointment.to_dict()["locator"],
|
||||
@@ -137,31 +138,31 @@ def test_add_appointment_with_invalid_signature(monkeypatch):
|
||||
}
|
||||
|
||||
responses.add(responses.POST, pisa_endpoint, json=response, status=200)
|
||||
result = pisa_cli.add_appointment([json.dumps(dummy_appointment_request)])
|
||||
result = wt_cli.add_appointment([json.dumps(dummy_appointment_request)])
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
def test_parse_add_appointment_args():
|
||||
# If no args are passed, function should fail.
|
||||
appt_data = pisa_cli.parse_add_appointment_args(None)
|
||||
appt_data = wt_cli.parse_add_appointment_args(None)
|
||||
assert not appt_data
|
||||
|
||||
# If file doesn't exist, function should fail.
|
||||
appt_data = pisa_cli.parse_add_appointment_args(["-f", "nonexistent_file"])
|
||||
appt_data = wt_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_request, f)
|
||||
|
||||
appt_data = pisa_cli.parse_add_appointment_args(["-f", "appt_test_file"])
|
||||
appt_data = wt_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 = pisa_cli.parse_add_appointment_args([json.dumps(dummy_appointment_request)])
|
||||
appt_data = wt_cli.parse_add_appointment_args([json.dumps(dummy_appointment_request)])
|
||||
assert appt_data
|
||||
|
||||
|
||||
@@ -173,7 +174,7 @@ def test_post_appointment():
|
||||
}
|
||||
|
||||
responses.add(responses.POST, pisa_endpoint, json=response, status=200)
|
||||
response = pisa_cli.post_appointment(json.dumps(dummy_appointment_request))
|
||||
response = wt_cli.post_appointment(json.dumps(dummy_appointment_request))
|
||||
|
||||
assert len(responses.calls) == 1
|
||||
assert responses.calls[0].request.url == pisa_endpoint
|
||||
@@ -190,27 +191,27 @@ def test_process_post_appointment_response():
|
||||
|
||||
# A 200 OK with a correct json response should return the json of the response
|
||||
responses.add(responses.POST, pisa_endpoint, json=response, status=200)
|
||||
r = pisa_cli.post_appointment(json.dumps(dummy_appointment_request))
|
||||
assert pisa_cli.process_post_appointment_response(r) == r.json()
|
||||
r = wt_cli.post_appointment(json.dumps(dummy_appointment_request))
|
||||
assert wt_cli.process_post_appointment_response(r) == r.json()
|
||||
|
||||
# If we modify the response code tor a rejection (lets say 404) we should get None
|
||||
responses.replace(responses.POST, pisa_endpoint, json=response, status=404)
|
||||
r = pisa_cli.post_appointment(json.dumps(dummy_appointment_request))
|
||||
assert pisa_cli.process_post_appointment_response(r) is None
|
||||
r = wt_cli.post_appointment(json.dumps(dummy_appointment_request))
|
||||
assert wt_cli.process_post_appointment_response(r) is None
|
||||
|
||||
# The same should happen if the response is not in json
|
||||
responses.replace(responses.POST, pisa_endpoint, status=404)
|
||||
r = pisa_cli.post_appointment(json.dumps(dummy_appointment_request))
|
||||
assert pisa_cli.process_post_appointment_response(r) is None
|
||||
r = wt_cli.post_appointment(json.dumps(dummy_appointment_request))
|
||||
assert wt_cli.process_post_appointment_response(r) is None
|
||||
|
||||
|
||||
def test_save_appointment_receipt(monkeypatch):
|
||||
appointments_folder = "test_appointments_receipts"
|
||||
pisa_cli.config["APPOINTMENTS_FOLDER_NAME"] = appointments_folder
|
||||
wt_cli.config["APPOINTMENTS_FOLDER_NAME"] = appointments_folder
|
||||
|
||||
# The functions creates a new directory if it does not exist
|
||||
assert not os.path.exists(appointments_folder)
|
||||
pisa_cli.save_appointment_receipt(dummy_appointment.to_dict(), get_dummy_signature())
|
||||
wt_cli.save_appointment_receipt(dummy_appointment.to_dict(), get_dummy_signature())
|
||||
assert os.path.exists(appointments_folder)
|
||||
|
||||
# Check that the receipt has been saved by checking the file names
|
||||
@@ -228,7 +229,7 @@ def test_get_appointment():
|
||||
|
||||
request_url = "{}get_appointment?locator={}".format(pisa_endpoint, response.get("locator"))
|
||||
responses.add(responses.GET, request_url, json=response, status=200)
|
||||
result = pisa_cli.get_appointment(response.get("locator"))
|
||||
result = wt_cli.get_appointment(response.get("locator"))
|
||||
|
||||
assert len(responses.calls) == 1
|
||||
assert responses.calls[0].request.url == request_url
|
||||
@@ -243,4 +244,4 @@ def test_get_appointment_err():
|
||||
request_url = "{}get_appointment?locator=".format(pisa_endpoint, locator)
|
||||
responses.add(responses.GET, request_url, body=ConnectionError())
|
||||
|
||||
assert not pisa_cli.get_appointment(locator)
|
||||
assert not wt_cli.get_appointment(locator)
|
||||
@@ -4,7 +4,7 @@ from time import sleep
|
||||
from riemann.tx import Tx
|
||||
|
||||
from pisa import HOST, PORT
|
||||
from apps.cli import pisa_cli
|
||||
from apps.cli import wt_cli
|
||||
from apps.cli.blob import Blob
|
||||
from apps.cli import config as cli_conf
|
||||
from common.tools import compute_locator
|
||||
@@ -19,10 +19,10 @@ from test.pisa.e2e.conftest import (
|
||||
run_pisad,
|
||||
)
|
||||
|
||||
# We'll use pisa_cli to add appointments. The expected input format is a list of arguments with a json-encoded
|
||||
# We'll use wt_cli to add appointments. The expected input format is a list of arguments with a json-encoded
|
||||
# appointment
|
||||
pisa_cli.pisa_api_server = HOST
|
||||
pisa_cli.pisa_api_port = PORT
|
||||
wt_cli.pisa_api_server = HOST
|
||||
wt_cli.pisa_api_port = PORT
|
||||
|
||||
# Run pisad
|
||||
pisad_process = run_pisad()
|
||||
@@ -37,7 +37,7 @@ def broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, addr):
|
||||
def get_appointment_info(locator):
|
||||
# Check that the justice has been triggered (the appointment has moved from Watcher to Responder)
|
||||
sleep(1) # Let's add a bit of delay so the state can be updated
|
||||
return pisa_cli.get_appointment(locator)
|
||||
return wt_cli.get_appointment(locator)
|
||||
|
||||
|
||||
def test_appointment_life_cycle(bitcoin_cli, create_txs):
|
||||
@@ -46,13 +46,17 @@ def test_appointment_life_cycle(bitcoin_cli, create_txs):
|
||||
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
|
||||
locator = compute_locator(commitment_tx_id)
|
||||
|
||||
assert pisa_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
assert wt_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
|
||||
appointment_info = get_appointment_info(locator)
|
||||
assert appointment_info is not None
|
||||
assert len(appointment_info) == 1
|
||||
assert appointment_info[0].get("status") == "being_watched"
|
||||
|
||||
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 len(appointment_info) == 1
|
||||
assert appointment_info[0].get("status") == "dispute_responded"
|
||||
@@ -92,7 +96,7 @@ 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 pisa_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
assert wt_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
|
||||
# Broadcast the commitment transaction and mine a block
|
||||
new_addr = bitcoin_cli.getnewaddress()
|
||||
@@ -115,13 +119,13 @@ def test_appointment_wrong_key(bitcoin_cli, create_txs):
|
||||
# The appointment data is built using a random 32-byte value.
|
||||
appointment_data = build_appointment_data(bitcoin_cli, get_random_value_hex(32), penalty_tx)
|
||||
|
||||
# We can't use pisa_cli.add_appointment here since it computes the locator internally, so let's do it manually.
|
||||
# We can't use wt_cli.add_appointment here since it computes the locator internally, so let's do it manually.
|
||||
# We will encrypt the blob using the random value and derive the locator from the commitment tx.
|
||||
appointment_data["locator"] = compute_locator(bitcoin_cli.decoderawtransaction(commitment_tx).get("txid"))
|
||||
appointment_data["encrypted_blob"] = Cryptographer.encrypt(Blob(penalty_tx), appointment_data.get("tx_id"))
|
||||
appointment = Appointment.from_dict(appointment_data)
|
||||
|
||||
pisa_pk, cli_sk, cli_pk_der = pisa_cli.load_keys(
|
||||
pisa_pk, cli_sk, cli_pk_der = wt_cli.load_keys(
|
||||
cli_conf.get("PISA_PUBLIC_KEY"), cli_conf.get("CLI_PRIVATE_KEY"), cli_conf.get("CLI_PUBLIC_KEY")
|
||||
)
|
||||
hex_pk_der = binascii.hexlify(cli_pk_der)
|
||||
@@ -130,8 +134,8 @@ def test_appointment_wrong_key(bitcoin_cli, create_txs):
|
||||
data = {"appointment": appointment.to_dict(), "signature": signature, "public_key": hex_pk_der.decode("utf-8")}
|
||||
|
||||
# Send appointment to the server.
|
||||
response = pisa_cli.post_appointment(data)
|
||||
response_json = pisa_cli.process_post_appointment_response(response)
|
||||
response = wt_cli.post_appointment(data)
|
||||
response_json = wt_cli.process_post_appointment_response(response)
|
||||
|
||||
# Check that the server has accepted the appointment
|
||||
signature = response_json.get("signature")
|
||||
@@ -165,8 +169,8 @@ def test_two_identical_appointments(bitcoin_cli, create_txs):
|
||||
locator = compute_locator(commitment_tx_id)
|
||||
|
||||
# Send the appointment twice
|
||||
assert pisa_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
assert pisa_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
assert wt_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
assert wt_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
|
||||
# Broadcast the commitment transaction and mine a block
|
||||
new_addr = bitcoin_cli.getnewaddress()
|
||||
@@ -199,8 +203,8 @@ def test_two_appointment_same_locator_different_penalty(bitcoin_cli, create_txs)
|
||||
appointment2_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx2)
|
||||
locator = compute_locator(commitment_tx_id)
|
||||
|
||||
assert pisa_cli.add_appointment([json.dumps(appointment1_data)]) is True
|
||||
assert pisa_cli.add_appointment([json.dumps(appointment2_data)]) is True
|
||||
assert wt_cli.add_appointment([json.dumps(appointment1_data)]) is True
|
||||
assert wt_cli.add_appointment([json.dumps(appointment2_data)]) is True
|
||||
|
||||
# Broadcast the commitment transaction and mine a block
|
||||
new_addr = bitcoin_cli.getnewaddress()
|
||||
@@ -227,7 +231,7 @@ def test_appointment_shutdown_pisa_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 pisa_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
assert wt_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
|
||||
# Restart pisa
|
||||
pisad_process.terminate()
|
||||
@@ -265,7 +269,7 @@ def test_appointment_shutdown_pisa_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 pisa_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
assert wt_cli.add_appointment([json.dumps(appointment_data)]) is True
|
||||
|
||||
# Check that the appointment is still in the Watcher
|
||||
appointment_info = get_appointment_info(locator)
|
||||
|
||||
Reference in New Issue
Block a user