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

Merge branch 'simplify'

Browse Source
This commit is contained in:
Sergi Delgado Segura
2019-12-16 11:59:05 +01:00
parent c679d59451 a7eb22626e
commit 399c25044a
34 changed files with 566 additions and 733 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -13,6 +13,6 @@ test.py
*.pyc
.cache
.pytest_cache/
*.pem
*.der
.coverage
htmlcov

10
apps/cli/__init__.py
View File

@@ -9,13 +9,9 @@ DEFAULT_PISA_API_PORT = 9814
CLIENT_LOG_FILE = "pisa-cli.log"
APPOINTMENTS_FOLDER_NAME = "appointments"
# CRYPTO
SUPPORTED_HASH_FUNCTIONS = ["SHA256"]
SUPPORTED_CIPHERS = ["AES-GCM-128"]
CLI_PUBLIC_KEY = "cli_pk.pem"
CLI_PRIVATE_KEY = "cli_sk.pem"
PISA_PUBLIC_KEY = "pisa_pk.pem"
CLI_PUBLIC_KEY = "cli_pk.der"
CLI_PRIVATE_KEY = "cli_sk.der"
PISA_PUBLIC_KEY = "pisa_pk.der"
# Configure logging
logging.basicConfig(

55
apps/cli/blob.py
View File

@@ -1,65 +1,14 @@
import re
from hashlib import sha256
from binascii import hexlify, unhexlify
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
from apps.cli import SUPPORTED_HASH_FUNCTIONS, SUPPORTED_CIPHERS
from apps.cli import logger
class Blob:
def __init__(self, data, cipher, hash_function):
def __init__(self, data):
if type(data) is not str or re.search(r"^[0-9A-Fa-f]+$", data) is None:
raise ValueError("Non-Hex character found in txid.")
self.data = data
self.cipher = cipher
self.hash_function = hash_function
# FIXME: We only support SHA256 for now
if self.hash_function.upper() not in SUPPORTED_HASH_FUNCTIONS:
raise ValueError(
"Hash function not supported ({}). Supported Hash functions: {}".format(
self.hash_function, SUPPORTED_HASH_FUNCTIONS
)
)
# FIXME: We only support AES-GCM-128 for now
if self.cipher.upper() not in SUPPORTED_CIPHERS:
raise ValueError(
"Cipher not supported ({}). Supported ciphers: {}".format(self.hash_function, SUPPORTED_CIPHERS)
)
def encrypt(self, tx_id):
if len(tx_id) != 64:
raise ValueError("txid does not matches the expected size (32-byte / 64 hex chars).")
elif re.search(r"^[0-9A-Fa-f]+$", tx_id) is None:
raise ValueError("Non-Hex character found in txid.")
# Transaction to be encrypted
# FIXME: The blob data should contain more things that just the transaction. Leaving like this for now.
tx = unhexlify(self.data)
tx_id = unhexlify(tx_id)
# master_key = H(tx_id | tx_id)
master_key = sha256(tx_id + tx_id).digest()
# The 16 MSB of the master key will serve as the AES GCM 128 secret key. The 16 LSB will serve as the IV.
sk = master_key[:16]
nonce = master_key[16:]
# Encrypt the data
aesgcm = AESGCM(sk)
encrypted_blob = aesgcm.encrypt(nonce=nonce, data=tx, associated_data=None)
encrypted_blob = hexlify(encrypted_blob).decode()
logger.info(
"Creating new blob",
master_key=hexlify(master_key).decode(),
sk=hexlify(sk).decode(),
nonce=hexlify(nonce).decode(),
encrypted_blob=encrypted_blob,
)
return encrypted_blob

98
apps/cli/pisa_cli.py
View File

@@ -1,22 +1,13 @@
import re
import os
import sys
import json
import requests
import time
from sys import argv
from hashlib import sha256
from binascii import hexlify, unhexlify
from getopt import getopt, GetoptError
from requests import ConnectTimeout, ConnectionError
from uuid import uuid4
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.serialization import load_pem_public_key, load_pem_private_key
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.exceptions import InvalidSignature, UnsupportedAlgorithm
from apps.cli.blob import Blob
from apps.cli.help import help_add_appointment, help_get_appointment
from apps.cli import (
@@ -29,6 +20,10 @@ from apps.cli import (
logger,
)
from common.constants import LOCATOR_LEN_HEX
from common.cryptographer import Cryptographer
from common.tools import check_sha256_hex_format
HTTP_OK = 200
@@ -55,52 +50,19 @@ def generate_dummy_appointment():
print("\nData stored in dummy_appointment_data.json")
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")
# Loads and returns Pisa keys from disk
def load_key_file_data(file_name):
try:
with open(file_name, "r") as key_file:
key_pem = key_file.read().encode("utf-8")
return key_pem
with open(file_name, "rb") as key_file:
key = key_file.read()
return key
except FileNotFoundError:
raise FileNotFoundError("File not found.")
# Deserialize public key from pem data.
def load_public_key(pk_pem):
try:
pisa_pk = load_pem_public_key(pk_pem, backend=default_backend())
return pisa_pk
except UnsupportedAlgorithm:
raise ValueError("Could not deserialize the public key (unsupported algorithm).")
# Deserialize private key from pem data.
def load_private_key(sk_pem):
try:
cli_sk = load_pem_private_key(sk_pem, None, backend=default_backend())
return cli_sk
except UnsupportedAlgorithm:
raise ValueError("Could not deserialize the private key (unsupported algorithm).")
# returning True or False accordingly.
def is_appointment_signature_valid(appointment, signature, pk):
try:
sig_bytes = unhexlify(signature.encode("utf-8"))
data = json.dumps(appointment, sort_keys=True, separators=(",", ":")).encode("utf-8")
pk.verify(sig_bytes, data, ec.ECDSA(hashes.SHA256()))
return True
except InvalidSignature:
return False
def compute_locator(tx_id):
return tx_id[:LOCATOR_LEN_HEX]
# Makes sure that the folder APPOINTMENTS_FOLDER_NAME exists, then saves the appointment and signature in it.
@@ -157,7 +119,7 @@ def add_appointment(args):
logger.error("The provided JSON is empty.")
return False
valid_locator = check_txid_format(appointment_data.get("tx_id"))
valid_locator = check_sha256_hex_format(appointment_data.get("tx_id"))
if not valid_locator:
logger.error("The provided locator is not valid.")
@@ -173,8 +135,8 @@ def add_appointment(args):
)
try:
sk_pem = load_key_file_data(CLI_PRIVATE_KEY)
cli_sk = load_private_key(sk_pem)
sk_der = load_key_file_data(CLI_PRIVATE_KEY)
cli_sk = Cryptographer.load_private_key_der(sk_der)
except ValueError:
logger.error("Failed to deserialize the public key. It might be in an unsupported format.")
@@ -188,19 +150,20 @@ def add_appointment(args):
logger.error("I/O error({}): {}".format(e.errno, e.strerror))
return False
signature = sign_appointment(cli_sk, appointment)
signature = Cryptographer.sign(Cryptographer.signature_format(appointment), cli_sk)
try:
cli_pk_pem = load_key_file_data(CLI_PUBLIC_KEY)
cli_pk_der = load_key_file_data(CLI_PUBLIC_KEY)
except FileNotFoundError:
logger.error("Client's private key file not found. Please check your settings.")
logger.error("Client's public key file not found. Please check your settings.")
return False
except IOError as e:
logger.error("I/O error({}): {}".format(e.errno, e.strerror))
return False
data = {"appointment": appointment, "signature": signature, "public_key": cli_pk_pem.decode("utf-8")}
data = {"appointment": appointment, "signature": signature, "public_key": cli_pk_der.decode("utf-8")}
appointment_json = json.dumps(data, sort_keys=True, separators=(",", ":"))
@@ -240,9 +203,9 @@ def add_appointment(args):
signature = response_json["signature"]
# verify that the returned signature is valid
try:
pk_pem = load_key_file_data(PISA_PUBLIC_KEY)
pk = load_public_key(pk_pem)
is_sig_valid = is_appointment_signature_valid(appointment, signature, pk)
pisa_pk_der = load_key_file_data(PISA_PUBLIC_KEY)
pisa_pk = Cryptographer.load_public_key_der(pisa_pk_der)
is_sig_valid = Cryptographer.verify(Cryptographer.signature_format(appointment), signature, pisa_pk)
except ValueError:
logger.error("Failed to deserialize the public key. It might be in an unsupported format.")
@@ -283,7 +246,7 @@ def get_appointment(args):
sys.exit(help_get_appointment())
else:
locator = arg_opt
valid_locator = check_txid_format(locator)
valid_locator = check_sha256_hex_format(locator)
if not valid_locator:
logger.error("The provided locator is not valid: {}".format(locator))
@@ -308,14 +271,11 @@ def get_appointment(args):
def build_appointment(tx, tx_id, start_time, end_time, dispute_delta):
locator = sha256(unhexlify(tx_id)).hexdigest()
cipher = "AES-GCM-128"
hash_function = "SHA256"
locator = compute_locator(tx_id)
# FIXME: The blob data should contain more things that just the transaction. Leaving like this for now.
blob = Blob(tx, cipher, hash_function)
encrypted_blob = blob.encrypt(tx_id)
blob = Blob(tx)
encrypted_blob = Cryptographer.encrypt(blob, tx_id)
appointment = {
"locator": locator,
@@ -323,21 +283,11 @@ def build_appointment(tx, tx_id, start_time, end_time, dispute_delta):
"end_time": end_time,
"dispute_delta": dispute_delta,
"encrypted_blob": encrypted_blob,
"cipher": cipher,
"hash_function": hash_function,
}
return appointment
def check_txid_format(txid):
if len(txid) != 64:
sys.exit("locator does not matches the expected size (32-byte / 64 hex chars).")
# TODO: #12-check-txid-regexp
return re.search(r"^[0-9A-Fa-f]+$", txid) is not None
def show_usage():
return (
"USAGE: "

20
apps/generate_key.py
View File

@@ -7,27 +7,27 @@ from cryptography.hazmat.primitives.asymmetric import ec
# Simple tool to generate an ECDSA private key using the secp256k1 curve and save private and public keys
# as 'pisa_sk.pem' 'and pisa_pk.pem', respectively.
# as 'pisa_sk.der' 'and pisa_pk.der', respectively.
SK_FILE_NAME = "pisa_sk.pem"
PK_FILE_NAME = "pisa_pk.pem"
SK_FILE_NAME = "../pisa_sk.der"
PK_FILE_NAME = "../pisa_pk.der"
def save_sk(sk, filename):
pem = sk.private_bytes(
encoding=serialization.Encoding.PEM,
der = sk.private_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption(),
)
with open(filename, "wb") as pem_out:
pem_out.write(pem)
with open(filename, "wb") as der_out:
der_out.write(der)
def save_pk(pk, filename):
pem = pk.public_bytes(encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo)
with open(filename, "wb") as pem_out:
pem_out.write(pem)
der = pk.public_bytes(encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo)
with open(filename, "wb") as der_out:
der_out.write(der)
if __name__ == "__main__":

0
common/__init__.py Normal file
View File

8
common/constants.py Normal file
View File

@@ -0,0 +1,8 @@
# Locator
LOCATOR_LEN_HEX = 32
LOCATOR_LEN_BYTES = LOCATOR_LEN_HEX // 2
# HTTP
HTTP_OK = 200
HTTP_BAD_REQUEST = 400
HTTP_SERVICE_UNAVAILABLE = 503

153
common/cryptographer.py Normal file
View File

@@ -0,0 +1,153 @@
import json
from hashlib import sha256
from binascii import unhexlify, hexlify
from cryptography.exceptions import InvalidTag, UnsupportedAlgorithm
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
from cryptography.hazmat.primitives.serialization import load_der_public_key, load_der_private_key
from cryptography.exceptions import InvalidSignature
from common.tools import check_sha256_hex_format
from pisa.logger import Logger
logger = Logger("Cryptographer")
class Cryptographer:
@staticmethod
def check_data_key_format(data, key):
if len(data) % 2:
error = "Incorrect (Odd-length) value."
logger.error(error, data=data)
raise ValueError(error)
if not check_sha256_hex_format(key):
error = "Key must be a 32-byte hex value (64 hex chars)."
logger.error(error, key=key)
raise ValueError(error)
return True
@staticmethod
def encrypt(blob, key, rtype="hex"):
if rtype not in ["hex", "bytes"]:
raise ValueError("Wrong return type. Return type must be 'hex' or 'bytes'")
Cryptographer.check_data_key_format(blob.data, key)
# Transaction to be encrypted
# FIXME: The blob data should contain more things that just the transaction. Leaving like this for now.
tx = unhexlify(blob.data)
# sk is the H(txid) (32-byte) and nonce is set to 0 (12-byte)
sk = sha256(unhexlify(key)).digest()
nonce = bytearray(12)
logger.info("Encrypting blob.", sk=hexlify(sk).decode(), nonce=hexlify(nonce).decode(), blob=blob.data)
# Encrypt the data
cipher = ChaCha20Poly1305(sk)
encrypted_blob = cipher.encrypt(nonce=nonce, data=tx, associated_data=None)
if rtype == "hex":
encrypted_blob = hexlify(encrypted_blob).decode("utf8")
return encrypted_blob
@staticmethod
# ToDo: #20-test-tx-decrypting-edge-cases
def decrypt(encrypted_blob, key, rtype="hex"):
if rtype not in ["hex", "bytes"]:
raise ValueError("Wrong return type. Return type must be 'hex' or 'bytes'")
Cryptographer.check_data_key_format(encrypted_blob.data, key)
# sk is the H(txid) (32-byte) and nonce is set to 0 (12-byte)
sk = sha256(unhexlify(key)).digest()
nonce = bytearray(12)
logger.info(
"Decrypting Blob.",
sk=hexlify(sk).decode(),
nonce=hexlify(nonce).decode(),
encrypted_blob=encrypted_blob.data,
)
# Decrypt
cipher = ChaCha20Poly1305(sk)
data = unhexlify(encrypted_blob.data.encode())
try:
blob = cipher.decrypt(nonce=nonce, data=data, associated_data=None)
# Change the blob encoding to hex depending on the rtype (default)
if rtype == "hex":
blob = hexlify(blob).decode("utf8")
except InvalidTag:
blob = None
return blob
# NOTCOVERED
@staticmethod
def signature_format(data):
# FIXME: This is temporary serialization. A proper one is required. Data need to be unhexlified too (can't atm)
return json.dumps(data, sort_keys=True, separators=(",", ":")).encode("utf-8")
# Deserialize public key from der data.
@staticmethod
def load_public_key_der(pk_der):
try:
pk = load_der_public_key(pk_der, backend=default_backend())
return pk
except UnsupportedAlgorithm:
raise ValueError("Could not deserialize the public key (unsupported algorithm).")
# Deserialize private key from der data.
@staticmethod
def load_private_key_der(sk_der):
try:
sk = load_der_private_key(sk_der, None, backend=default_backend())
return sk
except UnsupportedAlgorithm:
raise ValueError("Could not deserialize the private key (unsupported algorithm).")
@staticmethod
def sign(data, sk, rtype="hex"):
if rtype not in ["hex", "bytes"]:
raise ValueError("Wrong return type. Return type must be 'hex' or 'bytes'")
if not isinstance(sk, ec.EllipticCurvePrivateKey):
logger.error("The value passed as sk is not a private key (EllipticCurvePrivateKey).")
return None
else:
signature = sk.sign(data, ec.ECDSA(hashes.SHA256()))
if rtype == "hex":
signature = hexlify(signature).decode("utf-8")
return signature
@staticmethod
def verify(message, signature, pk):
if not isinstance(pk, ec.EllipticCurvePublicKey):
logger.error("The value passed as pk is not a public key (EllipticCurvePublicKey).")
return False
if isinstance(signature, str):
signature = unhexlify(signature.encode("utf-8"))
try:
pk.verify(signature, message, ec.ECDSA(hashes.SHA256()))
return True
except InvalidSignature:
return False

5
common/tools.py Normal file
View File

@@ -0,0 +1,5 @@
import re
def check_sha256_hex_format(value):
return isinstance(value, str) and re.match(r"^[0-9A-Fa-f]{64}$", value) is not None

8
pisa/TODO.md
View File

@@ -1,8 +0,0 @@
- Start jobs according to the start time, jobs are now started when they are received
- Add DB
- Store jobs in DB until start time?
- Handle failures in the underlying system (i.e. bitcoind crashes)
- Add checks related with OP_CSV in justice tx and dispute_delta provided once the blob is decrypted
- Do not accept new appointments if the locator has already been used
- <s> Check all the interactions with core, figure out the edge cases and error codes i.e: The justice transaction can already be in the blockchain the first time we push it <s>
- <s> Handle reconnection with ZMQ in case of broken pipe. The current version of the code fails if it does happen <s>

13
pisa/api.py
View File

@@ -1,7 +1,6 @@
import os
import json
from flask import Flask, request, abort, jsonify
from binascii import hexlify
from pisa import HOST, PORT, logging
from pisa.logger import Logger
@@ -9,16 +8,12 @@ from pisa.inspector import Inspector
from pisa.appointment import Appointment
from pisa.block_processor import BlockProcessor
from common.constants import HTTP_OK, HTTP_BAD_REQUEST, HTTP_SERVICE_UNAVAILABLE, LOCATOR_LEN_HEX
# ToDo: #5-add-async-to-api
app = Flask(__name__)
HTTP_OK = 200
HTTP_BAD_REQUEST = 400
HTTP_SERVICE_UNAVAILABLE = 503
logger = Logger("API")
watcher = None
@@ -44,7 +39,7 @@ def add_appointment():
if appointment_added:
rcode = HTTP_OK
response = {"locator": appointment.locator, "signature": hexlify(signature).decode("utf-8")}
response = {"locator": appointment.locator, "signature": signature}
else:
rcode = HTTP_SERVICE_UNAVAILABLE
error = "appointment rejected"
@@ -79,7 +74,7 @@ def get_appointment():
response = []
# ToDo: #15-add-system-monitor
if not isinstance(locator, str) or len(locator) != 64:
if not isinstance(locator, str) or len(locator) != LOCATOR_LEN_HEX:
response.append({"locator": locator, "status": "not_found"})
return jsonify(response)

39
pisa/appointment.py
View File

@@ -6,16 +6,12 @@ from pisa.encrypted_blob import EncryptedBlob
# Basic appointment structure
class Appointment:
# DISCUSS: 35-appointment-checks
def __init__(
self, locator, start_time, end_time, dispute_delta, encrypted_blob, cipher, hash_function, triggered=False
):
def __init__(self, locator, start_time, end_time, dispute_delta, encrypted_blob, triggered=False):
self.locator = locator
self.start_time = start_time # ToDo: #4-standardize-appointment-fields
self.end_time = end_time # ToDo: #4-standardize-appointment-fields
self.dispute_delta = dispute_delta
self.encrypted_blob = EncryptedBlob(encrypted_blob)
self.cipher = cipher
self.hash_function = hash_function
self.triggered = triggered
@classmethod
@@ -25,34 +21,18 @@ class Appointment:
end_time = appointment_data.get("end_time") # ToDo: #4-standardize-appointment-fields
dispute_delta = appointment_data.get("dispute_delta")
encrypted_blob_data = appointment_data.get("encrypted_blob")
cipher = appointment_data.get("cipher")
hash_function = appointment_data.get("hash_function")
triggered = True if appointment_data.get("triggered") is True else False
if any(
v is None
for v in [
locator,
start_time,
end_time,
dispute_delta,
encrypted_blob_data,
cipher,
hash_function,
triggered,
]
):
if any(v is None for v in [locator, start_time, end_time, dispute_delta, encrypted_blob_data, triggered]):
raise ValueError("Wrong appointment data, some fields are missing")
else:
appointment = cls(
locator, start_time, end_time, dispute_delta, encrypted_blob_data, cipher, hash_function, triggered
)
appointment = cls(locator, start_time, end_time, dispute_delta, encrypted_blob_data, triggered)
return appointment
def to_dict(self):
def to_dict(self, include_triggered=True):
# ToDO: #3-improve-appointment-structure
appointment = {
"locator": self.locator,
@@ -60,17 +40,12 @@ class Appointment:
"end_time": self.end_time,
"dispute_delta": self.dispute_delta,
"encrypted_blob": self.encrypted_blob.data,
"cipher": self.cipher,
"hash_function": self.hash_function,
"triggered": self.triggered,
}
if include_triggered:
appointment["triggered"] = self.triggered
return appointment
def to_json(self):
return json.dumps(self.to_dict(), sort_keys=True, separators=(",", ":"))
def serialize(self):
data = self.to_dict()
data.pop("triggered")
return json.dumps(data, sort_keys=True, separators=(",", ":")).encode("utf-8")

1
pisa/cleaner.py
View File

@@ -53,7 +53,6 @@ class Cleaner:
confirmations=confirmations,
)
# ToDo: #9-add-data-persistence
justice_txid = jobs[uuid].justice_txid
locator = jobs[uuid].locator
jobs.pop(uuid)

56
pisa/cryptographer.py
View File

@@ -1,56 +0,0 @@
from hashlib import sha256
from binascii import unhexlify, hexlify
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from pisa.logger import Logger
logger = Logger("Cryptographer")
# FIXME: Cryptographer is assuming AES-128-GCM and SHA256 since they are the only pair accepted by the encrypted blob
# and the only pair programmed so far.
class Cryptographer:
@staticmethod
# ToDo: #20-test-tx-decrypting-edge-cases
def decrypt(encrypted_blob, key, rtype="hex"):
if rtype not in ["hex", "bytes"]:
raise ValueError("Wrong return type. Return type must be 'hex' or 'bytes'")
if len(encrypted_blob.data) % 2:
logger.info(
"Incorrect (Odd-length) value to be decrypted.", encrypted_blob=encrypted_blob.data, dispute_txid=key
)
return None
# master_key = H(tx_id | tx_id)
key = unhexlify(key)
master_key = sha256(key + key).digest()
# The 16 MSB of the master key will serve as the AES GCM 128 secret key. The 16 LSB will serve as the IV.
sk = master_key[:16]
nonce = master_key[16:]
logger.info(
"Creating new blob.",
master_key=hexlify(master_key).decode(),
sk=hexlify(sk).decode(),
nonce=hexlify(nonce).decode(),
encrypted_blob=encrypted_blob.data,
)
# Decrypt
cipher = AESGCM(sk)
data = unhexlify(encrypted_blob.data.encode())
try:
blob = cipher.decrypt(nonce=nonce, data=data, associated_data=None)
# Change the blob encoding to hex depending on the rtype (default)
if rtype == "hex":
blob = hexlify(blob).decode("utf8")
except InvalidTag:
blob = None
return blob

17
pisa/encrypted_blob.py
View File

@@ -1,20 +1,5 @@
from pisa.conf import SUPPORTED_CIPHERS, SUPPORTED_HASH_FUNCTIONS
class EncryptedBlob:
def __init__(self, data, cipher="AES-GCM-128", hash_function="SHA256"):
if cipher in SUPPORTED_CIPHERS:
self.cipher = cipher
else:
raise ValueError("Cipher not supported")
if hash_function in SUPPORTED_HASH_FUNCTIONS:
self.hash_function = hash_function
else:
raise ValueError("Hash function not supported")
def __init__(self, data):
self.data = data
def __eq__(self, other):

4
pisa/errors.py
View File

@@ -6,9 +6,7 @@ APPOINTMENT_WRONG_FIELD_FORMAT = -4
APPOINTMENT_FIELD_TOO_SMALL = -5
APPOINTMENT_FIELD_TOO_BIG = -6
APPOINTMENT_WRONG_FIELD = -7
APPOINTMENT_CIPHER_NOT_SUPPORTED = -8
APPOINTMENT_HASH_FUNCTION_NOT_SUPPORTED = -9
APPOINTMENT_INVALID_SIGNATURE = -10
APPOINTMENT_INVALID_SIGNATURE = -8
# Custom RPC errors
RPC_TX_REORGED_AFTER_BROADCAST = -98

73
pisa/inspector.py
View File

@@ -1,12 +1,8 @@
import json
import re
from binascii import unhexlify
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.serialization import load_pem_public_key
from cryptography.exceptions import InvalidSignature
from common.constants import LOCATOR_LEN_HEX
from common.cryptographer import Cryptographer
from pisa import errors
import pisa.conf as conf
@@ -37,10 +33,6 @@ class Inspector:
rcode, message = self.check_delta(appt.get("dispute_delta"))
if rcode == 0:
rcode, message = self.check_blob(appt.get("encrypted_blob"))
if rcode == 0:
rcode, message = self.check_cipher(appt.get("cipher"))
if rcode == 0:
rcode, message = self.check_hash_function(appt.get("hash_function"))
if rcode == 0:
rcode, message = self.check_appointment_signature(appt, signature, public_key)
@@ -68,7 +60,7 @@ class Inspector:
rcode = errors.APPOINTMENT_WRONG_FIELD_TYPE
message = "wrong locator data type ({})".format(type(locator))
elif len(locator) != 64:
elif len(locator) != LOCATOR_LEN_HEX:
rcode = errors.APPOINTMENT_WRONG_FIELD_SIZE
message = "wrong locator size ({})".format(len(locator))
# TODO: #12-check-txid-regexp
@@ -200,57 +192,9 @@ class Inspector:
return rcode, message
@staticmethod
def check_cipher(cipher):
message = None
rcode = 0
t = type(cipher)
if cipher is None:
rcode = errors.APPOINTMENT_EMPTY_FIELD
message = "empty cipher received"
elif t != str:
rcode = errors.APPOINTMENT_WRONG_FIELD_TYPE
message = "wrong cipher data type ({})".format(t)
elif cipher.upper() not in conf.SUPPORTED_CIPHERS:
rcode = errors.APPOINTMENT_CIPHER_NOT_SUPPORTED
message = "cipher not supported: {}".format(cipher)
if message is not None:
logger.error(message)
return rcode, message
@staticmethod
def check_hash_function(hash_function):
message = None
rcode = 0
t = type(hash_function)
if hash_function is None:
rcode = errors.APPOINTMENT_EMPTY_FIELD
message = "empty hash_function received"
elif t != str:
rcode = errors.APPOINTMENT_WRONG_FIELD_TYPE
message = "wrong hash_function data type ({})".format(t)
elif hash_function.upper() not in conf.SUPPORTED_HASH_FUNCTIONS:
rcode = errors.APPOINTMENT_HASH_FUNCTION_NOT_SUPPORTED
message = "hash_function not supported {}".format(hash_function)
if message is not None:
logger.error(message)
return rcode, message
@staticmethod
# Verifies that the appointment signature is a valid signature with public key
def check_appointment_signature(appointment, signature, pk_pem):
def check_appointment_signature(appointment, signature, pk_der):
message = None
rcode = 0
@@ -258,13 +202,10 @@ class Inspector:
rcode = errors.APPOINTMENT_EMPTY_FIELD
message = "empty signature received"
try:
sig_bytes = unhexlify(signature.encode("utf-8"))
client_pk = load_pem_public_key(pk_pem.encode("utf-8"), backend=default_backend())
data = json.dumps(appointment, sort_keys=True, separators=(",", ":")).encode("utf-8")
client_pk.verify(sig_bytes, data, ec.ECDSA(hashes.SHA256()))
pk = Cryptographer.load_public_key_der(unhexlify(pk_der.encode("utf-8")))
valid_sig = Cryptographer.verify(Cryptographer.signature_format(appointment), signature, pk)
except InvalidSignature:
if not valid_sig:
rcode = errors.APPOINTMENT_INVALID_SIGNATURE
message = "invalid signature"

39
pisa/responder.py
View File

@@ -1,8 +1,6 @@
import json
from queue import Queue
from hashlib import sha256
from threading import Thread
from binascii import unhexlify
from pisa.logger import Logger
from pisa.cleaner import Cleaner
@@ -17,28 +15,26 @@ logger = Logger("Responder")
class Job:
def __init__(self, dispute_txid, justice_txid, justice_rawtx, appointment_end):
def __init__(self, locator, dispute_txid, justice_txid, justice_rawtx, appointment_end):
self.locator = locator
self.dispute_txid = dispute_txid
self.justice_txid = justice_txid
self.justice_rawtx = justice_rawtx
self.appointment_end = appointment_end
# FIXME: locator is here so we can give info about jobs for now. It can be either passed from watcher or info
# can be directly got from DB
self.locator = sha256(unhexlify(dispute_txid)).hexdigest()
@classmethod
def from_dict(cls, job_data):
locator = job_data.get("locator")
dispute_txid = job_data.get("dispute_txid")
justice_txid = job_data.get("justice_txid")
justice_rawtx = job_data.get("justice_rawtx")
appointment_end = job_data.get("appointment_end")
if any(v is None for v in [dispute_txid, justice_txid, justice_rawtx, appointment_end]):
if any(v is None for v in [locator, dispute_txid, justice_txid, justice_rawtx, appointment_end]):
raise ValueError("Wrong job data, some fields are missing")
else:
job = cls(dispute_txid, justice_txid, justice_rawtx, appointment_end)
job = cls(locator, dispute_txid, justice_txid, justice_rawtx, appointment_end)
return job
@@ -81,7 +77,9 @@ class Responder:
return synchronized
def add_response(self, uuid, dispute_txid, justice_txid, justice_rawtx, appointment_end, block_hash, retry=False):
def add_response(
self, uuid, locator, dispute_txid, justice_txid, justice_rawtx, appointment_end, block_hash, retry=False
):
if self.asleep:
logger.info("Waking up")
@@ -92,7 +90,9 @@ class Responder:
# retry holds that information. If retry is true the job already exists
if receipt.delivered:
if not retry:
self.create_job(uuid, dispute_txid, justice_txid, justice_rawtx, appointment_end, receipt.confirmations)
self.create_job(
uuid, locator, dispute_txid, justice_txid, justice_rawtx, appointment_end, receipt.confirmations
)
else:
# TODO: Add the missing reasons (e.g. RPC_VERIFY_REJECTED)
@@ -102,8 +102,8 @@ class Responder:
return receipt
def create_job(self, uuid, dispute_txid, justice_txid, justice_rawtx, appointment_end, confirmations=0):
job = Job(dispute_txid, justice_txid, justice_rawtx, appointment_end)
def create_job(self, uuid, locator, dispute_txid, justice_txid, justice_rawtx, appointment_end, confirmations=0):
job = Job(locator, dispute_txid, justice_txid, justice_rawtx, appointment_end)
self.jobs[uuid] = job
if justice_txid in self.tx_job_map:
@@ -134,8 +134,7 @@ class Responder:
self.zmq_subscriber.handle(self.block_queue)
def do_watch(self):
# ToDo: #9-add-data-persistence
# change prev_block_hash to the last known tip when bootstrapping
# ToDo: change prev_block_hash to the last known tip when bootstrapping
prev_block_hash = BlockProcessor.get_best_block_hash()
while len(self.jobs) > 0:
@@ -150,7 +149,6 @@ class Responder:
"New block received", block_hash=block_hash, prev_block_hash=block.get("previousblockhash"), txs=txs
)
# ToDo: #9-add-data-persistence
if prev_block_hash == block.get("previousblockhash"):
self.check_confirmations(txs)
@@ -242,6 +240,7 @@ class Responder:
for uuid in self.tx_job_map[txid]:
job = self.jobs[uuid]
receipt = self.add_response(
job.locator,
uuid,
job.dispute_txid,
job.justice_txid,
@@ -290,7 +289,13 @@ class Responder:
# FIXME: Whether we decide to increase the retried counter or not, the current counter should be
# maintained. There is no way of doing so with the current approach. Update if required
self.add_response(
uuid, job.dispute_txid, job.justice_txid, job.justice_rawtx, job.appointment_end, block_hash
job.locator,
uuid,
job.dispute_txid,
job.justice_txid,
job.justice_rawtx,
job.appointment_end,
block_hash,
)
logger.warning("Justice transaction banished. Resetting the job", justice_tx=job.justice_txid)

6
pisa/sample_conf.py
View File

@@ -15,7 +15,7 @@ MAX_APPOINTMENTS = 100
EXPIRY_DELTA = 6
MIN_DISPUTE_DELTA = 20
SERVER_LOG_FILE = "pisa.log"
PISA_SECRET_KEY = "pisa_sk.pem"
PISA_SECRET_KEY = "pisa_sk.der"
# PISA-CLI
CLIENT_LOG_FILE = "pisa.log"
@@ -23,9 +23,5 @@ CLIENT_LOG_FILE = "pisa.log"
# TEST
TEST_LOG_FILE = "test.log"
# CRYPTO
SUPPORTED_HASH_FUNCTIONS = ["SHA256"]
SUPPORTED_CIPHERS = ["AES-GCM-128"]
# LEVELDB
DB_PATH = "appointments"

5
pisa/tools.py
View File

@@ -39,8 +39,3 @@ def in_correct_network(network):
correct_network = True
return correct_network
def check_txid_format(txid):
# TODO: #12-check-txid-regexp
return isinstance(txid, str) and re.search(r"^[0-9A-Fa-f]{64}$", txid) is not None

32
pisa/watcher.py
View File

@@ -1,18 +1,13 @@
from uuid import uuid4
from queue import Queue
from hashlib import sha256
from threading import Thread
from binascii import unhexlify
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from common.cryptographer import Cryptographer
from common.constants import LOCATOR_LEN_HEX
from pisa.logger import Logger
from pisa.cleaner import Cleaner
from pisa.responder import Responder
from pisa.cryptographer import Cryptographer
from pisa.block_processor import BlockProcessor
from pisa.utils.zmq_subscriber import ZMQHandler
from pisa.conf import EXPIRY_DELTA, MAX_APPOINTMENTS, PISA_SECRET_KEY
@@ -36,17 +31,13 @@ class Watcher:
if pisa_sk_file is None:
raise ValueError("No signing key provided. Please fix your pisa.conf")
else:
with open(PISA_SECRET_KEY, "r") as key_file:
secret_key_pem = key_file.read().encode("utf-8")
self.signing_key = load_pem_private_key(secret_key_pem, password=None, backend=default_backend())
with open(PISA_SECRET_KEY, "rb") as key_file:
secret_key_der = key_file.read()
self.signing_key = Cryptographer.load_private_key_der(secret_key_der)
@staticmethod
def compute_locator(tx_id):
return sha256(unhexlify(tx_id)).hexdigest()
def sign_appointment(self, appointment):
data = appointment.serialize()
return self.signing_key.sign(data, ec.ECDSA(hashes.SHA256()))
return tx_id[:LOCATOR_LEN_HEX]
def add_appointment(self, appointment):
# Rationale:
@@ -87,7 +78,8 @@ class Watcher:
logger.info("New appointment accepted.", locator=appointment.locator)
signature = self.sign_appointment(appointment)
signature = Cryptographer.sign(Cryptographer.signature_format(appointment.to_dict()), self.signing_key)
else:
appointment_added = False
signature = None
@@ -136,6 +128,7 @@ class Watcher:
self.responder.add_response(
uuid,
filtered_match["locator"],
filtered_match["dispute_txid"],
filtered_match["justice_txid"],
filtered_match["justice_rawtx"],
@@ -179,7 +172,12 @@ class Watcher:
for locator, dispute_txid in matches.items():
for uuid in self.locator_uuid_map[locator]:
justice_rawtx = Cryptographer.decrypt(self.appointments[uuid].encrypted_blob, dispute_txid)
try:
justice_rawtx = Cryptographer.decrypt(self.appointments[uuid].encrypted_blob, dispute_txid)
except ValueError:
justice_rawtx = None
justice_tx = BlockProcessor.decode_raw_transaction(justice_rawtx)
if justice_tx is not None:

14
test/apps/cli/test_pisa_cli.py
View File

@@ -34,23 +34,15 @@ dummy_appointment_request = {
}
dummy_appointment = build_appointment(**dummy_appointment_request)
# FIXME: USE CRYPTOGRAPHER
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_is_appointment_signature_valid():
# Verify that an appointment signed by Pisa is valid
signature = sign_appointment(pisa_sk, dummy_appointment)
assert pisa_cli.is_appointment_signature_valid(dummy_appointment, signature, pisa_pk)
# Test that a signature from a different key is indeed invalid
other_signature = sign_appointment(other_sk, dummy_appointment)
assert not pisa_cli.is_appointment_signature_valid(dummy_appointment, other_signature, pisa_pk)
def get_dummy_pisa_pk(pem_data):
def get_dummy_pisa_pk(der_data):
return pisa_pk

65
test/unit/conftest.py
View File

@@ -1,27 +1,29 @@
import json
import pytest
import random
import requests
from time import sleep
from shutil import rmtree
from threading import Thread
from hashlib import sha256
from binascii import hexlify, unhexlify
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
from apps.cli.blob import Blob
from pisa.responder import Job
from pisa.watcher import Watcher
from pisa.tools import bitcoin_cli
from pisa.db_manager import DBManager
from pisa.appointment import Appointment
from test.simulator.utils import sha256d
from test.simulator.transaction import TX
from test.simulator.bitcoind_sim import run_simulator, HOST, PORT
from common.constants import LOCATOR_LEN_HEX
from common.cryptographer import Cryptographer
@pytest.fixture(scope="session")
def run_bitcoind():
@@ -38,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")
@@ -59,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)
@@ -75,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()
@@ -99,21 +91,16 @@ def generate_dummy_appointment_data(real_height=True, start_time_offset=5, end_t
"dispute_delta": 20,
}
cipher = "AES-GCM-128"
hash_function = "SHA256"
# 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 = sha256(unhexlify(dispute_txid)).hexdigest()
blob = Blob(dummy_appointment_data.get("tx"), cipher, hash_function)
locator = Watcher.compute_locator(dispute_txid)
blob = Blob(dummy_appointment_data.get("tx"))
encrypted_blob = blob.encrypt((dummy_appointment_data.get("tx_id")))
encrypted_blob = Cryptographer.encrypt(blob, dummy_appointment_data.get("tx_id"))
appointment_data = {
"locator": locator,
@@ -121,13 +108,12 @@ def generate_dummy_appointment_data(real_height=True, start_time_offset=5, end_t
"end_time": dummy_appointment_data.get("end_time"),
"dispute_delta": dummy_appointment_data.get("dispute_delta"),
"encrypted_blob": encrypted_blob,
"cipher": cipher,
"hash_function": hash_function,
}
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
@@ -144,9 +130,14 @@ def generate_dummy_job():
dispute_txid = get_random_value_hex(32)
justice_txid = get_random_value_hex(32)
justice_rawtx = get_random_value_hex(100)
locator = dispute_txid[:LOCATOR_LEN_HEX]
job_data = dict(
dispute_txid=dispute_txid, justice_txid=justice_txid, justice_rawtx=justice_rawtx, appointment_end=100
locator=locator,
dispute_txid=dispute_txid,
justice_txid=justice_txid,
justice_rawtx=justice_rawtx,
appointment_end=100,
)
return Job.from_dict(job_data)

5
test/unit/test_api.py
View File

@@ -9,8 +9,11 @@ from pisa.watcher import Watcher
from pisa.tools import bitcoin_cli
from pisa import HOST, PORT, c_logger
from pisa.conf import MAX_APPOINTMENTS
from test.unit.conftest import generate_block, generate_blocks, get_random_value_hex, generate_dummy_appointment_data
from common.constants import LOCATOR_LEN_BYTES
c_logger.disabled = True
PISA_API = "http://{}:{}".format(HOST, PORT)
@@ -61,7 +64,7 @@ def test_add_appointment(run_api, run_bitcoind, new_appt_data):
def test_request_random_appointment():
r = requests.get(url=PISA_API + "/get_appointment?locator=" + get_random_value_hex(32))
r = requests.get(url=PISA_API + "/get_appointment?locator=" + get_random_value_hex(LOCATOR_LEN_BYTES))
assert r.status_code == 200
received_appointments = json.loads(r.content)

31
test/unit/test_appointment.py
View File

@@ -4,21 +4,23 @@ from pytest import fixture
from pisa import c_logger
from pisa.appointment import Appointment
from pisa.encrypted_blob import EncryptedBlob
from test.unit.conftest import get_random_value_hex
from common.constants import LOCATOR_LEN_BYTES
c_logger.disabled = True
# Not much to test here, adding it for completeness
@fixture
def appointment_data():
locator = get_random_value_hex(32)
locator = get_random_value_hex(LOCATOR_LEN_BYTES)
start_time = 100
end_time = 120
dispute_delta = 20
encrypted_blob_data = get_random_value_hex(100)
cipher = "AES-GCM-128"
hash_function = "SHA256"
return {
"locator": locator,
@@ -26,8 +28,6 @@ def appointment_data():
"end_time": end_time,
"dispute_delta": dispute_delta,
"encrypted_blob": encrypted_blob_data,
"cipher": cipher,
"hash_function": hash_function,
}
@@ -42,8 +42,6 @@ def test_init_appointment(appointment_data):
appointment_data["end_time"],
appointment_data["dispute_delta"],
appointment_data["encrypted_blob"],
appointment_data["cipher"],
appointment_data["hash_function"],
)
assert (
@@ -52,8 +50,6 @@ def test_init_appointment(appointment_data):
and appointment_data["end_time"] == appointment.end_time
and appointment_data["dispute_delta"] == appointment.dispute_delta
and EncryptedBlob(appointment_data["encrypted_blob"]) == appointment.encrypted_blob
and appointment_data["cipher"] == appointment.cipher
and appointment_data["hash_function"] == appointment.hash_function
)
@@ -64,8 +60,6 @@ def test_to_dict(appointment_data):
appointment_data["end_time"],
appointment_data["dispute_delta"],
appointment_data["encrypted_blob"],
appointment_data["cipher"],
appointment_data["hash_function"],
)
dict_appointment = appointment.to_dict()
@@ -76,8 +70,6 @@ def test_to_dict(appointment_data):
and appointment_data["end_time"] == dict_appointment["end_time"]
and appointment_data["dispute_delta"] == dict_appointment["dispute_delta"]
and EncryptedBlob(appointment_data["encrypted_blob"]) == EncryptedBlob(dict_appointment["encrypted_blob"])
and appointment_data["cipher"] == dict_appointment["cipher"]
and appointment_data["hash_function"] == dict_appointment["hash_function"]
)
@@ -88,8 +80,6 @@ def test_to_json(appointment_data):
appointment_data["end_time"],
appointment_data["dispute_delta"],
appointment_data["encrypted_blob"],
appointment_data["cipher"],
appointment_data["hash_function"],
)
dict_appointment = json.loads(appointment.to_json())
@@ -100,8 +90,6 @@ def test_to_json(appointment_data):
and appointment_data["end_time"] == dict_appointment["end_time"]
and appointment_data["dispute_delta"] == dict_appointment["dispute_delta"]
and EncryptedBlob(appointment_data["encrypted_blob"]) == EncryptedBlob(dict_appointment["encrypted_blob"])
and appointment_data["cipher"] == dict_appointment["cipher"]
and appointment_data["hash_function"] == dict_appointment["hash_function"]
)
@@ -122,12 +110,3 @@ def test_from_dict(appointment_data):
except ValueError:
appointment_data[key] = prev_val
assert True
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")

77
test/unit/test_blob.py
View File

@@ -3,88 +3,19 @@ from binascii import unhexlify
from pisa import c_logger
from apps.cli.blob import Blob
from test.unit.conftest import get_random_value_hex
from pisa.conf import SUPPORTED_CIPHERS, SUPPORTED_HASH_FUNCTIONS
c_logger.disabled = True
def test_init_blob():
data = get_random_value_hex(64)
blob = Blob(data)
assert isinstance(blob, Blob)
# Fixed (valid) hash function, try different valid ciphers
hash_function = SUPPORTED_HASH_FUNCTIONS[0]
for cipher in SUPPORTED_CIPHERS:
cipher_cases = [cipher, cipher.lower(), cipher.capitalize()]
for case in cipher_cases:
blob = Blob(data, case, hash_function)
assert blob.data == data and blob.cipher == case and blob.hash_function == hash_function
# Fixed (valid) cipher, try different valid hash functions
cipher = SUPPORTED_CIPHERS[0]
for hash_function in SUPPORTED_HASH_FUNCTIONS:
hash_function_cases = [hash_function, hash_function.lower(), hash_function.capitalize()]
for case in hash_function_cases:
blob = Blob(data, cipher, case)
assert blob.data == data and blob.cipher == cipher and blob.hash_function == case
# Invalid data
data = unhexlify(get_random_value_hex(64))
cipher = SUPPORTED_CIPHERS[0]
hash_function = SUPPORTED_HASH_FUNCTIONS[0]
# Wrong data
try:
Blob(data, cipher, hash_function)
Blob(unhexlify(get_random_value_hex(64)))
assert False, "Able to create blob with wrong data"
except ValueError:
assert True
# Invalid cipher
data = get_random_value_hex(64)
cipher = "A" * 10
hash_function = SUPPORTED_HASH_FUNCTIONS[0]
try:
Blob(data, cipher, hash_function)
assert False, "Able to create blob with wrong data"
except ValueError:
assert True
# Invalid hash function
data = get_random_value_hex(64)
cipher = SUPPORTED_CIPHERS[0]
hash_function = "A" * 10
try:
Blob(data, cipher, hash_function)
assert False, "Able to create blob with wrong data"
except ValueError:
assert True
def test_encrypt():
# Valid data, valid key
data = get_random_value_hex(64)
blob = Blob(data, SUPPORTED_CIPHERS[0], SUPPORTED_HASH_FUNCTIONS[0])
key = get_random_value_hex(32)
encrypted_blob = blob.encrypt(key)
# Invalid key (note that encrypt cannot be called with invalid data since that's checked when the Blob is created)
invalid_key = unhexlify(get_random_value_hex(32))
try:
blob.encrypt(invalid_key)
assert False, "Able to create encrypt with invalid key"
except ValueError:
assert True
# Check that two encryptions of the same data have the same result
encrypted_blob2 = blob.encrypt(key)
assert encrypted_blob == encrypted_blob2 and id(encrypted_blob) != id(encrypted_blob2)

16
test/unit/test_cleaner.py
View File

@@ -6,8 +6,11 @@ from pisa.responder import Job
from pisa.cleaner import Cleaner
from pisa.appointment import Appointment
from pisa.db_manager import WATCHER_PREFIX
from test.unit.conftest import get_random_value_hex
from common.constants import LOCATOR_LEN_BYTES, LOCATOR_LEN_HEX
CONFIRMATIONS = 6
ITEMS = 10
MAX_ITEMS = 100
@@ -23,9 +26,9 @@ def set_up_appointments(db_manager, total_appointments):
for i in range(total_appointments):
uuid = uuid4().hex
locator = get_random_value_hex(32)
locator = get_random_value_hex(LOCATOR_LEN_BYTES)
appointment = Appointment(locator, None, None, None, None, None, None)
appointment = Appointment(locator, None, None, None, None, None)
appointments[uuid] = appointment
locator_uuid_map[locator] = [uuid]
@@ -55,9 +58,10 @@ def set_up_jobs(db_manager, total_jobs):
# We use the same txid for justice and dispute here, it shouldn't matter
justice_txid = get_random_value_hex(32)
dispute_txid = get_random_value_hex(32)
locator = dispute_txid[:LOCATOR_LEN_HEX]
# Assign both justice_txid and dispute_txid the same id (it shouldn't matter)
job = Job(dispute_txid, justice_txid, None, None)
job = Job(locator, dispute_txid, justice_txid, None, None)
jobs[uuid] = job
tx_job_map[justice_txid] = [uuid]
@@ -131,9 +135,10 @@ def test_delete_completed_jobs_no_db_match(db_manager):
for uuid in selected_jobs[: ITEMS // 2]:
justice_txid = jobs[uuid].justice_txid
dispute_txid = get_random_value_hex(32)
locator = dispute_txid[:LOCATOR_LEN_HEX]
new_uuid = uuid4().hex
jobs[new_uuid] = Job(dispute_txid, justice_txid, None, None)
jobs[new_uuid] = Job(locator, dispute_txid, justice_txid, None, None)
tx_job_map[justice_txid].append(new_uuid)
selected_jobs.append(new_uuid)
@@ -142,8 +147,9 @@ def test_delete_completed_jobs_no_db_match(db_manager):
uuid = uuid4().hex
justice_txid = get_random_value_hex(32)
dispute_txid = get_random_value_hex(32)
locator = dispute_txid[:LOCATOR_LEN_HEX]
jobs[uuid] = Job(dispute_txid, justice_txid, None, None)
jobs[uuid] = Job(locator, dispute_txid, justice_txid, None, None)
tx_job_map[justice_txid] = [uuid]
selected_jobs.append(uuid)

117
test/unit/test_cryptographer.py
View File

@@ -1,50 +1,145 @@
import binascii
from pisa.cryptographer import Cryptographer
from apps.cli.blob import Blob
from common.cryptographer import Cryptographer
from pisa.encrypted_blob import EncryptedBlob
from test.unit.conftest import get_random_value_hex
data = "6097cdf52309b1b2124efeed36bd34f46dc1c25ad23ac86f28380f746254f777"
key = "b2e984a570f6f49bc38ace178e09147b0aa296cbb7c92eb01412f7e2d07b5659"
encrypted_data = "092e93d4a34aac4367075506f2c050ddfa1a201ee6669b65058572904dcea642aeb01ea4b57293618e8c46809dfadadc"
encrypted_blob = EncryptedBlob(encrypted_data)
encrypted_data = "8f31028097a8bf12a92e088caab5cf3fcddf0d35ed2b72c24b12269373efcdea04f9d2a820adafe830c20ff132d89810"
# TODO: The decryption tests are assuming the cipher is AES-GCM-128, since EncryptedBlob assumes the same. Fix this.
def test_decrypt_wrong_data():
def test_check_data_key_format_wrong_data():
data = get_random_value_hex(64)[:-1]
key = get_random_value_hex(32)
try:
Cryptographer.check_data_key_format(data, key)
assert False
except ValueError as e:
assert "Odd-length" in str(e)
def test_check_data_key_format_wrong_key():
data = get_random_value_hex(64)
key = get_random_value_hex(33)
try:
Cryptographer.check_data_key_format(data, key)
assert False
except ValueError as e:
assert "32-byte hex" in str(e)
def test_check_data_key_format():
data = get_random_value_hex(64)
key = get_random_value_hex(32)
assert Cryptographer.check_data_key_format(data, key) is True
def test_encrypt_odd_length_data():
blob = Blob(get_random_value_hex(64)[-1])
key = get_random_value_hex(32)
try:
Cryptographer.encrypt(blob, key)
assert False
except ValueError:
assert True
def test_encrypt_wrong_key_size():
blob = Blob(get_random_value_hex(64))
key = get_random_value_hex(31)
try:
Cryptographer.encrypt(blob, key)
assert False
except ValueError:
assert True
def test_encrypt_hex():
blob = Blob(data)
assert Cryptographer.encrypt(blob, key) == encrypted_data
def test_encrypt_bytes():
blob = Blob(data)
byte_blob = Cryptographer.encrypt(blob, key, rtype="bytes")
assert isinstance(byte_blob, bytes) and byte_blob == binascii.unhexlify(encrypted_data)
def test_encrypt_wrong_return():
# Any other type but "hex" (default) or "bytes" should fail
try:
Cryptographer.encrypt(Blob(data), key, rtype="random_value")
assert False
except ValueError:
assert True
def test_decrypt_invalid_tag():
random_key = get_random_value_hex(32)
random_encrypted_data = get_random_value_hex(64)
random_encrypted_blob = EncryptedBlob(random_encrypted_data)
# Trying to decrypt random data (in AES_GCM-128) should result in an InvalidTag exception. Our decrypt function
# Trying to decrypt random data should result in an InvalidTag exception. Our decrypt function
# returns None
hex_tx = Cryptographer.decrypt(random_encrypted_blob, random_key)
assert hex_tx is None
def test_decrypt_odd_length():
def test_decrypt_odd_length_data():
random_key = get_random_value_hex(32)
random_encrypted_data_odd = get_random_value_hex(64)[:-1]
random_encrypted_blob_odd = EncryptedBlob(random_encrypted_data_odd)
assert Cryptographer.decrypt(random_encrypted_blob_odd, random_key) is None
try:
Cryptographer.decrypt(random_encrypted_blob_odd, random_key)
assert False
except ValueError:
assert True
def test_decrypt_wrong_key_size():
random_key = get_random_value_hex(31)
random_encrypted_data_odd = get_random_value_hex(64)
random_encrypted_blob_odd = EncryptedBlob(random_encrypted_data_odd)
try:
Cryptographer.decrypt(random_encrypted_blob_odd, random_key)
assert False
except ValueError:
assert True
def test_decrypt_hex():
# Valid data should run with no InvalidTag and verify
assert Cryptographer.decrypt(encrypted_blob, key) == data
assert Cryptographer.decrypt(EncryptedBlob(encrypted_data), key) == data
def test_decrypt_bytes():
# We can also get the decryption in bytes
byte_blob = Cryptographer.decrypt(encrypted_blob, key, rtype="bytes")
byte_blob = Cryptographer.decrypt(EncryptedBlob(encrypted_data), key, rtype="bytes")
assert isinstance(byte_blob, bytes) and byte_blob == binascii.unhexlify(data)
def test_decrypt_wrong_return():
# Any other type but "hex" (default) or "bytes" should fail
try:
Cryptographer.decrypt(encrypted_blob, key, rtype="random_value")
Cryptographer.decrypt(EncryptedBlob(encrypted_data), key, rtype="random_value")
assert False
except ValueError:

21
test/unit/test_db_manager.py
View File

@@ -5,9 +5,12 @@ import shutil
from uuid import uuid4
from pisa.db_manager import DBManager
from test.unit.conftest import get_random_value_hex, generate_dummy_appointment
from pisa.db_manager import WATCHER_LAST_BLOCK_KEY, RESPONDER_LAST_BLOCK_KEY, LOCATOR_MAP_PREFIX
from common.constants import LOCATOR_LEN_BYTES
from test.unit.conftest import get_random_value_hex, generate_dummy_appointment
@pytest.fixture(scope="module")
def watcher_appointments():
@@ -16,7 +19,7 @@ def watcher_appointments():
@pytest.fixture(scope="module")
def responder_jobs():
return {get_random_value_hex(32): get_random_value_hex(32) for _ in range(10)}
return {get_random_value_hex(16): get_random_value_hex(32) for _ in range(10)}
def open_create_db(db_path):
@@ -40,7 +43,6 @@ def test_init():
# Check that the db can be created if it does not exist
db_manager = open_create_db(db_path)
assert isinstance(db_manager, DBManager)
print(type(db_manager))
db_manager.db.close()
# Check that we can open an already create db
@@ -65,7 +67,7 @@ def test_load_appointments_db(db_manager):
# We can add a bunch of data to the db and try again (data is stored in json by the manager)
local_appointments = {}
for _ in range(10):
key = get_random_value_hex(32)
key = get_random_value_hex(16)
value = get_random_value_hex(32)
local_appointments[key] = value
@@ -106,7 +108,7 @@ def test_get_last_known_block():
def test_create_entry(db_manager):
key = get_random_value_hex(32)
key = get_random_value_hex(16)
value = get_random_value_hex(32)
# Adding a value with no prefix (create entry encodes values in utf-8 internally)
@@ -116,7 +118,7 @@ def test_create_entry(db_manager):
assert db_manager.db.get(key.encode("utf-8")).decode("utf-8") == value
# If we prefix the key we should be able to get it if we add the prefix, but not otherwise
key = get_random_value_hex(32)
key = get_random_value_hex(16)
prefix = "w"
db_manager.create_entry(key, value, prefix=prefix)
@@ -140,7 +142,7 @@ def test_delete_entry(db_manager):
# Let's check that the same works if a prefix is provided.
prefix = "r"
key = get_random_value_hex(32)
key = get_random_value_hex(16)
value = get_random_value_hex(32)
db_manager.create_entry(key, value, prefix)
@@ -161,12 +163,12 @@ def test_load_responder_jobs_empty(db_manager):
def test_load_locator_map_empty(db_manager):
assert db_manager.load_locator_map(get_random_value_hex(32)) is None
assert db_manager.load_locator_map(get_random_value_hex(LOCATOR_LEN_BYTES)) is None
def test_store_update_locator_map_empty(db_manager):
uuid = uuid4().hex
locator = get_random_value_hex(32)
locator = get_random_value_hex(LOCATOR_LEN_BYTES)
db_manager.store_update_locator_map(locator, uuid)
# Check that the locator map has been properly stored
@@ -188,7 +190,6 @@ def test_delete_locator_map(db_manager):
assert len(locator_maps) != 0
for locator, uuids in locator_maps.items():
print(locator)
db_manager.delete_locator_map(locator)
locator_maps = db_manager.load_appointments_db(prefix=LOCATOR_MAP_PREFIX)

18
test/unit/test_encrypted_blob.py
View File

@@ -11,24 +11,6 @@ def test_init_encrypted_blob():
assert EncryptedBlob(data).data == data
def test_init_encrypted_blob_wrong_cipher():
try:
EncryptedBlob(get_random_value_hex(64), cipher="")
assert False
except ValueError:
assert True
def test_init_encrypted_blob_wrong_hash_function():
try:
EncryptedBlob(get_random_value_hex(64), hash_function="")
assert False
except ValueError:
assert True
def test_equal():
data = get_random_value_hex(64)
e_blob1 = EncryptedBlob(data)

139
test/unit/test_inspector.py
View File

@@ -1,46 +1,58 @@
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 apps.cli.pisa_cli import build_appointment
from pisa import c_logger
from pisa.errors import *
from pisa.inspector import Inspector
from pisa.appointment import Appointment
from pisa.block_processor import BlockProcessor
from test.unit.conftest import get_random_value_hex
from pisa.conf import MIN_DISPUTE_DELTA
from pisa.conf import MIN_DISPUTE_DELTA, SUPPORTED_CIPHERS, SUPPORTED_HASH_FUNCTIONS
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
inspector = Inspector()
APPOINTMENT_OK = (0, None)
NO_HEX_STRINGS = ["R" * 64, get_random_value_hex(31) + "PP", "$" * 64, " " * 64]
WRONG_TYPES = [[], "", get_random_value_hex(32), 3.2, 2.0, (), object, {}, " " * 32, object()]
WRONG_TYPES_NO_STR = [[], unhexlify(get_random_value_hex(32)), 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")
NO_HEX_STRINGS = [
"R" * LOCATOR_LEN_HEX,
get_random_value_hex(LOCATOR_LEN_BYTES - 1) + "PP",
"$" * LOCATOR_LEN_HEX,
" " * LOCATOR_LEN_HEX,
]
WRONG_TYPES = [
[],
"",
get_random_value_hex(LOCATOR_LEN_BYTES),
3.2,
2.0,
(),
object,
{},
" " * LOCATOR_LEN_HEX,
object(),
]
WRONG_TYPES_NO_STR = [[], unhexlify(get_random_value_hex(LOCATOR_LEN_BYTES)), 3.2, 2.0, (), object, {}, object()]
def test_check_locator():
# Right appointment type, size and format
locator = get_random_value_hex(32)
locator = get_random_value_hex(LOCATOR_LEN_BYTES)
assert Inspector.check_locator(locator) == APPOINTMENT_OK
# Wrong size (too big)
locator = get_random_value_hex(33)
locator = get_random_value_hex(LOCATOR_LEN_BYTES + 1)
assert Inspector.check_locator(locator)[0] == APPOINTMENT_WRONG_FIELD_SIZE
# Wrong size (too small)
locator = get_random_value_hex(31)
locator = get_random_value_hex(LOCATOR_LEN_BYTES - 1)
assert Inspector.check_locator(locator)[0] == APPOINTMENT_WRONG_FIELD_SIZE
# Empty
@@ -157,96 +169,51 @@ def test_check_blob():
assert Inspector.check_blob(encrypted_blob)[0] == APPOINTMENT_WRONG_FIELD_FORMAT
def test_check_cipher():
# Right format and content (any case combination should be accepted)
for cipher in SUPPORTED_CIPHERS:
cipher_cases = [cipher, cipher.lower(), cipher.capitalize()]
for case in cipher_cases:
assert Inspector.check_cipher(case) == APPOINTMENT_OK
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")
# Wrong type
ciphers = WRONG_TYPES_NO_STR
for cipher in ciphers:
assert Inspector.check_cipher(cipher)[0] == APPOINTMENT_WRONG_FIELD_TYPE
# Wrong value
ciphers = NO_HEX_STRINGS
for cipher in ciphers:
assert Inspector.check_cipher(cipher)[0] == APPOINTMENT_CIPHER_NOT_SUPPORTED
# Empty field
cipher = None
assert Inspector.check_cipher(cipher)[0] == APPOINTMENT_EMPTY_FIELD
def test_check_hash_function():
# Right format and content (any case combination should be accepted)
for hash_function in SUPPORTED_HASH_FUNCTIONS:
hash_function_cases = [hash_function, hash_function.lower(), hash_function.capitalize()]
for case in hash_function_cases:
assert Inspector.check_hash_function(case) == APPOINTMENT_OK
# Wrong type
hash_functions = WRONG_TYPES_NO_STR
for hash_function in hash_functions:
assert Inspector.check_hash_function(hash_function)[0] == APPOINTMENT_WRONG_FIELD_TYPE
# Wrong value
hash_functions = NO_HEX_STRINGS
for hash_function in hash_functions:
assert Inspector.check_hash_function(hash_function)[0] == APPOINTMENT_HASH_FUNCTION_NOT_SUPPORTED
# Empty field
hash_function = None
assert Inspector.check_hash_function(hash_function)[0] == APPOINTMENT_EMPTY_FIELD
def test_check_appointment_signature(generate_keypair):
client_sk, client_pk = generate_keypair
dummy_appointment_request = {
"tx": get_random_value_hex(192),
"tx_id": get_random_value_hex(32),
"start_time": 1500,
"end_time": 50000,
"dispute_delta": 200,
}
dummy_appointment = build_appointment(**dummy_appointment_request)
# Verify that an appointment signed by the client is valid
signature = sign_appointment(client_sk, dummy_appointment)
assert Inspector.check_appointment_signature(dummy_appointment, signature, client_pk)
dummy_appointment_data, _ = generate_dummy_appointment_data(real_height=False)
assert Inspector.check_appointment_signature(
dummy_appointment_data["appointment"], dummy_appointment_data["signature"], dummy_appointment_data["public_key"]
)
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)
bad_signature = Cryptographer.sign(Cryptographer.signature_format(dummy_appointment_data["appointment"]), fake_sk)
assert (
Inspector.check_appointment_signature(dummy_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
# Valid appointment
locator = get_random_value_hex(32)
locator = get_random_value_hex(LOCATOR_LEN_BYTES)
start_time = BlockProcessor.get_block_count() + 5
end_time = start_time + 20
dispute_delta = MIN_DISPUTE_DELTA
encrypted_blob = get_random_value_hex(64)
cipher = SUPPORTED_CIPHERS[0]
hash_function = SUPPORTED_HASH_FUNCTIONS[0]
appointment_data = {
"locator": locator,
@@ -254,13 +221,11 @@ def test_inspect(run_bitcoind, generate_keypair):
"end_time": end_time,
"dispute_delta": dispute_delta,
"encrypted_blob": encrypted_blob,
"cipher": cipher,
"hash_function": hash_function,
}
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
@@ -269,6 +234,4 @@ def test_inspect(run_bitcoind, generate_keypair):
and appointment.end_time == end_time
and appointment.dispute_delta == dispute_delta
and appointment.encrypted_blob.data == encrypted_blob
and appointment.cipher == cipher
and appointment.hash_function == hash_function
)

49
test/unit/test_responder.py
View File

@@ -9,11 +9,15 @@ from queue import Queue, Empty
from pisa import c_logger
from pisa.db_manager import DBManager
from test.simulator.utils import sha256d
from pisa.responder import Responder, Job
from test.simulator.bitcoind_sim import TX
from pisa.block_processor import BlockProcessor
from pisa.tools import check_txid_format, bitcoin_cli
from pisa.tools import bitcoin_cli
from common.constants import LOCATOR_LEN_HEX
from common.tools import check_sha256_hex_format
from test.simulator.utils import sha256d
from test.simulator.bitcoind_sim import TX
from test.unit.conftest import generate_block, generate_blocks, get_random_value_hex
c_logger.disabled = True
@@ -58,18 +62,21 @@ def create_dummy_job_data(random_txid=False, justice_rawtx=None):
justice_txid = get_random_value_hex(32)
appointment_end = bitcoin_cli().getblockcount() + 2
locator = dispute_txid[:LOCATOR_LEN_HEX]
return dispute_txid, justice_txid, justice_rawtx, appointment_end
return locator, dispute_txid, justice_txid, justice_rawtx, appointment_end
def create_dummy_job(random_txid=False, justice_rawtx=None):
dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data(random_txid, justice_rawtx)
return Job(dispute_txid, justice_txid, justice_rawtx, appointment_end)
locator, dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data(
random_txid, justice_rawtx
)
return Job(locator, dispute_txid, justice_txid, justice_rawtx, appointment_end)
def test_job_init(run_bitcoind):
dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data()
job = Job(dispute_txid, justice_txid, justice_rawtx, appointment_end)
locator, dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data()
job = Job(locator, dispute_txid, justice_txid, justice_rawtx, appointment_end)
assert (
job.dispute_txid == dispute_txid
@@ -157,6 +164,7 @@ def test_add_response(db_manager):
# The block_hash passed to add_response does not matter much now. It will in the future to deal with errors
receipt = responder.add_response(
job.locator,
uuid,
job.dispute_txid,
job.justice_txid,
@@ -187,6 +195,7 @@ def test_add_bad_response(responder):
# The block_hash passed to add_response does not matter much now. It will in the future to deal with errors
receipt = responder.add_response(
job.locator,
uuid,
job.dispute_txid,
job.justice_txid,
@@ -204,7 +213,7 @@ def test_create_job(responder):
for _ in range(20):
uuid = uuid4().hex
confirmations = 0
dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data(random_txid=True)
locator, dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data(random_txid=True)
# Check the job is not within the responder jobs before adding it
assert uuid not in responder.jobs
@@ -212,7 +221,7 @@ def test_create_job(responder):
assert justice_txid not in responder.unconfirmed_txs
# And that it is afterwards
responder.create_job(uuid, dispute_txid, justice_txid, justice_rawtx, appointment_end, confirmations)
responder.create_job(uuid, locator, dispute_txid, justice_txid, justice_rawtx, appointment_end, confirmations)
assert uuid in responder.jobs
assert justice_txid in responder.tx_job_map
assert justice_txid in responder.unconfirmed_txs
@@ -231,12 +240,12 @@ def test_create_job(responder):
def test_create_job_same_justice_txid(responder):
# Create the same job using two different uuids
confirmations = 0
dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data(random_txid=True)
locator, dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data(random_txid=True)
uuid_1 = uuid4().hex
uuid_2 = uuid4().hex
responder.create_job(uuid_1, dispute_txid, justice_txid, justice_rawtx, appointment_end, confirmations)
responder.create_job(uuid_2, dispute_txid, justice_txid, justice_rawtx, appointment_end, confirmations)
responder.create_job(uuid_1, locator, dispute_txid, justice_txid, justice_rawtx, appointment_end, confirmations)
responder.create_job(uuid_2, locator, dispute_txid, justice_txid, justice_rawtx, appointment_end, confirmations)
# Check that both jobs have been added
assert uuid_1 in responder.jobs and uuid_2 in responder.jobs
@@ -261,11 +270,11 @@ def test_create_job_already_confirmed(responder):
for i in range(20):
uuid = uuid4().hex
confirmations = i + 1
dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data(
locator, dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data(
justice_rawtx=TX.create_dummy_transaction()
)
responder.create_job(uuid, dispute_txid, justice_txid, justice_rawtx, appointment_end, confirmations)
responder.create_job(uuid, locator, dispute_txid, justice_txid, justice_rawtx, appointment_end, confirmations)
assert justice_txid not in responder.unconfirmed_txs
@@ -280,7 +289,7 @@ def test_do_subscribe(responder):
try:
generate_block()
block_hash = responder.block_queue.get()
assert check_txid_format(block_hash)
assert check_sha256_hex_format(block_hash)
except Empty:
assert False
@@ -360,7 +369,9 @@ def test_check_confirmations(temp_db_manager):
responder.unconfirmed_txs.extend(txs_subset)
# We also need to add them to the tx_job_map since they would be there in normal conditions
responder.tx_job_map = {txid: Job(txid, None, None, None) for txid in responder.unconfirmed_txs}
responder.tx_job_map = {
txid: Job(txid[:LOCATOR_LEN_HEX], txid, None, None, None) for txid in responder.unconfirmed_txs
}
# Let's make sure that there are no txs with missed confirmations yet
assert len(responder.missed_confirmations) == 0
@@ -457,11 +468,11 @@ def test_rebroadcast(db_manager):
# Rebroadcast calls add_response with retry=True. The job data is already in jobs.
for i in range(20):
uuid = uuid4().hex
dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data(
locator, dispute_txid, justice_txid, justice_rawtx, appointment_end = create_dummy_job_data(
justice_rawtx=TX.create_dummy_transaction()
)
responder.jobs[uuid] = Job(dispute_txid, justice_txid, justice_rawtx, appointment_end)
responder.jobs[uuid] = Job(locator, dispute_txid, justice_txid, justice_rawtx, appointment_end)
responder.tx_job_map[justice_txid] = [uuid]
responder.unconfirmed_txs.append(justice_txid)

42
test/unit/test_tools.py
View File

@@ -1,5 +1,7 @@
from pisa import c_logger
from pisa.tools import can_connect_to_bitcoind, in_correct_network, bitcoin_cli, check_txid_format
from pisa.tools import can_connect_to_bitcoind, in_correct_network, bitcoin_cli
from common.tools import check_sha256_hex_format
c_logger.disabled = True
@@ -30,14 +32,30 @@ def test_bitcoin_cli():
assert False
def test_check_txid_format():
assert check_txid_format(None) is False
assert check_txid_format("") is False
assert check_txid_format(0x0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF) is False # wrong type
assert check_txid_format("abcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcd") is True # lowercase
assert check_txid_format("ABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCD") is True # uppercase
assert check_txid_format("0123456789abcdef0123456789ABCDEF0123456789abcdef0123456789ABCDEF") is True # mixed case
assert check_txid_format("0123456789012345678901234567890123456789012345678901234567890123") is True # only nums
assert check_txid_format("0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdf") is False # too short
assert check_txid_format("0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0") is False # too long
assert check_txid_format("g123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef") is False # non-hex
def test_check_sha256_hex_format():
assert check_sha256_hex_format(None) is False
assert check_sha256_hex_format("") is False
assert (
check_sha256_hex_format(0x0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF) is False
) # wrong type
assert (
check_sha256_hex_format("abcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcd") is True
) # lowercase
assert (
check_sha256_hex_format("ABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCD") is True
) # uppercase
assert (
check_sha256_hex_format("0123456789abcdef0123456789ABCDEF0123456789abcdef0123456789ABCDEF") is True
) # mixed case
assert (
check_sha256_hex_format("0123456789012345678901234567890123456789012345678901234567890123") is True
) # only nums
assert (
check_sha256_hex_format("0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdf") is False
) # too short
assert (
check_sha256_hex_format("0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0") is False
) # too long
assert (
check_sha256_hex_format("g123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef") is False
) # non-hex

61
test/unit/test_watcher.py
View File

@@ -1,23 +1,18 @@
import pytest
from uuid import uuid4
from hashlib import sha256
from threading import Thread
from binascii import unhexlify
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
from pisa.tools import check_txid_format, bitcoin_cli
from pisa.tools import bitcoin_cli
from test.unit.conftest import generate_block, generate_blocks, generate_dummy_appointment, get_random_value_hex
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
APPOINTMENTS = 5
@@ -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()
@@ -44,7 +38,7 @@ def txids():
@pytest.fixture(scope="module")
def locator_uuid_map(txids):
return {sha256(unhexlify(txid)).hexdigest(): uuid4().hex for txid in txids}
return {Watcher.compute_locator(txid): uuid4().hex for txid in txids}
def create_appointments(n):
@@ -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
@@ -154,7 +132,7 @@ def test_do_subscribe(watcher):
try:
generate_block()
block_hash = watcher.block_queue.get()
assert check_txid_format(block_hash)
assert check_sha256_hex_format(block_hash)
except Empty:
assert False
@@ -232,18 +210,17 @@ def test_filter_valid_matches_random_data(watcher):
def test_filter_valid_matches(watcher):
dispute_txid = "0437cd7f8525ceed2324359c2d0ba26006d92d856a9c20fa0241106ee5a597c9"
encrypted_blob = (
"29f55518945408f567bb7feb4d7bb15ba88b7d8ca0223a44d5c67dfe32d038caee7613e35736025d95ad4ecd6538a50"
"74cbe8d7739705697a5dc4d19b8a6e4459ed2d1b0d0a9b18c49bc2187dcbfb4046b14d58a1add83235fc632efc398d5"
"0abcb7738f1a04b3783d025c1828b4e8a8dc8f13f2843e6bc3bf08eade02fc7e2c4dce7d2f83b055652e944ac114e0b"
"72a9abcd98fd1d785a5d976c05ed780e033e125fa083c6591b6029aa68dbc099f148a2bc2e0cb63733e68af717d48d5"
"a312b5f5b2fcca9561b2ff4191f9cdff936a43f6efef4ee45fbaf1f18d0a4b006f3fc8399dd8ecb21f709d4583bba14"
"4af6d49fa99d7be2ca21059a997475aa8642b66b921dc7fc0321b6a2f6927f6f9bab55c75e17a19dc3b2ae895b6d4a4"
"f64f8eb21b1e"
"a62aa9bb3c8591e4d5de10f1bd49db92432ce2341af55762cdc9242c08662f97f5f47da0a1aa88373508cd6e67e87eefddeca0cee98c1"
"967ec1c1ecbb4c5e8bf08aa26159214e6c0bc4b2c7c247f87e7601d15c746fc4e711be95ba0e363001280138ba9a65b06c4aa6f592b21"
"3635ee763984d522a4c225814510c8f7ab0801f36d4a68f5ee7dd3930710005074121a172c29beba79ed647ebaf7e7fab1bbd9a208251"
"ef5486feadf2c46e33a7d66adf9dbbc5f67b55a34b1b3c4909dd34a482d759b0bc25ecd2400f656db509466d7479b5b92a2fadabccc9e"
"c8918da8979a9feadea27531643210368fee494d3aaa4983e05d6cf082a49105e2f8a7c7821899239ba7dee12940acd7d8a629894b5d31"
"e94b439cfe8d2e9f21e974ae5342a70c91e8"
)
dummy_appointment, _ = generate_dummy_appointment()
dummy_appointment.encrypted_blob.data = encrypted_blob
dummy_appointment.locator = sha256(unhexlify(dispute_txid)).hexdigest()
dummy_appointment.locator = Watcher.compute_locator(dispute_txid)
uuid = uuid4().hex
appointments = {uuid: dummy_appointment}