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

Refactors signing/verifiying functionality to be part of the Cryptographer

Browse Source 
- All encryption/decryption and signing/verifying calls are performed by the cryptographer now.

- The current signature format is temporal. We should define something not base on json.

- Some Cryptographer tests are still missing.

- The cli tests should be modified to fit this too.
This commit is contained in:
Sergi Delgado Segura
2019-12-07 13:22:39 +01:00
parent ae676e6632
commit d39056a0cc
12 changed files with 144 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
test/unit/conftest.py
View File

@@ -1,4 +1,3 @@
import json
import pytest
import random
import requests
@@ -8,7 +7,6 @@ from threading import Thread
from binascii import hexlify
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization
@@ -42,18 +40,6 @@ def prng_seed():
random.seed(0)
@pytest.fixture(scope="module")
def generate_keypair():
client_sk = ec.generate_private_key(ec.SECP256K1, default_backend())
client_pk = (
client_sk.public_key()
.public_bytes(encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo)
.decode("utf-8")
)
return client_sk, client_pk
@pytest.fixture(scope="session")
def db_manager():
manager = DBManager("test_db")
@@ -63,6 +49,13 @@ def db_manager():
rmtree("test_db")
def generate_keypair():
client_sk = ec.generate_private_key(ec.SECP256K1, default_backend())
client_pk = client_sk.public_key()
return client_sk, client_pk
def get_random_value_hex(nbytes):
pseudo_random_value = random.getrandbits(8 * nbytes)
prv_hex = "{:x}".format(pseudo_random_value)
@@ -79,11 +72,6 @@ def generate_blocks(n):
generate_block()
def sign_appointment(sk, appointment):
data = json.dumps(appointment, sort_keys=True, separators=(",", ":")).encode("utf-8")
return hexlify(sk.sign(data, ec.ECDSA(hashes.SHA256()))).decode("utf-8")
def generate_dummy_appointment_data(real_height=True, start_time_offset=5, end_time_offset=30):
if real_height:
current_height = bitcoin_cli().getblockcount()
@@ -104,11 +92,9 @@ def generate_dummy_appointment_data(real_height=True, start_time_offset=5, end_t
}
# dummy keys for this test
client_sk = ec.generate_private_key(ec.SECP256K1, default_backend())
client_pk = (
client_sk.public_key()
.public_bytes(encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo)
.decode("utf-8")
client_sk, client_pk = generate_keypair()
client_pk_der = client_pk.public_bytes(
encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo
)
locator = Watcher.compute_locator(dispute_txid)
@@ -124,9 +110,10 @@ def generate_dummy_appointment_data(real_height=True, start_time_offset=5, end_t
"encrypted_blob": encrypted_blob,
}
signature = sign_appointment(client_sk, appointment_data)
signature = Cryptographer.sign(Cryptographer.signature_format(appointment_data), client_sk)
pk_hex = hexlify(client_pk_der).decode("utf-8")
data = {"appointment": appointment_data, "signature": signature, "public_key": client_pk}
data = {"appointment": appointment_data, "signature": signature, "public_key": pk_hex}
return data, dispute_tx

1
test/unit/test_api.py
View File

@@ -150,7 +150,6 @@ def test_request_appointment_watcher(new_appt_data):
appointment_status = [appointment.pop("status") for appointment in received_appointments]
# Check that the appointment is within the received appoints
print("AAA", new_appt_data["appointment"], received_appointments)
assert new_appt_data["appointment"] in received_appointments
# Check that all the appointments are being watched

3
test/unit/test_appointment.py
View File

@@ -112,10 +112,11 @@ def test_from_dict(appointment_data):
assert True
# This test is pretty worthless atm, it would make sense once we have a proper serialize function
def test_serialize(appointment_data):
appointment = Appointment.from_dict(appointment_data)
assert appointment.triggered is False
ser_appointment = appointment.serialize()
assert ser_appointment == json.dumps(appointment_data, sort_keys=True, separators=(",", ":")).encode("utf-8")
assert ser_appointment == appointment.to_dict(include_triggered=False)

38
test/unit/test_inspector.py
View File

@@ -1,9 +1,8 @@
import json
from binascii import hexlify, unhexlify
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization
from pisa import c_logger
from pisa.errors import *
@@ -12,9 +11,10 @@ from pisa.appointment import Appointment
from pisa.block_processor import BlockProcessor
from pisa.conf import MIN_DISPUTE_DELTA
from test.unit.conftest import get_random_value_hex, generate_dummy_appointment_data
from test.unit.conftest import get_random_value_hex, generate_dummy_appointment_data, generate_keypair
from common.constants import LOCATOR_LEN_BYTES, LOCATOR_LEN_HEX
from common.cryptographer import Cryptographer
c_logger.disabled = True
@@ -42,11 +42,6 @@ WRONG_TYPES = [
WRONG_TYPES_NO_STR = [[], unhexlify(get_random_value_hex(LOCATOR_LEN_BYTES)), 3.2, 2.0, (), object, {}, object()]
def sign_appointment(sk, appointment):
data = json.dumps(appointment, sort_keys=True, separators=(",", ":")).encode("utf-8")
return hexlify(sk.sign(data, ec.ECDSA(hashes.SHA256()))).decode("utf-8")
def test_check_locator():
# Right appointment type, size and format
locator = get_random_value_hex(LOCATOR_LEN_BYTES)
@@ -174,8 +169,13 @@ def test_check_blob():
assert Inspector.check_blob(encrypted_blob)[0] == APPOINTMENT_WRONG_FIELD_FORMAT
def test_check_appointment_signature(generate_keypair):
client_sk, client_pk = generate_keypair
def test_check_appointment_signature():
# The inspector receives the public key as hex
client_sk, client_pk = generate_keypair()
client_pk_der = client_pk.public_bytes(
encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo
)
client_pk_hex = hexlify(client_pk_der).decode("utf-8")
dummy_appointment_data, _ = generate_dummy_appointment_data(real_height=False)
assert Inspector.check_appointment_signature(
@@ -185,22 +185,26 @@ def test_check_appointment_signature(generate_keypair):
fake_sk = ec.generate_private_key(ec.SECP256K1, default_backend())
# Create a bad signature to make sure inspector rejects it
bad_signature = sign_appointment(fake_sk, dummy_appointment_data["appointment"])
bad_signature = Cryptographer.sign(Cryptographer.signature_format(dummy_appointment_data["appointment"]), fake_sk)
assert (
Inspector.check_appointment_signature(dummy_appointment_data["appointment"], bad_signature, client_pk)[0]
Inspector.check_appointment_signature(dummy_appointment_data["appointment"], bad_signature, client_pk_hex)[0]
== APPOINTMENT_INVALID_SIGNATURE
)
def test_inspect(run_bitcoind, generate_keypair):
def test_inspect(run_bitcoind):
# At this point every single check function has been already tested, let's test inspect with an invalid and a valid
# appointments.
client_sk, client_pk = generate_keypair
client_sk, client_pk = generate_keypair()
client_pk_der = client_pk.public_bytes(
encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo
)
client_pk_hex = hexlify(client_pk_der).decode("utf-8")
# Invalid appointment, every field is empty
appointment_data = dict()
signature = sign_appointment(client_sk, appointment_data)
signature = Cryptographer.sign(Cryptographer.signature_format(appointment_data), client_sk)
appointment = inspector.inspect(appointment_data, signature, client_pk)
assert type(appointment) == tuple and appointment[0] != 0
@@ -219,9 +223,9 @@ def test_inspect(run_bitcoind, generate_keypair):
"encrypted_blob": encrypted_blob,
}
signature = sign_appointment(client_sk, appointment_data)
signature = Cryptographer.sign(Cryptographer.signature_format(appointment_data), client_sk)
appointment = inspector.inspect(appointment_data, signature, client_pk)
appointment = inspector.inspect(appointment_data, signature, client_pk_hex)
assert (
type(appointment) == Appointment

36
test/unit/test_watcher.py
View File

@@ -3,12 +3,6 @@ from uuid import uuid4
from threading import Thread
from queue import Queue, Empty
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.exceptions import InvalidSignature
from pisa import c_logger
from pisa.watcher import Watcher
from pisa.responder import Responder
@@ -17,6 +11,7 @@ from test.unit.conftest import generate_block, generate_blocks, generate_dummy_a
from pisa.conf import EXPIRY_DELTA, PISA_SECRET_KEY, MAX_APPOINTMENTS
from common.tools import check_sha256_hex_format
from common.cryptographer import Cryptographer
c_logger.disabled = True
@@ -25,10 +20,9 @@ START_TIME_OFFSET = 1
END_TIME_OFFSET = 1
TEST_SET_SIZE = 200
with open(PISA_SECRET_KEY, "r") as key_file:
pubkey_pem = key_file.read().encode("utf-8")
# TODO: should use the public key file instead, but it is not currently exported in the configuration
signing_key = load_pem_private_key(pubkey_pem, password=None, backend=default_backend())
with open(PISA_SECRET_KEY, "rb") as key_file_der:
sk_der = key_file_der.read()
signing_key = Cryptographer.load_private_key_der(sk_der)
public_key = signing_key.public_key()
@@ -65,16 +59,6 @@ def create_appointments(n):
return appointments, locator_uuid_map, dispute_txs
def is_signature_valid(appointment, signature, pk):
# verify the signature
try:
data = appointment.serialize()
pk.verify(signature, data, ec.ECDSA(hashes.SHA256()))
except InvalidSignature:
return False
return True
def test_init(watcher):
assert type(watcher.appointments) is dict and len(watcher.appointments) == 0
assert type(watcher.locator_uuid_map) is dict and len(watcher.locator_uuid_map) == 0
@@ -107,19 +91,13 @@ def test_add_appointment(run_bitcoind, watcher):
added_appointment, sig = watcher.add_appointment(appointment)
assert added_appointment is True
assert is_signature_valid(appointment, sig, public_key)
assert Cryptographer.verify(Cryptographer.signature_format(appointment.to_dict()), sig, public_key)
# Check that we can also add an already added appointment (same locator)
added_appointment, sig = watcher.add_appointment(appointment)
assert added_appointment is True
assert is_signature_valid(appointment, sig, public_key)
def test_sign_appointment(watcher):
appointment, _ = generate_dummy_appointment(start_time_offset=START_TIME_OFFSET, end_time_offset=END_TIME_OFFSET)
signature = watcher.sign_appointment(appointment)
assert is_signature_valid(appointment, signature, public_key)
assert Cryptographer.verify(Cryptographer.signature_format(appointment.to_dict()), sig, public_key)
def test_add_too_many_appointments(watcher):
@@ -133,7 +111,7 @@ def test_add_too_many_appointments(watcher):
added_appointment, sig = watcher.add_appointment(appointment)
assert added_appointment is True
assert is_signature_valid(appointment, sig, public_key)
assert Cryptographer.verify(Cryptographer.signature_format(appointment.to_dict()), sig, public_key)
appointment, dispute_tx = generate_dummy_appointment(
start_time_offset=START_TIME_OFFSET, end_time_offset=END_TIME_OFFSET