tests/hwsim/test_sigma_dut.py

   1 # Test cases for sigma_dut
   2 # Copyright (c) 2017, Qualcomm Atheros, Inc.
   3 #
   4 # This software may be distributed under the terms of the BSD license.
   5 # See README for more details.
   6
   7 import logging
   8 logger = logging.getLogger()
   9 import os
  10 import socket
  11 import subprocess
  12 import threading
  13 import time
  14
  15 import hostapd
  16 from utils import HwsimSkip
  17 from hwsim import HWSimRadio
  18 from test_dpp import check_dpp_capab, update_hapd_config
  19 from test_suite_b import check_suite_b_192_capa, suite_b_as_params, suite_b_192_rsa_ap_params
  20
  21 def check_sigma_dut():
  22     if not os.path.exists("./sigma_dut"):
  23         raise HwsimSkip("sigma_dut not available")
  24
  25 def sigma_dut_cmd(cmd, port=9000, timeout=2):
  26     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM,
  27                          socket.IPPROTO_TCP)
  28     sock.settimeout(timeout)
  29     addr = ('127.0.0.1', port)
  30     sock.connect(addr)
  31     sock.send(cmd + "\r\n")
  32     try:
  33         res = sock.recv(1000)
  34         running = False
  35         done = False
  36         for line in res.splitlines():
  37             if line.startswith("status,RUNNING"):
  38                 running = True
  39             elif line.startswith("status,INVALID"):
  40                 done = True
  41             elif line.startswith("status,ERROR"):
  42                 done = True
  43             elif line.startswith("status,COMPLETE"):
  44                 done = True
  45         if running and not done:
  46             # Read the actual response
  47             res = sock.recv(1000)
  48     except:
  49         res = ''
  50         pass
  51     sock.close()
  52     res = res.rstrip()
  53     logger.debug("sigma_dut: '%s' --> '%s'" % (cmd, res))
  54     return res
  55
  56 def sigma_dut_cmd_check(cmd, port=9000, timeout=2):
  57     res = sigma_dut_cmd(cmd, port=port, timeout=timeout)
  58     if "COMPLETE" not in res:
  59         raise Exception("sigma_dut command failed: " + cmd)
  60     return res
  61
  62 def start_sigma_dut(ifname, debug=False, hostapd_logdir=None, cert_path=None):
  63     check_sigma_dut()
  64     cmd = [ './sigma_dut',
  65             '-M', ifname,
  66             '-S', ifname,
  67             '-F', '../../hostapd/hostapd',
  68             '-G',
  69             '-w', '/var/run/wpa_supplicant/',
  70             '-j', ifname ]
  71     if debug:
  72         cmd += [ '-d' ]
  73     if hostapd_logdir:
  74         cmd += [ '-H', hostapd_logdir ]
  75     if cert_path:
  76         cmd += [ '-C', cert_path ]
  77     sigma = subprocess.Popen(cmd, stdout=subprocess.PIPE,
  78                              stderr=subprocess.PIPE)
  79     for i in range(20):
  80         try:
  81             res = sigma_dut_cmd("HELLO")
  82             break
  83         except:
  84             time.sleep(0.05)
  85     return sigma
  86
  87 def stop_sigma_dut(sigma):
  88     sigma.terminate()
  89     sigma.wait()
  90     out, err = sigma.communicate()
  91     logger.debug("sigma_dut stdout: " + str(out))
  92     logger.debug("sigma_dut stderr: " + str(err))
  93
  94 def sigma_dut_wait_connected(ifname):
  95     for i in range(50):
  96         res = sigma_dut_cmd("sta_is_connected,interface," + ifname)
  97         if "connected,1" in res:
  98             break
  99         time.sleep(0.2)
 100         if i == 49:
 101             raise Exception("Connection did not complete")
 102
 103 def test_sigma_dut_basic(dev, apdev):
 104     """sigma_dut basic functionality"""
 105     sigma = start_sigma_dut(dev[0].ifname)
 106
 107     res = sigma_dut_cmd("UNKNOWN")
 108     if "status,INVALID,errorCode,Unknown command" not in res:
 109         raise Exception("Unexpected sigma_dut response to unknown command")
 110
 111     tests = [ ("ca_get_version", "status,COMPLETE,version,1.0"),
 112               ("device_get_info", "status,COMPLETE,vendor"),
 113               ("device_list_interfaces,interfaceType,foo", "status,ERROR"),
 114               ("device_list_interfaces,interfaceType,802.11",
 115                "status,COMPLETE,interfaceType,802.11,interfaceID," + dev[0].ifname) ]
 116     for cmd, response in tests:
 117         res = sigma_dut_cmd(cmd)
 118         if response not in res:
 119             raise Exception("Unexpected %s response: %s" % (cmd, res))
 120
 121     stop_sigma_dut(sigma)
 122
 123 def test_sigma_dut_open(dev, apdev):
 124     """sigma_dut controlled open network association"""
 125     try:
 126         run_sigma_dut_open(dev, apdev)
 127     finally:
 128         dev[0].set("ignore_old_scan_res", "0")
 129
 130 def run_sigma_dut_open(dev, apdev):
 131     ifname = dev[0].ifname
 132     sigma = start_sigma_dut(ifname)
 133
 134     hapd = hostapd.add_ap(apdev[0], { "ssid": "open" })
 135
 136     sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 137     sigma_dut_cmd_check("sta_set_encryption,interface,%s,ssid,%s,encpType,none" % (ifname, "open"))
 138     sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s" % (ifname, "open"))
 139     sigma_dut_wait_connected(ifname)
 140     sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 141     sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 142     sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 143
 144     stop_sigma_dut(sigma)
 145
 146 def test_sigma_dut_psk_pmf(dev, apdev):
 147     """sigma_dut controlled PSK+PMF association"""
 148     try:
 149         run_sigma_dut_psk_pmf(dev, apdev)
 150     finally:
 151         dev[0].set("ignore_old_scan_res", "0")
 152
 153 def run_sigma_dut_psk_pmf(dev, apdev):
 154     ifname = dev[0].ifname
 155     sigma = start_sigma_dut(ifname)
 156
 157     ssid = "test-pmf-required"
 158     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 159     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 160     params["ieee80211w"] = "2"
 161     hapd = hostapd.add_ap(apdev[0], params)
 162
 163     sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,PMF" % ifname)
 164     sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 165     sigma_dut_cmd_check("sta_set_psk,interface,%s,ssid,%s,passphrase,%s,encpType,aes-ccmp,keymgmttype,wpa2,PMF,Required" % (ifname, "test-pmf-required", "12345678"))
 166     sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-pmf-required"))
 167     sigma_dut_wait_connected(ifname)
 168     sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 169     sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 170     sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 171
 172     stop_sigma_dut(sigma)
 173
 174 def test_sigma_dut_psk_pmf_bip_cmac_128(dev, apdev):
 175     """sigma_dut controlled PSK+PMF association with BIP-CMAC-128"""
 176     try:
 177         run_sigma_dut_psk_pmf_cipher(dev, apdev, "BIP-CMAC-128", "AES-128-CMAC")
 178     finally:
 179         dev[0].set("ignore_old_scan_res", "0")
 180
 181 def test_sigma_dut_psk_pmf_bip_cmac_256(dev, apdev):
 182     """sigma_dut controlled PSK+PMF association with BIP-CMAC-256"""
 183     try:
 184         run_sigma_dut_psk_pmf_cipher(dev, apdev, "BIP-CMAC-256", "BIP-CMAC-256")
 185     finally:
 186         dev[0].set("ignore_old_scan_res", "0")
 187
 188 def test_sigma_dut_psk_pmf_bip_gmac_128(dev, apdev):
 189     """sigma_dut controlled PSK+PMF association with BIP-GMAC-128"""
 190     try:
 191         run_sigma_dut_psk_pmf_cipher(dev, apdev, "BIP-GMAC-128", "BIP-GMAC-128")
 192     finally:
 193         dev[0].set("ignore_old_scan_res", "0")
 194
 195 def test_sigma_dut_psk_pmf_bip_gmac_256(dev, apdev):
 196     """sigma_dut controlled PSK+PMF association with BIP-GMAC-256"""
 197     try:
 198         run_sigma_dut_psk_pmf_cipher(dev, apdev, "BIP-GMAC-256", "BIP-GMAC-256")
 199     finally:
 200         dev[0].set("ignore_old_scan_res", "0")
 201
 202 def test_sigma_dut_psk_pmf_bip_gmac_256_mismatch(dev, apdev):
 203     """sigma_dut controlled PSK+PMF association with BIP-GMAC-256 mismatch"""
 204     try:
 205         run_sigma_dut_psk_pmf_cipher(dev, apdev, "BIP-GMAC-256", "AES-128-CMAC",
 206                                      failure=True)
 207     finally:
 208         dev[0].set("ignore_old_scan_res", "0")
 209
 210 def run_sigma_dut_psk_pmf_cipher(dev, apdev, sigma_cipher, hostapd_cipher,
 211                                  failure=False):
 212     ifname = dev[0].ifname
 213     sigma = start_sigma_dut(ifname)
 214
 215     ssid = "test-pmf-required"
 216     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 217     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 218     params["ieee80211w"] = "2"
 219     params["group_mgmt_cipher"] = hostapd_cipher
 220     hapd = hostapd.add_ap(apdev[0], params)
 221
 222     sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,PMF" % ifname)
 223     sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 224     sigma_dut_cmd_check("sta_set_psk,interface,%s,ssid,%s,passphrase,%s,encpType,aes-ccmp,keymgmttype,wpa2,PMF,Required,GroupMgntCipher,%s" % (ifname, "test-pmf-required", "12345678", sigma_cipher))
 225     sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-pmf-required"))
 226     if failure:
 227         ev = dev[0].wait_event(["CTRL-EVENT-NETWORK-NOT-FOUND",
 228                                 "CTRL-EVENT-CONNECTED"], timeout=10)
 229         if ev is None:
 230             raise Exception("Network selection result not indicated")
 231         if "CTRL-EVENT-CONNECTED" in ev:
 232             raise Exception("Unexpected connection")
 233         res = sigma_dut_cmd("sta_is_connected,interface," + ifname)
 234         if "connected,1" in res:
 235             raise Exception("Connection reported")
 236     else:
 237         sigma_dut_wait_connected(ifname)
 238         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 239
 240     sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 241     sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 242
 243     stop_sigma_dut(sigma)
 244
 245 def test_sigma_dut_sae(dev, apdev):
 246     """sigma_dut controlled SAE association"""
 247     if "SAE" not in dev[0].get_capability("auth_alg"):
 248         raise HwsimSkip("SAE not supported")
 249
 250     ifname = dev[0].ifname
 251     sigma = start_sigma_dut(ifname)
 252
 253     ssid = "test-sae"
 254     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 255     params['wpa_key_mgmt'] = 'SAE'
 256     hapd = hostapd.add_ap(apdev[0], params)
 257
 258     sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
 259     sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 260     sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2" % (ifname, "test-sae", "12345678"))
 261     sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"))
 262     sigma_dut_wait_connected(ifname)
 263     sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 264     if dev[0].get_status_field('sae_group') != '19':
 265             raise Exception("Expected default SAE group not used")
 266     sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 267
 268     sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 269
 270     sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 271     sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2,ECGroupID,20" % (ifname, "test-sae", "12345678"))
 272     sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"))
 273     sigma_dut_wait_connected(ifname)
 274     sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 275     if dev[0].get_status_field('sae_group') != '20':
 276             raise Exception("Expected SAE group not used")
 277     sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 278     sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 279
 280     stop_sigma_dut(sigma)
 281
 282 def test_sigma_dut_sae_password(dev, apdev):
 283     """sigma_dut controlled SAE association and long password"""
 284     if "SAE" not in dev[0].get_capability("auth_alg"):
 285         raise HwsimSkip("SAE not supported")
 286
 287     ifname = dev[0].ifname
 288     sigma = start_sigma_dut(ifname)
 289
 290     try:
 291         ssid = "test-sae"
 292         params = hostapd.wpa2_params(ssid=ssid)
 293         params['sae_password'] = 100*'B'
 294         params['wpa_key_mgmt'] = 'SAE'
 295         hapd = hostapd.add_ap(apdev[0], params)
 296
 297         sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
 298         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 299         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2" % (ifname, "test-sae", 100*'B'))
 300         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"))
 301         sigma_dut_wait_connected(ifname)
 302         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 303         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 304         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 305     finally:
 306         stop_sigma_dut(sigma)
 307
 308 def test_sigma_dut_sta_override_rsne(dev, apdev):
 309     """sigma_dut and RSNE override on STA"""
 310     try:
 311         run_sigma_dut_sta_override_rsne(dev, apdev)
 312     finally:
 313         dev[0].set("ignore_old_scan_res", "0")
 314
 315 def run_sigma_dut_sta_override_rsne(dev, apdev):
 316     ifname = dev[0].ifname
 317     sigma = start_sigma_dut(ifname)
 318
 319     ssid = "test-psk"
 320     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 321     hapd = hostapd.add_ap(apdev[0], params)
 322
 323     sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 324
 325     tests = [ "30120100000fac040100000fac040100000fac02",
 326               "30140100000fac040100000fac040100000fac02ffff" ]
 327     for test in tests:
 328         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,type,PSK,passphrase,%s,EncpType,aes-ccmp,KeyMgmtType,wpa2" % (ifname, "test-psk", "12345678"))
 329         sigma_dut_cmd_check("dev_configure_ie,interface,%s,IE_Name,RSNE,Contents,%s" % (ifname, test))
 330         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-psk"))
 331         sigma_dut_wait_connected(ifname)
 332         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 333         dev[0].dump_monitor()
 334
 335     sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,type,PSK,passphrase,%s,EncpType,aes-ccmp,KeyMgmtType,wpa2" % (ifname, "test-psk", "12345678"))
 336     sigma_dut_cmd_check("dev_configure_ie,interface,%s,IE_Name,RSNE,Contents,300101" % ifname)
 337     sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-psk"))
 338
 339     ev = dev[0].wait_event(["CTRL-EVENT-ASSOC-REJECT"])
 340     if ev is None:
 341         raise Exception("Association rejection not reported")
 342     if "status_code=40" not in ev:
 343         raise Exception("Unexpected status code: " + ev)
 344
 345     sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 346
 347     stop_sigma_dut(sigma)
 348
 349 def test_sigma_dut_ap_psk(dev, apdev):
 350     """sigma_dut controlled AP"""
 351     with HWSimRadio() as (radio, iface):
 352         sigma = start_sigma_dut(iface)
 353         try:
 354             sigma_dut_cmd_check("ap_reset_default")
 355             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-psk,MODE,11ng")
 356             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-PSK,PSK,12345678")
 357             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 358
 359             dev[0].connect("test-psk", psk="12345678", scan_freq="2412")
 360
 361             sigma_dut_cmd_check("ap_reset_default")
 362         finally:
 363             stop_sigma_dut(sigma)
 364
 365 def test_sigma_dut_ap_pskhex(dev, apdev, params):
 366     """sigma_dut controlled AP and PSKHEX"""
 367     logdir = os.path.join(params['logdir'],
 368                           "sigma_dut_ap_pskhex.sigma-hostapd")
 369     with HWSimRadio() as (radio, iface):
 370         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
 371         try:
 372             psk = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
 373             sigma_dut_cmd_check("ap_reset_default")
 374             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-psk,MODE,11ng")
 375             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-PSK,PSKHEX," + psk)
 376             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 377
 378             dev[0].connect("test-psk", raw_psk=psk, scan_freq="2412")
 379
 380             sigma_dut_cmd_check("ap_reset_default")
 381         finally:
 382             stop_sigma_dut(sigma)
 383
 384 def test_sigma_dut_suite_b(dev, apdev, params):
 385     """sigma_dut controlled STA Suite B"""
 386     check_suite_b_192_capa(dev)
 387     logdir = params['logdir']
 388
 389     with open("auth_serv/ec2-ca.pem", "r") as f:
 390         with open(os.path.join(logdir, "suite_b_ca.pem"), "w") as f2:
 391             f2.write(f.read())
 392
 393     with open("auth_serv/ec2-user.pem", "r") as f:
 394         with open("auth_serv/ec2-user.key", "r") as f2:
 395             with open(os.path.join(logdir, "suite_b.pem"), "w") as f3:
 396                 f3.write(f.read())
 397                 f3.write(f2.read())
 398
 399     dev[0].flush_scan_cache()
 400     params = suite_b_as_params()
 401     params['ca_cert'] = 'auth_serv/ec2-ca.pem'
 402     params['server_cert'] = 'auth_serv/ec2-server.pem'
 403     params['private_key'] = 'auth_serv/ec2-server.key'
 404     params['openssl_ciphers'] = 'SUITEB192'
 405     hostapd.add_ap(apdev[1], params)
 406
 407     params = { "ssid": "test-suite-b",
 408                "wpa": "2",
 409                "wpa_key_mgmt": "WPA-EAP-SUITE-B-192",
 410                "rsn_pairwise": "GCMP-256",
 411                "group_mgmt_cipher": "BIP-GMAC-256",
 412                "ieee80211w": "2",
 413                "ieee8021x": "1",
 414                'auth_server_addr': "127.0.0.1",
 415                'auth_server_port': "18129",
 416                'auth_server_shared_secret': "radius",
 417                'nas_identifier': "nas.w1.fi" }
 418     hapd = hostapd.add_ap(apdev[0], params)
 419
 420     ifname = dev[0].ifname
 421     sigma = start_sigma_dut(ifname, cert_path=logdir)
 422
 423     sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,PMF" % ifname)
 424     sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 425     sigma_dut_cmd_check("sta_set_security,type,eaptls,interface,%s,ssid,%s,PairwiseCipher,AES-GCMP-256,GroupCipher,AES-GCMP-256,GroupMgntCipher,BIP-GMAC-256,keymgmttype,SuiteB,PMF,Required,clientCertificate,suite_b.pem,trustedRootCA,suite_b_ca.pem,CertType,ECC" % (ifname, "test-suite-b"))
 426     sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-suite-b"))
 427     sigma_dut_wait_connected(ifname)
 428     sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 429     sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 430     sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 431
 432     stop_sigma_dut(sigma)
 433
 434 def test_sigma_dut_suite_b_rsa(dev, apdev, params):
 435     """sigma_dut controlled STA Suite B (RSA)"""
 436     check_suite_b_192_capa(dev)
 437     logdir = params['logdir']
 438
 439     with open("auth_serv/rsa3072-ca.pem", "r") as f:
 440         with open(os.path.join(logdir, "suite_b_ca_rsa.pem"), "w") as f2:
 441             f2.write(f.read())
 442
 443     with open("auth_serv/rsa3072-user.pem", "r") as f:
 444         with open("auth_serv/rsa3072-user.key", "r") as f2:
 445             with open(os.path.join(logdir, "suite_b_rsa.pem"), "w") as f3:
 446                 f3.write(f.read())
 447                 f3.write(f2.read())
 448
 449     dev[0].flush_scan_cache()
 450     params = suite_b_192_rsa_ap_params()
 451     hapd = hostapd.add_ap(apdev[0], params)
 452
 453     ifname = dev[0].ifname
 454     sigma = start_sigma_dut(ifname, cert_path=logdir)
 455
 456     cmd = "sta_set_security,type,eaptls,interface,%s,ssid,%s,PairwiseCipher,AES-GCMP-256,GroupCipher,AES-GCMP-256,GroupMgntCipher,BIP-GMAC-256,keymgmttype,SuiteB,PMF,Required,clientCertificate,suite_b_rsa.pem,trustedRootCA,suite_b_ca_rsa.pem,CertType,RSA" % (ifname, "test-suite-b")
 457
 458     tests = [ "",
 459               ",TLSCipher,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384",
 460               ",TLSCipher,TLS_DHE_RSA_WITH_AES_256_GCM_SHA384" ]
 461     for extra in tests:
 462         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,PMF" % ifname)
 463         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 464         sigma_dut_cmd_check(cmd + extra)
 465         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-suite-b"))
 466         sigma_dut_wait_connected(ifname)
 467         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 468         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 469         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 470
 471     stop_sigma_dut(sigma)
 472
 473 def test_sigma_dut_ap_suite_b(dev, apdev, params):
 474     """sigma_dut controlled AP Suite B"""
 475     check_suite_b_192_capa(dev)
 476     logdir = os.path.join(params['logdir'],
 477                           "sigma_dut_ap_suite_b.sigma-hostapd")
 478     params = suite_b_as_params()
 479     params['ca_cert'] = 'auth_serv/ec2-ca.pem'
 480     params['server_cert'] = 'auth_serv/ec2-server.pem'
 481     params['private_key'] = 'auth_serv/ec2-server.key'
 482     params['openssl_ciphers'] = 'SUITEB192'
 483     hostapd.add_ap(apdev[1], params)
 484     with HWSimRadio() as (radio, iface):
 485         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
 486         try:
 487             sigma_dut_cmd_check("ap_reset_default")
 488             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-suite-b,MODE,11ng")
 489             sigma_dut_cmd_check("ap_set_radius,NAME,AP,IPADDR,127.0.0.1,PORT,18129,PASSWORD,radius")
 490             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,SuiteB,PMF,Required")
 491             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 492
 493             dev[0].connect("test-suite-b", key_mgmt="WPA-EAP-SUITE-B-192",
 494                            ieee80211w="2",
 495                            openssl_ciphers="SUITEB192",
 496                            eap="TLS", identity="tls user",
 497                            ca_cert="auth_serv/ec2-ca.pem",
 498                            client_cert="auth_serv/ec2-user.pem",
 499                            private_key="auth_serv/ec2-user.key",
 500                            pairwise="GCMP-256", group="GCMP-256",
 501                            scan_freq="2412")
 502
 503             sigma_dut_cmd_check("ap_reset_default")
 504         finally:
 505             stop_sigma_dut(sigma)
 506
 507 def test_sigma_dut_ap_cipher_gcmp_128(dev, apdev, params):
 508     """sigma_dut controlled AP with GCMP-128/BIP-GMAC-128 cipher"""
 509     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-GCMP-128", "BIP-GMAC-128",
 510                             "GCMP")
 511
 512 def test_sigma_dut_ap_cipher_gcmp_256(dev, apdev, params):
 513     """sigma_dut controlled AP with GCMP-256/BIP-GMAC-256 cipher"""
 514     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-GCMP-256", "BIP-GMAC-256",
 515                             "GCMP-256")
 516
 517 def test_sigma_dut_ap_cipher_ccmp_128(dev, apdev, params):
 518     """sigma_dut controlled AP with CCMP-128/BIP-CMAC-128 cipher"""
 519     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-CCMP-128", "BIP-CMAC-128",
 520                             "CCMP")
 521
 522 def test_sigma_dut_ap_cipher_ccmp_256(dev, apdev, params):
 523     """sigma_dut controlled AP with CCMP-256/BIP-CMAC-256 cipher"""
 524     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-CCMP-256", "BIP-CMAC-256",
 525                             "CCMP-256")
 526
 527 def test_sigma_dut_ap_cipher_ccmp_gcmp_1(dev, apdev, params):
 528     """sigma_dut controlled AP with CCMP-128+GCMP-256 ciphers (1)"""
 529     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-CCMP-128 AES-GCMP-256",
 530                             "BIP-GMAC-256", "CCMP")
 531
 532 def test_sigma_dut_ap_cipher_ccmp_gcmp_2(dev, apdev, params):
 533     """sigma_dut controlled AP with CCMP-128+GCMP-256 ciphers (2)"""
 534     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-CCMP-128 AES-GCMP-256",
 535                             "BIP-GMAC-256", "GCMP-256", "CCMP")
 536
 537 def test_sigma_dut_ap_cipher_gcmp_256_group_ccmp(dev, apdev, params):
 538     """sigma_dut controlled AP with GCMP-256/CCMP/BIP-GMAC-256 cipher"""
 539     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-GCMP-256", "BIP-GMAC-256",
 540                             "GCMP-256", "CCMP", "AES-CCMP-128")
 541
 542 def run_sigma_dut_ap_cipher(dev, apdev, params, ap_pairwise, ap_group_mgmt,
 543                             sta_cipher, sta_cipher_group=None, ap_group=None):
 544     check_suite_b_192_capa(dev)
 545     logdir = os.path.join(params['logdir'],
 546                           "sigma_dut_ap_cipher.sigma-hostapd")
 547     params = suite_b_as_params()
 548     params['ca_cert'] = 'auth_serv/ec2-ca.pem'
 549     params['server_cert'] = 'auth_serv/ec2-server.pem'
 550     params['private_key'] = 'auth_serv/ec2-server.key'
 551     params['openssl_ciphers'] = 'SUITEB192'
 552     hostapd.add_ap(apdev[1], params)
 553     with HWSimRadio() as (radio, iface):
 554         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
 555         try:
 556             sigma_dut_cmd_check("ap_reset_default")
 557             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-suite-b,MODE,11ng")
 558             sigma_dut_cmd_check("ap_set_radius,NAME,AP,IPADDR,127.0.0.1,PORT,18129,PASSWORD,radius")
 559             cmd = "ap_set_security,NAME,AP,KEYMGNT,SuiteB,PMF,Required,PairwiseCipher,%s,GroupMgntCipher,%s" % (ap_pairwise, ap_group_mgmt)
 560             if ap_group:
 561                 cmd += ",GroupCipher,%s" % ap_group
 562             sigma_dut_cmd_check(cmd)
 563             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 564
 565             if sta_cipher_group is None:
 566                 sta_cipher_group = sta_cipher
 567             dev[0].connect("test-suite-b", key_mgmt="WPA-EAP-SUITE-B-192",
 568                            ieee80211w="2",
 569                            openssl_ciphers="SUITEB192",
 570                            eap="TLS", identity="tls user",
 571                            ca_cert="auth_serv/ec2-ca.pem",
 572                            client_cert="auth_serv/ec2-user.pem",
 573                            private_key="auth_serv/ec2-user.key",
 574                            pairwise=sta_cipher, group=sta_cipher_group,
 575                            scan_freq="2412")
 576
 577             sigma_dut_cmd_check("ap_reset_default")
 578         finally:
 579             stop_sigma_dut(sigma)
 580
 581 def test_sigma_dut_ap_override_rsne(dev, apdev):
 582     """sigma_dut controlled AP overriding RSNE"""
 583     with HWSimRadio() as (radio, iface):
 584         sigma = start_sigma_dut(iface)
 585         try:
 586             sigma_dut_cmd_check("ap_reset_default")
 587             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-psk,MODE,11ng")
 588             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-PSK,PSK,12345678")
 589             sigma_dut_cmd_check("dev_configure_ie,NAME,AP,interface,%s,IE_Name,RSNE,Contents,30180100000fac040200ffffffff000fac040100000fac020c00" % iface)
 590             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 591
 592             dev[0].connect("test-psk", psk="12345678", scan_freq="2412")
 593
 594             sigma_dut_cmd_check("ap_reset_default")
 595         finally:
 596             stop_sigma_dut(sigma)
 597
 598 def test_sigma_dut_ap_sae(dev, apdev):
 599     """sigma_dut controlled AP with SAE"""
 600     with HWSimRadio() as (radio, iface):
 601         sigma = start_sigma_dut(iface)
 602         try:
 603             sigma_dut_cmd_check("ap_reset_default")
 604             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
 605             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK,12345678")
 606             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 607
 608             dev[0].request("SET sae_groups ")
 609             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
 610                            scan_freq="2412")
 611             if dev[0].get_status_field('sae_group') != '19':
 612                 raise Exception("Expected default SAE group not used")
 613
 614             sigma_dut_cmd_check("ap_reset_default")
 615         finally:
 616             stop_sigma_dut(sigma)
 617
 618 def test_sigma_dut_ap_sae_password(dev, apdev):
 619     """sigma_dut controlled AP with SAE and long password"""
 620     with HWSimRadio() as (radio, iface):
 621         sigma = start_sigma_dut(iface)
 622         try:
 623             sigma_dut_cmd_check("ap_reset_default")
 624             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
 625             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK," + 100*'C')
 626             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 627
 628             dev[0].request("SET sae_groups ")
 629             dev[0].connect("test-sae", key_mgmt="SAE", sae_password=100*'C',
 630                            scan_freq="2412")
 631             if dev[0].get_status_field('sae_group') != '19':
 632                 raise Exception("Expected default SAE group not used")
 633
 634             sigma_dut_cmd_check("ap_reset_default")
 635         finally:
 636             stop_sigma_dut(sigma)
 637
 638 def test_sigma_dut_ap_sae_group(dev, apdev):
 639     """sigma_dut controlled AP with SAE and specific group"""
 640     with HWSimRadio() as (radio, iface):
 641         sigma = start_sigma_dut(iface)
 642         try:
 643             sigma_dut_cmd_check("ap_reset_default")
 644             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
 645             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK,12345678,ECGroupID,20")
 646             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 647
 648             dev[0].request("SET sae_groups ")
 649             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
 650                            scan_freq="2412")
 651             if dev[0].get_status_field('sae_group') != '20':
 652                 raise Exception("Expected SAE group not used")
 653
 654             sigma_dut_cmd_check("ap_reset_default")
 655         finally:
 656             stop_sigma_dut(sigma)
 657
 658 def test_sigma_dut_ap_psk_sae(dev, apdev):
 659     """sigma_dut controlled AP with PSK+SAE"""
 660     with HWSimRadio() as (radio, iface):
 661         sigma = start_sigma_dut(iface)
 662         try:
 663             sigma_dut_cmd_check("ap_reset_default")
 664             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
 665             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-PSK-SAE,PSK,12345678")
 666             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 667
 668             dev[0].request("SET sae_groups ")
 669             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
 670                            scan_freq="2412")
 671             dev[1].connect("test-sae", psk="12345678", scan_freq="2412")
 672
 673             sigma_dut_cmd_check("ap_reset_default")
 674         finally:
 675             stop_sigma_dut(sigma)
 676
 677 def test_sigma_dut_owe(dev, apdev):
 678     """sigma_dut controlled OWE station"""
 679     try:
 680         run_sigma_dut_owe(dev, apdev)
 681     finally:
 682         dev[0].set("ignore_old_scan_res", "0")
 683
 684 def run_sigma_dut_owe(dev, apdev):
 685     if "OWE" not in dev[0].get_capability("key_mgmt"):
 686         raise HwsimSkip("OWE not supported")
 687
 688     ifname = dev[0].ifname
 689     sigma = start_sigma_dut(ifname)
 690
 691     try:
 692         params = { "ssid": "owe",
 693                    "wpa": "2",
 694                    "wpa_key_mgmt": "OWE",
 695                    "rsn_pairwise": "CCMP" }
 696         hapd = hostapd.add_ap(apdev[0], params)
 697         bssid = hapd.own_addr()
 698
 699         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
 700         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 701         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,owe,Type,OWE" % ifname)
 702         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,owe,channel,1" % ifname)
 703         sigma_dut_wait_connected(ifname)
 704         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 705
 706         dev[0].dump_monitor()
 707         sigma_dut_cmd("sta_reassoc,interface,%s,Channel,1,bssid,%s" % (ifname, bssid))
 708         dev[0].wait_connected()
 709         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 710         dev[0].wait_disconnected()
 711         dev[0].dump_monitor()
 712
 713         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
 714         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 715         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,owe,Type,OWE,ECGroupID,20" % ifname)
 716         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,owe,channel,1" % ifname)
 717         sigma_dut_wait_connected(ifname)
 718         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 719         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 720         dev[0].wait_disconnected()
 721         dev[0].dump_monitor()
 722
 723         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
 724         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 725         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,owe,Type,OWE,ECGroupID,0" % ifname)
 726         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,owe,channel,1" % ifname)
 727         ev = dev[0].wait_event(["CTRL-EVENT-ASSOC-REJECT"], timeout=10)
 728         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 729         if ev is None:
 730             raise Exception("Association not rejected")
 731         if "status_code=77" not in ev:
 732             raise Exception("Unexpected rejection reason: " + ev)
 733
 734         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 735     finally:
 736         stop_sigma_dut(sigma)
 737
 738 def test_sigma_dut_ap_owe(dev, apdev):
 739     """sigma_dut controlled AP with OWE"""
 740     if "OWE" not in dev[0].get_capability("key_mgmt"):
 741         raise HwsimSkip("OWE not supported")
 742     with HWSimRadio() as (radio, iface):
 743         sigma = start_sigma_dut(iface)
 744         try:
 745             sigma_dut_cmd_check("ap_reset_default,NAME,AP,Program,WPA3")
 746             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,owe,MODE,11ng")
 747             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,OWE")
 748             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 749
 750             dev[0].connect("owe", key_mgmt="OWE", scan_freq="2412")
 751
 752             sigma_dut_cmd_check("ap_reset_default")
 753         finally:
 754             stop_sigma_dut(sigma)
 755
 756 def test_sigma_dut_ap_owe_ecgroupid(dev, apdev):
 757     """sigma_dut controlled AP with OWE and ECGroupID"""
 758     if "OWE" not in dev[0].get_capability("key_mgmt"):
 759         raise HwsimSkip("OWE not supported")
 760     with HWSimRadio() as (radio, iface):
 761         sigma = start_sigma_dut(iface)
 762         try:
 763             sigma_dut_cmd_check("ap_reset_default,NAME,AP,Program,WPA3")
 764             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,owe,MODE,11ng")
 765             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,OWE,ECGroupID,20 21,PMF,Required")
 766             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 767
 768             dev[0].connect("owe", key_mgmt="OWE", ieee80211w="2",
 769                            owe_group="20", scan_freq="2412")
 770             dev[0].request("REMOVE_NETWORK all")
 771             dev[0].wait_disconnected()
 772
 773             dev[0].connect("owe", key_mgmt="OWE", ieee80211w="2",
 774                            owe_group="21", scan_freq="2412")
 775             dev[0].request("REMOVE_NETWORK all")
 776             dev[0].wait_disconnected()
 777
 778             dev[0].connect("owe", key_mgmt="OWE", ieee80211w="2",
 779                            owe_group="19", scan_freq="2412", wait_connect=False)
 780             ev = dev[0].wait_event(["CTRL-EVENT-ASSOC-REJECT"], timeout=10)
 781             dev[0].request("DISCONNECT")
 782             if ev is None:
 783                 raise Exception("Association not rejected")
 784             if "status_code=77" not in ev:
 785                 raise Exception("Unexpected rejection reason: " + ev)
 786             dev[0].dump_monitor()
 787
 788             sigma_dut_cmd_check("ap_reset_default")
 789         finally:
 790             stop_sigma_dut(sigma)
 791
 792 def test_sigma_dut_ap_owe_transition_mode(dev, apdev, params):
 793     """sigma_dut controlled AP with OWE and transition mode"""
 794     if "OWE" not in dev[0].get_capability("key_mgmt"):
 795         raise HwsimSkip("OWE not supported")
 796     logdir = os.path.join(params['logdir'],
 797                           "sigma_dut_ap_owe_transition_mode.sigma-hostapd")
 798     with HWSimRadio() as (radio, iface):
 799         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
 800         try:
 801             sigma_dut_cmd_check("ap_reset_default,NAME,AP,Program,WPA3")
 802             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,WLAN_TAG,1,CHANNEL,1,SSID,owe,MODE,11ng")
 803             sigma_dut_cmd_check("ap_set_security,NAME,AP,WLAN_TAG,1,KEYMGNT,OWE")
 804             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,WLAN_TAG,2,CHANNEL,1,SSID,owe,MODE,11ng")
 805             sigma_dut_cmd_check("ap_set_security,NAME,AP,WLAN_TAG,2,KEYMGNT,NONE")
 806             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 807
 808             res1 = sigma_dut_cmd_check("ap_get_mac_address,NAME,AP,WLAN_TAG,1,Interface,24G")
 809             res2 = sigma_dut_cmd_check("ap_get_mac_address,NAME,AP,WLAN_TAG,2,Interface,24G")
 810
 811             dev[0].connect("owe", key_mgmt="OWE", scan_freq="2412")
 812             dev[1].connect("owe", key_mgmt="NONE", scan_freq="2412")
 813             if dev[0].get_status_field('bssid') not in res1:
 814                 raise Exception("Unexpected ap_get_mac_address WLAN_TAG,1: " + res1)
 815             if dev[1].get_status_field('bssid') not in res2:
 816                 raise Exception("Unexpected ap_get_mac_address WLAN_TAG,2: " + res2)
 817
 818             sigma_dut_cmd_check("ap_reset_default")
 819         finally:
 820             stop_sigma_dut(sigma)
 821
 822 def dpp_init_enrollee(dev, id1):
 823     logger.info("Starting DPP initiator/enrollee in a thread")
 824     time.sleep(1)
 825     cmd = "DPP_AUTH_INIT peer=%d role=enrollee" % id1
 826     if "OK" not in dev.request(cmd):
 827         raise Exception("Failed to initiate DPP Authentication")
 828     ev = dev.wait_event(["DPP-CONF-RECEIVED"], timeout=5)
 829     if ev is None:
 830         raise Exception("DPP configuration not completed (Enrollee)")
 831     logger.info("DPP initiator/enrollee done")
 832
 833 def test_sigma_dut_dpp_qr_resp_1(dev, apdev):
 834     """sigma_dut DPP/QR responder (conf index 1)"""
 835     run_sigma_dut_dpp_qr_resp(dev, apdev, 1)
 836
 837 def test_sigma_dut_dpp_qr_resp_2(dev, apdev):
 838     """sigma_dut DPP/QR responder (conf index 2)"""
 839     run_sigma_dut_dpp_qr_resp(dev, apdev, 2)
 840
 841 def test_sigma_dut_dpp_qr_resp_3(dev, apdev):
 842     """sigma_dut DPP/QR responder (conf index 3)"""
 843     run_sigma_dut_dpp_qr_resp(dev, apdev, 3)
 844
 845 def test_sigma_dut_dpp_qr_resp_4(dev, apdev):
 846     """sigma_dut DPP/QR responder (conf index 4)"""
 847     run_sigma_dut_dpp_qr_resp(dev, apdev, 4)
 848
 849 def test_sigma_dut_dpp_qr_resp_chan_list(dev, apdev):
 850     """sigma_dut DPP/QR responder (channel list override)"""
 851     run_sigma_dut_dpp_qr_resp(dev, apdev, 1, chan_list='81/2 81/6 81/1',
 852                               listen_chan=2)
 853
 854 def run_sigma_dut_dpp_qr_resp(dev, apdev, conf_idx, chan_list=None,
 855                               listen_chan=None):
 856     check_dpp_capab(dev[0])
 857     check_dpp_capab(dev[1])
 858     sigma = start_sigma_dut(dev[0].ifname)
 859     try:
 860         cmd = "dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR"
 861         if chan_list:
 862             cmd += ",DPPChannelList," + chan_list
 863         res = sigma_dut_cmd(cmd)
 864         if "status,COMPLETE" not in res:
 865             raise Exception("dev_exec_action did not succeed: " + res)
 866         hex = res.split(',')[3]
 867         uri = hex.decode('hex')
 868         logger.info("URI from sigma_dut: " + uri)
 869
 870         res = dev[1].request("DPP_QR_CODE " + uri)
 871         if "FAIL" in res:
 872             raise Exception("Failed to parse QR Code URI")
 873         id1 = int(res)
 874
 875         t = threading.Thread(target=dpp_init_enrollee, args=(dev[1], id1))
 876         t.start()
 877         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPConfIndex,%d,DPPAuthDirection,Single,DPPProvisioningRole,Configurator,DPPConfEnrolleeRole,STA,DPPSigningKeyECC,P-256,DPPBS,QR,DPPTimeout,6" % conf_idx
 878         if listen_chan:
 879             cmd += ",DPPListenChannel," + str(listen_chan)
 880         res = sigma_dut_cmd(cmd, timeout=10)
 881         t.join()
 882         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
 883             raise Exception("Unexpected result: " + res)
 884     finally:
 885         stop_sigma_dut(sigma)
 886
 887 def test_sigma_dut_dpp_qr_init_enrollee(dev, apdev):
 888     """sigma_dut DPP/QR initiator as Enrollee"""
 889     check_dpp_capab(dev[0])
 890     check_dpp_capab(dev[1])
 891
 892     csign = "30770201010420768240a3fc89d6662d9782f120527fe7fb9edc6366ab0b9c7dde96125cfd250fa00a06082a8648ce3d030107a144034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
 893     csign_pub = "3059301306072a8648ce3d020106082a8648ce3d030107034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
 894     ap_connector = "eyJ0eXAiOiJkcHBDb24iLCJraWQiOiJwYWtZbXVzd1dCdWpSYTl5OEsweDViaTVrT3VNT3dzZHRlaml2UG55ZHZzIiwiYWxnIjoiRVMyNTYifQ.eyJncm91cHMiOlt7Imdyb3VwSWQiOiIqIiwibmV0Um9sZSI6ImFwIn1dLCJuZXRBY2Nlc3NLZXkiOnsia3R5IjoiRUMiLCJjcnYiOiJQLTI1NiIsIngiOiIybU5vNXZuRkI5bEw3d1VWb1hJbGVPYzBNSEE1QXZKbnpwZXZULVVTYzVNIiwieSI6IlhzS3dqVHJlLTg5WWdpU3pKaG9CN1haeUttTU05OTl3V2ZaSVl0bi01Q3MifX0.XhjFpZgcSa7G2lHy0OCYTvaZFRo5Hyx6b7g7oYyusLC7C_73AJ4_BxEZQVYJXAtDuGvb3dXSkHEKxREP9Q6Qeg"
 895     ap_netaccesskey = "30770201010420ceba752db2ad5200fa7bc565b9c05c69b7eb006751b0b329b0279de1c19ca67ca00a06082a8648ce3d030107a14403420004da6368e6f9c507d94bef0515a1722578e73430703902f267ce97af4fe51273935ec2b08d3adefbcf588224b3261a01ed76722a630cf7df7059f64862d9fee42b"
 896
 897     params = { "ssid": "DPPNET01",
 898                "wpa": "2",
 899                "wpa_key_mgmt": "DPP",
 900                "rsn_pairwise": "CCMP",
 901                "dpp_connector": ap_connector,
 902                "dpp_csign": csign_pub,
 903                "dpp_netaccesskey": ap_netaccesskey }
 904     try:
 905         hapd = hostapd.add_ap(apdev[0], params)
 906     except:
 907         raise HwsimSkip("DPP not supported")
 908
 909     sigma = start_sigma_dut(dev[0].ifname)
 910     try:
 911         dev[0].set("dpp_config_processing", "2")
 912
 913         cmd = "DPP_CONFIGURATOR_ADD key=" + csign
 914         res = dev[1].request(cmd);
 915         if "FAIL" in res:
 916             raise Exception("Failed to add configurator")
 917         conf_id = int(res)
 918
 919         addr = dev[1].own_addr().replace(':', '')
 920         cmd = "DPP_BOOTSTRAP_GEN type=qrcode chan=81/6 mac=" + addr
 921         res = dev[1].request(cmd)
 922         if "FAIL" in res:
 923             raise Exception("Failed to generate bootstrapping info")
 924         id0 = int(res)
 925         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
 926
 927         dev[1].set("dpp_configurator_params",
 928                    " conf=sta-dpp ssid=%s configurator=%d" % ("DPPNET01".encode("hex"), conf_id));
 929         cmd = "DPP_LISTEN 2437 role=configurator"
 930         if "OK" not in dev[1].request(cmd):
 931             raise Exception("Failed to start listen operation")
 932
 933         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % uri0.encode('hex'))
 934         if "status,COMPLETE" not in res:
 935             raise Exception("dev_exec_action did not succeed: " + res)
 936
 937         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Single,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,6,DPPWaitForConnect,Yes", timeout=10)
 938         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,OK,NetworkConnectResult,OK" not in res:
 939             raise Exception("Unexpected result: " + res)
 940     finally:
 941         dev[0].set("dpp_config_processing", "0")
 942         stop_sigma_dut(sigma)
 943
 944 def test_sigma_dut_dpp_qr_mutual_init_enrollee(dev, apdev):
 945     """sigma_dut DPP/QR (mutual) initiator as Enrollee"""
 946     check_dpp_capab(dev[0])
 947     check_dpp_capab(dev[1])
 948
 949     csign = "30770201010420768240a3fc89d6662d9782f120527fe7fb9edc6366ab0b9c7dde96125cfd250fa00a06082a8648ce3d030107a144034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
 950     csign_pub = "3059301306072a8648ce3d020106082a8648ce3d030107034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
 951     ap_connector = "eyJ0eXAiOiJkcHBDb24iLCJraWQiOiJwYWtZbXVzd1dCdWpSYTl5OEsweDViaTVrT3VNT3dzZHRlaml2UG55ZHZzIiwiYWxnIjoiRVMyNTYifQ.eyJncm91cHMiOlt7Imdyb3VwSWQiOiIqIiwibmV0Um9sZSI6ImFwIn1dLCJuZXRBY2Nlc3NLZXkiOnsia3R5IjoiRUMiLCJjcnYiOiJQLTI1NiIsIngiOiIybU5vNXZuRkI5bEw3d1VWb1hJbGVPYzBNSEE1QXZKbnpwZXZULVVTYzVNIiwieSI6IlhzS3dqVHJlLTg5WWdpU3pKaG9CN1haeUttTU05OTl3V2ZaSVl0bi01Q3MifX0.XhjFpZgcSa7G2lHy0OCYTvaZFRo5Hyx6b7g7oYyusLC7C_73AJ4_BxEZQVYJXAtDuGvb3dXSkHEKxREP9Q6Qeg"
 952     ap_netaccesskey = "30770201010420ceba752db2ad5200fa7bc565b9c05c69b7eb006751b0b329b0279de1c19ca67ca00a06082a8648ce3d030107a14403420004da6368e6f9c507d94bef0515a1722578e73430703902f267ce97af4fe51273935ec2b08d3adefbcf588224b3261a01ed76722a630cf7df7059f64862d9fee42b"
 953
 954     params = { "ssid": "DPPNET01",
 955                "wpa": "2",
 956                "wpa_key_mgmt": "DPP",
 957                "rsn_pairwise": "CCMP",
 958                "dpp_connector": ap_connector,
 959                "dpp_csign": csign_pub,
 960                "dpp_netaccesskey": ap_netaccesskey }
 961     try:
 962         hapd = hostapd.add_ap(apdev[0], params)
 963     except:
 964         raise HwsimSkip("DPP not supported")
 965
 966     sigma = start_sigma_dut(dev[0].ifname)
 967     try:
 968         dev[0].set("dpp_config_processing", "2")
 969
 970         cmd = "DPP_CONFIGURATOR_ADD key=" + csign
 971         res = dev[1].request(cmd);
 972         if "FAIL" in res:
 973             raise Exception("Failed to add configurator")
 974         conf_id = int(res)
 975
 976         addr = dev[1].own_addr().replace(':', '')
 977         cmd = "DPP_BOOTSTRAP_GEN type=qrcode chan=81/6 mac=" + addr
 978         res = dev[1].request(cmd)
 979         if "FAIL" in res:
 980             raise Exception("Failed to generate bootstrapping info")
 981         id0 = int(res)
 982         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
 983
 984         dev[1].set("dpp_configurator_params",
 985                    " conf=sta-dpp ssid=%s configurator=%d" % ("DPPNET01".encode("hex"), conf_id));
 986         cmd = "DPP_LISTEN 2437 role=configurator qr=mutual"
 987         if "OK" not in dev[1].request(cmd):
 988             raise Exception("Failed to start listen operation")
 989
 990         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
 991         if "status,COMPLETE" not in res:
 992             raise Exception("dev_exec_action did not succeed: " + res)
 993         hex = res.split(',')[3]
 994         uri = hex.decode('hex')
 995         logger.info("URI from sigma_dut: " + uri)
 996
 997         res = dev[1].request("DPP_QR_CODE " + uri)
 998         if "FAIL" in res:
 999             raise Exception("Failed to parse QR Code URI")
1000         id1 = int(res)
1001
1002         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % uri0.encode('hex'))
1003         if "status,COMPLETE" not in res:
1004             raise Exception("dev_exec_action did not succeed: " + res)
1005
1006         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Mutual,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,6,DPPWaitForConnect,Yes", timeout=10)
1007         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,OK,NetworkConnectResult,OK" not in res:
1008             raise Exception("Unexpected result: " + res)
1009     finally:
1010         dev[0].set("dpp_config_processing", "0")
1011         stop_sigma_dut(sigma)
1012
1013 def dpp_init_conf_mutual(dev, id1, conf_id, own_id=None):
1014     time.sleep(1)
1015     logger.info("Starting DPP initiator/configurator in a thread")
1016     cmd = "DPP_AUTH_INIT peer=%d conf=sta-dpp ssid=%s configurator=%d" % (id1, "DPPNET01".encode("hex"), conf_id)
1017     if own_id is not None:
1018         cmd += " own=%d" % own_id
1019     if "OK" not in dev.request(cmd):
1020         raise Exception("Failed to initiate DPP Authentication")
1021     ev = dev.wait_event(["DPP-CONF-SENT"], timeout=10)
1022     if ev is None:
1023         raise Exception("DPP configuration not completed (Configurator)")
1024     logger.info("DPP initiator/configurator done")
1025
1026 def test_sigma_dut_dpp_qr_mutual_resp_enrollee(dev, apdev):
1027     """sigma_dut DPP/QR (mutual) responder as Enrollee"""
1028     run_sigma_dut_dpp_qr_mutual_resp_enrollee(dev, apdev)
1029
1030 def test_sigma_dut_dpp_qr_mutual_resp_enrollee_pending(dev, apdev):
1031     """sigma_dut DPP/QR (mutual) responder as Enrollee (response pending)"""
1032     run_sigma_dut_dpp_qr_mutual_resp_enrollee(dev, apdev, ',DPPDelayQRResponse,1')
1033
1034 def run_sigma_dut_dpp_qr_mutual_resp_enrollee(dev, apdev, extra=None):
1035     check_dpp_capab(dev[0])
1036     check_dpp_capab(dev[1])
1037
1038     csign = "30770201010420768240a3fc89d6662d9782f120527fe7fb9edc6366ab0b9c7dde96125cfd250fa00a06082a8648ce3d030107a144034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1039     csign_pub = "3059301306072a8648ce3d020106082a8648ce3d030107034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1040     ap_connector = "eyJ0eXAiOiJkcHBDb24iLCJraWQiOiJwYWtZbXVzd1dCdWpSYTl5OEsweDViaTVrT3VNT3dzZHRlaml2UG55ZHZzIiwiYWxnIjoiRVMyNTYifQ.eyJncm91cHMiOlt7Imdyb3VwSWQiOiIqIiwibmV0Um9sZSI6ImFwIn1dLCJuZXRBY2Nlc3NLZXkiOnsia3R5IjoiRUMiLCJjcnYiOiJQLTI1NiIsIngiOiIybU5vNXZuRkI5bEw3d1VWb1hJbGVPYzBNSEE1QXZKbnpwZXZULVVTYzVNIiwieSI6IlhzS3dqVHJlLTg5WWdpU3pKaG9CN1haeUttTU05OTl3V2ZaSVl0bi01Q3MifX0.XhjFpZgcSa7G2lHy0OCYTvaZFRo5Hyx6b7g7oYyusLC7C_73AJ4_BxEZQVYJXAtDuGvb3dXSkHEKxREP9Q6Qeg"
1041     ap_netaccesskey = "30770201010420ceba752db2ad5200fa7bc565b9c05c69b7eb006751b0b329b0279de1c19ca67ca00a06082a8648ce3d030107a14403420004da6368e6f9c507d94bef0515a1722578e73430703902f267ce97af4fe51273935ec2b08d3adefbcf588224b3261a01ed76722a630cf7df7059f64862d9fee42b"
1042
1043     params = { "ssid": "DPPNET01",
1044                "wpa": "2",
1045                "wpa_key_mgmt": "DPP",
1046                "rsn_pairwise": "CCMP",
1047                "dpp_connector": ap_connector,
1048                "dpp_csign": csign_pub,
1049                "dpp_netaccesskey": ap_netaccesskey }
1050     try:
1051         hapd = hostapd.add_ap(apdev[0], params)
1052     except:
1053         raise HwsimSkip("DPP not supported")
1054
1055     sigma = start_sigma_dut(dev[0].ifname)
1056     try:
1057         dev[0].set("dpp_config_processing", "2")
1058
1059         cmd = "DPP_CONFIGURATOR_ADD key=" + csign
1060         res = dev[1].request(cmd);
1061         if "FAIL" in res:
1062             raise Exception("Failed to add configurator")
1063         conf_id = int(res)
1064
1065         addr = dev[1].own_addr().replace(':', '')
1066         cmd = "DPP_BOOTSTRAP_GEN type=qrcode chan=81/6 mac=" + addr
1067         res = dev[1].request(cmd)
1068         if "FAIL" in res:
1069             raise Exception("Failed to generate bootstrapping info")
1070         id0 = int(res)
1071         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1072
1073         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
1074         if "status,COMPLETE" not in res:
1075             raise Exception("dev_exec_action did not succeed: " + res)
1076         hex = res.split(',')[3]
1077         uri = hex.decode('hex')
1078         logger.info("URI from sigma_dut: " + uri)
1079
1080         res = dev[1].request("DPP_QR_CODE " + uri)
1081         if "FAIL" in res:
1082             raise Exception("Failed to parse QR Code URI")
1083         id1 = int(res)
1084
1085         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % uri0.encode('hex'))
1086         if "status,COMPLETE" not in res:
1087             raise Exception("dev_exec_action did not succeed: " + res)
1088
1089         t = threading.Thread(target=dpp_init_conf_mutual,
1090                              args=(dev[1], id1, conf_id, id0))
1091         t.start()
1092
1093         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPAuthDirection,Mutual,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,20,DPPWaitForConnect,Yes"
1094         if extra:
1095             cmd += extra
1096         res = sigma_dut_cmd(cmd, timeout=25)
1097         t.join()
1098         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,OK,NetworkConnectResult,OK" not in res:
1099             raise Exception("Unexpected result: " + res)
1100     finally:
1101         dev[0].set("dpp_config_processing", "0")
1102         stop_sigma_dut(sigma)
1103
1104 def dpp_resp_conf_mutual(dev, conf_id, uri):
1105     logger.info("Starting DPP responder/configurator in a thread")
1106     dev.set("dpp_configurator_params",
1107             " conf=sta-dpp ssid=%s configurator=%d" % ("DPPNET01".encode("hex"), conf_id));
1108     cmd = "DPP_LISTEN 2437 role=configurator qr=mutual"
1109     if "OK" not in dev.request(cmd):
1110         raise Exception("Failed to initiate DPP listen")
1111     if uri:
1112         ev = dev.wait_event(["DPP-SCAN-PEER-QR-CODE"], timeout=10)
1113         if ev is None:
1114             raise Exception("QR Code scan for mutual authentication not requested")
1115         res = dev.request("DPP_QR_CODE " + uri)
1116         if "FAIL" in res:
1117             raise Exception("Failed to parse QR Code URI")
1118     ev = dev.wait_event(["DPP-CONF-SENT"], timeout=10)
1119     if ev is None:
1120         raise Exception("DPP configuration not completed (Configurator)")
1121     logger.info("DPP responder/configurator done")
1122
1123 def test_sigma_dut_dpp_qr_mutual_init_enrollee(dev, apdev):
1124     """sigma_dut DPP/QR (mutual) initiator as Enrollee"""
1125     run_sigma_dut_dpp_qr_mutual_init_enrollee(dev, apdev, False)
1126
1127 def test_sigma_dut_dpp_qr_mutual_init_enrollee_pending(dev, apdev):
1128     """sigma_dut DPP/QR (mutual) initiator as Enrollee (response pending)"""
1129     run_sigma_dut_dpp_qr_mutual_init_enrollee(dev, apdev, True)
1130
1131 def run_sigma_dut_dpp_qr_mutual_init_enrollee(dev, apdev, resp_pending):
1132     check_dpp_capab(dev[0])
1133     check_dpp_capab(dev[1])
1134
1135     csign = "30770201010420768240a3fc89d6662d9782f120527fe7fb9edc6366ab0b9c7dde96125cfd250fa00a06082a8648ce3d030107a144034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1136     csign_pub = "3059301306072a8648ce3d020106082a8648ce3d030107034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1137     ap_connector = "eyJ0eXAiOiJkcHBDb24iLCJraWQiOiJwYWtZbXVzd1dCdWpSYTl5OEsweDViaTVrT3VNT3dzZHRlaml2UG55ZHZzIiwiYWxnIjoiRVMyNTYifQ.eyJncm91cHMiOlt7Imdyb3VwSWQiOiIqIiwibmV0Um9sZSI6ImFwIn1dLCJuZXRBY2Nlc3NLZXkiOnsia3R5IjoiRUMiLCJjcnYiOiJQLTI1NiIsIngiOiIybU5vNXZuRkI5bEw3d1VWb1hJbGVPYzBNSEE1QXZKbnpwZXZULVVTYzVNIiwieSI6IlhzS3dqVHJlLTg5WWdpU3pKaG9CN1haeUttTU05OTl3V2ZaSVl0bi01Q3MifX0.XhjFpZgcSa7G2lHy0OCYTvaZFRo5Hyx6b7g7oYyusLC7C_73AJ4_BxEZQVYJXAtDuGvb3dXSkHEKxREP9Q6Qeg"
1138     ap_netaccesskey = "30770201010420ceba752db2ad5200fa7bc565b9c05c69b7eb006751b0b329b0279de1c19ca67ca00a06082a8648ce3d030107a14403420004da6368e6f9c507d94bef0515a1722578e73430703902f267ce97af4fe51273935ec2b08d3adefbcf588224b3261a01ed76722a630cf7df7059f64862d9fee42b"
1139
1140     params = { "ssid": "DPPNET01",
1141                "wpa": "2",
1142                "wpa_key_mgmt": "DPP",
1143                "rsn_pairwise": "CCMP",
1144                "dpp_connector": ap_connector,
1145                "dpp_csign": csign_pub,
1146                "dpp_netaccesskey": ap_netaccesskey }
1147     try:
1148         hapd = hostapd.add_ap(apdev[0], params)
1149     except:
1150         raise HwsimSkip("DPP not supported")
1151
1152     sigma = start_sigma_dut(dev[0].ifname)
1153     try:
1154         dev[0].set("dpp_config_processing", "2")
1155
1156         cmd = "DPP_CONFIGURATOR_ADD key=" + csign
1157         res = dev[1].request(cmd);
1158         if "FAIL" in res:
1159             raise Exception("Failed to add configurator")
1160         conf_id = int(res)
1161
1162         addr = dev[1].own_addr().replace(':', '')
1163         cmd = "DPP_BOOTSTRAP_GEN type=qrcode chan=81/6 mac=" + addr
1164         res = dev[1].request(cmd)
1165         if "FAIL" in res:
1166             raise Exception("Failed to generate bootstrapping info")
1167         id0 = int(res)
1168         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1169
1170         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
1171         if "status,COMPLETE" not in res:
1172             raise Exception("dev_exec_action did not succeed: " + res)
1173         hex = res.split(',')[3]
1174         uri = hex.decode('hex')
1175         logger.info("URI from sigma_dut: " + uri)
1176
1177         if not resp_pending:
1178             res = dev[1].request("DPP_QR_CODE " + uri)
1179             if "FAIL" in res:
1180                 raise Exception("Failed to parse QR Code URI")
1181             uri = None
1182
1183         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % uri0.encode('hex'))
1184         if "status,COMPLETE" not in res:
1185             raise Exception("dev_exec_action did not succeed: " + res)
1186
1187         t = threading.Thread(target=dpp_resp_conf_mutual,
1188                              args=(dev[1], conf_id, uri))
1189         t.start()
1190
1191         time.sleep(1)
1192         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Mutual,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,10,DPPWaitForConnect,Yes"
1193         res = sigma_dut_cmd(cmd, timeout=15)
1194         t.join()
1195         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,OK,NetworkConnectResult,OK" not in res:
1196             raise Exception("Unexpected result: " + res)
1197     finally:
1198         dev[0].set("dpp_config_processing", "0")
1199         stop_sigma_dut(sigma)
1200
1201 def test_sigma_dut_dpp_qr_init_enrollee_psk(dev, apdev):
1202     """sigma_dut DPP/QR initiator as Enrollee (PSK)"""
1203     check_dpp_capab(dev[0])
1204     check_dpp_capab(dev[1])
1205
1206     params = hostapd.wpa2_params(ssid="DPPNET01",
1207                                  passphrase="ThisIsDppPassphrase")
1208     hapd = hostapd.add_ap(apdev[0], params)
1209
1210     sigma = start_sigma_dut(dev[0].ifname)
1211     try:
1212         dev[0].set("dpp_config_processing", "2")
1213
1214         cmd = "DPP_CONFIGURATOR_ADD"
1215         res = dev[1].request(cmd);
1216         if "FAIL" in res:
1217             raise Exception("Failed to add configurator")
1218         conf_id = int(res)
1219
1220         addr = dev[1].own_addr().replace(':', '')
1221         cmd = "DPP_BOOTSTRAP_GEN type=qrcode chan=81/6 mac=" + addr
1222         res = dev[1].request(cmd)
1223         if "FAIL" in res:
1224             raise Exception("Failed to generate bootstrapping info")
1225         id0 = int(res)
1226         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1227
1228         dev[1].set("dpp_configurator_params",
1229                    " conf=sta-psk ssid=%s pass=%s configurator=%d" % ("DPPNET01".encode("hex"), "ThisIsDppPassphrase".encode("hex"), conf_id));
1230         cmd = "DPP_LISTEN 2437 role=configurator"
1231         if "OK" not in dev[1].request(cmd):
1232             raise Exception("Failed to start listen operation")
1233
1234         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % uri0.encode('hex'))
1235         if "status,COMPLETE" not in res:
1236             raise Exception("dev_exec_action did not succeed: " + res)
1237
1238         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Single,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,6,DPPWaitForConnect,Yes", timeout=10)
1239         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkConnectResult,OK" not in res:
1240             raise Exception("Unexpected result: " + res)
1241     finally:
1242         dev[0].set("dpp_config_processing", "0")
1243         stop_sigma_dut(sigma)
1244
1245 def test_sigma_dut_dpp_qr_init_configurator_1(dev, apdev):
1246     """sigma_dut DPP/QR initiator as Configurator (conf index 1)"""
1247     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 1)
1248
1249 def test_sigma_dut_dpp_qr_init_configurator_2(dev, apdev):
1250     """sigma_dut DPP/QR initiator as Configurator (conf index 2)"""
1251     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 2)
1252
1253 def test_sigma_dut_dpp_qr_init_configurator_3(dev, apdev):
1254     """sigma_dut DPP/QR initiator as Configurator (conf index 3)"""
1255     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 3)
1256
1257 def test_sigma_dut_dpp_qr_init_configurator_4(dev, apdev):
1258     """sigma_dut DPP/QR initiator as Configurator (conf index 4)"""
1259     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 4)
1260
1261 def test_sigma_dut_dpp_qr_init_configurator_both(dev, apdev):
1262     """sigma_dut DPP/QR initiator as Configurator or Enrollee (conf index 1)"""
1263     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 1, "Both")
1264
1265 def test_sigma_dut_dpp_qr_init_configurator_neg_freq(dev, apdev):
1266     """sigma_dut DPP/QR initiator as Configurator (neg_freq)"""
1267     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 1, extra='DPPSubsequentChannel,81/11')
1268
1269 def run_sigma_dut_dpp_qr_init_configurator(dev, apdev, conf_idx,
1270                                            prov_role="Configurator",
1271                                            extra=None):
1272     check_dpp_capab(dev[0])
1273     check_dpp_capab(dev[1])
1274     sigma = start_sigma_dut(dev[0].ifname)
1275     try:
1276         addr = dev[1].own_addr().replace(':', '')
1277         cmd = "DPP_BOOTSTRAP_GEN type=qrcode chan=81/6 mac=" + addr
1278         res = dev[1].request(cmd)
1279         if "FAIL" in res:
1280             raise Exception("Failed to generate bootstrapping info")
1281         id0 = int(res)
1282         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1283
1284         cmd = "DPP_LISTEN 2437 role=enrollee"
1285         if "OK" not in dev[1].request(cmd):
1286             raise Exception("Failed to start listen operation")
1287
1288         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % uri0.encode('hex'))
1289         if "status,COMPLETE" not in res:
1290             raise Exception("dev_exec_action did not succeed: " + res)
1291
1292         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Single,DPPProvisioningRole,%s,DPPConfIndex,%d,DPPSigningKeyECC,P-256,DPPConfEnrolleeRole,STA,DPPBS,QR,DPPTimeout,6" % (prov_role, conf_idx)
1293         if extra:
1294             cmd += "," + extra
1295         res = sigma_dut_cmd(cmd)
1296         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
1297             raise Exception("Unexpected result: " + res)
1298     finally:
1299         stop_sigma_dut(sigma)
1300
1301 def test_sigma_dut_dpp_pkex_init_configurator(dev, apdev):
1302     """sigma_dut DPP/PKEX initiator as Configurator"""
1303     check_dpp_capab(dev[0])
1304     check_dpp_capab(dev[1])
1305     sigma = start_sigma_dut(dev[0].ifname)
1306     try:
1307         cmd = "DPP_BOOTSTRAP_GEN type=pkex"
1308         res = dev[1].request(cmd)
1309         if "FAIL" in res:
1310             raise Exception("Failed to generate bootstrapping info")
1311         id1 = int(res)
1312         cmd = "DPP_PKEX_ADD own=%d identifier=test code=secret" % (id1)
1313         res = dev[1].request(cmd)
1314         if "FAIL" in res:
1315             raise Exception("Failed to set PKEX data (responder)")
1316         cmd = "DPP_LISTEN 2437 role=enrollee"
1317         if "OK" not in dev[1].request(cmd):
1318             raise Exception("Failed to start listen operation")
1319
1320         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Mutual,DPPProvisioningRole,Configurator,DPPConfIndex,1,DPPSigningKeyECC,P-256,DPPConfEnrolleeRole,STA,DPPBS,PKEX,DPPPKEXCodeIdentifier,test,DPPPKEXCode,secret,DPPTimeout,6")
1321         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
1322             raise Exception("Unexpected result: " + res)
1323     finally:
1324         stop_sigma_dut(sigma)
1325
1326 def dpp_init_conf(dev, id1, conf, conf_id, extra):
1327     logger.info("Starting DPP initiator/configurator in a thread")
1328     cmd = "DPP_AUTH_INIT peer=%d conf=%s %s configurator=%d" % (id1, conf, extra, conf_id)
1329     if "OK" not in dev.request(cmd):
1330         raise Exception("Failed to initiate DPP Authentication")
1331     ev = dev.wait_event(["DPP-CONF-SENT"], timeout=5)
1332     if ev is None:
1333         raise Exception("DPP configuration not completed (Configurator)")
1334     logger.info("DPP initiator/configurator done")
1335
1336 def test_sigma_dut_ap_dpp_qr(dev, apdev, params):
1337     """sigma_dut controlled AP (DPP)"""
1338     run_sigma_dut_ap_dpp_qr(dev, apdev, params, "ap-dpp", "sta-dpp")
1339
1340 def test_sigma_dut_ap_dpp_qr_legacy(dev, apdev, params):
1341     """sigma_dut controlled AP (legacy)"""
1342     run_sigma_dut_ap_dpp_qr(dev, apdev, params, "ap-psk", "sta-psk",
1343                             extra="pass=%s" % "qwertyuiop".encode("hex"))
1344
1345 def test_sigma_dut_ap_dpp_qr_legacy_psk(dev, apdev, params):
1346     """sigma_dut controlled AP (legacy)"""
1347     run_sigma_dut_ap_dpp_qr(dev, apdev, params, "ap-psk", "sta-psk",
1348                             extra="psk=%s" % (32*"12"))
1349
1350 def run_sigma_dut_ap_dpp_qr(dev, apdev, params, ap_conf, sta_conf, extra=""):
1351     logdir = os.path.join(params['logdir'], "sigma_dut_ap_dpp_qr.sigma-hostapd")
1352     with HWSimRadio() as (radio, iface):
1353         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
1354         try:
1355             sigma_dut_cmd_check("ap_reset_default,program,DPP")
1356             res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
1357             if "status,COMPLETE" not in res:
1358                 raise Exception("dev_exec_action did not succeed: " + res)
1359             hex = res.split(',')[3]
1360             uri = hex.decode('hex')
1361             logger.info("URI from sigma_dut: " + uri)
1362
1363             cmd = "DPP_CONFIGURATOR_ADD"
1364             res = dev[0].request(cmd);
1365             if "FAIL" in res:
1366                 raise Exception("Failed to add configurator")
1367             conf_id = int(res)
1368
1369             res = dev[0].request("DPP_QR_CODE " + uri)
1370             if "FAIL" in res:
1371                 raise Exception("Failed to parse QR Code URI")
1372             id1 = int(res)
1373
1374             t = threading.Thread(target=dpp_init_conf,
1375                                  args=(dev[0], id1, ap_conf, conf_id, extra))
1376             t.start()
1377             res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPAuthDirection,Single,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,6")
1378             t.join()
1379             if "ConfResult,OK" not in res:
1380                 raise Exception("Unexpected result: " + res)
1381
1382             addr = dev[1].own_addr().replace(':', '')
1383             cmd = "DPP_BOOTSTRAP_GEN type=qrcode chan=81/1 mac=" + addr
1384             res = dev[1].request(cmd)
1385             if "FAIL" in res:
1386                 raise Exception("Failed to generate bootstrapping info")
1387             id1 = int(res)
1388             uri1 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id1)
1389
1390             res = dev[0].request("DPP_QR_CODE " + uri1)
1391             if "FAIL" in res:
1392                 raise Exception("Failed to parse QR Code URI")
1393             id0b = int(res)
1394
1395             dev[1].set("dpp_config_processing", "2")
1396             cmd = "DPP_LISTEN 2412"
1397             if "OK" not in dev[1].request(cmd):
1398                 raise Exception("Failed to start listen operation")
1399             cmd = "DPP_AUTH_INIT peer=%d conf=%s %s configurator=%d" % (id0b, sta_conf, extra, conf_id)
1400             if "OK" not in dev[0].request(cmd):
1401                 raise Exception("Failed to initiate DPP Authentication")
1402             dev[1].wait_connected()
1403
1404             sigma_dut_cmd_check("ap_reset_default")
1405         finally:
1406             dev[1].set("dpp_config_processing", "0")
1407             stop_sigma_dut(sigma)
1408
1409 def test_sigma_dut_ap_dpp_pkex_responder(dev, apdev, params):
1410     """sigma_dut controlled AP as DPP PKEX responder"""
1411     logdir = os.path.join(params['logdir'],
1412                           "sigma_dut_ap_dpp_pkex_responder.sigma-hostapd")
1413     with HWSimRadio() as (radio, iface):
1414         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
1415         try:
1416             run_sigma_dut_ap_dpp_pkex_responder(dev, apdev)
1417         finally:
1418             stop_sigma_dut(sigma)
1419
1420 def dpp_init_conf_pkex(dev, conf_id):
1421     logger.info("Starting DPP PKEX initiator/configurator in a thread")
1422     time.sleep(1.5)
1423     cmd = "DPP_BOOTSTRAP_GEN type=pkex"
1424     res = dev.request(cmd)
1425     if "FAIL" in res:
1426         raise Exception("Failed to generate bootstrapping info")
1427     id = int(res)
1428     cmd = "DPP_PKEX_ADD own=%d init=1 conf=ap-dpp configurator=%d code=password" % (id, conf_id)
1429     res = dev.request(cmd)
1430     if "FAIL" in res:
1431         raise Exception("Failed to initiate DPP PKEX")
1432     ev = dev.wait_event(["DPP-CONF-SENT"], timeout=5)
1433     if ev is None:
1434         raise Exception("DPP configuration not completed (Configurator)")
1435     logger.info("DPP initiator/configurator done")
1436
1437 def run_sigma_dut_ap_dpp_pkex_responder(dev, apdev):
1438     sigma_dut_cmd_check("ap_reset_default,program,DPP")
1439
1440     cmd = "DPP_CONFIGURATOR_ADD"
1441     res = dev[0].request(cmd);
1442     if "FAIL" in res:
1443         raise Exception("Failed to add configurator")
1444     conf_id = int(res)
1445
1446     t = threading.Thread(target=dpp_init_conf_pkex, args=(dev[0], conf_id))
1447     t.start()
1448     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPAuthDirection,Mutual,DPPProvisioningRole,Enrollee,DPPBS,PKEX,DPPPKEXCode,password,DPPTimeout,6", timeout=10)
1449     t.join()
1450     if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
1451         raise Exception("Unexpected result: " + res)
1452
1453     sigma_dut_cmd_check("ap_reset_default")
1454
1455 def dpp_proto_init(dev, id1):
1456     time.sleep(1)
1457     logger.info("Starting DPP initiator/configurator in a thread")
1458     cmd = "DPP_CONFIGURATOR_ADD"
1459     res = dev.request(cmd);
1460     if "FAIL" in res:
1461         raise Exception("Failed to add configurator")
1462     conf_id = int(res)
1463
1464     cmd = "DPP_AUTH_INIT peer=%d conf=sta-dpp configurator=%d" % (id1, conf_id)
1465     if "OK" not in dev.request(cmd):
1466         raise Exception("Failed to initiate DPP Authentication")
1467
1468 def test_sigma_dut_dpp_proto_initiator(dev, apdev):
1469     """sigma_dut DPP protocol testing - Initiator"""
1470     check_dpp_capab(dev[0])
1471     check_dpp_capab(dev[1])
1472     tests = [ ("InvalidValue", "AuthenticationRequest", "WrappedData",
1473                "BootstrapResult,OK,AuthResult,Errorsent",
1474                None),
1475               ("InvalidValue", "AuthenticationConfirm", "WrappedData",
1476                "BootstrapResult,OK,AuthResult,Errorsent",
1477                None),
1478               ("MissingAttribute", "AuthenticationRequest", "InitCapabilities",
1479                "BootstrapResult,OK,AuthResult,Errorsent",
1480                "Missing or invalid I-capabilities"),
1481               ("InvalidValue", "AuthenticationConfirm", "InitAuthTag",
1482                "BootstrapResult,OK,AuthResult,Errorsent",
1483                "Mismatching Initiator Authenticating Tag"),
1484               ("MissingAttribute", "ConfigurationResponse", "EnrolleeNonce",
1485                "BootstrapResult,OK,AuthResult,OK,ConfResult,Errorsent",
1486                "Missing or invalid Enrollee Nonce attribute") ]
1487     for step, frame, attr, result, fail in tests:
1488         dev[0].request("FLUSH")
1489         dev[1].request("FLUSH")
1490         sigma = start_sigma_dut(dev[0].ifname)
1491         try:
1492             run_sigma_dut_dpp_proto_initiator(dev, step, frame, attr, result,
1493                                               fail)
1494         finally:
1495             stop_sigma_dut(sigma)
1496
1497 def run_sigma_dut_dpp_proto_initiator(dev, step, frame, attr, result, fail):
1498     addr = dev[1].own_addr().replace(':', '')
1499     cmd = "DPP_BOOTSTRAP_GEN type=qrcode chan=81/6 mac=" + addr
1500     res = dev[1].request(cmd)
1501     if "FAIL" in res:
1502         raise Exception("Failed to generate bootstrapping info")
1503     id0 = int(res)
1504     uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1505
1506     cmd = "DPP_LISTEN 2437 role=enrollee"
1507     if "OK" not in dev[1].request(cmd):
1508         raise Exception("Failed to start listen operation")
1509
1510     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % uri0.encode('hex'))
1511     if "status,COMPLETE" not in res:
1512         raise Exception("dev_exec_action did not succeed: " + res)
1513
1514     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Single,DPPProvisioningRole,Configurator,DPPConfIndex,1,DPPSigningKeyECC,P-256,DPPConfEnrolleeRole,STA,DPPBS,QR,DPPTimeout,6,DPPStep,%s,DPPFrameType,%s,DPPIEAttribute,%s" % (step, frame, attr))
1515     if result not in res:
1516         raise Exception("Unexpected result: " + res)
1517     if fail:
1518         ev = dev[1].wait_event(["DPP-FAIL"], timeout=5)
1519         if ev is None or fail not in ev:
1520             raise Exception("Failure not reported correctly: " + str(ev))
1521
1522     dev[1].request("DPP_STOP_LISTEN")
1523     dev[0].dump_monitor()
1524     dev[1].dump_monitor()
1525
1526 def test_sigma_dut_dpp_proto_responder(dev, apdev):
1527     """sigma_dut DPP protocol testing - Responder"""
1528     check_dpp_capab(dev[0])
1529     check_dpp_capab(dev[1])
1530     tests = [ ("MissingAttribute", "AuthenticationResponse", "DPPStatus",
1531                "BootstrapResult,OK,AuthResult,Errorsent",
1532                "Missing or invalid required DPP Status attribute"),
1533               ("MissingAttribute", "ConfigurationRequest", "EnrolleeNonce",
1534                "BootstrapResult,OK,AuthResult,OK,ConfResult,Errorsent",
1535                "Missing or invalid Enrollee Nonce attribute") ]
1536     for step, frame, attr, result, fail in tests:
1537         dev[0].request("FLUSH")
1538         dev[1].request("FLUSH")
1539         sigma = start_sigma_dut(dev[0].ifname)
1540         try:
1541             run_sigma_dut_dpp_proto_responder(dev, step, frame, attr, result,
1542                                               fail)
1543         finally:
1544             stop_sigma_dut(sigma)
1545
1546 def run_sigma_dut_dpp_proto_responder(dev, step, frame, attr, result, fail):
1547     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
1548     if "status,COMPLETE" not in res:
1549         raise Exception("dev_exec_action did not succeed: " + res)
1550     hex = res.split(',')[3]
1551     uri = hex.decode('hex')
1552     logger.info("URI from sigma_dut: " + uri)
1553
1554     res = dev[1].request("DPP_QR_CODE " + uri)
1555     if "FAIL" in res:
1556         raise Exception("Failed to parse QR Code URI")
1557     id1 = int(res)
1558
1559     t = threading.Thread(target=dpp_proto_init, args=(dev[1], id1))
1560     t.start()
1561     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPAuthDirection,Single,DPPProvisioningRole,Enrollee,DPPConfIndex,1,DPPSigningKeyECC,P-256,DPPConfEnrolleeRole,STA,DPPBS,QR,DPPTimeout,6,DPPStep,%s,DPPFrameType,%s,DPPIEAttribute,%s" % (step, frame, attr), timeout=10)
1562     t.join()
1563     if result not in res:
1564         raise Exception("Unexpected result: " + res)
1565     if fail:
1566         ev = dev[1].wait_event(["DPP-FAIL"], timeout=5)
1567         if ev is None or fail not in ev:
1568             raise Exception("Failure not reported correctly:" + str(ev))
1569
1570     dev[1].request("DPP_STOP_LISTEN")
1571     dev[0].dump_monitor()
1572     dev[1].dump_monitor()
1573
1574 def dpp_proto_init_pkex(dev):
1575     time.sleep(1)
1576     logger.info("Starting DPP PKEX initiator/configurator in a thread")
1577     cmd = "DPP_CONFIGURATOR_ADD"
1578     res = dev.request(cmd);
1579     if "FAIL" in res:
1580         raise Exception("Failed to add configurator")
1581     conf_id = int(res)
1582
1583     cmd = "DPP_BOOTSTRAP_GEN type=pkex"
1584     res = dev.request(cmd)
1585     if "FAIL" in res:
1586         raise Exception("Failed to generate bootstrapping info")
1587     id = int(res)
1588
1589     cmd = "DPP_PKEX_ADD own=%d init=1 conf=sta-dpp configurator=%d code=secret" % (id, conf_id)
1590     if "FAIL" in dev.request(cmd):
1591         raise Exception("Failed to initiate DPP PKEX")
1592
1593 def test_sigma_dut_dpp_proto_initiator_pkex(dev, apdev):
1594     """sigma_dut DPP protocol testing - Initiator (PKEX)"""
1595     check_dpp_capab(dev[0])
1596     check_dpp_capab(dev[1])
1597     tests = [ ("InvalidValue", "PKEXCRRequest", "WrappedData",
1598                "BootstrapResult,Errorsent",
1599                None),
1600               ("MissingAttribute", "PKEXExchangeRequest", "FiniteCyclicGroup",
1601                "BootstrapResult,Errorsent",
1602                "Missing or invalid Finite Cyclic Group attribute"),
1603               ("MissingAttribute", "PKEXCRRequest", "BSKey",
1604                "BootstrapResult,Errorsent",
1605                "No valid peer bootstrapping key found") ]
1606     for step, frame, attr, result, fail in tests:
1607         dev[0].request("FLUSH")
1608         dev[1].request("FLUSH")
1609         sigma = start_sigma_dut(dev[0].ifname)
1610         try:
1611             run_sigma_dut_dpp_proto_initiator_pkex(dev, step, frame, attr,
1612                                                    result, fail)
1613         finally:
1614             stop_sigma_dut(sigma)
1615
1616 def run_sigma_dut_dpp_proto_initiator_pkex(dev, step, frame, attr, result, fail):
1617     cmd = "DPP_BOOTSTRAP_GEN type=pkex"
1618     res = dev[1].request(cmd)
1619     if "FAIL" in res:
1620         raise Exception("Failed to generate bootstrapping info")
1621     id1 = int(res)
1622
1623     cmd = "DPP_PKEX_ADD own=%d code=secret" % (id1)
1624     res = dev[1].request(cmd)
1625     if "FAIL" in res:
1626         raise Exception("Failed to set PKEX data (responder)")
1627
1628     cmd = "DPP_LISTEN 2437 role=enrollee"
1629     if "OK" not in dev[1].request(cmd):
1630         raise Exception("Failed to start listen operation")
1631
1632     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Single,DPPProvisioningRole,Configurator,DPPConfIndex,1,DPPSigningKeyECC,P-256,DPPConfEnrolleeRole,STA,DPPBS,PKEX,DPPPKEXCode,secret,DPPTimeout,6,DPPStep,%s,DPPFrameType,%s,DPPIEAttribute,%s" % (step, frame, attr))
1633     if result not in res:
1634         raise Exception("Unexpected result: " + res)
1635     if fail:
1636         ev = dev[1].wait_event(["DPP-FAIL"], timeout=5)
1637         if ev is None or fail not in ev:
1638             raise Exception("Failure not reported correctly: " + str(ev))
1639
1640     dev[1].request("DPP_STOP_LISTEN")
1641     dev[0].dump_monitor()
1642     dev[1].dump_monitor()
1643
1644 def test_sigma_dut_dpp_proto_responder_pkex(dev, apdev):
1645     """sigma_dut DPP protocol testing - Responder (PKEX)"""
1646     check_dpp_capab(dev[0])
1647     check_dpp_capab(dev[1])
1648     tests = [ ("InvalidValue", "PKEXCRResponse", "WrappedData",
1649                "BootstrapResult,Errorsent",
1650                None),
1651               ("MissingAttribute", "PKEXExchangeResponse", "DPPStatus",
1652                "BootstrapResult,Errorsent",
1653                "No DPP Status attribute"),
1654               ("MissingAttribute", "PKEXCRResponse", "BSKey",
1655                "BootstrapResult,Errorsent",
1656                "No valid peer bootstrapping key found") ]
1657     for step, frame, attr, result, fail in tests:
1658         dev[0].request("FLUSH")
1659         dev[1].request("FLUSH")
1660         sigma = start_sigma_dut(dev[0].ifname)
1661         try:
1662             run_sigma_dut_dpp_proto_responder_pkex(dev, step, frame, attr,
1663                                                    result, fail)
1664         finally:
1665             stop_sigma_dut(sigma)
1666
1667 def run_sigma_dut_dpp_proto_responder_pkex(dev, step, frame, attr, result, fail):
1668     t = threading.Thread(target=dpp_proto_init_pkex, args=(dev[1],))
1669     t.start()
1670     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPAuthDirection,Single,DPPProvisioningRole,Enrollee,DPPConfIndex,1,DPPSigningKeyECC,P-256,DPPConfEnrolleeRole,STA,DPPBS,PKEX,DPPPKEXCode,secret,DPPTimeout,6,DPPStep,%s,DPPFrameType,%s,DPPIEAttribute,%s" % (step, frame, attr), timeout=10)
1671     t.join()
1672     if result not in res:
1673         raise Exception("Unexpected result: " + res)
1674     if fail:
1675         ev = dev[1].wait_event(["DPP-FAIL"], timeout=5)
1676         if ev is None or fail not in ev:
1677             raise Exception("Failure not reported correctly:" + str(ev))
1678
1679     dev[1].request("DPP_STOP_LISTEN")
1680     dev[0].dump_monitor()
1681     dev[1].dump_monitor()
1682
1683 def init_sigma_dut_dpp_proto_peer_disc_req(dev, apdev):
1684     check_dpp_capab(dev[0])
1685     check_dpp_capab(dev[1])
1686
1687     csign = "30770201010420768240a3fc89d6662d9782f120527fe7fb9edc6366ab0b9c7dde96125cfd250fa00a06082a8648ce3d030107a144034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1688     csign_pub = "3059301306072a8648ce3d020106082a8648ce3d030107034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1689     ap_connector = "eyJ0eXAiOiJkcHBDb24iLCJraWQiOiJwYWtZbXVzd1dCdWpSYTl5OEsweDViaTVrT3VNT3dzZHRlaml2UG55ZHZzIiwiYWxnIjoiRVMyNTYifQ.eyJncm91cHMiOlt7Imdyb3VwSWQiOiIqIiwibmV0Um9sZSI6ImFwIn1dLCJuZXRBY2Nlc3NLZXkiOnsia3R5IjoiRUMiLCJjcnYiOiJQLTI1NiIsIngiOiIybU5vNXZuRkI5bEw3d1VWb1hJbGVPYzBNSEE1QXZKbnpwZXZULVVTYzVNIiwieSI6IlhzS3dqVHJlLTg5WWdpU3pKaG9CN1haeUttTU05OTl3V2ZaSVl0bi01Q3MifX0.XhjFpZgcSa7G2lHy0OCYTvaZFRo5Hyx6b7g7oYyusLC7C_73AJ4_BxEZQVYJXAtDuGvb3dXSkHEKxREP9Q6Qeg"
1690     ap_netaccesskey = "30770201010420ceba752db2ad5200fa7bc565b9c05c69b7eb006751b0b329b0279de1c19ca67ca00a06082a8648ce3d030107a14403420004da6368e6f9c507d94bef0515a1722578e73430703902f267ce97af4fe51273935ec2b08d3adefbcf588224b3261a01ed76722a630cf7df7059f64862d9fee42b"
1691
1692     params = { "ssid": "DPPNET01",
1693                "wpa": "2",
1694                "wpa_key_mgmt": "DPP",
1695                "rsn_pairwise": "CCMP",
1696                "dpp_connector": ap_connector,
1697                "dpp_csign": csign_pub,
1698                "dpp_netaccesskey": ap_netaccesskey }
1699     try:
1700         hapd = hostapd.add_ap(apdev[0], params)
1701     except:
1702         raise HwsimSkip("DPP not supported")
1703
1704     dev[0].set("dpp_config_processing", "2")
1705
1706     cmd = "DPP_CONFIGURATOR_ADD key=" + csign
1707     res = dev[1].request(cmd);
1708     if "FAIL" in res:
1709         raise Exception("Failed to add configurator")
1710     conf_id = int(res)
1711
1712     addr = dev[1].own_addr().replace(':', '')
1713     cmd = "DPP_BOOTSTRAP_GEN type=qrcode chan=81/6 mac=" + addr
1714     res = dev[1].request(cmd)
1715     if "FAIL" in res:
1716         raise Exception("Failed to generate bootstrapping info")
1717     id0 = int(res)
1718     uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1719
1720     dev[1].set("dpp_configurator_params",
1721                " conf=sta-dpp ssid=%s configurator=%d" % ("DPPNET01".encode("hex"), conf_id));
1722     cmd = "DPP_LISTEN 2437 role=configurator"
1723     if "OK" not in dev[1].request(cmd):
1724         raise Exception("Failed to start listen operation")
1725
1726     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % uri0.encode('hex'))
1727     if "status,COMPLETE" not in res:
1728         raise Exception("dev_exec_action did not succeed: " + res)
1729
1730 def test_sigma_dut_dpp_proto_peer_disc_req(dev, apdev):
1731     """sigma_dut DPP protocol testing - Peer Discovery Request"""
1732     sigma = start_sigma_dut(dev[0].ifname)
1733     try:
1734         init_sigma_dut_dpp_proto_peer_disc_req(dev, apdev)
1735
1736         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Single,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,6,DPPWaitForConnect,Yes,DPPStep,MissingAttribute,DPPFrameType,PeerDiscoveryRequest,DPPIEAttribute,TransactionID", timeout=10)
1737         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,Errorsent" not in res:
1738             raise Exception("Unexpected result: " + res)
1739     finally:
1740         dev[0].set("dpp_config_processing", "0")
1741         stop_sigma_dut(sigma)
1742
1743 def test_sigma_dut_dpp_self_config(dev, apdev):
1744     """sigma_dut DPP Configurator enrolling an AP and using self-configuration"""
1745     check_dpp_capab(dev[0])
1746
1747     hapd = hostapd.add_ap(apdev[0], { "ssid": "unconfigured" })
1748     check_dpp_capab(hapd)
1749
1750     sigma = start_sigma_dut(dev[0].ifname)
1751     try:
1752         dev[0].set("dpp_config_processing", "2")
1753         addr = hapd.own_addr().replace(':', '')
1754         cmd = "DPP_BOOTSTRAP_GEN type=qrcode chan=81/1 mac=" + addr
1755         res = hapd.request(cmd)
1756         if "FAIL" in res:
1757             raise Exception("Failed to generate bootstrapping info")
1758         id = int(res)
1759         uri = hapd.request("DPP_BOOTSTRAP_GET_URI %d" % id)
1760
1761         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % uri.encode('hex'))
1762         if "status,COMPLETE" not in res:
1763             raise Exception("dev_exec_action did not succeed: " + res)
1764
1765         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Single,DPPProvisioningRole,Configurator,DPPConfIndex,1,DPPSigningKeyECC,P-256,DPPConfEnrolleeRole,AP,DPPBS,QR,DPPTimeout,6")
1766         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
1767             raise Exception("Unexpected result: " + res)
1768         update_hapd_config(hapd)
1769
1770         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPCryptoIdentifier,P-256,DPPBS,QR,DPPAuthRole,Initiator,DPPProvisioningRole,Configurator,DPPAuthDirection,Single,DPPConfIndex,1,DPPTimeout,6,DPPWaitForConnect,Yes,DPPSelfConfigure,Yes"
1771         res = sigma_dut_cmd(cmd, timeout=10)
1772         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,OK,NetworkConnectResult,OK" not in res:
1773             raise Exception("Unexpected result: " + res)
1774     finally:
1775         stop_sigma_dut(sigma)
1776         dev[0].set("dpp_config_processing", "0")