tests: DPP enterprise provisioning
Signed-off-by: Jouni Malinen <jouni@codeaurora.org>
This commit is contained in:
Jouni Malinen
Jouni Malinen
parent
ace3723d98
commit
5abf8ad9b2
2 changed files with 151 additions and 1 deletions
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
Loading…
Reference in a new issue