#!/usr/bin/env python import ctypes import json import os from binascii import hexlify, unhexlify import pytest from pyasn1.codec.ber.decoder import decode as ber_decode from pyasn1.codec.der.decoder import decode as der_decode from pyasn1.codec.der.encoder import encode as der_encode from pyasn1.type import namedtype, univ class EcSignature(univ.Sequence): componentType = namedtype.NamedTypes( namedtype.NamedType("r", univ.Integer()), namedtype.NamedType("s", univ.Integer()), ) class EcKeyInfo(univ.Sequence): componentType = namedtype.NamedTypes( namedtype.NamedType("key_type", univ.ObjectIdentifier()), namedtype.NamedType("curve_name", univ.ObjectIdentifier()), ) class EcPublicKey(univ.Sequence): componentType = namedtype.NamedTypes( namedtype.NamedType("key_info", EcKeyInfo()), namedtype.NamedType("public_key", univ.BitString()), ) class EdKeyInfo(univ.Sequence): componentType = namedtype.NamedTypes( namedtype.NamedType("key_type", univ.ObjectIdentifier()) ) class EdPublicKey(univ.Sequence): componentType = namedtype.NamedTypes( namedtype.NamedType("key_info", EdKeyInfo()), namedtype.NamedType("public_key", univ.BitString()), ) class ParseError(Exception): pass class NotSupported(Exception): pass class DataError(Exception): pass class curve_info(ctypes.Structure): _fields_ = [("bip32_name", ctypes.c_char_p), ("params", ctypes.c_void_p)] def keys_in_dict(dictionary, keys): return keys <= set(dictionary.keys()) def parse_eddsa_signature(signature): if len(signature) != 64: raise ParseError("Not a valid EdDSA signature") return signature def parse_ecdh256_privkey(private_key): if private_key < 0 or private_key.bit_length() > 256: raise ParseError("Not a valid 256 bit ECDH private key") return private_key.to_bytes(32, byteorder="big") def parse_signed_hex(string): if len(string) % 2 == 1: string = "0" + string number = int(string, 16) if int(string[0], 16) & 8: return -number else: return number def parse_result(result): if result == "valid": return True elif result == "invalid": return False elif result == "acceptable": return None else: raise DataError() def is_valid_der(data): try: structure, _ = der_decode(data) return data == der_encode(structure) except Exception: return False def parse_ed_pubkey(public_key): try: public_key, _ = ber_decode(public_key, asn1Spec=EdPublicKey()) except Exception: raise ParseError("Not a BER encoded Edwards curve public key") if not public_key["key_info"]["key_type"] == univ.ObjectIdentifier("1.3.101.112"): raise ParseError("Not a BER encoded Edwards curve public key") public_key = bytes(public_key["public_key"].asOctets()) return public_key def parse_ec_pubkey(public_key): try: public_key, _ = ber_decode(public_key, asn1Spec=EcPublicKey()) except Exception: raise ParseError("Not a BER encoded named elliptic curve public key") if not public_key["key_info"]["key_type"] == univ.ObjectIdentifier( "1.2.840.10045.2.1" ): raise ParseError("Not a BER encoded named elliptic curve public key") curve_identifier = public_key["key_info"]["curve_name"] curve_name = get_curve_name_by_identifier(curve_identifier) if curve_name is None: raise NotSupported( "Unsupported named elliptic curve: {}".format(curve_identifier) ) try: public_key = bytes(public_key["public_key"].asOctets()) except Exception: raise ParseError("Not a BER encoded named elliptic curve public key") return curve_name, public_key def parse_ecdsa256_signature(signature): s = signature if not is_valid_der(signature): raise ParseError("Not a valid DER") try: signature, _ = der_decode(signature, asn1Spec=EcSignature()) except Exception: raise ParseError("Not a valid DER encoded ECDSA signature") try: r = int(signature["r"]).to_bytes(32, byteorder="big") s = int(signature["s"]).to_bytes(32, byteorder="big") signature = r + s except Exception: raise ParseError("Not a valid DER encoded 256 bit ECDSA signature") return signature def parse_digest(name): if name == "SHA-256": return 0 else: raise NotSupported("Unsupported hash function: {}".format(name)) def get_curve_by_name(name): lib.get_curve_by_name.restype = ctypes.c_void_p curve = lib.get_curve_by_name(bytes(name, "ascii")) if curve is None: return None curve = ctypes.cast(curve, ctypes.POINTER(curve_info)) return ctypes.c_void_p(curve.contents.params) def parse_curve_name(name): if name == "secp256r1": return "nist256p1" elif name == "secp256k1": return "secp256k1" elif name == "curve25519": return "curve25519" else: return None def get_curve_name_by_identifier(identifier): if identifier == univ.ObjectIdentifier("1.3.132.0.10"): return "secp256k1" elif identifier == univ.ObjectIdentifier("1.2.840.10045.3.1.7"): return "nist256p1" else: return None def chacha_poly_encrypt(key, iv, associated_data, plaintext): context = bytes(context_structure_length) tag = bytes(16) ciphertext = bytes(len(plaintext)) lib.rfc7539_init(context, key, iv) lib.rfc7539_auth(context, associated_data, len(associated_data)) lib.chacha20poly1305_encrypt(context, plaintext, ciphertext, len(plaintext)) lib.rfc7539_finish(context, len(associated_data), len(plaintext), tag) return ciphertext, tag def chacha_poly_decrypt(key, iv, associated_data, ciphertext, tag): context = bytes(context_structure_length) computed_tag = bytes(16) plaintext = bytes(len(ciphertext)) lib.rfc7539_init(context, key, iv) lib.rfc7539_auth(context, associated_data, len(associated_data)) lib.chacha20poly1305_decrypt(context, ciphertext, plaintext, len(ciphertext)) lib.rfc7539_finish(context, len(associated_data), len(ciphertext), computed_tag) return plaintext if tag == computed_tag else False def add_pkcs_padding(data): padding_length = 16 - len(data) % 16 return data + bytes([padding_length] * padding_length) def remove_pkcs_padding(data): padding_length = data[-1] if not ( 0 < padding_length <= 16 and data[-padding_length:] == bytes([padding_length] * padding_length) ): return False else: return data[:-padding_length] def aes_encrypt_initialise(key, context): if len(key) == (128 / 8): lib.aes_encrypt_key128(key, context) elif len(key) == (192 / 8): lib.aes_encrypt_key192(key, context) elif len(key) == (256 / 8): lib.aes_encrypt_key256(key, context) else: raise NotSupported("Unsupported key length: {}".format(len(key) * 8)) def aes_cbc_encrypt(key, iv, plaintext): plaintext = add_pkcs_padding(plaintext) context = bytes(context_structure_length) ciphertext = bytes(len(plaintext)) aes_encrypt_initialise(key, context) lib.aes_cbc_encrypt( plaintext, ciphertext, len(plaintext), bytes(bytearray(iv)), context ) return ciphertext def aes_decrypt_initialise(key, context): if len(key) == (128 / 8): lib.aes_decrypt_key128(key, context) elif len(key) == (192 / 8): lib.aes_decrypt_key192(key, context) elif len(key) == (256 / 8): lib.aes_decrypt_key256(key, context) else: raise NotSupported("Unsupported AES key length: {}".format(len(key) * 8)) def aes_cbc_decrypt(key, iv, ciphertext): context = bytes(context_structure_length) plaintext = bytes(len(ciphertext)) aes_decrypt_initialise(key, context) lib.aes_cbc_decrypt(ciphertext, plaintext, len(ciphertext), iv, context) return remove_pkcs_padding(plaintext) def load_json_testvectors(filename): try: result = json.loads(open(os.path.join(testvectors_directory, filename)).read()) except Exception: raise DataError() return result def generate_aes(filename): vectors = [] data = load_json_testvectors(filename) if not keys_in_dict(data, {"algorithm", "testGroups"}): raise DataError() if data["algorithm"] != "AES-CBC-PKCS5": raise DataError() for test_group in data["testGroups"]: if not keys_in_dict(test_group, {"tests"}): raise DataError() for test in test_group["tests"]: if not keys_in_dict(test, {"key", "iv", "msg", "ct", "result"}): raise DataError() try: key = unhexlify(test["key"]) iv = unhexlify(test["iv"]) plaintext = unhexlify(test["msg"]) ciphertext = unhexlify(test["ct"]) result = parse_result(test["result"]) except Exception: raise DataError() if len(key) not in [128 / 8, 192 / 8, 256 / 8]: continue if result is None: continue vectors.append( ( hexlify(key), hexlify(iv), hexlify(plaintext), hexlify(ciphertext), result, ) ) return vectors def generate_chacha_poly(filename): vectors = [] data = load_json_testvectors(filename) if not keys_in_dict(data, {"algorithm", "testGroups"}): raise DataError() if data["algorithm"] != "CHACHA20-POLY1305": raise DataError() for test_group in data["testGroups"]: if not keys_in_dict(test_group, {"tests"}): raise DataError() for test in test_group["tests"]: if not keys_in_dict( test, {"key", "iv", "aad", "msg", "ct", "tag", "result"} ): raise DataError() try: key = unhexlify(test["key"]) iv = unhexlify(test["iv"]) associated_data = unhexlify(test["aad"]) plaintext = unhexlify(test["msg"]) ciphertext = unhexlify(test["ct"]) tag = unhexlify(test["tag"]) result = parse_result(test["result"]) except Exception: raise DataError() if result is None: continue vectors.append( ( hexlify(key), hexlify(iv), hexlify(associated_data), hexlify(plaintext), hexlify(ciphertext), hexlify(tag), result, ) ) return vectors def generate_curve25519_dh(filename): vectors = [] data = load_json_testvectors(filename) if not keys_in_dict(data, {"algorithm", "testGroups"}): raise DataError() if data["algorithm"] != "XDH": raise DataError() for test_group in data["testGroups"]: if not keys_in_dict(test_group, {"tests", "curve"}): raise DataError() try: curve_name = parse_curve_name(test_group["curve"]) except Exception: raise DataError() for test in test_group["tests"]: if not keys_in_dict(test, {"public", "private", "shared", "result"}): raise DataError() try: public_key = unhexlify(test["public"]) private_key = unhexlify(test["private"]) shared = unhexlify(test["shared"]) result = parse_result(test["result"]) except Exception: raise DataError() if curve_name != "curve25519": continue if result is None: continue vectors.append( (hexlify(public_key), hexlify(private_key), hexlify(shared), result) ) return vectors def generate_ecdh(filename): vectors = [] data = load_json_testvectors(filename) if not keys_in_dict(data, {"algorithm", "testGroups"}): raise DataError() if data["algorithm"] != "ECDH": raise DataError() for test_group in data["testGroups"]: if not keys_in_dict(test_group, {"tests", "curve"}): raise DataError() try: curve_name = parse_curve_name(test_group["curve"]) except Exception: raise DataError() for test in test_group["tests"]: if not keys_in_dict(test, {"public", "private", "shared", "result"}): raise DataError() try: public_key = unhexlify(test["public"]) private_key = parse_signed_hex(test["private"]) shared = unhexlify(test["shared"]) result = parse_result(test["result"]) except Exception: raise DataError() try: private_key = parse_ecdh256_privkey(private_key) except ParseError: continue try: key_curve_name, public_key = parse_ec_pubkey(public_key) except NotSupported: continue except ParseError: continue if key_curve_name != curve_name: continue if result is None: continue vectors.append( ( curve_name, hexlify(public_key), hexlify(private_key), hexlify(shared), result, ) ) return vectors def generate_ecdsa(filename): vectors = [] data = load_json_testvectors(filename) if not keys_in_dict(data, {"algorithm", "testGroups"}): raise DataError() if data["algorithm"] != "ECDSA": raise DataError() for test_group in data["testGroups"]: if not keys_in_dict(test_group, {"tests", "keyDer", "sha"}): raise DataError() try: public_key = unhexlify(test_group["keyDer"]) except Exception: raise DataError() try: curve_name, public_key = parse_ec_pubkey(public_key) except NotSupported: continue except ParseError: continue try: hasher = parse_digest(test_group["sha"]) except NotSupported: continue for test in test_group["tests"]: if not keys_in_dict(test, {"sig", "msg", "result"}): raise DataError() try: signature = unhexlify(test["sig"]) message = unhexlify(test["msg"]) result = parse_result(test["result"]) except Exception: raise DataError() if result is None: continue try: signature = parse_ecdsa256_signature(signature) except ParseError: continue vectors.append( ( curve_name, hexlify(public_key), hasher, hexlify(message), hexlify(signature), result, ) ) return vectors def generate_eddsa(filename): vectors = [] data = load_json_testvectors(filename) if not keys_in_dict(data, {"algorithm", "testGroups"}): raise DataError() if data["algorithm"] != "EDDSA": raise DataError() for test_group in data["testGroups"]: if not keys_in_dict(test_group, {"tests", "keyDer"}): raise DataError() try: public_key = unhexlify(test_group["keyDer"]) except Exception: raise DataError() try: public_key = parse_ed_pubkey(public_key) except ParseError: continue for test in test_group["tests"]: if not keys_in_dict(test, {"sig", "msg", "result"}): raise DataError() try: signature = unhexlify(test["sig"]) message = unhexlify(test["msg"]) result = parse_result(test["result"]) except Exception: raise DataError() if result is None: continue try: signature = parse_eddsa_signature(signature) except ParseError: continue vectors.append( (hexlify(public_key), hexlify(message), hexlify(signature), result) ) return vectors dir = os.path.abspath(os.path.dirname(__file__)) lib = ctypes.cdll.LoadLibrary(os.path.join(dir, "libtrezor-crypto.so")) if not lib.zkp_context_is_initialized(): assert lib.zkp_context_init() == 0 testvectors_directory = os.path.join(dir, "wycheproof/testvectors") context_structure_length = 1024 curve25519_dh_vectors = generate_curve25519_dh("x25519_test.json") eddsa_vectors = generate_eddsa("eddsa_test.json") ecdsa_vectors = ( generate_ecdsa("ecdsa_test.json") + generate_ecdsa("ecdsa_secp256k1_sha256_test.json") + generate_ecdsa("ecdsa_secp256r1_sha256_test.json") ) ecdh_vectors = ( generate_ecdh("ecdh_test.json") + generate_ecdh("ecdh_secp256k1_test.json") + generate_ecdh("ecdh_secp256r1_test.json") ) chacha_poly_vectors = generate_chacha_poly("chacha20_poly1305_test.json") aes_vectors = generate_aes("aes_cbc_pkcs5_test.json") @pytest.mark.parametrize("public_key, message, signature, result", eddsa_vectors) def test_eddsa(public_key, message, signature, result): public_key = unhexlify(public_key) signature = unhexlify(signature) message = unhexlify(message) computed_result = ( lib.ed25519_sign_open(message, len(message), public_key, signature) == 0 ) assert result == computed_result @pytest.mark.parametrize( "curve_name, public_key, hasher, message, signature, result", ecdsa_vectors ) def test_ecdsa(curve_name, public_key, hasher, message, signature, result): curve = get_curve_by_name(curve_name) if curve is None: raise NotSupported("Curve not supported: {}".format(curve_name)) public_key = unhexlify(public_key) signature = unhexlify(signature) message = unhexlify(message) computed_result = ( lib.ecdsa_verify(curve, hasher, public_key, signature, message, len(message)) == 0 ) assert result == computed_result @pytest.mark.parametrize( "curve_name, public_key, hasher, message, signature, result", filter(lambda v: v[0] == "secp256k1", ecdsa_vectors), ) def test_ecdsa_zkp(curve_name, public_key, hasher, message, signature, result): curve = get_curve_by_name(curve_name) if curve is None: raise NotSupported("Curve not supported: {}".format(curve_name)) public_key = unhexlify(public_key) signature = unhexlify(signature) message = unhexlify(message) computed_result = ( lib.zkp_ecdsa_verify( curve, hasher, public_key, signature, message, len(message) ) == 0 ) assert result == computed_result @pytest.mark.parametrize( "public_key, private_key, shared, result", curve25519_dh_vectors ) def test_curve25519_dh(public_key, private_key, shared, result): public_key = unhexlify(public_key) private_key = unhexlify(private_key) shared = unhexlify(shared) computed_shared = bytes([0] * 32) lib.curve25519_scalarmult(computed_shared, private_key, public_key) computed_result = shared == computed_shared assert result == computed_result @pytest.mark.parametrize( "curve_name, public_key, private_key, shared, result", ecdh_vectors ) def test_ecdh(curve_name, public_key, private_key, shared, result): curve = get_curve_by_name(curve_name) if curve is None: raise NotSupported("Curve not supported: {}".format(curve_name)) public_key = unhexlify(public_key) private_key = unhexlify(private_key) shared = unhexlify(shared) computed_shared = bytes([0] * 2 * 32) lib.ecdh_multiply(curve, private_key, public_key, computed_shared) computed_shared = computed_shared[1:33] computed_result = shared == computed_shared assert result == computed_result @pytest.mark.parametrize( "key, iv, associated_data, plaintext, ciphertext, tag, result", chacha_poly_vectors ) def test_chacha_poly(key, iv, associated_data, plaintext, ciphertext, tag, result): key = unhexlify(key) iv = unhexlify(iv) associated_data = unhexlify(associated_data) plaintext = unhexlify(plaintext) ciphertext = unhexlify(ciphertext) tag = unhexlify(tag) computed_ciphertext, computed_tag = chacha_poly_encrypt( key, iv, associated_data, plaintext ) computed_result = ciphertext == computed_ciphertext and tag == computed_tag assert result == computed_result computed_plaintext = chacha_poly_decrypt(key, iv, associated_data, ciphertext, tag) computed_result = plaintext == computed_plaintext assert result == computed_result @pytest.mark.parametrize("key, iv, plaintext, ciphertext, result", aes_vectors) def test_aes(key, iv, plaintext, ciphertext, result): key = unhexlify(key) iv = unhexlify(iv) plaintext = unhexlify(plaintext) ciphertext = unhexlify(ciphertext) computed_ciphertext = aes_cbc_encrypt(key, iv, plaintext) computed_result = ciphertext == computed_ciphertext assert result == computed_result computed_plaintext = aes_cbc_decrypt(key, bytes(iv), ciphertext) computed_result = plaintext == computed_plaintext assert result == computed_result