@@ -29,10 +29,11 @@ from wpasupplicant import WpaSupplicant
from test_ap_psk import check_mib, find_wpas_process, read_process_memory, verify_not_present, get_key_locations, set_test_assoc_ie
try:
- import OpenSSL
- openssl_imported = True
+ from cryptography import x509
+ from cryptography.x509.oid import NameOID
+ cryptography_imported = True
except ImportError:
- openssl_imported = False
+ cryptography_imported = False
def check_hlr_auc_gw_support():
if not os.path.exists("/tmp/hlr_auc_gw.sock"):
@@ -7372,8 +7373,8 @@ def test_eap_fast_ext_cert_check(dev, apdev):
def run_ext_cert_check(dev, apdev, net_id):
check_ext_cert_check_support(dev[0])
- if not openssl_imported:
- raise HwsimSkip("OpenSSL python method not available")
+ if not cryptography_imported:
+ raise HwsimSkip("cryptography python module not available")
params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
hapd = hostapd.add_ap(apdev[0], params)
@@ -7407,14 +7408,17 @@ def run_ext_cert_check(dev, apdev, net_id):
if 1 not in certs:
raise Exception("Server certificate issuer not received")
- cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_ASN1,
- certs[0])
- cn = cert.get_subject().commonName
+ def cert_cn(der):
+ cert = x509.load_der_x509_certificate(der)
+ cn = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
+ if not cn:
+ raise Exception("No CN in certificate")
+ return cn[0].value
+
+ cn = cert_cn(certs[0])
logger.info("Server certificate CN=" + cn)
- issuer = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_ASN1,
- certs[1])
- icn = issuer.get_subject().commonName
+ icn = cert_cn(certs[1])
logger.info("Issuer certificate CN=" + icn)
if cn != "server.w1.fi":
@@ -4,20 +4,23 @@
# This software may be distributed under the terms of the BSD license.
# See README for more details.
+import datetime
import os
try:
- import OpenSSL
- openssl_imported = True
+ from cryptography import x509
+ from cryptography.hazmat.primitives import hashes, serialization
+ from cryptography.hazmat.primitives.asymmetric import padding, rsa
+ cryptography_imported = True
except ImportError:
- openssl_imported = False
+ cryptography_imported = False
from utils import HwsimSkip
import hostapd
from test_ap_eap import check_domain_suffix_match, check_altsubject_match_support, check_domain_match
def check_cert_check_support():
- if not openssl_imported:
- raise HwsimSkip("OpenSSL python method not available")
+ if not cryptography_imported:
+ raise HwsimSkip("cryptography python module not available")
def start_hapd(apdev, server_cert="auth_serv/server.pem"):
params = {"ssid": "cert-check", "wpa": "2", "wpa_key_mgmt": "WPA-EAP",
@@ -32,47 +35,51 @@ def start_hapd(apdev, server_cert="auth_serv/server.pem"):
def load_certs():
with open("auth_serv/ca.pem", "rb") as f:
- res = f.read()
- cacert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
- res)
+ cacert = x509.load_pem_x509_certificate(f.read())
with open("auth_serv/ca-key.pem", "rb") as f:
- res = f.read()
- cakey = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, res)
+ cakey = serialization.load_pem_private_key(f.read(), None)
with open("auth_serv/server.pem", "rb") as f:
- res = f.read()
- servercert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, res)
+ servercert = x509.load_pem_x509_certificate(f.read())
return cacert, cakey, servercert
-def start_cert(servercert, cacert, cn='server.w1.fi', v3=True):
- cert = OpenSSL.crypto.X509()
- cert.set_serial_number(12345)
- cert.gmtime_adj_notBefore(-10)
- cert.gmtime_adj_notAfter(1000)
- cert.set_pubkey(servercert.get_pubkey())
- dn = cert.get_subject()
- dn.CN = cn
- cert.set_subject(dn)
- if v3:
- cert.set_version(2)
- cert.add_extensions([
- OpenSSL.crypto.X509Extension(b"basicConstraints", True,
- b"CA:FALSE"),
- OpenSSL.crypto.X509Extension(b"subjectKeyIdentifier", False,
- b"hash", subject=cert),
- OpenSSL.crypto.X509Extension(b"authorityKeyIdentifier", False,
- b"keyid:always", issuer=cacert),
- ])
- return cert
+def authority_key_identifier(cacert):
+ # Mimic OpenSSL's "keyid:always": use the issuer's subjectKeyIdentifier
+ # when present, otherwise derive the key id from its public key.
+ try:
+ ski = cacert.extensions.get_extension_for_class(
+ x509.SubjectKeyIdentifier).value
+ return x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(ski)
+ except x509.ExtensionNotFound:
+ return x509.AuthorityKeyIdentifier.from_issuer_public_key(cacert.public_key())
+
+def add_server_cert_extensions(builder, pubkey, cacert):
+ builder = builder.add_extension(
+ x509.BasicConstraints(ca=False, path_length=None), critical=True)
+ builder = builder.add_extension(
+ x509.SubjectKeyIdentifier.from_public_key(pubkey), critical=False)
+ builder = builder.add_extension(authority_key_identifier(cacert),
+ critical=False)
+ return builder
+
+def start_cert(servercert, cacert, cn='server.w1.fi'):
+ now = datetime.datetime.now(datetime.timezone.utc)
+ builder = x509.CertificateBuilder()
+ builder = builder.serial_number(12345)
+ builder = builder.not_valid_before(now - datetime.timedelta(seconds=10))
+ builder = builder.not_valid_after(now + datetime.timedelta(seconds=1000))
+ builder = builder.public_key(servercert.public_key())
+ builder = builder.subject_name(
+ x509.Name([x509.NameAttribute(x509.NameOID.COMMON_NAME, cn)]))
+ return add_server_cert_extensions(builder, servercert.public_key(), cacert)
def sign_cert(cert, cert_file, cakey, cacert):
- cert.set_issuer(cacert.get_subject())
- cert.sign(cakey, "sha256")
+ cert = cert.issuer_name(cacert.subject)
+ signed = cert.sign(private_key=cakey, algorithm=hashes.SHA256())
with open(cert_file, 'wb') as f:
- f.write(OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM,
- cert))
+ f.write(signed.public_bytes(serialization.Encoding.PEM))
def check_connect(dev, fail=False, wait_error=None, **kwargs):
dev.connect("cert-check", key_mgmt="WPA-EAP", eap="TTLS",
@@ -98,14 +105,65 @@ def check_connect(dev, fail=False, wait_error=None, **kwargs):
dev.wait_disconnected()
dev.dump_monitor()
+def der_len(n):
+ if n < 0x80:
+ return bytes([n])
+ b = n.to_bytes((n.bit_length() + 7) // 8, 'big')
+ return bytes([0x80 | len(b)]) + b
+
+def der_tlv(tag, value):
+ return bytes([tag]) + der_len(len(value)) + value
+
+def der_integer(n):
+ content = n.to_bytes(max((n.bit_length() + 7) // 8, 1), 'big')
+ if content[0] & 0x80:
+ content = b'\x00' + content
+ return der_tlv(0x02, content)
+
+# AlgorithmIdentifier for sha256WithRSAEncryption (1.2.840.113549.1.1.11).
+SHA256_RSA_ALG_ID = bytes.fromhex('300d06092a864886f70d01010b0500')
+
+def sign_v1_cert(cert_file, cn='server.w1.fi', key_file="auth_serv/server.key",
+ ca_cert="auth_serv/ca.pem", ca_key="auth_serv/ca-key.pem"):
+ # cryptography's CertificateBuilder always emits a v3 certificate and the
+ # "openssl x509 -req" is version-dependent, so do it here manually
+ with open(key_file, "rb") as f:
+ pkey = serialization.load_pem_private_key(f.read(), None)
+ with open(ca_cert, "rb") as f:
+ cacert = x509.load_pem_x509_certificate(f.read())
+ with open(ca_key, "rb") as f:
+ cakey = serialization.load_pem_private_key(f.read(), None)
+ if not isinstance(cakey, rsa.RSAPrivateKey):
+ raise Exception("sign_v1_cert() only supports an RSA ca_key")
+
+ now = datetime.datetime.now(datetime.timezone.utc)
+ not_before = now - datetime.timedelta(seconds=10)
+ not_after = now + datetime.timedelta(seconds=1000)
+ subject = x509.Name([x509.NameAttribute(x509.NameOID.COMMON_NAME, cn)])
+ pubkey_der = pkey.public_key().public_bytes(
+ serialization.Encoding.DER,
+ serialization.PublicFormat.SubjectPublicKeyInfo)
+ validity = der_tlv(0x30,
+ der_tlv(0x17, not_before.strftime('%y%m%d%H%M%SZ').encode()) +
+ der_tlv(0x17, not_after.strftime('%y%m%d%H%M%SZ').encode()))
+ tbs = der_tlv(0x30, der_integer(12345) + SHA256_RSA_ALG_ID +
+ cacert.subject.public_bytes() + validity +
+ subject.public_bytes() + pubkey_der)
+ signature = cakey.sign(tbs, padding.PKCS1v15(), hashes.SHA256())
+ cert_der = der_tlv(0x30, tbs + SHA256_RSA_ALG_ID +
+ der_tlv(0x03, b'\x00' + signature))
+
+ signed = x509.load_der_x509_certificate(cert_der)
+ if signed.version != x509.Version.v1:
+ raise Exception("sign_v1_cert() did not produce an X.509v1 certificate")
+ with open(cert_file, "wb") as f:
+ f.write(signed.public_bytes(serialization.Encoding.PEM))
+
def test_cert_check_basic(dev, apdev, params):
- """Basic test with generated X.509 server certificate"""
+ """Certificate check with a plain X.509v1 server certificate (no extensions)"""
check_cert_check_support()
cert_file = os.path.join(params['logdir'], "cert_check_basic.pem")
- cacert, cakey, servercert = load_certs()
-
- cert = start_cert(servercert, cacert, v3=False)
- sign_cert(cert, cert_file, cakey, cacert)
+ sign_v1_cert(cert_file)
hapd = start_hapd(apdev[0], server_cert=cert_file)
check_connect(dev[0])
@@ -129,10 +187,10 @@ def test_cert_check_dnsname(dev, apdev, params):
cacert, cakey, servercert = load_certs()
cert = start_cert(servercert, cacert, cn="server")
- dns = ["DNS:one.example.com", "DNS:two.example.com",
- "DNS:three.example.com"]
- cert.add_extensions([OpenSSL.crypto.X509Extension(b"subjectAltName", False,
- ",".join(dns).encode())])
+ dns = ["one.example.com", "two.example.com", "three.example.com"]
+ cert = cert.add_extension(
+ x509.SubjectAlternativeName([x509.DNSName(d) for d in dns]),
+ critical=False)
sign_cert(cert, cert_file, cakey, cacert)
hapd = start_hapd(apdev[0], server_cert=cert_file)
check_connect(dev[0])
@@ -186,10 +244,10 @@ def test_cert_check_dnsname_wildcard(dev, apdev, params):
cacert, cakey, servercert = load_certs()
cert = start_cert(servercert, cacert, cn="server")
- dns = ["DNS:*.one.example.com", "DNS:two.example.com",
- "DNS:*.three.example.com"]
- cert.add_extensions([OpenSSL.crypto.X509Extension(b"subjectAltName", False,
- ",".join(dns).encode())])
+ dns = ["*.one.example.com", "two.example.com", "*.three.example.com"]
+ cert = cert.add_extension(
+ x509.SubjectAlternativeName([x509.DNSName(d) for d in dns]),
+ critical=False)
sign_cert(cert, cert_file, cakey, cacert)
hapd = start_hapd(apdev[0], server_cert=cert_file)
check_connect(dev[0])
@@ -242,10 +300,10 @@ def test_cert_check_dnsname_alt(dev, apdev, params):
cacert, cakey, servercert = load_certs()
cert = start_cert(servercert, cacert, cn="server")
- dns = ["DNS:*.one.example.com", "DNS:two.example.com",
- "DNS:*.three.example.com"]
- cert.add_extensions([OpenSSL.crypto.X509Extension(b"subjectAltName", False,
- ",".join(dns).encode())])
+ dns = ["*.one.example.com", "two.example.com", "*.three.example.com"]
+ cert = cert.add_extension(
+ x509.SubjectAlternativeName([x509.DNSName(d) for d in dns]),
+ critical=False)
sign_cert(cert, cert_file, cakey, cacert)
hapd = start_hapd(apdev[0], server_cert=cert_file)
@@ -27,12 +27,17 @@ from hwsim import HWSimRadio
from utils import *
from wpasupplicant import WpaSupplicant
from wlantest import WlantestCapture
+from test_cert_check import add_server_cert_extensions
+import datetime
try:
- import OpenSSL
- openssl_imported = True
+ from cryptography import x509
+ from cryptography.hazmat.primitives import hashes, serialization
+ from cryptography.hazmat.primitives.asymmetric import ec
+ from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature
+ cryptography_imported = True
except ImportError:
- openssl_imported = False
+ cryptography_imported = False
def check_dpp_capab(dev, brainpool=False, min_ver=1):
if "UNKNOWN COMMAND" in dev.request("DPP_BOOTSTRAP_GET_URI 0"):
@@ -1491,39 +1496,20 @@ def test_dpp_config_error_legacy_too_short_psk(dev, apdev):
conf = '{"wi-fi_tech":"infra","discovery":{"ssid":"test"},"cred":{"akm":"psk","psk_hex":"%s"}}' % (31*"12")
run_dpp_config_error(dev, apdev, conf)
-def get_der_int_32(val):
- a, b = struct.unpack('BB', val[0:2])
- if a != 0x02:
- raise Exception("Invalid DER encoding of INTEGER")
- if b > len(val) - 2:
- raise Exception("Invalid length of INTEGER (truncated)")
- val = val[2:]
- if b == 32:
- r = val[0:32]
- elif b == 33:
- if val[0] != 0:
- raise Exception("Too large INTEGER (32)")
- r = val[1:33]
- elif b < 32:
- r = (32 - b) * b'\x00' + val[0:b]
- else:
- raise Exception("Invalid length of INTEGER (32): %d" % b)
- return r, val[b:]
+if cryptography_imported:
+ _ecdsa_hash_algs = {"sha256": hashes.SHA256(),
+ "sha384": hashes.SHA384(),
+ "sha512": hashes.SHA512()}
def ecdsa_sign(pkey, message, alg="sha256"):
- sign = OpenSSL.crypto.sign(pkey, message, alg)
+ if isinstance(message, str):
+ message = message.encode()
+ sign = pkey.sign(message, ec.ECDSA(_ecdsa_hash_algs[alg]))
logger.debug("sign=" + binascii.hexlify(sign).decode())
- a, b = struct.unpack('BB', sign[0:2])
- if a != 0x30:
- raise Exception("Invalid DER encoding of ECDSA signature")
- if b != len(sign) - 2:
- raise Exception("Invalid length of ECDSA signature")
- sign = sign[2:]
-
- r, sign = get_der_int_32(sign)
- s, sign = get_der_int_32(sign)
- if len(sign) != 0:
- raise Exception("Extra data at the end of ECDSA signature")
+ r, s = decode_dss_signature(sign)
+ size = (pkey.curve.key_size + 7) // 8
+ r = r.to_bytes(size, 'big')
+ s = s.to_bytes(size, 'big')
logger.info("r=" + binascii.hexlify(r).decode())
logger.info("s=" + binascii.hexlify(s).decode())
@@ -1541,10 +1527,9 @@ p256_pub_key_y = binascii.unhexlify("c9e75a76984a169f4dcde9746f4c2f86ed63e897d36
def run_dpp_config_connector(dev, apdev, expiry=None, payload=None,
skip_net_access_key_mismatch=True,
conf_failure=True):
- if not openssl_imported:
- raise HwsimSkip("OpenSSL python method not available")
- pkey = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM,
- p256_priv_key)
+ if not cryptography_imported:
+ raise HwsimSkip("cryptography python module not available")
+ pkey = serialization.load_pem_private_key(p256_priv_key.encode(), None)
x = base64.urlsafe_b64encode(p256_pub_key_x).decode().rstrip('=')
y = base64.urlsafe_b64encode(p256_pub_key_y).decode().rstrip('=')
@@ -7441,27 +7426,17 @@ def test_dpp_qr_code_auth_rand_mac_addr(dev, apdev):
dev[0].set("gas_rand_mac_addr", "0")
def dpp_sign_cert(cacert, cakey, csr_der):
- csr = OpenSSL.crypto.load_certificate_request(OpenSSL.crypto.FILETYPE_ASN1,
- csr_der)
- cert = OpenSSL.crypto.X509()
- cert.set_serial_number(12345)
- cert.gmtime_adj_notBefore(-10)
- cert.gmtime_adj_notAfter(100000)
- cert.set_pubkey(csr.get_pubkey())
- dn = csr.get_subject()
- cert.set_subject(dn)
- cert.set_version(2)
- cert.add_extensions([
- OpenSSL.crypto.X509Extension(b"basicConstraints", True,
- b"CA:FALSE"),
- OpenSSL.crypto.X509Extension(b"subjectKeyIdentifier", False,
- b"hash", subject=cert),
- OpenSSL.crypto.X509Extension(b"authorityKeyIdentifier", False,
- b"keyid:always", issuer=cacert),
- ])
- cert.set_issuer(cacert.get_subject())
- cert.sign(cakey, "sha256")
- return cert
+ csr = x509.load_der_x509_csr(csr_der)
+ now = datetime.datetime.now(datetime.timezone.utc)
+ builder = x509.CertificateBuilder()
+ builder = builder.subject_name(csr.subject)
+ builder = builder.issuer_name(cacert.subject)
+ builder = builder.public_key(csr.public_key())
+ builder = builder.serial_number(12345)
+ builder = builder.not_valid_before(now - datetime.timedelta(seconds=10))
+ builder = builder.not_valid_after(now + datetime.timedelta(seconds=100000))
+ builder = add_server_cert_extensions(builder, csr.public_key(), cacert)
+ return builder.sign(private_key=cakey, algorithm=hashes.SHA256())
def test_dpp_enterprise(dev, apdev, params):
"""DPP and enterprise EAP-TLS provisioning"""
@@ -7488,8 +7463,8 @@ def dpp_enterprise(dev, apdev, params, ca_delay):
dev[0].set("dpp_config_processing", "0", allow_fail=True)
def run_dpp_enterprise(dev, apdev, params, ca_delay=0):
- if not openssl_imported:
- raise HwsimSkip("OpenSSL python method not available")
+ if not cryptography_imported:
+ raise HwsimSkip("cryptography python module not available")
check_dpp_capab(dev[0])
check_dpp_capab(dev[1])
@@ -7509,13 +7484,10 @@ def run_dpp_enterprise(dev, apdev, params, ca_delay=0):
hapd = hostapd.add_ap(apdev[0], params)
with open("auth_serv/ec-ca.pem", "rb") as f:
- res = f.read()
- cacert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
- res)
+ cacert = x509.load_pem_x509_certificate(f.read())
with open("auth_serv/ec-ca.key", "rb") as f:
- res = f.read()
- cakey = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, res)
+ cakey = serialization.load_pem_private_key(f.read(), None)
conf_id = dev[1].dpp_configurator_add()
id0 = dev[0].dpp_bootstrap_gen(chan="81/1", mac=True)
@@ -7544,8 +7516,7 @@ def run_dpp_enterprise(dev, apdev, params, ca_delay=0):
cert = dpp_sign_cert(cacert, cakey, csr)
with open(cert_file, 'wb') as f:
- f.write(OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM,
- cert))
+ f.write(cert.public_bytes(serialization.Encoding.PEM))
subprocess.check_call(['openssl', 'crl2pkcs7', '-nocrl',
'-certfile', cert_file,
'-certfile', 'auth_serv/ec-ca.pem',
@@ -7641,8 +7612,8 @@ def test_dpp_enterprise_reject(dev, apdev, params):
def test_dpp_enterprise_tcp(dev, apdev, params):
"""DPP over TCP for enterprise provisioning"""
- if not openssl_imported:
- raise HwsimSkip("OpenSSL python method not available")
+ if not cryptography_imported:
+ raise HwsimSkip("cryptography python module not available")
try:
run_dpp_enterprise_tcp(dev, apdev, params)
@@ -7679,13 +7650,10 @@ def run_dpp_enterprise_tcp_end(params, dev, wt):
pkcs7_file = params['prefix'] + ".pkcs7.der"
with open("auth_serv/ec-ca.pem", "rb") as f:
- res = f.read()
- cacert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
- res)
+ cacert = x509.load_pem_x509_certificate(f.read())
with open("auth_serv/ec-ca.key", "rb") as f:
- res = f.read()
- cakey = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, res)
+ cakey = serialization.load_pem_private_key(f.read(), None)
ev = dev[1].wait_event(["DPP-CSR"], timeout=10)
if ev is None:
@@ -7700,8 +7668,7 @@ def run_dpp_enterprise_tcp_end(params, dev, wt):
cert = dpp_sign_cert(cacert, cakey, csr)
with open(cert_file, 'wb') as f:
- f.write(OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM,
- cert))
+ f.write(cert.public_bytes(serialization.Encoding.PEM))
subprocess.check_call(['openssl', 'crl2pkcs7', '-nocrl',
'-certfile', cert_file,
'-certfile', 'auth_serv/ec-ca.pem',
@@ -7729,8 +7696,8 @@ def run_dpp_enterprise_tcp_end(params, dev, wt):
def test_dpp_enterprise_tcp2(dev, apdev, params):
"""DPP over TCP for enterprise provisioning (Controller initiating)"""
- if not openssl_imported:
- raise HwsimSkip("OpenSSL python method not available")
+ if not cryptography_imported:
+ raise HwsimSkip("cryptography python module not available")
try:
run_dpp_enterprise_tcp2(dev, apdev, params)