diff --git a/tests/hwsim/test_dpp.py b/tests/hwsim/test_dpp.py index a15a8a689..9e79e4e00 100644 --- a/tests/hwsim/test_dpp.py +++ b/tests/hwsim/test_dpp.py @@ -5619,3 +5619,149 @@ def test_dpp_qr_code_auth_rand_mac_addr(dev, apdev): run_dpp_qr_code_auth_unicast(dev, apdev, None) finally: 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 = cert.get_subject() + dn.CN = "dpp-tls-test" + 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 + +def test_dpp_enterprise(dev, apdev, params): + """DPP and enterprise EAP-TLS provisioning""" + try: + dev[0].set("dpp_config_processing", "2") + run_dpp_enterprise(dev, apdev, params) + finally: + dev[0].set("dpp_config_processing", "0", allow_fail=True) + +def run_dpp_enterprise(dev, apdev, params): + if not openssl_imported: + raise HwsimSkip("OpenSSL python method not available") + check_dpp_capab(dev[0]) + check_dpp_capab(dev[1]) + + cert_file = params['prefix'] + ".cert.pem" + pkcs7_file = params['prefix'] + ".pkcs7.der" + + params = {"ssid": "dpp-ent", + "wpa": "2", + "wpa_key_mgmt": "WPA-EAP", + "rsn_pairwise": "CCMP", + "ieee8021x": "1", + "eap_server": "1", + "eap_user_file": "auth_serv/eap_user.conf", + "ca_cert": "auth_serv/ec-ca.pem", + "server_cert": "auth_serv/ec-server.pem", + "private_key": "auth_serv/ec-server.key"} + 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) + + with open("auth_serv/ec-ca.key", "rb") as f: + res = f.read() + cakey = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, res) + + conf_id = dev[1].dpp_configurator_add() + id0 = dev[0].dpp_bootstrap_gen(chan="81/1", mac=True) + uri0 = dev[0].request("DPP_BOOTSTRAP_GET_URI %d" % id0) + dev[0].dpp_listen(2412) + csrattrs = "MAsGCSqGSIb3DQEJBw==" + id1 = dev[1].dpp_auth_init(uri=uri0, configurator=conf_id, conf="sta-dot1x", + csrattrs=csrattrs, ssid="dpp-ent") + + ev = dev[1].wait_event(["DPP-CSR"], timeout=10) + if ev is None: + raise Exception("Configurator did not receive CSR") + id1_csr = int(ev.split(' ')[1].split('=')[1]) + if id1 != id1_csr: + raise Exception("Peer bootstrapping ID mismatch in CSR event") + csr = ev.split(' ')[2] + if not csr.startswith("csr="): + raise Exception("Could not parse CSR event: " + ev) + csr = csr[4:] + csr = base64.b64decode(csr.encode()) + logger.info("CSR: " + binascii.hexlify(csr).decode()) + + 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)) + subprocess.check_call(['openssl', 'crl2pkcs7', '-nocrl', + '-certfile', cert_file, + '-certfile', 'auth_serv/ec-ca.pem', + '-outform', 'DER', '-out', pkcs7_file]) + + #caCert = base64.b64encode(b"TODO").decode() + #res = dev[1].request("DPP_CA_SET peer=%d name=caCert value=%s" % (id1, caCert)) + #if "OK" not in res: + # raise Exception("Failed to set caCert") + + name = "server.w1.fi" + res = dev[1].request("DPP_CA_SET peer=%d name=trustedEapServerName value=%s" % (id1, name)) + if "OK" not in res: + raise Exception("Failed to set trustedEapServerName") + + with open(pkcs7_file, 'rb') as f: + pkcs7_der = f.read() + certbag = base64.b64encode(pkcs7_der).decode() + res = dev[1].request("DPP_CA_SET peer=%d name=certBag value=%s" % (id1, certbag)) + if "OK" not in res: + raise Exception("Failed to set certBag") + + ev = dev[1].wait_event(["DPP-CONF-SENT", "DPP-CONF-FAILED"], timeout=5) + if ev is None: + raise Exception("DPP configuration not completed (Configurator)") + if "DPP-CONF-FAILED" in ev: + raise Exception("DPP configuration did not succeed (Configurator)") + + ev = dev[0].wait_event(["DPP-CONF-RECEIVED", "DPP-CONF-FAILED"], + timeout=1) + if ev is None: + raise Exception("DPP configuration not completed (Enrollee)") + if "DPP-CONF-FAILED" in ev: + raise Exception("DPP configuration did not succeed (Enrollee)") + + ev = dev[0].wait_event(["DPP-CERTBAG"], timeout=1) + if ev is None: + raise Exception("DPP-CERTBAG not reported") + certbag = base64.b64decode(ev.split(' ')[1].encode()) + if certbag != pkcs7_der: + raise Exception("DPP-CERTBAG mismatch") + + #ev = dev[0].wait_event(["DPP-CACERT"], timeout=1) + #if ev is None: + # raise Exception("DPP-CACERT not reported") + + ev = dev[0].wait_event(["DPP-SERVER-NAME"], timeout=1) + if ev is None: + raise Exception("DPP-SERVER-NAME not reported") + if ev.split(' ')[1] != name: + raise Exception("DPP-SERVER-NAME mismatch: " + ev) + + ev = dev[0].wait_event(["DPP-NETWORK-ID"], timeout=1) + if ev is None: + raise Exception("DPP network profile not generated") + id = ev.split(' ')[1] + + dev[0].wait_connected() diff --git a/tests/hwsim/wpasupplicant.py b/tests/hwsim/wpasupplicant.py index 88e64340a..92c855227 100644 --- a/tests/hwsim/wpasupplicant.py +++ b/tests/hwsim/wpasupplicant.py @@ -1529,7 +1529,8 @@ class WpaSupplicant: extra=None, own=None, role=None, neg_freq=None, ssid=None, passphrase=None, expect_fail=False, tcp_addr=None, tcp_port=None, conn_status=False, - ssid_charset=None, nfc_uri=None, netrole=None): + ssid_charset=None, nfc_uri=None, netrole=None, + csrattrs=None): cmd = "DPP_AUTH_INIT" if peer is None: if nfc_uri: @@ -1563,6 +1564,8 @@ class WpaSupplicant: cmd += " conn_status=1" if netrole: cmd += " netrole=" + netrole + if csrattrs: + cmd += " csrattrs=" + csrattrs res = self.request(cmd) if expect_fail: if "FAIL" not in res: @@ -1570,6 +1573,7 @@ class WpaSupplicant: return if "OK" not in res: raise Exception("Failed to initiate DPP Authentication") + return int(peer) def dpp_pkex_init(self, identifier, code, role=None, key=None, curve=None, extra=None, use_id=None, allow_fail=False):