# This file is dual licensed under the terms of the Apache License, Version # 2.0, and the BSD License. See the LICENSE file in the root of this repository # for complete details. from __future__ import absolute_import, division, print_function import itertools import os import subprocess import sys import textwrap from pkg_resources import parse_version import pytest from cryptography import x509 from cryptography.exceptions import InternalError, _Reasons from cryptography.hazmat.backends.interfaces import DHBackend, RSABackend from cryptography.hazmat.backends.openssl.backend import ( Backend, backend ) from cryptography.hazmat.backends.openssl.ec import _sn_to_elliptic_curve from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import dh, dsa, padding from cryptography.hazmat.primitives.ciphers import Cipher from cryptography.hazmat.primitives.ciphers.algorithms import AES from cryptography.hazmat.primitives.ciphers.modes import CBC from ..primitives.fixtures_rsa import RSA_KEY_2048, RSA_KEY_512 from ...doubles import ( DummyAsymmetricPadding, DummyCipherAlgorithm, DummyHashAlgorithm, DummyMode ) from ...utils import ( load_nist_vectors, load_vectors_from_file, raises_unsupported_algorithm ) from ...x509.test_x509 import _load_cert def skip_if_libre_ssl(openssl_version): if u'LibreSSL' in openssl_version: pytest.skip("LibreSSL hard-codes RAND_bytes to use arc4random.") class TestLibreSkip(object): def test_skip_no(self): assert skip_if_libre_ssl(u"OpenSSL 1.0.2h 3 May 2016") is None def test_skip_yes(self): with pytest.raises(pytest.skip.Exception): skip_if_libre_ssl(u"LibreSSL 2.1.6") class DummyMGF(object): _salt_length = 0 class TestOpenSSL(object): def test_backend_exists(self): assert backend def test_openssl_version_text(self): """ This test checks the value of OPENSSL_VERSION_TEXT. Unfortunately, this define does not appear to have a formal content definition, so for now we'll test to see if it starts with OpenSSL or LibreSSL as that appears to be true for every OpenSSL-alike. """ assert ( backend.openssl_version_text().startswith("OpenSSL") or backend.openssl_version_text().startswith("LibreSSL") ) def test_openssl_version_number(self): assert backend.openssl_version_number() > 0 def test_supports_cipher(self): assert backend.cipher_supported(None, None) is False def test_register_duplicate_cipher_adapter(self): with pytest.raises(ValueError): backend.register_cipher_adapter(AES, CBC, None) @pytest.mark.parametrize("mode", [DummyMode(), None]) def test_nonexistent_cipher(self, mode): b = Backend() b.register_cipher_adapter( DummyCipherAlgorithm, type(mode), lambda backend, cipher, mode: backend._ffi.NULL ) cipher = Cipher( DummyCipherAlgorithm(), mode, backend=b, ) with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_CIPHER): cipher.encryptor() def test_openssl_assert(self): backend.openssl_assert(True) with pytest.raises(InternalError): backend.openssl_assert(False) def test_consume_errors(self): for i in range(10): backend._lib.ERR_put_error(backend._lib.ERR_LIB_EVP, 0, 0, b"test_openssl.py", -1) assert backend._lib.ERR_peek_error() != 0 errors = backend._consume_errors() assert backend._lib.ERR_peek_error() == 0 assert len(errors) == 10 def test_ssl_ciphers_registered(self): meth = backend._lib.TLSv1_method() ctx = backend._lib.SSL_CTX_new(meth) assert ctx != backend._ffi.NULL backend._lib.SSL_CTX_free(ctx) def test_evp_ciphers_registered(self): cipher = backend._lib.EVP_get_cipherbyname(b"aes-256-cbc") assert cipher != backend._ffi.NULL def test_error_strings_loaded(self): buf = backend._ffi.new("char[]", 256) backend._lib.ERR_error_string_n(101183626, buf, len(buf)) assert b"data not multiple of block length" in backend._ffi.string(buf) def test_unknown_error_in_cipher_finalize(self): cipher = Cipher(AES(b"\0" * 16), CBC(b"\0" * 16), backend=backend) enc = cipher.encryptor() enc.update(b"\0") backend._lib.ERR_put_error(0, 0, 1, b"test_openssl.py", -1) with pytest.raises(InternalError): enc.finalize() def test_large_key_size_on_new_openssl(self): parameters = dsa.generate_parameters(2048, backend) param_num = parameters.parameter_numbers() assert param_num.p.bit_length() == 2048 parameters = dsa.generate_parameters(3072, backend) param_num = parameters.parameter_numbers() assert param_num.p.bit_length() == 3072 def test_int_to_bn(self): value = (2 ** 4242) - 4242 bn = backend._int_to_bn(value) assert bn != backend._ffi.NULL bn = backend._ffi.gc(bn, backend._lib.BN_clear_free) assert bn assert backend._bn_to_int(bn) == value def test_int_to_bn_inplace(self): value = (2 ** 4242) - 4242 bn_ptr = backend._lib.BN_new() assert bn_ptr != backend._ffi.NULL bn_ptr = backend._ffi.gc(bn_ptr, backend._lib.BN_free) bn = backend._int_to_bn(value, bn_ptr) assert bn == bn_ptr assert backend._bn_to_int(bn_ptr) == value def test_bn_to_int(self): bn = backend._int_to_bn(0) assert backend._bn_to_int(bn) == 0 class TestOpenSSLRandomEngine(object): def setup(self): # The default RAND engine is global and shared between # tests. We make sure that the default engine is osrandom # before we start each test and restore the global state to # that engine in teardown. current_default = backend._lib.ENGINE_get_default_RAND() name = backend._lib.ENGINE_get_name(current_default) assert name == backend._binding._osrandom_engine_name def teardown(self): # we need to reset state to being default. backend is a shared global # for all these tests. backend.activate_osrandom_engine() current_default = backend._lib.ENGINE_get_default_RAND() name = backend._lib.ENGINE_get_name(current_default) assert name == backend._binding._osrandom_engine_name @pytest.mark.skipif(sys.executable is None, reason="No Python interpreter available.") def test_osrandom_engine_is_default(self, tmpdir): engine_printer = textwrap.dedent( """ import sys from cryptography.hazmat.backends.openssl.backend import backend e = backend._lib.ENGINE_get_default_RAND() name = backend._lib.ENGINE_get_name(e) sys.stdout.write(backend._ffi.string(name).decode('ascii')) res = backend._lib.ENGINE_free(e) assert res == 1 """ ) engine_name = tmpdir.join('engine_name') # If we're running tests via ``python setup.py test`` in a clean # environment then all of our dependencies are going to be installed # into either the current directory or the .eggs directory. However the # subprocess won't know to activate these dependencies, so we'll get it # to do so by passing our entire sys.path into the subprocess via the # PYTHONPATH environment variable. env = os.environ.copy() env["PYTHONPATH"] = os.pathsep.join(sys.path) with engine_name.open('w') as out: subprocess.check_call( [sys.executable, "-c", engine_printer], env=env, stdout=out, stderr=subprocess.PIPE, ) osrandom_engine_name = backend._ffi.string( backend._binding._osrandom_engine_name ) assert engine_name.read().encode('ascii') == osrandom_engine_name def test_osrandom_sanity_check(self): # This test serves as a check against catastrophic failure. buf = backend._ffi.new("unsigned char[]", 500) res = backend._lib.RAND_bytes(buf, 500) assert res == 1 assert backend._ffi.buffer(buf)[:] != "\x00" * 500 def test_activate_osrandom_no_default(self): backend.activate_builtin_random() e = backend._lib.ENGINE_get_default_RAND() assert e == backend._ffi.NULL backend.activate_osrandom_engine() e = backend._lib.ENGINE_get_default_RAND() name = backend._lib.ENGINE_get_name(e) assert name == backend._binding._osrandom_engine_name res = backend._lib.ENGINE_free(e) assert res == 1 def test_activate_builtin_random(self): e = backend._lib.ENGINE_get_default_RAND() assert e != backend._ffi.NULL name = backend._lib.ENGINE_get_name(e) assert name == backend._binding._osrandom_engine_name res = backend._lib.ENGINE_free(e) assert res == 1 backend.activate_builtin_random() e = backend._lib.ENGINE_get_default_RAND() assert e == backend._ffi.NULL def test_activate_builtin_random_already_active(self): backend.activate_builtin_random() e = backend._lib.ENGINE_get_default_RAND() assert e == backend._ffi.NULL backend.activate_builtin_random() e = backend._lib.ENGINE_get_default_RAND() assert e == backend._ffi.NULL def test_osrandom_engine_implementation(self): name = backend.osrandom_engine_implementation() assert name in ['/dev/urandom', 'CryptGenRandom', 'getentropy', 'getrandom'] if sys.platform.startswith('linux'): assert name in ['getrandom', '/dev/urandom'] if sys.platform == 'darwin': # macOS 10.12+ supports getentropy if parse_version(os.uname()[2]) >= parse_version("16.0"): assert name == 'getentropy' else: assert name == '/dev/urandom' if 'bsd' in sys.platform: assert name in ['getentropy', '/dev/urandom'] if sys.platform == 'win32': assert name == 'CryptGenRandom' def test_activate_osrandom_already_default(self): e = backend._lib.ENGINE_get_default_RAND() name = backend._lib.ENGINE_get_name(e) assert name == backend._binding._osrandom_engine_name res = backend._lib.ENGINE_free(e) assert res == 1 backend.activate_osrandom_engine() e = backend._lib.ENGINE_get_default_RAND() name = backend._lib.ENGINE_get_name(e) assert name == backend._binding._osrandom_engine_name res = backend._lib.ENGINE_free(e) assert res == 1 class TestOpenSSLRSA(object): def test_generate_rsa_parameters_supported(self): assert backend.generate_rsa_parameters_supported(1, 1024) is False assert backend.generate_rsa_parameters_supported(4, 1024) is False assert backend.generate_rsa_parameters_supported(3, 1024) is True assert backend.generate_rsa_parameters_supported(3, 511) is False def test_generate_bad_public_exponent(self): with pytest.raises(ValueError): backend.generate_rsa_private_key(public_exponent=1, key_size=2048) with pytest.raises(ValueError): backend.generate_rsa_private_key(public_exponent=4, key_size=2048) def test_cant_generate_insecure_tiny_key(self): with pytest.raises(ValueError): backend.generate_rsa_private_key(public_exponent=65537, key_size=511) with pytest.raises(ValueError): backend.generate_rsa_private_key(public_exponent=65537, key_size=256) def test_rsa_padding_unsupported_pss_mgf1_hash(self): assert backend.rsa_padding_supported( padding.PSS(mgf=padding.MGF1(DummyHashAlgorithm()), salt_length=0) ) is False def test_rsa_padding_unsupported(self): assert backend.rsa_padding_supported(DummyAsymmetricPadding()) is False def test_rsa_padding_supported_pkcs1v15(self): assert backend.rsa_padding_supported(padding.PKCS1v15()) is True def test_rsa_padding_supported_pss(self): assert backend.rsa_padding_supported( padding.PSS(mgf=padding.MGF1(hashes.SHA1()), salt_length=0) ) is True def test_rsa_padding_supported_oaep(self): assert backend.rsa_padding_supported( padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA1()), algorithm=hashes.SHA1(), label=None ), ) is True @pytest.mark.skipif( backend._lib.Cryptography_HAS_RSA_OAEP_MD == 0, reason="Requires OpenSSL with rsa_oaep_md (1.0.2+)" ) def test_rsa_padding_supported_oaep_sha2_combinations(self): hashalgs = [ hashes.SHA1(), hashes.SHA224(), hashes.SHA256(), hashes.SHA384(), hashes.SHA512(), ] for mgf1alg, oaepalg in itertools.product(hashalgs, hashalgs): assert backend.rsa_padding_supported( padding.OAEP( mgf=padding.MGF1(algorithm=mgf1alg), algorithm=oaepalg, label=None ), ) is True def test_rsa_padding_unsupported_mgf(self): assert backend.rsa_padding_supported( padding.OAEP( mgf=DummyMGF(), algorithm=hashes.SHA1(), label=None ), ) is False assert backend.rsa_padding_supported( padding.PSS(mgf=DummyMGF(), salt_length=0) ) is False @pytest.mark.skipif( backend._lib.Cryptography_HAS_RSA_OAEP_MD == 1, reason="Requires OpenSSL without rsa_oaep_md (< 1.0.2)" ) def test_unsupported_mgf1_hash_algorithm_decrypt(self): private_key = RSA_KEY_512.private_key(backend) with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_PADDING): private_key.decrypt( b"0" * 64, padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA1(), label=None ) ) @pytest.mark.skipif( backend._lib.Cryptography_HAS_RSA_OAEP_MD == 1, reason="Requires OpenSSL without rsa_oaep_md (< 1.0.2)" ) def test_unsupported_oaep_hash_algorithm_decrypt(self): private_key = RSA_KEY_512.private_key(backend) with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_PADDING): private_key.decrypt( b"0" * 64, padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA1()), algorithm=hashes.SHA256(), label=None ) ) def test_unsupported_mgf1_hash_algorithm_md5_decrypt(self): private_key = RSA_KEY_512.private_key(backend) with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_PADDING): private_key.decrypt( b"0" * 64, padding.OAEP( mgf=padding.MGF1(algorithm=hashes.MD5()), algorithm=hashes.MD5(), label=None ) ) class TestOpenSSLCMAC(object): def test_unsupported_cipher(self): with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_CIPHER): backend.create_cmac_ctx(DummyCipherAlgorithm()) class TestOpenSSLSignX509Certificate(object): def test_requires_certificate_builder(self): private_key = RSA_KEY_2048.private_key(backend) with pytest.raises(TypeError): backend.create_x509_certificate( object(), private_key, DummyHashAlgorithm() ) class TestOpenSSLSignX509CertificateRevocationList(object): def test_invalid_builder(self): private_key = RSA_KEY_2048.private_key(backend) with pytest.raises(TypeError): backend.create_x509_crl(object(), private_key, hashes.SHA256()) class TestOpenSSLCreateRevokedCertificate(object): def test_invalid_builder(self): with pytest.raises(TypeError): backend.create_x509_revoked_certificate(object()) class TestOpenSSLSerializationWithOpenSSL(object): def test_pem_password_cb(self): userdata = backend._ffi.new("CRYPTOGRAPHY_PASSWORD_DATA *") pw = b"abcdefg" password = backend._ffi.new("char []", pw) userdata.password = password userdata.length = len(pw) buflen = 10 buf = backend._ffi.new("char []", buflen) res = backend._lib.Cryptography_pem_password_cb( buf, buflen, 0, userdata ) assert res == len(pw) assert userdata.called == 1 assert backend._ffi.buffer(buf, len(pw))[:] == pw assert userdata.maxsize == buflen assert userdata.error == 0 def test_pem_password_cb_no_password(self): userdata = backend._ffi.new("CRYPTOGRAPHY_PASSWORD_DATA *") buflen = 10 buf = backend._ffi.new("char []", buflen) res = backend._lib.Cryptography_pem_password_cb( buf, buflen, 0, userdata ) assert res == 0 assert userdata.error == -1 def test_unsupported_evp_pkey_type(self): key = backend._create_evp_pkey_gc() with raises_unsupported_algorithm(None): backend._evp_pkey_to_private_key(key) with raises_unsupported_algorithm(None): backend._evp_pkey_to_public_key(key) def test_very_long_pem_serialization_password(self): password = b"x" * 1024 with pytest.raises(ValueError): load_vectors_from_file( os.path.join( "asymmetric", "Traditional_OpenSSL_Serialization", "key1.pem" ), lambda pemfile: ( backend.load_pem_private_key( pemfile.read().encode(), password ) ) ) class TestOpenSSLEllipticCurve(object): def test_sn_to_elliptic_curve_not_supported(self): with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_ELLIPTIC_CURVE): _sn_to_elliptic_curve(backend, b"fake") @pytest.mark.requires_backend_interface(interface=RSABackend) class TestRSAPEMSerialization(object): def test_password_length_limit(self): password = b"x" * 1024 key = RSA_KEY_2048.private_key(backend) with pytest.raises(ValueError): key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.BestAvailableEncryption(password) ) class TestGOSTCertificate(object): def test_numeric_string_x509_name_entry(self): cert = _load_cert( os.path.join("x509", "e-trust.ru.der"), x509.load_der_x509_certificate, backend ) if backend._lib.CRYPTOGRAPHY_OPENSSL_LESS_THAN_102I: with pytest.raises(ValueError) as exc: cert.subject # We assert on the message in this case because if the certificate # fails to load it will also raise a ValueError and this test could # erroneously pass. assert str(exc.value) == "Unsupported ASN1 string type. Type: 18" else: assert cert.subject.get_attributes_for_oid( x509.ObjectIdentifier("1.2.643.3.131.1.1") )[0].value == "007710474375" @pytest.mark.skipif( backend._lib.Cryptography_HAS_EVP_PKEY_DHX == 1, reason="Requires OpenSSL without EVP_PKEY_DHX (< 1.0.2)") @pytest.mark.requires_backend_interface(interface=DHBackend) class TestOpenSSLDHSerialization(object): @pytest.mark.parametrize( "vector", load_vectors_from_file( os.path.join("asymmetric", "DH", "RFC5114.txt"), load_nist_vectors)) def test_dh_serialization_with_q_unsupported(self, backend, vector): parameters = dh.DHParameterNumbers(int(vector["p"], 16), int(vector["g"], 16), int(vector["q"], 16)) public = dh.DHPublicNumbers(int(vector["ystatcavs"], 16), parameters) private = dh.DHPrivateNumbers(int(vector["xstatcavs"], 16), public) private_key = private.private_key(backend) public_key = private_key.public_key() with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_SERIALIZATION): private_key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()) with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_SERIALIZATION): public_key.public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_SERIALIZATION): parameters.parameters(backend).parameter_bytes( serialization.Encoding.PEM, serialization.ParameterFormat.PKCS3) @pytest.mark.parametrize( ("key_path", "loader_func"), [ ( os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.pem"), serialization.load_pem_private_key, ), ( os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.der"), serialization.load_der_private_key, ) ] ) def test_private_load_dhx_unsupported(self, key_path, loader_func, backend): key_bytes = load_vectors_from_file( key_path, lambda pemfile: pemfile.read(), mode="rb" ) with pytest.raises(ValueError): loader_func(key_bytes, None, backend) @pytest.mark.parametrize( ("key_path", "loader_func"), [ ( os.path.join("asymmetric", "DH", "dhpub_rfc5114_2.pem"), serialization.load_pem_public_key, ), ( os.path.join("asymmetric", "DH", "dhpub_rfc5114_2.der"), serialization.load_der_public_key, ) ] ) def test_public_load_dhx_unsupported(self, key_path, loader_func, backend): key_bytes = load_vectors_from_file( key_path, lambda pemfile: pemfile.read(), mode="rb" ) with pytest.raises(ValueError): loader_func(key_bytes, backend)