tests/hwsim/test_sigma_dut.py

   1 # Test cases for sigma_dut
   2 # Copyright (c) 2017, Qualcomm Atheros, Inc.
   3 # Copyright (c) 2018-2019, The Linux Foundation
   4 #
   5 # This software may be distributed under the terms of the BSD license.
   6 # See README for more details.
   7
   8 import binascii
   9 import errno
  10 import fcntl
  11 import hashlib
  12 import logging
  13 logger = logging.getLogger()
  14 import os
  15 import socket
  16 import struct
  17 import subprocess
  18 import threading
  19 import time
  20
  21 import hostapd
  22 from utils import HwsimSkip
  23 from hwsim import HWSimRadio
  24 import hwsim_utils
  25 from test_dpp import check_dpp_capab, update_hapd_config, wait_auth_success
  26 from test_suite_b import check_suite_b_192_capa, suite_b_as_params, suite_b_192_rsa_ap_params
  27 from test_ap_eap import check_eap_capa, int_eap_server_params, check_domain_match, check_domain_suffix_match
  28 from test_ap_hs20 import hs20_ap_params
  29
  30 def check_sigma_dut():
  31     if not os.path.exists("./sigma_dut"):
  32         raise HwsimSkip("sigma_dut not available")
  33
  34 def to_hex(s):
  35     return binascii.hexlify(s.encode()).decode()
  36
  37 def from_hex(s):
  38     return binascii.unhexlify(s).decode()
  39
  40 def sigma_log_output(cmd):
  41     try:
  42         out = cmd.stdout.read()
  43         if out:
  44             logger.debug("sigma_dut stdout: " + str(out.decode()))
  45     except IOError as e:
  46         if e.errno != errno.EAGAIN:
  47             raise
  48     try:
  49         out = cmd.stderr.read()
  50         if out:
  51             logger.debug("sigma_dut stderr: " + str(out.decode()))
  52     except IOError as e:
  53         if e.errno != errno.EAGAIN:
  54             raise
  55
  56 sigma_prog = None
  57
  58 def sigma_dut_cmd(cmd, port=9000, timeout=2):
  59     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM,
  60                          socket.IPPROTO_TCP)
  61     sock.settimeout(timeout)
  62     addr = ('127.0.0.1', port)
  63     sock.connect(addr)
  64     sock.send(cmd.encode() + b"\r\n")
  65     try:
  66         res = sock.recv(1000).decode()
  67         running = False
  68         done = False
  69         for line in res.splitlines():
  70             if line.startswith("status,RUNNING"):
  71                 running = True
  72             elif line.startswith("status,INVALID"):
  73                 done = True
  74             elif line.startswith("status,ERROR"):
  75                 done = True
  76             elif line.startswith("status,COMPLETE"):
  77                 done = True
  78         if running and not done:
  79             # Read the actual response
  80             res = sock.recv(1000).decode()
  81     except:
  82         res = ''
  83         pass
  84     sock.close()
  85     res = res.rstrip()
  86     logger.debug("sigma_dut: '%s' --> '%s'" % (cmd, res))
  87     global sigma_prog
  88     if sigma_prog:
  89         sigma_log_output(sigma_prog)
  90     return res
  91
  92 def sigma_dut_cmd_check(cmd, port=9000, timeout=2):
  93     res = sigma_dut_cmd(cmd, port=port, timeout=timeout)
  94     if "COMPLETE" not in res:
  95         raise Exception("sigma_dut command failed: " + cmd)
  96     return res
  97
  98 def start_sigma_dut(ifname, hostapd_logdir=None, cert_path=None,
  99                     bridge=None, sae_h2e=False):
 100     check_sigma_dut()
 101     cmd = ['./sigma_dut',
 102            '-d',
 103            '-M', ifname,
 104            '-S', ifname,
 105            '-F', '../../hostapd/hostapd',
 106            '-G',
 107            '-w', '/var/run/wpa_supplicant/',
 108            '-j', ifname]
 109     if hostapd_logdir:
 110         cmd += ['-H', hostapd_logdir]
 111     if cert_path:
 112         cmd += ['-C', cert_path]
 113     if bridge:
 114         cmd += ['-b', bridge]
 115     if sae_h2e:
 116         cmd += ['-2']
 117     sigma = subprocess.Popen(cmd, stdout=subprocess.PIPE,
 118                              stderr=subprocess.PIPE)
 119     for stream in [sigma.stdout, sigma.stderr]:
 120         fd = stream.fileno()
 121         fl = fcntl.fcntl(fd, fcntl.F_GETFL)
 122         fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
 123
 124     global sigma_prog
 125     sigma_prog = sigma
 126     res = None
 127     for i in range(20):
 128         try:
 129             res = sigma_dut_cmd("HELLO")
 130             break
 131         except:
 132             time.sleep(0.05)
 133     if res is None or "errorCode,Unknown command" not in res:
 134         raise Exception("Failed to start sigma_dut")
 135     return {'cmd': sigma, 'ifname': ifname}
 136
 137 def stop_sigma_dut(sigma):
 138     global sigma_prog
 139     sigma_prog = None
 140     cmd = sigma['cmd']
 141     sigma_log_output(cmd)
 142     logger.debug("Terminating sigma_dut process")
 143     cmd.terminate()
 144     cmd.wait()
 145     out, err = cmd.communicate()
 146     logger.debug("sigma_dut stdout: " + str(out.decode()))
 147     logger.debug("sigma_dut stderr: " + str(err.decode()))
 148     subprocess.call(["ip", "addr", "del", "dev", sigma['ifname'],
 149                      "127.0.0.11/24"],
 150                     stderr=open('/dev/null', 'w'))
 151
 152 def sigma_dut_wait_connected(ifname):
 153     for i in range(50):
 154         res = sigma_dut_cmd("sta_is_connected,interface," + ifname)
 155         if "connected,1" in res:
 156             break
 157         time.sleep(0.2)
 158         if i == 49:
 159             raise Exception("Connection did not complete")
 160
 161 def test_sigma_dut_basic(dev, apdev):
 162     """sigma_dut basic functionality"""
 163     sigma = start_sigma_dut(dev[0].ifname)
 164
 165     tests = [("ca_get_version", "status,COMPLETE,version,1.0"),
 166              ("device_get_info", "status,COMPLETE,vendor"),
 167              ("device_list_interfaces,interfaceType,foo", "status,ERROR"),
 168              ("device_list_interfaces,interfaceType,802.11",
 169               "status,COMPLETE,interfaceType,802.11,interfaceID," + dev[0].ifname)]
 170     try:
 171         res = sigma_dut_cmd("UNKNOWN")
 172         if "status,INVALID,errorCode,Unknown command" not in res:
 173             raise Exception("Unexpected sigma_dut response to unknown command")
 174
 175         for cmd, response in tests:
 176             res = sigma_dut_cmd(cmd)
 177             if response not in res:
 178                 raise Exception("Unexpected %s response: %s" % (cmd, res))
 179     finally:
 180         stop_sigma_dut(sigma)
 181
 182 def test_sigma_dut_open(dev, apdev):
 183     """sigma_dut controlled open network association"""
 184     try:
 185         run_sigma_dut_open(dev, apdev)
 186     finally:
 187         dev[0].set("ignore_old_scan_res", "0")
 188
 189 def run_sigma_dut_open(dev, apdev):
 190     ifname = dev[0].ifname
 191     sigma = start_sigma_dut(ifname)
 192
 193     try:
 194         hapd = hostapd.add_ap(apdev[0], {"ssid": "open"})
 195
 196         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 197         sigma_dut_cmd_check("sta_set_encryption,interface,%s,ssid,%s,encpType,none" % (ifname, "open"))
 198         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s" % (ifname, "open"),
 199                             timeout=10)
 200         sigma_dut_wait_connected(ifname)
 201         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 202         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 203         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 204     finally:
 205         stop_sigma_dut(sigma)
 206
 207 def test_sigma_dut_psk_pmf(dev, apdev):
 208     """sigma_dut controlled PSK+PMF association"""
 209     try:
 210         run_sigma_dut_psk_pmf(dev, apdev)
 211     finally:
 212         dev[0].set("ignore_old_scan_res", "0")
 213
 214 def run_sigma_dut_psk_pmf(dev, apdev):
 215     ifname = dev[0].ifname
 216     sigma = start_sigma_dut(ifname)
 217
 218     try:
 219         ssid = "test-pmf-required"
 220         params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 221         params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 222         params["ieee80211w"] = "2"
 223         hapd = hostapd.add_ap(apdev[0], params)
 224
 225         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,PMF" % ifname)
 226         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 227         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"))
 228         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-pmf-required"),
 229                             timeout=10)
 230         sigma_dut_wait_connected(ifname)
 231         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 232         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 233         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 234     finally:
 235         stop_sigma_dut(sigma)
 236
 237 def test_sigma_dut_psk_pmf_bip_cmac_128(dev, apdev):
 238     """sigma_dut controlled PSK+PMF association with BIP-CMAC-128"""
 239     try:
 240         run_sigma_dut_psk_pmf_cipher(dev, apdev, "BIP-CMAC-128", "AES-128-CMAC")
 241     finally:
 242         dev[0].set("ignore_old_scan_res", "0")
 243
 244 def test_sigma_dut_psk_pmf_bip_cmac_256(dev, apdev):
 245     """sigma_dut controlled PSK+PMF association with BIP-CMAC-256"""
 246     try:
 247         run_sigma_dut_psk_pmf_cipher(dev, apdev, "BIP-CMAC-256", "BIP-CMAC-256")
 248     finally:
 249         dev[0].set("ignore_old_scan_res", "0")
 250
 251 def test_sigma_dut_psk_pmf_bip_gmac_128(dev, apdev):
 252     """sigma_dut controlled PSK+PMF association with BIP-GMAC-128"""
 253     try:
 254         run_sigma_dut_psk_pmf_cipher(dev, apdev, "BIP-GMAC-128", "BIP-GMAC-128")
 255     finally:
 256         dev[0].set("ignore_old_scan_res", "0")
 257
 258 def test_sigma_dut_psk_pmf_bip_gmac_256(dev, apdev):
 259     """sigma_dut controlled PSK+PMF association with BIP-GMAC-256"""
 260     try:
 261         run_sigma_dut_psk_pmf_cipher(dev, apdev, "BIP-GMAC-256", "BIP-GMAC-256")
 262     finally:
 263         dev[0].set("ignore_old_scan_res", "0")
 264
 265 def test_sigma_dut_psk_pmf_bip_gmac_256_mismatch(dev, apdev):
 266     """sigma_dut controlled PSK+PMF association with BIP-GMAC-256 mismatch"""
 267     try:
 268         run_sigma_dut_psk_pmf_cipher(dev, apdev, "BIP-GMAC-256", "AES-128-CMAC",
 269                                      failure=True)
 270     finally:
 271         dev[0].set("ignore_old_scan_res", "0")
 272
 273 def run_sigma_dut_psk_pmf_cipher(dev, apdev, sigma_cipher, hostapd_cipher,
 274                                  failure=False):
 275     ifname = dev[0].ifname
 276     sigma = start_sigma_dut(ifname)
 277
 278     try:
 279         ssid = "test-pmf-required"
 280         params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 281         params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 282         params["ieee80211w"] = "2"
 283         params["group_mgmt_cipher"] = hostapd_cipher
 284         hapd = hostapd.add_ap(apdev[0], params)
 285
 286         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,PMF" % ifname)
 287         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 288         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))
 289         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-pmf-required"),
 290                             timeout=2 if failure else 10)
 291         if failure:
 292             ev = dev[0].wait_event(["CTRL-EVENT-NETWORK-NOT-FOUND",
 293                                     "CTRL-EVENT-CONNECTED"], timeout=10)
 294             if ev is None:
 295                 raise Exception("Network selection result not indicated")
 296             if "CTRL-EVENT-CONNECTED" in ev:
 297                 raise Exception("Unexpected connection")
 298             res = sigma_dut_cmd("sta_is_connected,interface," + ifname)
 299             if "connected,1" in res:
 300                 raise Exception("Connection reported")
 301         else:
 302             sigma_dut_wait_connected(ifname)
 303             sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 304
 305         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 306         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 307     finally:
 308         stop_sigma_dut(sigma)
 309
 310 def test_sigma_dut_sae(dev, apdev):
 311     """sigma_dut controlled SAE association"""
 312     if "SAE" not in dev[0].get_capability("auth_alg"):
 313         raise HwsimSkip("SAE not supported")
 314
 315     ifname = dev[0].ifname
 316     sigma = start_sigma_dut(ifname)
 317
 318     try:
 319         ssid = "test-sae"
 320         params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 321         params['wpa_key_mgmt'] = 'SAE'
 322         params["ieee80211w"] = "2"
 323         params['sae_groups'] = '19 20 21'
 324         hapd = hostapd.add_ap(apdev[0], params)
 325
 326         sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
 327         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 328         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2" % (ifname, "test-sae", "12345678"))
 329         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
 330                             timeout=10)
 331         sigma_dut_wait_connected(ifname)
 332         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 333         if dev[0].get_status_field('sae_group') != '19':
 334             raise Exception("Expected default SAE group not used")
 335         res = sigma_dut_cmd_check("sta_get_parameter,interface,%s,Parameter,PMK" % ifname)
 336         logger.info("Reported PMK: " + res)
 337         if ",PMK," not in res:
 338             raise Exception("PMK not reported");
 339         if hapd.request("GET_PMK " + dev[0].own_addr()) != res.split(',')[3]:
 340             raise Exception("Mismatch in reported PMK")
 341         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 342
 343         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 344
 345         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 346         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"))
 347         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
 348                             timeout=10)
 349         sigma_dut_wait_connected(ifname)
 350         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 351         if dev[0].get_status_field('sae_group') != '20':
 352             raise Exception("Expected SAE group not used")
 353         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 354         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 355     finally:
 356         stop_sigma_dut(sigma)
 357
 358 def test_sigma_dut_sae_groups(dev, apdev):
 359     """sigma_dut controlled SAE association with group negotiation"""
 360     if "SAE" not in dev[0].get_capability("auth_alg"):
 361         raise HwsimSkip("SAE not supported")
 362
 363     ifname = dev[0].ifname
 364     sigma = start_sigma_dut(ifname)
 365
 366     try:
 367         ssid = "test-sae"
 368         params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 369         params['wpa_key_mgmt'] = 'SAE'
 370         params["ieee80211w"] = "2"
 371         params['sae_groups'] = '19'
 372         hapd = hostapd.add_ap(apdev[0], params)
 373
 374         sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
 375         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 376         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2,ECGroupID,21 20 19" % (ifname, "test-sae", "12345678"))
 377         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
 378                             timeout=10)
 379         sigma_dut_wait_connected(ifname)
 380         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 381         if dev[0].get_status_field('sae_group') != '19':
 382             raise Exception("Expected default SAE group not used")
 383         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 384
 385         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 386     finally:
 387         stop_sigma_dut(sigma)
 388
 389 def test_sigma_dut_sae_pmkid_include(dev, apdev):
 390     """sigma_dut controlled SAE association with PMKID"""
 391     if "SAE" not in dev[0].get_capability("auth_alg"):
 392         raise HwsimSkip("SAE not supported")
 393
 394     ifname = dev[0].ifname
 395     sigma = start_sigma_dut(ifname)
 396
 397     try:
 398         ssid = "test-sae"
 399         params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 400         params['wpa_key_mgmt'] = 'SAE'
 401         params["ieee80211w"] = "2"
 402         params["sae_confirm_immediate"] = "1"
 403         hapd = hostapd.add_ap(apdev[0], params)
 404
 405         sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
 406         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 407         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2,PMKID_Include,enable" % (ifname, "test-sae", "12345678"))
 408         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
 409                             timeout=10)
 410         sigma_dut_wait_connected(ifname)
 411         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 412     finally:
 413         stop_sigma_dut(sigma)
 414
 415 def test_sigma_dut_sae_password(dev, apdev):
 416     """sigma_dut controlled SAE association and long password"""
 417     if "SAE" not in dev[0].get_capability("auth_alg"):
 418         raise HwsimSkip("SAE not supported")
 419
 420     ifname = dev[0].ifname
 421     sigma = start_sigma_dut(ifname)
 422
 423     try:
 424         ssid = "test-sae"
 425         params = hostapd.wpa2_params(ssid=ssid)
 426         params['sae_password'] = 100*'B'
 427         params['wpa_key_mgmt'] = 'SAE'
 428         params["ieee80211w"] = "2"
 429         hapd = hostapd.add_ap(apdev[0], params)
 430
 431         sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
 432         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 433         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'))
 434         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
 435                             timeout=10)
 436         sigma_dut_wait_connected(ifname)
 437         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 438         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 439         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 440     finally:
 441         stop_sigma_dut(sigma)
 442
 443 def test_sigma_dut_sae_pw_id(dev, apdev):
 444     """sigma_dut controlled SAE association with Password Identifier"""
 445     if "SAE" not in dev[0].get_capability("auth_alg"):
 446         raise HwsimSkip("SAE not supported")
 447
 448     ifname = dev[0].ifname
 449     sigma = start_sigma_dut(ifname)
 450
 451     try:
 452         ssid = "test-sae"
 453         params = hostapd.wpa2_params(ssid=ssid)
 454         params['wpa_key_mgmt'] = 'SAE'
 455         params["ieee80211w"] = "2"
 456         params['sae_password'] = 'secret|id=pw id'
 457         params['sae_groups'] = '19'
 458         hapd = hostapd.add_ap(apdev[0], params)
 459
 460         sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
 461         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 462         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,AKMSuiteType,8;9,PasswordID,pw id" % (ifname, "test-sae", "secret"))
 463         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
 464                             timeout=10)
 465         sigma_dut_wait_connected(ifname)
 466         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 467         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 468     finally:
 469         stop_sigma_dut(sigma)
 470
 471 def test_sigma_dut_sae_pw_id_ft(dev, apdev):
 472     """sigma_dut controlled SAE association with Password Identifier and FT"""
 473     run_sigma_dut_sae_pw_id_ft(dev, apdev)
 474
 475 def test_sigma_dut_sae_pw_id_ft_over_ds(dev, apdev):
 476     """sigma_dut controlled SAE association with Password Identifier and FT-over-DS"""
 477     run_sigma_dut_sae_pw_id_ft(dev, apdev, over_ds=True)
 478
 479 def run_sigma_dut_sae_pw_id_ft(dev, apdev, over_ds=False):
 480     if "SAE" not in dev[0].get_capability("auth_alg"):
 481         raise HwsimSkip("SAE not supported")
 482
 483     ifname = dev[0].ifname
 484     sigma = start_sigma_dut(ifname)
 485
 486     try:
 487         ssid = "test-sae"
 488         params = hostapd.wpa2_params(ssid=ssid)
 489         params['wpa_key_mgmt'] = 'SAE FT-SAE'
 490         params["ieee80211w"] = "2"
 491         params['sae_password'] = ['pw1|id=id1', 'pw2|id=id2', 'pw3', 'pw4|id=id4']
 492         params['mobility_domain'] = 'aabb'
 493         params['ft_over_ds'] = '1' if over_ds else '0'
 494         bssid = apdev[0]['bssid'].replace(':', '')
 495         params['nas_identifier'] = bssid + '.nas.example.com'
 496         params['r1_key_holder'] = bssid
 497         params['pmk_r1_push'] = '0'
 498         params['r0kh'] = 'ff:ff:ff:ff:ff:ff * 00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff'
 499         params['r1kh'] = '00:00:00:00:00:00 00:00:00:00:00:00 00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff'
 500         hapd = hostapd.add_ap(apdev[0], params)
 501
 502         sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
 503         if over_ds:
 504             sigma_dut_cmd_check("sta_preset_testparameters,interface,%s,FT_DS,Enable" % ifname)
 505         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 506         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,AKMSuiteType,8;9,PasswordID,id2" % (ifname, "test-sae", "pw2"))
 507         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
 508                             timeout=10)
 509         sigma_dut_wait_connected(ifname)
 510
 511         bssid = apdev[1]['bssid'].replace(':', '')
 512         params['nas_identifier'] = bssid + '.nas.example.com'
 513         params['r1_key_holder'] = bssid
 514         hapd2 = hostapd.add_ap(apdev[1], params)
 515         bssid = hapd2.own_addr()
 516         sigma_dut_cmd_check("sta_reassoc,interface,%s,Channel,1,bssid,%s" % (ifname, bssid))
 517         dev[0].wait_connected()
 518
 519         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 520         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 521     finally:
 522         stop_sigma_dut(sigma)
 523
 524 def test_sigma_dut_sta_override_rsne(dev, apdev):
 525     """sigma_dut and RSNE override on STA"""
 526     try:
 527         run_sigma_dut_sta_override_rsne(dev, apdev)
 528     finally:
 529         dev[0].set("ignore_old_scan_res", "0")
 530
 531 def run_sigma_dut_sta_override_rsne(dev, apdev):
 532     ifname = dev[0].ifname
 533     sigma = start_sigma_dut(ifname)
 534
 535     try:
 536         ssid = "test-psk"
 537         params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 538         hapd = hostapd.add_ap(apdev[0], params)
 539
 540         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 541
 542         tests = ["30120100000fac040100000fac040100000fac02",
 543                  "30140100000fac040100000fac040100000fac02ffff"]
 544         for test in tests:
 545             sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,type,PSK,passphrase,%s,EncpType,aes-ccmp,KeyMgmtType,wpa2" % (ifname, "test-psk", "12345678"))
 546             sigma_dut_cmd_check("dev_configure_ie,interface,%s,IE_Name,RSNE,Contents,%s" % (ifname, test))
 547             sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-psk"),
 548                                 timeout=10)
 549             sigma_dut_wait_connected(ifname)
 550             sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 551             dev[0].dump_monitor()
 552
 553         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,type,PSK,passphrase,%s,EncpType,aes-ccmp,KeyMgmtType,wpa2" % (ifname, "test-psk", "12345678"))
 554         sigma_dut_cmd_check("dev_configure_ie,interface,%s,IE_Name,RSNE,Contents,300101" % ifname)
 555         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-psk"),
 556                             timeout=10)
 557
 558         ev = dev[0].wait_event(["CTRL-EVENT-ASSOC-REJECT"])
 559         if ev is None:
 560             raise Exception("Association rejection not reported")
 561         if "status_code=40" not in ev:
 562             raise Exception("Unexpected status code: " + ev)
 563
 564         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 565     finally:
 566         stop_sigma_dut(sigma)
 567
 568 def test_sigma_dut_ap_psk(dev, apdev):
 569     """sigma_dut controlled AP"""
 570     with HWSimRadio() as (radio, iface):
 571         sigma = start_sigma_dut(iface)
 572         try:
 573             sigma_dut_cmd_check("ap_reset_default")
 574             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-psk,MODE,11ng")
 575             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-PSK,PSK,12345678")
 576             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 577
 578             dev[0].connect("test-psk", psk="12345678", scan_freq="2412")
 579
 580             sigma_dut_cmd_check("ap_reset_default")
 581         finally:
 582             stop_sigma_dut(sigma)
 583
 584 def test_sigma_dut_ap_pskhex(dev, apdev, params):
 585     """sigma_dut controlled AP and PSKHEX"""
 586     logdir = os.path.join(params['logdir'],
 587                           "sigma_dut_ap_pskhex.sigma-hostapd")
 588     with HWSimRadio() as (radio, iface):
 589         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
 590         try:
 591             psk = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
 592             sigma_dut_cmd_check("ap_reset_default")
 593             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-psk,MODE,11ng")
 594             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-PSK,PSKHEX," + psk)
 595             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 596
 597             dev[0].connect("test-psk", raw_psk=psk, scan_freq="2412")
 598
 599             sigma_dut_cmd_check("ap_reset_default")
 600         finally:
 601             stop_sigma_dut(sigma)
 602
 603 def test_sigma_dut_ap_psk_sha256(dev, apdev, params):
 604     """sigma_dut controlled AP PSK SHA256"""
 605     logdir = os.path.join(params['logdir'],
 606                           "sigma_dut_ap_psk_sha256.sigma-hostapd")
 607     with HWSimRadio() as (radio, iface):
 608         sigma = start_sigma_dut(iface)
 609         try:
 610             sigma_dut_cmd_check("ap_reset_default")
 611             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-psk,MODE,11ng")
 612             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-PSK-256,PSK,12345678")
 613             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 614
 615             dev[0].connect("test-psk", key_mgmt="WPA-PSK-SHA256",
 616                            psk="12345678", scan_freq="2412")
 617
 618             sigma_dut_cmd_check("ap_reset_default")
 619         finally:
 620             stop_sigma_dut(sigma)
 621
 622 def test_sigma_dut_ap_psk_deauth(dev, apdev, params):
 623     """sigma_dut controlled AP and deauth commands"""
 624     logdir = os.path.join(params['logdir'],
 625                           "sigma_dut_ap_psk_deauth.sigma-hostapd")
 626     with HWSimRadio() as (radio, iface):
 627         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
 628         try:
 629             sigma_dut_cmd_check("ap_reset_default")
 630             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-psk,MODE,11ng")
 631             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-PSK,PSK,12345678,PMF,Required")
 632             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 633
 634             dev[0].connect("test-psk", key_mgmt="WPA-PSK-SHA256",
 635                            psk="12345678", ieee80211w="2", scan_freq="2412")
 636             addr = dev[0].own_addr()
 637             dev[0].dump_monitor()
 638
 639             sigma_dut_cmd_check("ap_deauth_sta,NAME,AP,sta_mac_address," + addr)
 640             ev = dev[0].wait_disconnected()
 641             dev[0].dump_monitor()
 642             if "locally_generated=1" in ev:
 643                 raise Exception("Unexpected disconnection reason")
 644             dev[0].wait_connected()
 645             dev[0].dump_monitor()
 646
 647             sigma_dut_cmd_check("ap_deauth_sta,NAME,AP,sta_mac_address," + addr + ",disconnect,silent")
 648             ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
 649             if ev and "locally_generated=1" not in ev:
 650                 raise Exception("Unexpected disconnection")
 651
 652             sigma_dut_cmd_check("ap_reset_default")
 653         finally:
 654             stop_sigma_dut(sigma)
 655
 656 def test_sigma_dut_eap_ttls(dev, apdev, params):
 657     """sigma_dut controlled STA and EAP-TTLS parameters"""
 658     check_domain_match(dev[0])
 659     logdir = params['logdir']
 660
 661     with open("auth_serv/ca.pem", "r") as f:
 662         with open(os.path.join(logdir, "sigma_dut_eap_ttls.ca.pem"), "w") as f2:
 663             f2.write(f.read())
 664
 665     src = "auth_serv/server.pem"
 666     dst = os.path.join(logdir, "sigma_dut_eap_ttls.server.der")
 667     hashdst = os.path.join(logdir, "sigma_dut_eap_ttls.server.pem.sha256")
 668     subprocess.check_call(["openssl", "x509", "-in", src, "-out", dst,
 669                            "-outform", "DER"],
 670                           stderr=open('/dev/null', 'w'))
 671     with open(dst, "rb") as f:
 672         der = f.read()
 673     hash = hashlib.sha256(der).digest()
 674     with open(hashdst, "w") as f:
 675         f.write(binascii.hexlify(hash).decode())
 676
 677     dst = os.path.join(logdir, "sigma_dut_eap_ttls.incorrect.pem.sha256")
 678     with open(dst, "w") as f:
 679         f.write(32*"00")
 680
 681     ssid = "test-wpa2-eap"
 682     params = hostapd.wpa2_eap_params(ssid=ssid)
 683     hapd = hostapd.add_ap(apdev[0], params)
 684
 685     ifname = dev[0].ifname
 686     sigma = start_sigma_dut(ifname, cert_path=logdir)
 687
 688     cmd = "sta_set_security,type,eapttls,interface,%s,ssid,%s,keymgmttype,wpa2,encType,AES-CCMP,PairwiseCipher,AES-CCMP-128,trustedRootCA,sigma_dut_eap_ttls.ca.pem,username,DOMAIN\mschapv2 user,password,password" % (ifname, ssid)
 689
 690     try:
 691         tests = ["",
 692                  ",Domain,server.w1.fi",
 693                  ",DomainSuffix,w1.fi",
 694                  ",DomainSuffix,server.w1.fi",
 695                  ",ServerCert,sigma_dut_eap_ttls.server.pem"]
 696         for extra in tests:
 697             sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
 698             sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 699             sigma_dut_cmd_check(cmd + extra)
 700             sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, ssid),
 701                                 timeout=10)
 702             sigma_dut_wait_connected(ifname)
 703             sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 704             sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 705             sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 706             dev[0].dump_monitor()
 707
 708         tests = [",Domain,w1.fi",
 709                  ",DomainSuffix,example.com",
 710                  ",ServerCert,sigma_dut_eap_ttls.incorrect.pem"]
 711         for extra in tests:
 712             sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
 713             sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 714             sigma_dut_cmd_check(cmd + extra)
 715             sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, ssid),
 716                                 timeout=10)
 717             ev = dev[0].wait_event(["CTRL-EVENT-EAP-TLS-CERT-ERROR"], timeout=10)
 718             if ev is None:
 719                 raise Exception("Server certificate error not reported")
 720             res = sigma_dut_cmd("sta_is_connected,interface," + ifname)
 721             if "connected,1" in res:
 722                 raise Exception("Unexpected connection reported")
 723             sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 724             sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 725             dev[0].dump_monitor()
 726     finally:
 727         stop_sigma_dut(sigma)
 728
 729 def test_sigma_dut_suite_b(dev, apdev, params):
 730     """sigma_dut controlled STA Suite B"""
 731     check_suite_b_192_capa(dev)
 732     logdir = params['logdir']
 733
 734     with open("auth_serv/ec2-ca.pem", "r") as f:
 735         with open(os.path.join(logdir, "suite_b_ca.pem"), "w") as f2:
 736             f2.write(f.read())
 737
 738     with open("auth_serv/ec2-user.pem", "r") as f:
 739         with open("auth_serv/ec2-user.key", "r") as f2:
 740             with open(os.path.join(logdir, "suite_b.pem"), "w") as f3:
 741                 f3.write(f.read())
 742                 f3.write(f2.read())
 743
 744     dev[0].flush_scan_cache()
 745     params = suite_b_as_params()
 746     params['ca_cert'] = 'auth_serv/ec2-ca.pem'
 747     params['server_cert'] = 'auth_serv/ec2-server.pem'
 748     params['private_key'] = 'auth_serv/ec2-server.key'
 749     params['openssl_ciphers'] = 'SUITEB192'
 750     hostapd.add_ap(apdev[1], params)
 751
 752     params = {"ssid": "test-suite-b",
 753               "wpa": "2",
 754               "wpa_key_mgmt": "WPA-EAP-SUITE-B-192",
 755               "rsn_pairwise": "GCMP-256",
 756               "group_mgmt_cipher": "BIP-GMAC-256",
 757               "ieee80211w": "2",
 758               "ieee8021x": "1",
 759               'auth_server_addr': "127.0.0.1",
 760               'auth_server_port': "18129",
 761               'auth_server_shared_secret': "radius",
 762               'nas_identifier': "nas.w1.fi"}
 763     hapd = hostapd.add_ap(apdev[0], params)
 764
 765     ifname = dev[0].ifname
 766     sigma = start_sigma_dut(ifname, cert_path=logdir)
 767
 768     try:
 769         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,PMF" % ifname)
 770         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 771         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,clientCertificate,suite_b.pem,trustedRootCA,suite_b_ca.pem,CertType,ECC" % (ifname, "test-suite-b"))
 772         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-suite-b"),
 773                             timeout=10)
 774         sigma_dut_wait_connected(ifname)
 775         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 776         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 777         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 778     finally:
 779         stop_sigma_dut(sigma)
 780
 781 def test_sigma_dut_suite_b_rsa(dev, apdev, params):
 782     """sigma_dut controlled STA Suite B (RSA)"""
 783     check_suite_b_192_capa(dev)
 784     logdir = params['logdir']
 785
 786     with open("auth_serv/rsa3072-ca.pem", "r") as f:
 787         with open(os.path.join(logdir, "suite_b_ca_rsa.pem"), "w") as f2:
 788             f2.write(f.read())
 789
 790     with open("auth_serv/rsa3072-user.pem", "r") as f:
 791         with open("auth_serv/rsa3072-user.key", "r") as f2:
 792             with open(os.path.join(logdir, "suite_b_rsa.pem"), "w") as f3:
 793                 f3.write(f.read())
 794                 f3.write(f2.read())
 795
 796     dev[0].flush_scan_cache()
 797     params = suite_b_192_rsa_ap_params()
 798     hapd = hostapd.add_ap(apdev[0], params)
 799
 800     ifname = dev[0].ifname
 801     sigma = start_sigma_dut(ifname, cert_path=logdir)
 802
 803     cmd = "sta_set_security,type,eaptls,interface,%s,ssid,%s,PairwiseCipher,AES-GCMP-256,GroupCipher,AES-GCMP-256,GroupMgntCipher,BIP-GMAC-256,keymgmttype,SuiteB,clientCertificate,suite_b_rsa.pem,trustedRootCA,suite_b_ca_rsa.pem,CertType,RSA" % (ifname, "test-suite-b")
 804
 805     try:
 806         tests = ["",
 807                  ",TLSCipher,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384",
 808                  ",TLSCipher,TLS_DHE_RSA_WITH_AES_256_GCM_SHA384"]
 809         for extra in tests:
 810             sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,PMF" % ifname)
 811             sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
 812             sigma_dut_cmd_check(cmd + extra)
 813             sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-suite-b"),
 814                                 timeout=10)
 815             sigma_dut_wait_connected(ifname)
 816             sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
 817             sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
 818             sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
 819     finally:
 820         stop_sigma_dut(sigma)
 821
 822 def test_sigma_dut_ap_suite_b(dev, apdev, params):
 823     """sigma_dut controlled AP Suite B"""
 824     check_suite_b_192_capa(dev)
 825     logdir = os.path.join(params['logdir'],
 826                           "sigma_dut_ap_suite_b.sigma-hostapd")
 827     params = suite_b_as_params()
 828     params['ca_cert'] = 'auth_serv/ec2-ca.pem'
 829     params['server_cert'] = 'auth_serv/ec2-server.pem'
 830     params['private_key'] = 'auth_serv/ec2-server.key'
 831     params['openssl_ciphers'] = 'SUITEB192'
 832     hostapd.add_ap(apdev[1], params)
 833     with HWSimRadio() as (radio, iface):
 834         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
 835         try:
 836             sigma_dut_cmd_check("ap_reset_default")
 837             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-suite-b,MODE,11ng")
 838             sigma_dut_cmd_check("ap_set_radius,NAME,AP,IPADDR,127.0.0.1,PORT,18129,PASSWORD,radius")
 839             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,SuiteB")
 840             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 841
 842             dev[0].connect("test-suite-b", key_mgmt="WPA-EAP-SUITE-B-192",
 843                            ieee80211w="2",
 844                            openssl_ciphers="SUITEB192",
 845                            eap="TLS", identity="tls user",
 846                            ca_cert="auth_serv/ec2-ca.pem",
 847                            client_cert="auth_serv/ec2-user.pem",
 848                            private_key="auth_serv/ec2-user.key",
 849                            pairwise="GCMP-256", group="GCMP-256",
 850                            scan_freq="2412")
 851
 852             sigma_dut_cmd_check("ap_reset_default")
 853         finally:
 854             stop_sigma_dut(sigma)
 855
 856 def test_sigma_dut_ap_cipher_gcmp_128(dev, apdev, params):
 857     """sigma_dut controlled AP with GCMP-128/BIP-GMAC-128 cipher"""
 858     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-GCMP-128", "BIP-GMAC-128",
 859                             "GCMP")
 860
 861 def test_sigma_dut_ap_cipher_gcmp_256(dev, apdev, params):
 862     """sigma_dut controlled AP with GCMP-256/BIP-GMAC-256 cipher"""
 863     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-GCMP-256", "BIP-GMAC-256",
 864                             "GCMP-256")
 865
 866 def test_sigma_dut_ap_cipher_ccmp_128(dev, apdev, params):
 867     """sigma_dut controlled AP with CCMP-128/BIP-CMAC-128 cipher"""
 868     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-CCMP-128", "BIP-CMAC-128",
 869                             "CCMP")
 870
 871 def test_sigma_dut_ap_cipher_ccmp_256(dev, apdev, params):
 872     """sigma_dut controlled AP with CCMP-256/BIP-CMAC-256 cipher"""
 873     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-CCMP-256", "BIP-CMAC-256",
 874                             "CCMP-256")
 875
 876 def test_sigma_dut_ap_cipher_ccmp_gcmp_1(dev, apdev, params):
 877     """sigma_dut controlled AP with CCMP-128+GCMP-256 ciphers (1)"""
 878     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-CCMP-128 AES-GCMP-256",
 879                             "BIP-GMAC-256", "CCMP")
 880
 881 def test_sigma_dut_ap_cipher_ccmp_gcmp_2(dev, apdev, params):
 882     """sigma_dut controlled AP with CCMP-128+GCMP-256 ciphers (2)"""
 883     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-CCMP-128 AES-GCMP-256",
 884                             "BIP-GMAC-256", "GCMP-256", "CCMP")
 885
 886 def test_sigma_dut_ap_cipher_gcmp_256_group_ccmp(dev, apdev, params):
 887     """sigma_dut controlled AP with GCMP-256/CCMP/BIP-GMAC-256 cipher"""
 888     run_sigma_dut_ap_cipher(dev, apdev, params, "AES-GCMP-256", "BIP-GMAC-256",
 889                             "GCMP-256", "CCMP", "AES-CCMP-128")
 890
 891 def run_sigma_dut_ap_cipher(dev, apdev, params, ap_pairwise, ap_group_mgmt,
 892                             sta_cipher, sta_cipher_group=None, ap_group=None):
 893     check_suite_b_192_capa(dev)
 894     logdir = os.path.join(params['logdir'],
 895                           "sigma_dut_ap_cipher.sigma-hostapd")
 896     params = suite_b_as_params()
 897     params['ca_cert'] = 'auth_serv/ec2-ca.pem'
 898     params['server_cert'] = 'auth_serv/ec2-server.pem'
 899     params['private_key'] = 'auth_serv/ec2-server.key'
 900     params['openssl_ciphers'] = 'SUITEB192'
 901     hostapd.add_ap(apdev[1], params)
 902     with HWSimRadio() as (radio, iface):
 903         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
 904         try:
 905             sigma_dut_cmd_check("ap_reset_default")
 906             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-suite-b,MODE,11ng")
 907             sigma_dut_cmd_check("ap_set_radius,NAME,AP,IPADDR,127.0.0.1,PORT,18129,PASSWORD,radius")
 908             cmd = "ap_set_security,NAME,AP,KEYMGNT,SuiteB,PMF,Required,PairwiseCipher,%s,GroupMgntCipher,%s" % (ap_pairwise, ap_group_mgmt)
 909             if ap_group:
 910                 cmd += ",GroupCipher,%s" % ap_group
 911             sigma_dut_cmd_check(cmd)
 912             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 913
 914             if sta_cipher_group is None:
 915                 sta_cipher_group = sta_cipher
 916             dev[0].connect("test-suite-b", key_mgmt="WPA-EAP-SUITE-B-192",
 917                            ieee80211w="2",
 918                            openssl_ciphers="SUITEB192",
 919                            eap="TLS", identity="tls user",
 920                            ca_cert="auth_serv/ec2-ca.pem",
 921                            client_cert="auth_serv/ec2-user.pem",
 922                            private_key="auth_serv/ec2-user.key",
 923                            pairwise=sta_cipher, group=sta_cipher_group,
 924                            scan_freq="2412")
 925
 926             sigma_dut_cmd_check("ap_reset_default")
 927         finally:
 928             stop_sigma_dut(sigma)
 929
 930 def test_sigma_dut_ap_override_rsne(dev, apdev):
 931     """sigma_dut controlled AP overriding RSNE"""
 932     with HWSimRadio() as (radio, iface):
 933         sigma = start_sigma_dut(iface)
 934         try:
 935             sigma_dut_cmd_check("ap_reset_default")
 936             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-psk,MODE,11ng")
 937             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-PSK,PSK,12345678")
 938             sigma_dut_cmd_check("dev_configure_ie,NAME,AP,interface,%s,IE_Name,RSNE,Contents,30180100000fac040200ffffffff000fac040100000fac020c00" % iface)
 939             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 940
 941             dev[0].connect("test-psk", psk="12345678", scan_freq="2412")
 942
 943             sigma_dut_cmd_check("ap_reset_default")
 944         finally:
 945             stop_sigma_dut(sigma)
 946
 947 def test_sigma_dut_ap_sae(dev, apdev, params):
 948     """sigma_dut controlled AP with SAE"""
 949     logdir = os.path.join(params['logdir'],
 950                           "sigma_dut_ap_sae.sigma-hostapd")
 951     if "SAE" not in dev[0].get_capability("auth_alg"):
 952         raise HwsimSkip("SAE not supported")
 953     with HWSimRadio() as (radio, iface):
 954         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
 955         try:
 956             sigma_dut_cmd_check("ap_reset_default")
 957             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
 958             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK,12345678")
 959             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 960
 961             dev[0].request("SET sae_groups ")
 962             id = dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
 963                                 ieee80211w="2", scan_freq="2412")
 964             if dev[0].get_status_field('sae_group') != '19':
 965                 raise Exception("Expected default SAE group not used")
 966
 967             res = sigma_dut_cmd_check("ap_get_parameter,name,AP,STA_MAC_Address,%s,Parameter,PMK" % dev[0].own_addr())
 968             logger.info("Reported PMK: " + res)
 969             if ",PMK," not in res:
 970                 raise Exception("PMK not reported");
 971             if dev[0].get_pmk(id) != res.split(',')[3]:
 972                 raise Exception("Mismatch in reported PMK")
 973
 974             sigma_dut_cmd_check("ap_reset_default")
 975         finally:
 976             stop_sigma_dut(sigma)
 977
 978 def test_sigma_dut_ap_sae_confirm_immediate(dev, apdev, params):
 979     """sigma_dut controlled AP with SAE Confirm immediate"""
 980     logdir = os.path.join(params['logdir'],
 981                           "sigma_dut_ap_sae_confirm_immediate.sigma-hostapd")
 982     if "SAE" not in dev[0].get_capability("auth_alg"):
 983         raise HwsimSkip("SAE not supported")
 984     with HWSimRadio() as (radio, iface):
 985         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
 986         try:
 987             sigma_dut_cmd_check("ap_reset_default")
 988             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
 989             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK,12345678,SAE_Confirm_Immediate,enable")
 990             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
 991
 992             dev[0].request("SET sae_groups ")
 993             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
 994                            ieee80211w="2", scan_freq="2412")
 995             if dev[0].get_status_field('sae_group') != '19':
 996                 raise Exception("Expected default SAE group not used")
 997
 998             sigma_dut_cmd_check("ap_reset_default")
 999         finally:
1000             stop_sigma_dut(sigma)
1001
1002 def test_sigma_dut_ap_sae_password(dev, apdev, params):
1003     """sigma_dut controlled AP with SAE and long password"""
1004     logdir = os.path.join(params['logdir'],
1005                           "sigma_dut_ap_sae_password.sigma-hostapd")
1006     if "SAE" not in dev[0].get_capability("auth_alg"):
1007         raise HwsimSkip("SAE not supported")
1008     with HWSimRadio() as (radio, iface):
1009         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
1010         try:
1011             sigma_dut_cmd_check("ap_reset_default")
1012             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
1013             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK," + 100*'C')
1014             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
1015
1016             dev[0].request("SET sae_groups ")
1017             dev[0].connect("test-sae", key_mgmt="SAE", sae_password=100*'C',
1018                            ieee80211w="2", scan_freq="2412")
1019             if dev[0].get_status_field('sae_group') != '19':
1020                 raise Exception("Expected default SAE group not used")
1021
1022             sigma_dut_cmd_check("ap_reset_default")
1023         finally:
1024             stop_sigma_dut(sigma)
1025
1026 def test_sigma_dut_ap_sae_pw_id(dev, apdev, params):
1027     """sigma_dut controlled AP with SAE Password Identifier"""
1028     logdir = os.path.join(params['logdir'],
1029                           "sigma_dut_ap_sae_pw_id.sigma-hostapd")
1030     conffile = os.path.join(params['logdir'],
1031                             "sigma_dut_ap_sae_pw_id.sigma-conf")
1032     if "SAE" not in dev[0].get_capability("auth_alg"):
1033         raise HwsimSkip("SAE not supported")
1034     with HWSimRadio() as (radio, iface):
1035         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
1036         try:
1037             sigma_dut_cmd_check("ap_reset_default")
1038             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
1039             sigma_dut_cmd_check("ap_set_security,NAME,AP,AKMSuiteType,8,SAEPasswords,pw1:id1;pw2:id2;pw3;pw4:id4,PMF,Required")
1040             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
1041
1042             with open("/tmp/sigma_dut-ap.conf", "rb") as f:
1043                 with open(conffile, "wb") as f2:
1044                     f2.write(f.read())
1045
1046             dev[0].request("SET sae_groups ")
1047             tests = [("pw1", "id1"),
1048                      ("pw2", "id2"),
1049                      ("pw3", None),
1050                      ("pw4", "id4")]
1051             for pw, pw_id in tests:
1052                 dev[0].connect("test-sae", key_mgmt="SAE", sae_password=pw,
1053                                sae_password_id=pw_id,
1054                                ieee80211w="2", scan_freq="2412")
1055                 dev[0].request("REMOVE_NETWORK all")
1056                 dev[0].wait_disconnected()
1057
1058             sigma_dut_cmd_check("ap_reset_default")
1059         finally:
1060             stop_sigma_dut(sigma)
1061
1062 def test_sigma_dut_ap_sae_pw_id_ft(dev, apdev, params):
1063     """sigma_dut controlled AP with SAE Password Identifier and FT"""
1064     logdir = os.path.join(params['logdir'],
1065                           "sigma_dut_ap_sae_pw_id_ft.sigma-hostapd")
1066     conffile = os.path.join(params['logdir'],
1067                             "sigma_dut_ap_sae_pw_id_ft.sigma-conf")
1068     if "SAE" not in dev[0].get_capability("auth_alg"):
1069         raise HwsimSkip("SAE not supported")
1070     with HWSimRadio() as (radio, iface):
1071         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
1072         try:
1073             sigma_dut_cmd_check("ap_reset_default")
1074             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng,DOMAIN,aabb")
1075             sigma_dut_cmd_check("ap_set_security,NAME,AP,AKMSuiteType,8;9,SAEPasswords,pw1:id1;pw2:id2;pw3;pw4:id4,PMF,Required")
1076             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
1077
1078             with open("/tmp/sigma_dut-ap.conf", "rb") as f:
1079                 with open(conffile, "wb") as f2:
1080                     f2.write(f.read())
1081
1082             dev[0].request("SET sae_groups ")
1083             tests = [("pw1", "id1", "SAE"),
1084                      ("pw2", "id2", "FT-SAE"),
1085                      ("pw3", None, "FT-SAE"),
1086                      ("pw4", "id4", "SAE")]
1087             for pw, pw_id, key_mgmt in tests:
1088                 dev[0].connect("test-sae", key_mgmt=key_mgmt, sae_password=pw,
1089                                sae_password_id=pw_id,
1090                                ieee80211w="2", scan_freq="2412")
1091                 dev[0].request("REMOVE_NETWORK all")
1092                 dev[0].wait_disconnected()
1093
1094             sigma_dut_cmd_check("ap_reset_default")
1095         finally:
1096             stop_sigma_dut(sigma)
1097
1098 def test_sigma_dut_ap_sae_group(dev, apdev, params):
1099     """sigma_dut controlled AP with SAE and specific group"""
1100     logdir = os.path.join(params['logdir'],
1101                           "sigma_dut_ap_sae_group.sigma-hostapd")
1102     if "SAE" not in dev[0].get_capability("auth_alg"):
1103         raise HwsimSkip("SAE not supported")
1104     with HWSimRadio() as (radio, iface):
1105         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
1106         try:
1107             sigma_dut_cmd_check("ap_reset_default")
1108             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
1109             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK,12345678,ECGroupID,20")
1110             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
1111
1112             dev[0].request("SET sae_groups ")
1113             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
1114                            ieee80211w="2", scan_freq="2412")
1115             if dev[0].get_status_field('sae_group') != '20':
1116                 raise Exception("Expected SAE group not used")
1117
1118             sigma_dut_cmd_check("ap_reset_default")
1119         finally:
1120             stop_sigma_dut(sigma)
1121
1122 def test_sigma_dut_ap_psk_sae(dev, apdev, params):
1123     """sigma_dut controlled AP with PSK+SAE"""
1124     if "SAE" not in dev[0].get_capability("auth_alg"):
1125         raise HwsimSkip("SAE not supported")
1126     logdir = os.path.join(params['logdir'],
1127                           "sigma_dut_ap_psk_sae.sigma-hostapd")
1128     with HWSimRadio() as (radio, iface):
1129         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
1130         try:
1131             sigma_dut_cmd_check("ap_reset_default")
1132             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
1133             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-PSK-SAE,PSK,12345678")
1134             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
1135
1136             dev[2].request("SET sae_groups ")
1137             dev[2].connect("test-sae", key_mgmt="SAE", psk="12345678",
1138                            scan_freq="2412", ieee80211w="0", wait_connect=False)
1139             dev[0].request("SET sae_groups ")
1140             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
1141                            scan_freq="2412", ieee80211w="2")
1142             dev[1].connect("test-sae", psk="12345678", scan_freq="2412")
1143
1144             ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED"], timeout=0.1)
1145             dev[2].request("DISCONNECT")
1146             if ev is not None:
1147                 raise Exception("Unexpected connection without PMF")
1148
1149             sigma_dut_cmd_check("ap_reset_default")
1150         finally:
1151             stop_sigma_dut(sigma)
1152
1153 def test_sigma_dut_ap_psk_sae_ft(dev, apdev, params):
1154     """sigma_dut controlled AP with PSK, SAE, FT"""
1155     logdir = os.path.join(params['logdir'],
1156                           "sigma_dut_ap_psk_sae_ft.sigma-hostapd")
1157     conffile = os.path.join(params['logdir'],
1158                             "sigma_dut_ap_psk_sae_ft.sigma-conf")
1159     if "SAE" not in dev[0].get_capability("auth_alg"):
1160         raise HwsimSkip("SAE not supported")
1161     with HWSimRadio() as (radio, iface):
1162         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
1163         try:
1164             sigma_dut_cmd_check("ap_reset_default,NAME,AP,Program,WPA3")
1165             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae-psk,MODE,11ng,DOMAIN,aabb")
1166             sigma_dut_cmd_check("ap_set_security,NAME,AP,AKMSuiteType,2;4;6;8;9,PSK,12345678,PairwiseCipher,AES-CCMP-128,GroupCipher,AES-CCMP-128")
1167             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,DOMAIN,0101,FT_OA,Enable")
1168             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,FT_BSS_LIST," + apdev[1]['bssid'])
1169             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
1170
1171             with open("/tmp/sigma_dut-ap.conf", "rb") as f:
1172                 with open(conffile, "wb") as f2:
1173                     f2.write(f.read())
1174
1175             dev[0].request("SET sae_groups ")
1176             dev[0].connect("test-sae-psk", key_mgmt="SAE FT-SAE",
1177                            sae_password="12345678", scan_freq="2412")
1178             dev[1].connect("test-sae-psk", key_mgmt="WPA-PSK FT-PSK",
1179                            psk="12345678", scan_freq="2412")
1180             dev[2].connect("test-sae-psk", key_mgmt="WPA-PSK",
1181                            psk="12345678", scan_freq="2412")
1182
1183             sigma_dut_cmd_check("ap_reset_default")
1184         finally:
1185             stop_sigma_dut(sigma)
1186
1187 def test_sigma_dut_owe(dev, apdev):
1188     """sigma_dut controlled OWE station"""
1189     try:
1190         run_sigma_dut_owe(dev, apdev)
1191     finally:
1192         dev[0].set("ignore_old_scan_res", "0")
1193
1194 def run_sigma_dut_owe(dev, apdev):
1195     if "OWE" not in dev[0].get_capability("key_mgmt"):
1196         raise HwsimSkip("OWE not supported")
1197
1198     ifname = dev[0].ifname
1199     sigma = start_sigma_dut(ifname)
1200
1201     try:
1202         params = {"ssid": "owe",
1203                   "wpa": "2",
1204                   "wpa_key_mgmt": "OWE",
1205                   "ieee80211w": "2",
1206                   "rsn_pairwise": "CCMP"}
1207         hapd = hostapd.add_ap(apdev[0], params)
1208         bssid = hapd.own_addr()
1209
1210         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
1211         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
1212         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,owe,Type,OWE" % ifname)
1213         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,owe,channel,1" % ifname,
1214                             timeout=10)
1215         sigma_dut_wait_connected(ifname)
1216         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
1217         res = sigma_dut_cmd_check("sta_get_parameter,interface,%s,Parameter,PMK" % ifname)
1218         logger.info("Reported PMK: " + res)
1219         if ",PMK," not in res:
1220             raise Exception("PMK not reported");
1221         if hapd.request("GET_PMK " + dev[0].own_addr()) != res.split(',')[3]:
1222             raise Exception("Mismatch in reported PMK")
1223
1224         dev[0].dump_monitor()
1225         sigma_dut_cmd("sta_reassoc,interface,%s,Channel,1,bssid,%s" % (ifname, bssid))
1226         dev[0].wait_connected()
1227         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
1228         dev[0].wait_disconnected()
1229         dev[0].dump_monitor()
1230
1231         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
1232         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
1233         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,owe,Type,OWE,ECGroupID,20" % ifname)
1234         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,owe,channel,1" % ifname,
1235                             timeout=10)
1236         sigma_dut_wait_connected(ifname)
1237         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
1238         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
1239         dev[0].wait_disconnected()
1240         dev[0].dump_monitor()
1241
1242         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
1243         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
1244         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,owe,Type,OWE,ECGroupID,0" % ifname)
1245         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,owe,channel,1" % ifname,
1246                             timeout=10)
1247         ev = dev[0].wait_event(["CTRL-EVENT-ASSOC-REJECT"], timeout=10)
1248         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
1249         if ev is None:
1250             raise Exception("Association not rejected")
1251         if "status_code=77" not in ev:
1252             raise Exception("Unexpected rejection reason: " + ev)
1253
1254         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
1255     finally:
1256         stop_sigma_dut(sigma)
1257
1258 def test_sigma_dut_ap_owe(dev, apdev, params):
1259     """sigma_dut controlled AP with OWE"""
1260     logdir = os.path.join(params['logdir'],
1261                           "sigma_dut_ap_owe.sigma-hostapd")
1262     if "OWE" not in dev[0].get_capability("key_mgmt"):
1263         raise HwsimSkip("OWE not supported")
1264     with HWSimRadio() as (radio, iface):
1265         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
1266         try:
1267             sigma_dut_cmd_check("ap_reset_default,NAME,AP,Program,WPA3")
1268             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,owe,MODE,11ng")
1269             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,OWE")
1270             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
1271
1272             id = dev[0].connect("owe", key_mgmt="OWE", ieee80211w="2",
1273                                 scan_freq="2412")
1274
1275             res = sigma_dut_cmd_check("ap_get_parameter,name,AP,STA_MAC_Address,%s,Parameter,PMK" % dev[0].own_addr())
1276             logger.info("Reported PMK: " + res)
1277             if ",PMK," not in res:
1278                 raise Exception("PMK not reported");
1279             if dev[0].get_pmk(id) != res.split(',')[3]:
1280                 raise Exception("Mismatch in reported PMK")
1281
1282             sigma_dut_cmd_check("ap_reset_default")
1283         finally:
1284             stop_sigma_dut(sigma)
1285
1286 def test_sigma_dut_ap_owe_ecgroupid(dev, apdev):
1287     """sigma_dut controlled AP with OWE and ECGroupID"""
1288     if "OWE" not in dev[0].get_capability("key_mgmt"):
1289         raise HwsimSkip("OWE not supported")
1290     with HWSimRadio() as (radio, iface):
1291         sigma = start_sigma_dut(iface)
1292         try:
1293             sigma_dut_cmd_check("ap_reset_default,NAME,AP,Program,WPA3")
1294             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,owe,MODE,11ng")
1295             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,OWE,ECGroupID,20 21,PMF,Required")
1296             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
1297
1298             dev[0].connect("owe", key_mgmt="OWE", ieee80211w="2",
1299                            owe_group="20", scan_freq="2412")
1300             dev[0].request("REMOVE_NETWORK all")
1301             dev[0].wait_disconnected()
1302
1303             dev[0].connect("owe", key_mgmt="OWE", ieee80211w="2",
1304                            owe_group="21", scan_freq="2412")
1305             dev[0].request("REMOVE_NETWORK all")
1306             dev[0].wait_disconnected()
1307
1308             dev[0].connect("owe", key_mgmt="OWE", ieee80211w="2",
1309                            owe_group="19", scan_freq="2412", wait_connect=False)
1310             ev = dev[0].wait_event(["CTRL-EVENT-ASSOC-REJECT"], timeout=10)
1311             dev[0].request("DISCONNECT")
1312             if ev is None:
1313                 raise Exception("Association not rejected")
1314             if "status_code=77" not in ev:
1315                 raise Exception("Unexpected rejection reason: " + ev)
1316             dev[0].dump_monitor()
1317
1318             sigma_dut_cmd_check("ap_reset_default")
1319         finally:
1320             stop_sigma_dut(sigma)
1321
1322 def test_sigma_dut_ap_owe_transition_mode(dev, apdev, params):
1323     """sigma_dut controlled AP with OWE and transition mode"""
1324     if "OWE" not in dev[0].get_capability("key_mgmt"):
1325         raise HwsimSkip("OWE not supported")
1326     logdir = os.path.join(params['logdir'],
1327                           "sigma_dut_ap_owe_transition_mode.sigma-hostapd")
1328     with HWSimRadio() as (radio, iface):
1329         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
1330         try:
1331             sigma_dut_cmd_check("ap_reset_default,NAME,AP,Program,WPA3")
1332             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,WLAN_TAG,1,CHANNEL,1,SSID,owe,MODE,11ng")
1333             sigma_dut_cmd_check("ap_set_security,NAME,AP,WLAN_TAG,1,KEYMGNT,OWE")
1334             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,WLAN_TAG,2,CHANNEL,1,SSID,owe,MODE,11ng")
1335             sigma_dut_cmd_check("ap_set_security,NAME,AP,WLAN_TAG,2,KEYMGNT,NONE")
1336             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
1337
1338             res1 = sigma_dut_cmd_check("ap_get_mac_address,NAME,AP,WLAN_TAG,1,Interface,24G")
1339             res2 = sigma_dut_cmd_check("ap_get_mac_address,NAME,AP,WLAN_TAG,2,Interface,24G")
1340
1341             dev[0].connect("owe", key_mgmt="OWE", ieee80211w="2",
1342                            scan_freq="2412")
1343             dev[1].connect("owe", key_mgmt="NONE", scan_freq="2412")
1344             if dev[0].get_status_field('bssid') not in res1:
1345                 raise Exception("Unexpected ap_get_mac_address WLAN_TAG,1: " + res1)
1346             if dev[1].get_status_field('bssid') not in res2:
1347                 raise Exception("Unexpected ap_get_mac_address WLAN_TAG,2: " + res2)
1348
1349             sigma_dut_cmd_check("ap_reset_default")
1350         finally:
1351             stop_sigma_dut(sigma)
1352
1353 def test_sigma_dut_ap_owe_transition_mode_2(dev, apdev, params):
1354     """sigma_dut controlled AP with OWE and transition mode (2)"""
1355     if "OWE" not in dev[0].get_capability("key_mgmt"):
1356         raise HwsimSkip("OWE not supported")
1357     logdir = os.path.join(params['logdir'],
1358                           "sigma_dut_ap_owe_transition_mode_2.sigma-hostapd")
1359     with HWSimRadio() as (radio, iface):
1360         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
1361         try:
1362             sigma_dut_cmd_check("ap_reset_default,NAME,AP,Program,WPA3")
1363             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,WLAN_TAG,1,CHANNEL,1,SSID,owe,MODE,11ng")
1364             sigma_dut_cmd_check("ap_set_security,NAME,AP,WLAN_TAG,1,KEYMGNT,NONE")
1365             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,WLAN_TAG,2,CHANNEL,1,MODE,11ng")
1366             sigma_dut_cmd_check("ap_set_security,NAME,AP,WLAN_TAG,2,KEYMGNT,OWE")
1367             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
1368
1369             res1 = sigma_dut_cmd_check("ap_get_mac_address,NAME,AP,WLAN_TAG,1,Interface,24G")
1370             res2 = sigma_dut_cmd_check("ap_get_mac_address,NAME,AP,WLAN_TAG,2,Interface,24G")
1371
1372             dev[0].connect("owe", key_mgmt="OWE", ieee80211w="2",
1373                            scan_freq="2412")
1374             dev[1].connect("owe", key_mgmt="NONE", scan_freq="2412")
1375             if dev[0].get_status_field('bssid') not in res2:
1376                 raise Exception("Unexpected ap_get_mac_address WLAN_TAG,2: " + res1)
1377             if dev[1].get_status_field('bssid') not in res1:
1378                 raise Exception("Unexpected ap_get_mac_address WLAN_TAG,1: " + res2)
1379
1380             sigma_dut_cmd_check("ap_reset_default")
1381         finally:
1382             stop_sigma_dut(sigma)
1383
1384 def dpp_init_enrollee(dev, id1, enrollee_role):
1385     logger.info("Starting DPP initiator/enrollee in a thread")
1386     time.sleep(1)
1387     cmd = "DPP_AUTH_INIT peer=%d role=enrollee" % id1
1388     if enrollee_role == "Configurator":
1389         cmd += " netrole=configurator"
1390     if "OK" not in dev.request(cmd):
1391         raise Exception("Failed to initiate DPP Authentication")
1392     ev = dev.wait_event(["DPP-CONF-RECEIVED"], timeout=5)
1393     if ev is None:
1394         raise Exception("DPP configuration not completed (Enrollee)")
1395     logger.info("DPP initiator/enrollee done")
1396
1397 def test_sigma_dut_dpp_qr_resp_1(dev, apdev):
1398     """sigma_dut DPP/QR responder (conf index 1)"""
1399     run_sigma_dut_dpp_qr_resp(dev, apdev, 1)
1400
1401 def test_sigma_dut_dpp_qr_resp_2(dev, apdev):
1402     """sigma_dut DPP/QR responder (conf index 2)"""
1403     run_sigma_dut_dpp_qr_resp(dev, apdev, 2)
1404
1405 def test_sigma_dut_dpp_qr_resp_3(dev, apdev):
1406     """sigma_dut DPP/QR responder (conf index 3)"""
1407     run_sigma_dut_dpp_qr_resp(dev, apdev, 3)
1408
1409 def test_sigma_dut_dpp_qr_resp_4(dev, apdev):
1410     """sigma_dut DPP/QR responder (conf index 4)"""
1411     run_sigma_dut_dpp_qr_resp(dev, apdev, 4)
1412
1413 def test_sigma_dut_dpp_qr_resp_5(dev, apdev):
1414     """sigma_dut DPP/QR responder (conf index 5)"""
1415     run_sigma_dut_dpp_qr_resp(dev, apdev, 5)
1416
1417 def test_sigma_dut_dpp_qr_resp_6(dev, apdev):
1418     """sigma_dut DPP/QR responder (conf index 6)"""
1419     run_sigma_dut_dpp_qr_resp(dev, apdev, 6)
1420
1421 def test_sigma_dut_dpp_qr_resp_7(dev, apdev):
1422     """sigma_dut DPP/QR responder (conf index 7)"""
1423     run_sigma_dut_dpp_qr_resp(dev, apdev, 7)
1424
1425 def test_sigma_dut_dpp_qr_resp_8(dev, apdev):
1426     """sigma_dut DPP/QR responder (conf index 8)"""
1427     run_sigma_dut_dpp_qr_resp(dev, apdev, 8)
1428
1429 def test_sigma_dut_dpp_qr_resp_9(dev, apdev):
1430     """sigma_dut DPP/QR responder (conf index 9)"""
1431     run_sigma_dut_dpp_qr_resp(dev, apdev, 9)
1432
1433 def test_sigma_dut_dpp_qr_resp_10(dev, apdev):
1434     """sigma_dut DPP/QR responder (conf index 10)"""
1435     run_sigma_dut_dpp_qr_resp(dev, apdev, 10)
1436
1437 def test_sigma_dut_dpp_qr_resp_chan_list(dev, apdev):
1438     """sigma_dut DPP/QR responder (channel list override)"""
1439     run_sigma_dut_dpp_qr_resp(dev, apdev, 1, chan_list='81/2 81/6 81/1',
1440                               listen_chan=2)
1441
1442 def test_sigma_dut_dpp_qr_resp_status_query(dev, apdev):
1443     """sigma_dut DPP/QR responder status query"""
1444     check_dpp_capab(dev[1])
1445     params = hostapd.wpa2_params(ssid="DPPNET01",
1446                                  passphrase="ThisIsDppPassphrase")
1447     hapd = hostapd.add_ap(apdev[0], params)
1448
1449     try:
1450         dev[1].set("dpp_config_processing", "2")
1451         run_sigma_dut_dpp_qr_resp(dev, apdev, 3, status_query=True)
1452     finally:
1453         dev[1].set("dpp_config_processing", "0", allow_fail=True)
1454
1455 def test_sigma_dut_dpp_qr_resp_configurator(dev, apdev):
1456     """sigma_dut DPP/QR responder (configurator provisioning)"""
1457     run_sigma_dut_dpp_qr_resp(dev, apdev, -1, enrollee_role="Configurator")
1458
1459 def run_sigma_dut_dpp_qr_resp(dev, apdev, conf_idx, chan_list=None,
1460                               listen_chan=None, status_query=False,
1461                               enrollee_role="STA"):
1462     check_dpp_capab(dev[0])
1463     check_dpp_capab(dev[1])
1464     sigma = start_sigma_dut(dev[0].ifname)
1465     try:
1466         cmd = "dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR"
1467         if chan_list:
1468             cmd += ",DPPChannelList," + chan_list
1469         res = sigma_dut_cmd(cmd)
1470         if "status,COMPLETE" not in res:
1471             raise Exception("dev_exec_action did not succeed: " + res)
1472         hex = res.split(',')[3]
1473         uri = from_hex(hex)
1474         logger.info("URI from sigma_dut: " + uri)
1475
1476         id1 = dev[1].dpp_qr_code(uri)
1477
1478         t = threading.Thread(target=dpp_init_enrollee, args=(dev[1], id1,
1479                                                              enrollee_role))
1480         t.start()
1481         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPAuthDirection,Single,DPPProvisioningRole,Configurator,DPPConfEnrolleeRole,%s,DPPSigningKeyECC,P-256,DPPBS,QR,DPPTimeout,6" % enrollee_role
1482         if conf_idx is not None:
1483             cmd += ",DPPConfIndex,%d" % conf_idx
1484         if listen_chan:
1485             cmd += ",DPPListenChannel," + str(listen_chan)
1486         if status_query:
1487             cmd += ",DPPStatusQuery,Yes"
1488         res = sigma_dut_cmd(cmd, timeout=10)
1489         t.join()
1490         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
1491             raise Exception("Unexpected result: " + res)
1492         if status_query and "StatusResult,0" not in res:
1493             raise Exception("Status query did not succeed: " + res)
1494     finally:
1495         stop_sigma_dut(sigma)
1496
1497 def test_sigma_dut_dpp_qr_init_enrollee(dev, apdev):
1498     """sigma_dut DPP/QR initiator as Enrollee"""
1499     check_dpp_capab(dev[0])
1500     check_dpp_capab(dev[1])
1501
1502     csign = "30770201010420768240a3fc89d6662d9782f120527fe7fb9edc6366ab0b9c7dde96125cfd250fa00a06082a8648ce3d030107a144034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1503     csign_pub = "3059301306072a8648ce3d020106082a8648ce3d030107034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1504     ap_connector = "eyJ0eXAiOiJkcHBDb24iLCJraWQiOiJwYWtZbXVzd1dCdWpSYTl5OEsweDViaTVrT3VNT3dzZHRlaml2UG55ZHZzIiwiYWxnIjoiRVMyNTYifQ.eyJncm91cHMiOlt7Imdyb3VwSWQiOiIqIiwibmV0Um9sZSI6ImFwIn1dLCJuZXRBY2Nlc3NLZXkiOnsia3R5IjoiRUMiLCJjcnYiOiJQLTI1NiIsIngiOiIybU5vNXZuRkI5bEw3d1VWb1hJbGVPYzBNSEE1QXZKbnpwZXZULVVTYzVNIiwieSI6IlhzS3dqVHJlLTg5WWdpU3pKaG9CN1haeUttTU05OTl3V2ZaSVl0bi01Q3MifX0.XhjFpZgcSa7G2lHy0OCYTvaZFRo5Hyx6b7g7oYyusLC7C_73AJ4_BxEZQVYJXAtDuGvb3dXSkHEKxREP9Q6Qeg"
1505     ap_netaccesskey = "30770201010420ceba752db2ad5200fa7bc565b9c05c69b7eb006751b0b329b0279de1c19ca67ca00a06082a8648ce3d030107a14403420004da6368e6f9c507d94bef0515a1722578e73430703902f267ce97af4fe51273935ec2b08d3adefbcf588224b3261a01ed76722a630cf7df7059f64862d9fee42b"
1506
1507     params = {"ssid": "DPPNET01",
1508               "wpa": "2",
1509               "ieee80211w": "2",
1510               "wpa_key_mgmt": "DPP",
1511               "rsn_pairwise": "CCMP",
1512               "dpp_connector": ap_connector,
1513               "dpp_csign": csign_pub,
1514               "dpp_netaccesskey": ap_netaccesskey}
1515     try:
1516         hapd = hostapd.add_ap(apdev[0], params)
1517     except:
1518         raise HwsimSkip("DPP not supported")
1519
1520     sigma = start_sigma_dut(dev[0].ifname)
1521     try:
1522         dev[0].set("dpp_config_processing", "2")
1523
1524         cmd = "DPP_CONFIGURATOR_ADD key=" + csign
1525         res = dev[1].request(cmd)
1526         if "FAIL" in res:
1527             raise Exception("Failed to add configurator")
1528         conf_id = int(res)
1529
1530         id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
1531         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1532
1533         dev[1].set("dpp_configurator_params",
1534                    " conf=sta-dpp ssid=%s configurator=%d" % (to_hex("DPPNET01"), conf_id))
1535         cmd = "DPP_LISTEN 2437 role=configurator"
1536         if "OK" not in dev[1].request(cmd):
1537             raise Exception("Failed to start listen operation")
1538
1539         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
1540         if "status,COMPLETE" not in res:
1541             raise Exception("dev_exec_action did not succeed: " + res)
1542
1543         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)
1544         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,OK,NetworkConnectResult,OK" not in res:
1545             raise Exception("Unexpected result: " + res)
1546     finally:
1547         dev[0].set("dpp_config_processing", "0")
1548         stop_sigma_dut(sigma)
1549
1550 def test_sigma_dut_dpp_qr_init_enrollee_configurator(dev, apdev):
1551     """sigma_dut DPP/QR initiator as Enrollee (to become Configurator)"""
1552     check_dpp_capab(dev[0])
1553     check_dpp_capab(dev[1])
1554
1555     sigma = start_sigma_dut(dev[0].ifname)
1556     try:
1557         cmd = "DPP_CONFIGURATOR_ADD"
1558         res = dev[1].request(cmd)
1559         if "FAIL" in res:
1560             raise Exception("Failed to add configurator")
1561         conf_id = int(res)
1562
1563         id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
1564         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1565
1566         dev[1].set("dpp_configurator_params",
1567                    " conf=configurator ssid=%s configurator=%d" % (to_hex("DPPNET01"), conf_id))
1568         cmd = "DPP_LISTEN 2437 role=configurator"
1569         if "OK" not in dev[1].request(cmd):
1570             raise Exception("Failed to start listen operation")
1571
1572         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
1573         if "status,COMPLETE" not in res:
1574             raise Exception("dev_exec_action did not succeed: " + res)
1575
1576         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Single,DPPProvisioningRole,Enrollee,DPPNetworkRole,Configurator,DPPBS,QR,DPPTimeout,6", timeout=10)
1577         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
1578             raise Exception("Unexpected result: " + res)
1579     finally:
1580         dev[0].set("dpp_config_processing", "0")
1581         stop_sigma_dut(sigma)
1582
1583 def test_sigma_dut_dpp_qr_mutual_init_enrollee(dev, apdev):
1584     """sigma_dut DPP/QR (mutual) initiator as Enrollee"""
1585     run_sigma_dut_dpp_qr_mutual_init_enrollee_check(dev, apdev)
1586
1587 def test_sigma_dut_dpp_qr_mutual_init_enrollee_check(dev, apdev):
1588     """sigma_dut DPP/QR (mutual) initiator as Enrollee (extra check)"""
1589     run_sigma_dut_dpp_qr_mutual_init_enrollee_check(dev, apdev,
1590                                                     extra="DPPAuthDirection,Mutual,")
1591
1592 def run_sigma_dut_dpp_qr_mutual_init_enrollee_check(dev, apdev, extra=''):
1593     check_dpp_capab(dev[0])
1594     check_dpp_capab(dev[1])
1595
1596     csign = "30770201010420768240a3fc89d6662d9782f120527fe7fb9edc6366ab0b9c7dde96125cfd250fa00a06082a8648ce3d030107a144034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1597     csign_pub = "3059301306072a8648ce3d020106082a8648ce3d030107034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1598     ap_connector = "eyJ0eXAiOiJkcHBDb24iLCJraWQiOiJwYWtZbXVzd1dCdWpSYTl5OEsweDViaTVrT3VNT3dzZHRlaml2UG55ZHZzIiwiYWxnIjoiRVMyNTYifQ.eyJncm91cHMiOlt7Imdyb3VwSWQiOiIqIiwibmV0Um9sZSI6ImFwIn1dLCJuZXRBY2Nlc3NLZXkiOnsia3R5IjoiRUMiLCJjcnYiOiJQLTI1NiIsIngiOiIybU5vNXZuRkI5bEw3d1VWb1hJbGVPYzBNSEE1QXZKbnpwZXZULVVTYzVNIiwieSI6IlhzS3dqVHJlLTg5WWdpU3pKaG9CN1haeUttTU05OTl3V2ZaSVl0bi01Q3MifX0.XhjFpZgcSa7G2lHy0OCYTvaZFRo5Hyx6b7g7oYyusLC7C_73AJ4_BxEZQVYJXAtDuGvb3dXSkHEKxREP9Q6Qeg"
1599     ap_netaccesskey = "30770201010420ceba752db2ad5200fa7bc565b9c05c69b7eb006751b0b329b0279de1c19ca67ca00a06082a8648ce3d030107a14403420004da6368e6f9c507d94bef0515a1722578e73430703902f267ce97af4fe51273935ec2b08d3adefbcf588224b3261a01ed76722a630cf7df7059f64862d9fee42b"
1600
1601     params = {"ssid": "DPPNET01",
1602               "wpa": "2",
1603               "ieee80211w": "2",
1604               "wpa_key_mgmt": "DPP",
1605               "rsn_pairwise": "CCMP",
1606               "dpp_connector": ap_connector,
1607               "dpp_csign": csign_pub,
1608               "dpp_netaccesskey": ap_netaccesskey}
1609     try:
1610         hapd = hostapd.add_ap(apdev[0], params)
1611     except:
1612         raise HwsimSkip("DPP not supported")
1613
1614     sigma = start_sigma_dut(dev[0].ifname)
1615     try:
1616         dev[0].set("dpp_config_processing", "2")
1617
1618         cmd = "DPP_CONFIGURATOR_ADD key=" + csign
1619         res = dev[1].request(cmd)
1620         if "FAIL" in res:
1621             raise Exception("Failed to add configurator")
1622         conf_id = int(res)
1623
1624         id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
1625         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1626
1627         dev[1].set("dpp_configurator_params",
1628                    " conf=sta-dpp ssid=%s configurator=%d" % (to_hex("DPPNET01"), conf_id))
1629         cmd = "DPP_LISTEN 2437 role=configurator qr=mutual"
1630         if "OK" not in dev[1].request(cmd):
1631             raise Exception("Failed to start listen operation")
1632
1633         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
1634         if "status,COMPLETE" not in res:
1635             raise Exception("dev_exec_action did not succeed: " + res)
1636         hex = res.split(',')[3]
1637         uri = from_hex(hex)
1638         logger.info("URI from sigma_dut: " + uri)
1639
1640         id1 = dev[1].dpp_qr_code(uri)
1641
1642         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
1643         if "status,COMPLETE" not in res:
1644             raise Exception("dev_exec_action did not succeed: " + res)
1645
1646         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,%sDPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,6,DPPWaitForConnect,Yes" % extra, timeout=10)
1647         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,OK,NetworkConnectResult,OK" not in res:
1648             raise Exception("Unexpected result: " + res)
1649     finally:
1650         dev[0].set("dpp_config_processing", "0")
1651         stop_sigma_dut(sigma)
1652
1653 def dpp_init_conf_mutual(dev, id1, conf_id, own_id=None):
1654     time.sleep(1)
1655     logger.info("Starting DPP initiator/configurator in a thread")
1656     cmd = "DPP_AUTH_INIT peer=%d conf=sta-dpp ssid=%s configurator=%d" % (id1, to_hex("DPPNET01"), conf_id)
1657     if own_id is not None:
1658         cmd += " own=%d" % own_id
1659     if "OK" not in dev.request(cmd):
1660         raise Exception("Failed to initiate DPP Authentication")
1661     ev = dev.wait_event(["DPP-CONF-SENT"], timeout=10)
1662     if ev is None:
1663         raise Exception("DPP configuration not completed (Configurator)")
1664     logger.info("DPP initiator/configurator done")
1665
1666 def test_sigma_dut_dpp_qr_mutual_resp_enrollee(dev, apdev):
1667     """sigma_dut DPP/QR (mutual) responder as Enrollee"""
1668     run_sigma_dut_dpp_qr_mutual_resp_enrollee(dev, apdev)
1669
1670 def test_sigma_dut_dpp_qr_mutual_resp_enrollee_pending(dev, apdev):
1671     """sigma_dut DPP/QR (mutual) responder as Enrollee (response pending)"""
1672     run_sigma_dut_dpp_qr_mutual_resp_enrollee(dev, apdev, ',DPPDelayQRResponse,1')
1673
1674 def run_sigma_dut_dpp_qr_mutual_resp_enrollee(dev, apdev, extra=None):
1675     check_dpp_capab(dev[0])
1676     check_dpp_capab(dev[1])
1677
1678     csign = "30770201010420768240a3fc89d6662d9782f120527fe7fb9edc6366ab0b9c7dde96125cfd250fa00a06082a8648ce3d030107a144034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1679     csign_pub = "3059301306072a8648ce3d020106082a8648ce3d030107034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1680     ap_connector = "eyJ0eXAiOiJkcHBDb24iLCJraWQiOiJwYWtZbXVzd1dCdWpSYTl5OEsweDViaTVrT3VNT3dzZHRlaml2UG55ZHZzIiwiYWxnIjoiRVMyNTYifQ.eyJncm91cHMiOlt7Imdyb3VwSWQiOiIqIiwibmV0Um9sZSI6ImFwIn1dLCJuZXRBY2Nlc3NLZXkiOnsia3R5IjoiRUMiLCJjcnYiOiJQLTI1NiIsIngiOiIybU5vNXZuRkI5bEw3d1VWb1hJbGVPYzBNSEE1QXZKbnpwZXZULVVTYzVNIiwieSI6IlhzS3dqVHJlLTg5WWdpU3pKaG9CN1haeUttTU05OTl3V2ZaSVl0bi01Q3MifX0.XhjFpZgcSa7G2lHy0OCYTvaZFRo5Hyx6b7g7oYyusLC7C_73AJ4_BxEZQVYJXAtDuGvb3dXSkHEKxREP9Q6Qeg"
1681     ap_netaccesskey = "30770201010420ceba752db2ad5200fa7bc565b9c05c69b7eb006751b0b329b0279de1c19ca67ca00a06082a8648ce3d030107a14403420004da6368e6f9c507d94bef0515a1722578e73430703902f267ce97af4fe51273935ec2b08d3adefbcf588224b3261a01ed76722a630cf7df7059f64862d9fee42b"
1682
1683     params = {"ssid": "DPPNET01",
1684               "wpa": "2",
1685               "ieee80211w": "2",
1686               "wpa_key_mgmt": "DPP",
1687               "rsn_pairwise": "CCMP",
1688               "dpp_connector": ap_connector,
1689               "dpp_csign": csign_pub,
1690               "dpp_netaccesskey": ap_netaccesskey}
1691     try:
1692         hapd = hostapd.add_ap(apdev[0], params)
1693     except:
1694         raise HwsimSkip("DPP not supported")
1695
1696     sigma = start_sigma_dut(dev[0].ifname)
1697     try:
1698         dev[0].set("dpp_config_processing", "2")
1699
1700         cmd = "DPP_CONFIGURATOR_ADD key=" + csign
1701         res = dev[1].request(cmd)
1702         if "FAIL" in res:
1703             raise Exception("Failed to add configurator")
1704         conf_id = int(res)
1705
1706         id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
1707         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1708
1709         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
1710         if "status,COMPLETE" not in res:
1711             raise Exception("dev_exec_action did not succeed: " + res)
1712         hex = res.split(',')[3]
1713         uri = from_hex(hex)
1714         logger.info("URI from sigma_dut: " + uri)
1715
1716         id1 = dev[1].dpp_qr_code(uri)
1717
1718         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
1719         if "status,COMPLETE" not in res:
1720             raise Exception("dev_exec_action did not succeed: " + res)
1721
1722         t = threading.Thread(target=dpp_init_conf_mutual,
1723                              args=(dev[1], id1, conf_id, id0))
1724         t.start()
1725
1726         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPAuthDirection,Mutual,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,20,DPPWaitForConnect,Yes"
1727         if extra:
1728             cmd += extra
1729         res = sigma_dut_cmd(cmd, timeout=25)
1730         t.join()
1731         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,OK,NetworkConnectResult,OK" not in res:
1732             raise Exception("Unexpected result: " + res)
1733     finally:
1734         dev[0].set("dpp_config_processing", "0")
1735         stop_sigma_dut(sigma)
1736
1737 def dpp_resp_conf_mutual(dev, conf_id, uri):
1738     logger.info("Starting DPP responder/configurator in a thread")
1739     dev.set("dpp_configurator_params",
1740             " conf=sta-dpp ssid=%s configurator=%d" % (to_hex("DPPNET01"),
1741                                                        conf_id))
1742     cmd = "DPP_LISTEN 2437 role=configurator qr=mutual"
1743     if "OK" not in dev.request(cmd):
1744         raise Exception("Failed to initiate DPP listen")
1745     if uri:
1746         ev = dev.wait_event(["DPP-SCAN-PEER-QR-CODE"], timeout=10)
1747         if ev is None:
1748             raise Exception("QR Code scan for mutual authentication not requested")
1749         dev.dpp_qr_code(uri)
1750     ev = dev.wait_event(["DPP-CONF-SENT"], timeout=10)
1751     if ev is None:
1752         raise Exception("DPP configuration not completed (Configurator)")
1753     logger.info("DPP responder/configurator done")
1754
1755 def test_sigma_dut_dpp_qr_mutual_init_enrollee(dev, apdev):
1756     """sigma_dut DPP/QR (mutual) initiator as Enrollee"""
1757     run_sigma_dut_dpp_qr_mutual_init_enrollee(dev, apdev, False)
1758
1759 def test_sigma_dut_dpp_qr_mutual_init_enrollee_pending(dev, apdev):
1760     """sigma_dut DPP/QR (mutual) initiator as Enrollee (response pending)"""
1761     run_sigma_dut_dpp_qr_mutual_init_enrollee(dev, apdev, True)
1762
1763 def run_sigma_dut_dpp_qr_mutual_init_enrollee(dev, apdev, resp_pending):
1764     check_dpp_capab(dev[0])
1765     check_dpp_capab(dev[1])
1766
1767     csign = "30770201010420768240a3fc89d6662d9782f120527fe7fb9edc6366ab0b9c7dde96125cfd250fa00a06082a8648ce3d030107a144034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1768     csign_pub = "3059301306072a8648ce3d020106082a8648ce3d030107034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
1769     ap_connector = "eyJ0eXAiOiJkcHBDb24iLCJraWQiOiJwYWtZbXVzd1dCdWpSYTl5OEsweDViaTVrT3VNT3dzZHRlaml2UG55ZHZzIiwiYWxnIjoiRVMyNTYifQ.eyJncm91cHMiOlt7Imdyb3VwSWQiOiIqIiwibmV0Um9sZSI6ImFwIn1dLCJuZXRBY2Nlc3NLZXkiOnsia3R5IjoiRUMiLCJjcnYiOiJQLTI1NiIsIngiOiIybU5vNXZuRkI5bEw3d1VWb1hJbGVPYzBNSEE1QXZKbnpwZXZULVVTYzVNIiwieSI6IlhzS3dqVHJlLTg5WWdpU3pKaG9CN1haeUttTU05OTl3V2ZaSVl0bi01Q3MifX0.XhjFpZgcSa7G2lHy0OCYTvaZFRo5Hyx6b7g7oYyusLC7C_73AJ4_BxEZQVYJXAtDuGvb3dXSkHEKxREP9Q6Qeg"
1770     ap_netaccesskey = "30770201010420ceba752db2ad5200fa7bc565b9c05c69b7eb006751b0b329b0279de1c19ca67ca00a06082a8648ce3d030107a14403420004da6368e6f9c507d94bef0515a1722578e73430703902f267ce97af4fe51273935ec2b08d3adefbcf588224b3261a01ed76722a630cf7df7059f64862d9fee42b"
1771
1772     params = {"ssid": "DPPNET01",
1773               "wpa": "2",
1774               "ieee80211w": "2",
1775               "wpa_key_mgmt": "DPP",
1776               "rsn_pairwise": "CCMP",
1777               "dpp_connector": ap_connector,
1778               "dpp_csign": csign_pub,
1779               "dpp_netaccesskey": ap_netaccesskey}
1780     try:
1781         hapd = hostapd.add_ap(apdev[0], params)
1782     except:
1783         raise HwsimSkip("DPP not supported")
1784
1785     sigma = start_sigma_dut(dev[0].ifname)
1786     try:
1787         dev[0].set("dpp_config_processing", "2")
1788
1789         cmd = "DPP_CONFIGURATOR_ADD key=" + csign
1790         res = dev[1].request(cmd)
1791         if "FAIL" in res:
1792             raise Exception("Failed to add configurator")
1793         conf_id = int(res)
1794
1795         id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
1796         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1797
1798         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
1799         if "status,COMPLETE" not in res:
1800             raise Exception("dev_exec_action did not succeed: " + res)
1801         hex = res.split(',')[3]
1802         uri = from_hex(hex)
1803         logger.info("URI from sigma_dut: " + uri)
1804
1805         if not resp_pending:
1806             dev[1].dpp_qr_code(uri)
1807             uri = None
1808
1809         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
1810         if "status,COMPLETE" not in res:
1811             raise Exception("dev_exec_action did not succeed: " + res)
1812
1813         t = threading.Thread(target=dpp_resp_conf_mutual,
1814                              args=(dev[1], conf_id, uri))
1815         t.start()
1816
1817         time.sleep(1)
1818         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,10,DPPWaitForConnect,Yes"
1819         res = sigma_dut_cmd(cmd, timeout=15)
1820         t.join()
1821         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,OK,NetworkConnectResult,OK" not in res:
1822             raise Exception("Unexpected result: " + res)
1823     finally:
1824         dev[0].set("dpp_config_processing", "0")
1825         stop_sigma_dut(sigma)
1826
1827 def test_sigma_dut_dpp_qr_init_enrollee_psk(dev, apdev):
1828     """sigma_dut DPP/QR initiator as Enrollee (PSK)"""
1829     check_dpp_capab(dev[0])
1830     check_dpp_capab(dev[1])
1831
1832     params = hostapd.wpa2_params(ssid="DPPNET01",
1833                                  passphrase="ThisIsDppPassphrase")
1834     hapd = hostapd.add_ap(apdev[0], params)
1835
1836     sigma = start_sigma_dut(dev[0].ifname)
1837     try:
1838         dev[0].set("dpp_config_processing", "2")
1839
1840         cmd = "DPP_CONFIGURATOR_ADD"
1841         res = dev[1].request(cmd)
1842         if "FAIL" in res:
1843             raise Exception("Failed to add configurator")
1844         conf_id = int(res)
1845
1846         id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
1847         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1848
1849         dev[1].set("dpp_configurator_params",
1850                    " conf=sta-psk ssid=%s pass=%s configurator=%d" % (to_hex("DPPNET01"), to_hex("ThisIsDppPassphrase"), conf_id))
1851         cmd = "DPP_LISTEN 2437 role=configurator"
1852         if "OK" not in dev[1].request(cmd):
1853             raise Exception("Failed to start listen operation")
1854
1855         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
1856         if "status,COMPLETE" not in res:
1857             raise Exception("dev_exec_action did not succeed: " + res)
1858
1859         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)
1860         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkConnectResult,OK" not in res:
1861             raise Exception("Unexpected result: " + res)
1862     finally:
1863         dev[0].set("dpp_config_processing", "0")
1864         stop_sigma_dut(sigma)
1865
1866 def test_sigma_dut_dpp_qr_init_enrollee_sae(dev, apdev):
1867     """sigma_dut DPP/QR initiator as Enrollee (SAE)"""
1868     check_dpp_capab(dev[0])
1869     check_dpp_capab(dev[1])
1870     if "SAE" not in dev[0].get_capability("auth_alg"):
1871         raise HwsimSkip("SAE not supported")
1872
1873     params = hostapd.wpa2_params(ssid="DPPNET01",
1874                                  passphrase="ThisIsDppPassphrase")
1875     params['wpa_key_mgmt'] = 'SAE'
1876     params["ieee80211w"] = "2"
1877     hapd = hostapd.add_ap(apdev[0], params)
1878
1879     sigma = start_sigma_dut(dev[0].ifname)
1880     try:
1881         dev[0].set("dpp_config_processing", "2")
1882         dev[0].set("sae_groups", "")
1883
1884         cmd = "DPP_CONFIGURATOR_ADD"
1885         res = dev[1].request(cmd)
1886         if "FAIL" in res:
1887             raise Exception("Failed to add configurator")
1888         conf_id = int(res)
1889
1890         id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
1891         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1892
1893         dev[1].set("dpp_configurator_params",
1894                    " conf=sta-sae ssid=%s pass=%s configurator=%d" % (to_hex("DPPNET01"), to_hex("ThisIsDppPassphrase"), conf_id))
1895         cmd = "DPP_LISTEN 2437 role=configurator"
1896         if "OK" not in dev[1].request(cmd):
1897             raise Exception("Failed to start listen operation")
1898
1899         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
1900         if "status,COMPLETE" not in res:
1901             raise Exception("dev_exec_action did not succeed: " + res)
1902
1903         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)
1904         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkConnectResult,OK" not in res:
1905             raise Exception("Unexpected result: " + res)
1906     finally:
1907         dev[0].set("dpp_config_processing", "0")
1908         stop_sigma_dut(sigma)
1909
1910 def test_sigma_dut_dpp_qr_init_configurator_1(dev, apdev):
1911     """sigma_dut DPP/QR initiator as Configurator (conf index 1)"""
1912     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 1)
1913
1914 def test_sigma_dut_dpp_qr_init_configurator_2(dev, apdev):
1915     """sigma_dut DPP/QR initiator as Configurator (conf index 2)"""
1916     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 2)
1917
1918 def test_sigma_dut_dpp_qr_init_configurator_3(dev, apdev):
1919     """sigma_dut DPP/QR initiator as Configurator (conf index 3)"""
1920     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 3)
1921
1922 def test_sigma_dut_dpp_qr_init_configurator_4(dev, apdev):
1923     """sigma_dut DPP/QR initiator as Configurator (conf index 4)"""
1924     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 4)
1925
1926 def test_sigma_dut_dpp_qr_init_configurator_5(dev, apdev):
1927     """sigma_dut DPP/QR initiator as Configurator (conf index 5)"""
1928     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 5)
1929
1930 def test_sigma_dut_dpp_qr_init_configurator_6(dev, apdev):
1931     """sigma_dut DPP/QR initiator as Configurator (conf index 6)"""
1932     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 6)
1933
1934 def test_sigma_dut_dpp_qr_init_configurator_7(dev, apdev):
1935     """sigma_dut DPP/QR initiator as Configurator (conf index 7)"""
1936     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 7)
1937
1938 def test_sigma_dut_dpp_qr_init_configurator_both(dev, apdev):
1939     """sigma_dut DPP/QR initiator as Configurator or Enrollee (conf index 1)"""
1940     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 1, "Both")
1941
1942 def test_sigma_dut_dpp_qr_init_configurator_neg_freq(dev, apdev):
1943     """sigma_dut DPP/QR initiator as Configurator (neg_freq)"""
1944     run_sigma_dut_dpp_qr_init_configurator(dev, apdev, 1, extra='DPPSubsequentChannel,81/11')
1945
1946 def run_sigma_dut_dpp_qr_init_configurator(dev, apdev, conf_idx,
1947                                            prov_role="Configurator",
1948                                            extra=None):
1949     check_dpp_capab(dev[0])
1950     check_dpp_capab(dev[1])
1951     sigma = start_sigma_dut(dev[0].ifname)
1952     try:
1953         id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
1954         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1955
1956         cmd = "DPP_LISTEN 2437 role=enrollee"
1957         if "OK" not in dev[1].request(cmd):
1958             raise Exception("Failed to start listen operation")
1959
1960         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
1961         if "status,COMPLETE" not in res:
1962             raise Exception("dev_exec_action did not succeed: " + res)
1963
1964         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)
1965         if extra:
1966             cmd += "," + extra
1967         res = sigma_dut_cmd(cmd)
1968         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
1969             raise Exception("Unexpected result: " + res)
1970     finally:
1971         stop_sigma_dut(sigma)
1972
1973 def test_sigma_dut_dpp_incompatible_roles_init(dev, apdev):
1974     """sigma_dut DPP roles incompatible (Initiator)"""
1975     check_dpp_capab(dev[0])
1976     check_dpp_capab(dev[1])
1977     sigma = start_sigma_dut(dev[0].ifname)
1978     try:
1979         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
1980         if "status,COMPLETE" not in res:
1981             raise Exception("dev_exec_action did not succeed: " + res)
1982         hex = res.split(',')[3]
1983         uri = from_hex(hex)
1984         logger.info("URI from sigma_dut: " + uri)
1985
1986         id1 = dev[1].dpp_qr_code(uri)
1987
1988         id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
1989         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
1990
1991         cmd = "DPP_LISTEN 2437 role=enrollee"
1992         if "OK" not in dev[1].request(cmd):
1993             raise Exception("Failed to start listen operation")
1994
1995         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
1996         if "status,COMPLETE" not in res:
1997             raise Exception("dev_exec_action did not succeed: " + res)
1998
1999         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Mutual,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,6"
2000         res = sigma_dut_cmd(cmd)
2001         if "BootstrapResult,OK,AuthResult,ROLES_NOT_COMPATIBLE" not in res:
2002             raise Exception("Unexpected result: " + res)
2003     finally:
2004         stop_sigma_dut(sigma)
2005
2006 def dpp_init_enrollee_mutual(dev, id1, own_id):
2007     logger.info("Starting DPP initiator/enrollee in a thread")
2008     time.sleep(1)
2009     cmd = "DPP_AUTH_INIT peer=%d own=%d role=enrollee" % (id1, own_id)
2010     if "OK" not in dev.request(cmd):
2011         raise Exception("Failed to initiate DPP Authentication")
2012     ev = dev.wait_event(["DPP-CONF-RECEIVED",
2013                          "DPP-NOT-COMPATIBLE"], timeout=5)
2014     if ev is None:
2015         raise Exception("DPP configuration not completed (Enrollee)")
2016     logger.info("DPP initiator/enrollee done")
2017
2018 def test_sigma_dut_dpp_incompatible_roles_resp(dev, apdev):
2019     """sigma_dut DPP roles incompatible (Responder)"""
2020     check_dpp_capab(dev[0])
2021     check_dpp_capab(dev[1])
2022     sigma = start_sigma_dut(dev[0].ifname)
2023     try:
2024         cmd = "dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR"
2025         res = sigma_dut_cmd(cmd)
2026         if "status,COMPLETE" not in res:
2027             raise Exception("dev_exec_action did not succeed: " + res)
2028         hex = res.split(',')[3]
2029         uri = from_hex(hex)
2030         logger.info("URI from sigma_dut: " + uri)
2031
2032         id1 = dev[1].dpp_qr_code(uri)
2033
2034         id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
2035         uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
2036
2037         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
2038         if "status,COMPLETE" not in res:
2039             raise Exception("dev_exec_action did not succeed: " + res)
2040
2041         t = threading.Thread(target=dpp_init_enrollee_mutual, args=(dev[1], id1, id0))
2042         t.start()
2043         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPAuthDirection,Mutual,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,6"
2044         res = sigma_dut_cmd(cmd, timeout=10)
2045         t.join()
2046         if "BootstrapResult,OK,AuthResult,ROLES_NOT_COMPATIBLE" not in res:
2047             raise Exception("Unexpected result: " + res)
2048     finally:
2049         stop_sigma_dut(sigma)
2050
2051 def test_sigma_dut_dpp_pkex_init_configurator(dev, apdev):
2052     """sigma_dut DPP/PKEX initiator as Configurator"""
2053     check_dpp_capab(dev[0])
2054     check_dpp_capab(dev[1])
2055     sigma = start_sigma_dut(dev[0].ifname)
2056     try:
2057         id1 = dev[1].dpp_bootstrap_gen(type="pkex")
2058         cmd = "DPP_PKEX_ADD own=%d identifier=test code=secret" % (id1)
2059         res = dev[1].request(cmd)
2060         if "FAIL" in res:
2061             raise Exception("Failed to set PKEX data (responder)")
2062         cmd = "DPP_LISTEN 2437 role=enrollee"
2063         if "OK" not in dev[1].request(cmd):
2064             raise Exception("Failed to start listen operation")
2065
2066         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPProvisioningRole,Configurator,DPPConfIndex,1,DPPSigningKeyECC,P-256,DPPConfEnrolleeRole,STA,DPPBS,PKEX,DPPPKEXCodeIdentifier,test,DPPPKEXCode,secret,DPPTimeout,6")
2067         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
2068             raise Exception("Unexpected result: " + res)
2069     finally:
2070         stop_sigma_dut(sigma)
2071
2072 def dpp_init_conf(dev, id1, conf, conf_id, extra):
2073     logger.info("Starting DPP initiator/configurator in a thread")
2074     cmd = "DPP_AUTH_INIT peer=%d conf=%s %s configurator=%d" % (id1, conf, extra, conf_id)
2075     if "OK" not in dev.request(cmd):
2076         raise Exception("Failed to initiate DPP Authentication")
2077     ev = dev.wait_event(["DPP-CONF-SENT"], timeout=5)
2078     if ev is None:
2079         raise Exception("DPP configuration not completed (Configurator)")
2080     logger.info("DPP initiator/configurator done")
2081
2082 def test_sigma_dut_ap_dpp_qr(dev, apdev, params):
2083     """sigma_dut controlled AP (DPP)"""
2084     run_sigma_dut_ap_dpp_qr(dev, apdev, params, "ap-dpp", "sta-dpp")
2085
2086 def test_sigma_dut_ap_dpp_qr_legacy(dev, apdev, params):
2087     """sigma_dut controlled AP (legacy)"""
2088     run_sigma_dut_ap_dpp_qr(dev, apdev, params, "ap-psk", "sta-psk",
2089                             extra="pass=%s" % to_hex("qwertyuiop"))
2090
2091 def test_sigma_dut_ap_dpp_qr_legacy_psk(dev, apdev, params):
2092     """sigma_dut controlled AP (legacy)"""
2093     run_sigma_dut_ap_dpp_qr(dev, apdev, params, "ap-psk", "sta-psk",
2094                             extra="psk=%s" % (32*"12"))
2095
2096 def run_sigma_dut_ap_dpp_qr(dev, apdev, params, ap_conf, sta_conf, extra=""):
2097     check_dpp_capab(dev[0])
2098     logdir = os.path.join(params['logdir'], "sigma_dut_ap_dpp_qr.sigma-hostapd")
2099     with HWSimRadio() as (radio, iface):
2100         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
2101         try:
2102             sigma_dut_cmd_check("ap_reset_default,program,DPP")
2103             res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
2104             if "status,COMPLETE" not in res:
2105                 raise Exception("dev_exec_action did not succeed: " + res)
2106             hex = res.split(',')[3]
2107             uri = from_hex(hex)
2108             logger.info("URI from sigma_dut: " + uri)
2109
2110             cmd = "DPP_CONFIGURATOR_ADD"
2111             res = dev[0].request(cmd)
2112             if "FAIL" in res:
2113                 raise Exception("Failed to add configurator")
2114             conf_id = int(res)
2115
2116             id1 = dev[0].dpp_qr_code(uri)
2117
2118             t = threading.Thread(target=dpp_init_conf,
2119                                  args=(dev[0], id1, ap_conf, conf_id, extra))
2120             t.start()
2121             res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPAuthDirection,Single,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,6")
2122             t.join()
2123             if "ConfResult,OK" not in res:
2124                 raise Exception("Unexpected result: " + res)
2125
2126             id1 = dev[1].dpp_bootstrap_gen(chan="81/1", mac=True)
2127             uri1 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id1)
2128
2129             id0b = dev[0].dpp_qr_code(uri1)
2130
2131             dev[1].set("dpp_config_processing", "2")
2132             cmd = "DPP_LISTEN 2412"
2133             if "OK" not in dev[1].request(cmd):
2134                 raise Exception("Failed to start listen operation")
2135             cmd = "DPP_AUTH_INIT peer=%d conf=%s %s configurator=%d" % (id0b, sta_conf, extra, conf_id)
2136             if "OK" not in dev[0].request(cmd):
2137                 raise Exception("Failed to initiate DPP Authentication")
2138             dev[1].wait_connected()
2139
2140             sigma_dut_cmd_check("ap_reset_default")
2141         finally:
2142             dev[1].set("dpp_config_processing", "0")
2143             stop_sigma_dut(sigma)
2144
2145 def test_sigma_dut_ap_dpp_pkex_responder(dev, apdev, params):
2146     """sigma_dut controlled AP as DPP PKEX responder"""
2147     check_dpp_capab(dev[0])
2148     logdir = os.path.join(params['logdir'],
2149                           "sigma_dut_ap_dpp_pkex_responder.sigma-hostapd")
2150     with HWSimRadio() as (radio, iface):
2151         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
2152         try:
2153             run_sigma_dut_ap_dpp_pkex_responder(dev, apdev)
2154         finally:
2155             stop_sigma_dut(sigma)
2156
2157 def dpp_init_conf_pkex(dev, conf_id, check_config=True):
2158     logger.info("Starting DPP PKEX initiator/configurator in a thread")
2159     time.sleep(1.5)
2160     id = dev.dpp_bootstrap_gen(type="pkex")
2161     cmd = "DPP_PKEX_ADD own=%d init=1 conf=ap-dpp configurator=%d code=password" % (id, conf_id)
2162     res = dev.request(cmd)
2163     if "FAIL" in res:
2164         raise Exception("Failed to initiate DPP PKEX")
2165     if not check_config:
2166         return
2167     ev = dev.wait_event(["DPP-CONF-SENT"], timeout=5)
2168     if ev is None:
2169         raise Exception("DPP configuration not completed (Configurator)")
2170     logger.info("DPP initiator/configurator done")
2171
2172 def run_sigma_dut_ap_dpp_pkex_responder(dev, apdev):
2173     sigma_dut_cmd_check("ap_reset_default,program,DPP")
2174
2175     cmd = "DPP_CONFIGURATOR_ADD"
2176     res = dev[0].request(cmd)
2177     if "FAIL" in res:
2178         raise Exception("Failed to add configurator")
2179     conf_id = int(res)
2180
2181     t = threading.Thread(target=dpp_init_conf_pkex, args=(dev[0], conf_id))
2182     t.start()
2183     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPAuthDirection,Mutual,DPPProvisioningRole,Enrollee,DPPBS,PKEX,DPPPKEXCode,password,DPPTimeout,6,DPPWaitForConnect,No", timeout=10)
2184     t.join()
2185     if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
2186         raise Exception("Unexpected result: " + res)
2187
2188     sigma_dut_cmd_check("ap_reset_default")
2189
2190 def test_sigma_dut_dpp_pkex_responder_proto(dev, apdev):
2191     """sigma_dut controlled STA as DPP PKEX responder and error case"""
2192     check_dpp_capab(dev[0])
2193     sigma = start_sigma_dut(dev[0].ifname)
2194     try:
2195         run_sigma_dut_dpp_pkex_responder_proto(dev, apdev)
2196     finally:
2197         stop_sigma_dut(sigma)
2198
2199 def run_sigma_dut_dpp_pkex_responder_proto(dev, apdev):
2200     cmd = "DPP_CONFIGURATOR_ADD"
2201     res = dev[1].request(cmd)
2202     if "FAIL" in res:
2203         raise Exception("Failed to add configurator")
2204     conf_id = int(res)
2205
2206     dev[1].set("dpp_test", "44")
2207
2208     t = threading.Thread(target=dpp_init_conf_pkex, args=(dev[1], conf_id,
2209                                                           False))
2210     t.start()
2211     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPProvisioningRole,Enrollee,DPPBS,PKEX,DPPPKEXCode,password,DPPTimeout,6", timeout=10)
2212     t.join()
2213     if "BootstrapResult,Timeout" not in res:
2214         raise Exception("Unexpected result: " + res)
2215
2216 def dpp_proto_init(dev, id1):
2217     time.sleep(1)
2218     logger.info("Starting DPP initiator/configurator in a thread")
2219     cmd = "DPP_CONFIGURATOR_ADD"
2220     res = dev.request(cmd)
2221     if "FAIL" in res:
2222         raise Exception("Failed to add configurator")
2223     conf_id = int(res)
2224
2225     cmd = "DPP_AUTH_INIT peer=%d conf=sta-dpp configurator=%d" % (id1, conf_id)
2226     if "OK" not in dev.request(cmd):
2227         raise Exception("Failed to initiate DPP Authentication")
2228
2229 def test_sigma_dut_dpp_proto_initiator(dev, apdev):
2230     """sigma_dut DPP protocol testing - Initiator"""
2231     check_dpp_capab(dev[0])
2232     check_dpp_capab(dev[1])
2233     tests = [("InvalidValue", "AuthenticationRequest", "WrappedData",
2234               "BootstrapResult,OK,AuthResult,Errorsent",
2235               None),
2236              ("InvalidValue", "AuthenticationConfirm", "WrappedData",
2237               "BootstrapResult,OK,AuthResult,Errorsent",
2238               None),
2239              ("MissingAttribute", "AuthenticationRequest", "InitCapabilities",
2240               "BootstrapResult,OK,AuthResult,Errorsent",
2241               "Missing or invalid I-capabilities"),
2242              ("InvalidValue", "AuthenticationConfirm", "InitAuthTag",
2243               "BootstrapResult,OK,AuthResult,Errorsent",
2244               "Mismatching Initiator Authenticating Tag"),
2245              ("MissingAttribute", "ConfigurationResponse", "EnrolleeNonce",
2246               "BootstrapResult,OK,AuthResult,OK,ConfResult,Errorsent",
2247               "Missing or invalid Enrollee Nonce attribute")]
2248     for step, frame, attr, result, fail in tests:
2249         dev[0].request("FLUSH")
2250         dev[1].request("FLUSH")
2251         sigma = start_sigma_dut(dev[0].ifname)
2252         try:
2253             run_sigma_dut_dpp_proto_initiator(dev, step, frame, attr, result,
2254                                               fail)
2255         finally:
2256             stop_sigma_dut(sigma)
2257
2258 def run_sigma_dut_dpp_proto_initiator(dev, step, frame, attr, result, fail):
2259     id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
2260     uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
2261
2262     cmd = "DPP_LISTEN 2437 role=enrollee"
2263     if "OK" not in dev[1].request(cmd):
2264         raise Exception("Failed to start listen operation")
2265
2266     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
2267     if "status,COMPLETE" not in res:
2268         raise Exception("dev_exec_action did not succeed: " + res)
2269
2270     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),
2271                         timeout=10)
2272     if result not in res:
2273         raise Exception("Unexpected result: " + res)
2274     if fail:
2275         ev = dev[1].wait_event(["DPP-FAIL"], timeout=5)
2276         if ev is None or fail not in ev:
2277             raise Exception("Failure not reported correctly: " + str(ev))
2278
2279     dev[1].request("DPP_STOP_LISTEN")
2280     dev[0].dump_monitor()
2281     dev[1].dump_monitor()
2282
2283 def test_sigma_dut_dpp_proto_responder(dev, apdev):
2284     """sigma_dut DPP protocol testing - Responder"""
2285     check_dpp_capab(dev[0])
2286     check_dpp_capab(dev[1])
2287     tests = [("MissingAttribute", "AuthenticationResponse", "DPPStatus",
2288               "BootstrapResult,OK,AuthResult,Errorsent",
2289               "Missing or invalid required DPP Status attribute"),
2290              ("MissingAttribute", "ConfigurationRequest", "EnrolleeNonce",
2291               "BootstrapResult,OK,AuthResult,OK,ConfResult,Errorsent",
2292               "Missing or invalid Enrollee Nonce attribute")]
2293     for step, frame, attr, result, fail in tests:
2294         dev[0].request("FLUSH")
2295         dev[1].request("FLUSH")
2296         sigma = start_sigma_dut(dev[0].ifname)
2297         try:
2298             run_sigma_dut_dpp_proto_responder(dev, step, frame, attr, result,
2299                                               fail)
2300         finally:
2301             stop_sigma_dut(sigma)
2302
2303 def run_sigma_dut_dpp_proto_responder(dev, step, frame, attr, result, fail):
2304     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
2305     if "status,COMPLETE" not in res:
2306         raise Exception("dev_exec_action did not succeed: " + res)
2307     hex = res.split(',')[3]
2308     uri = from_hex(hex)
2309     logger.info("URI from sigma_dut: " + uri)
2310
2311     id1 = dev[1].dpp_qr_code(uri)
2312
2313     t = threading.Thread(target=dpp_proto_init, args=(dev[1], id1))
2314     t.start()
2315     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)
2316     t.join()
2317     if result not in res:
2318         raise Exception("Unexpected result: " + res)
2319     if fail:
2320         ev = dev[1].wait_event(["DPP-FAIL"], timeout=5)
2321         if ev is None or fail not in ev:
2322             raise Exception("Failure not reported correctly:" + str(ev))
2323
2324     dev[1].request("DPP_STOP_LISTEN")
2325     dev[0].dump_monitor()
2326     dev[1].dump_monitor()
2327
2328 def test_sigma_dut_dpp_proto_stop_at_initiator(dev, apdev):
2329     """sigma_dut DPP protocol testing - Stop at RX on Initiator"""
2330     check_dpp_capab(dev[0])
2331     check_dpp_capab(dev[1])
2332     tests = [("AuthenticationResponse",
2333               "BootstrapResult,OK,AuthResult,Errorsent",
2334               None),
2335              ("ConfigurationRequest",
2336               "BootstrapResult,OK,AuthResult,OK,ConfResult,Errorsent",
2337               None)]
2338     for frame, result, fail in tests:
2339         dev[0].request("FLUSH")
2340         dev[1].request("FLUSH")
2341         sigma = start_sigma_dut(dev[0].ifname)
2342         try:
2343             run_sigma_dut_dpp_proto_stop_at_initiator(dev, frame, result, fail)
2344         finally:
2345             stop_sigma_dut(sigma)
2346
2347 def run_sigma_dut_dpp_proto_stop_at_initiator(dev, frame, result, fail):
2348     id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
2349     uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
2350
2351     cmd = "DPP_LISTEN 2437 role=enrollee"
2352     if "OK" not in dev[1].request(cmd):
2353         raise Exception("Failed to start listen operation")
2354
2355     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
2356     if "status,COMPLETE" not in res:
2357         raise Exception("dev_exec_action did not succeed: " + res)
2358
2359     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,Timeout,DPPFrameType,%s" % (frame))
2360     if result not in res:
2361         raise Exception("Unexpected result: " + res)
2362     if fail:
2363         ev = dev[1].wait_event(["DPP-FAIL"], timeout=5)
2364         if ev is None or fail not in ev:
2365             raise Exception("Failure not reported correctly: " + str(ev))
2366
2367     dev[1].request("DPP_STOP_LISTEN")
2368     dev[0].dump_monitor()
2369     dev[1].dump_monitor()
2370
2371 def test_sigma_dut_dpp_proto_stop_at_initiator_enrollee(dev, apdev):
2372     """sigma_dut DPP protocol testing - Stop at TX on Initiator/Enrollee"""
2373     check_dpp_capab(dev[0])
2374     check_dpp_capab(dev[1])
2375     tests = [("AuthenticationConfirm",
2376               "BootstrapResult,OK,AuthResult,Errorsent,LastFrameReceived,AuthenticationResponse",
2377               None)]
2378     for frame, result, fail in tests:
2379         dev[0].request("FLUSH")
2380         dev[1].request("FLUSH")
2381         sigma = start_sigma_dut(dev[0].ifname)
2382         try:
2383             run_sigma_dut_dpp_proto_stop_at_initiator_enrollee(dev, frame,
2384                                                                result, fail)
2385         finally:
2386             stop_sigma_dut(sigma)
2387
2388 def run_sigma_dut_dpp_proto_stop_at_initiator_enrollee(dev, frame, result,
2389                                                        fail):
2390     id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
2391     uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
2392
2393     cmd = "DPP_LISTEN 2437 role=configurator"
2394     if "OK" not in dev[1].request(cmd):
2395         raise Exception("Failed to start listen operation")
2396
2397     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
2398     if "status,COMPLETE" not in res:
2399         raise Exception("dev_exec_action did not succeed: " + res)
2400
2401     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Single,DPPProvisioningRole,Enrollee,DPPBS,QR,DPPTimeout,6,DPPStep,Timeout,DPPFrameType,%s" % (frame), timeout=10)
2402     if result not in res:
2403         raise Exception("Unexpected result: " + res)
2404     if fail:
2405         ev = dev[1].wait_event(["DPP-FAIL"], timeout=5)
2406         if ev is None or fail not in ev:
2407             raise Exception("Failure not reported correctly: " + str(ev))
2408
2409     dev[1].request("DPP_STOP_LISTEN")
2410     dev[0].dump_monitor()
2411     dev[1].dump_monitor()
2412
2413 def test_sigma_dut_dpp_proto_stop_at_responder(dev, apdev):
2414     """sigma_dut DPP protocol testing - Stop at RX on Responder"""
2415     check_dpp_capab(dev[0])
2416     check_dpp_capab(dev[1])
2417     tests = [("AuthenticationRequest",
2418               "BootstrapResult,OK,AuthResult,Errorsent",
2419               None),
2420              ("AuthenticationConfirm",
2421               "BootstrapResult,OK,AuthResult,Errorsent",
2422               None)]
2423     for frame, result, fail in tests:
2424         dev[0].request("FLUSH")
2425         dev[1].request("FLUSH")
2426         sigma = start_sigma_dut(dev[0].ifname)
2427         try:
2428             run_sigma_dut_dpp_proto_stop_at_responder(dev, frame, result, fail)
2429         finally:
2430             stop_sigma_dut(sigma)
2431
2432 def run_sigma_dut_dpp_proto_stop_at_responder(dev, frame, result, fail):
2433     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
2434     if "status,COMPLETE" not in res:
2435         raise Exception("dev_exec_action did not succeed: " + res)
2436     hex = res.split(',')[3]
2437     uri = from_hex(hex)
2438     logger.info("URI from sigma_dut: " + uri)
2439
2440     id1 = dev[1].dpp_qr_code(uri)
2441
2442     t = threading.Thread(target=dpp_proto_init, args=(dev[1], id1))
2443     t.start()
2444     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,Timeout,DPPFrameType,%s" % (frame), timeout=10)
2445     t.join()
2446     if result not in res:
2447         raise Exception("Unexpected result: " + res)
2448     if fail:
2449         ev = dev[1].wait_event(["DPP-FAIL"], timeout=5)
2450         if ev is None or fail not in ev:
2451             raise Exception("Failure not reported correctly:" + str(ev))
2452
2453     dev[1].request("DPP_STOP_LISTEN")
2454     dev[0].dump_monitor()
2455     dev[1].dump_monitor()
2456
2457 def dpp_proto_init_pkex(dev):
2458     time.sleep(1)
2459     logger.info("Starting DPP PKEX initiator/configurator in a thread")
2460     cmd = "DPP_CONFIGURATOR_ADD"
2461     res = dev.request(cmd)
2462     if "FAIL" in res:
2463         raise Exception("Failed to add configurator")
2464     conf_id = int(res)
2465
2466     id = dev.dpp_bootstrap_gen(type="pkex")
2467
2468     cmd = "DPP_PKEX_ADD own=%d init=1 conf=sta-dpp configurator=%d code=secret" % (id, conf_id)
2469     if "FAIL" in dev.request(cmd):
2470         raise Exception("Failed to initiate DPP PKEX")
2471
2472 def test_sigma_dut_dpp_proto_initiator_pkex(dev, apdev):
2473     """sigma_dut DPP protocol testing - Initiator (PKEX)"""
2474     check_dpp_capab(dev[0])
2475     check_dpp_capab(dev[1])
2476     tests = [("InvalidValue", "PKEXCRRequest", "WrappedData",
2477               "BootstrapResult,Errorsent",
2478               None),
2479              ("MissingAttribute", "PKEXExchangeRequest", "FiniteCyclicGroup",
2480               "BootstrapResult,Errorsent",
2481               "Missing or invalid Finite Cyclic Group attribute"),
2482              ("MissingAttribute", "PKEXCRRequest", "BSKey",
2483               "BootstrapResult,Errorsent",
2484               "No valid peer bootstrapping key found")]
2485     for step, frame, attr, result, fail in tests:
2486         dev[0].request("FLUSH")
2487         dev[1].request("FLUSH")
2488         sigma = start_sigma_dut(dev[0].ifname)
2489         try:
2490             run_sigma_dut_dpp_proto_initiator_pkex(dev, step, frame, attr,
2491                                                    result, fail)
2492         finally:
2493             stop_sigma_dut(sigma)
2494
2495 def run_sigma_dut_dpp_proto_initiator_pkex(dev, step, frame, attr, result, fail):
2496     id1 = dev[1].dpp_bootstrap_gen(type="pkex")
2497
2498     cmd = "DPP_PKEX_ADD own=%d code=secret" % (id1)
2499     res = dev[1].request(cmd)
2500     if "FAIL" in res:
2501         raise Exception("Failed to set PKEX data (responder)")
2502
2503     cmd = "DPP_LISTEN 2437 role=enrollee"
2504     if "OK" not in dev[1].request(cmd):
2505         raise Exception("Failed to start listen operation")
2506
2507     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))
2508     if result not in res:
2509         raise Exception("Unexpected result: " + res)
2510     if fail:
2511         ev = dev[1].wait_event(["DPP-FAIL"], timeout=5)
2512         if ev is None or fail not in ev:
2513             raise Exception("Failure not reported correctly: " + str(ev))
2514
2515     dev[1].request("DPP_STOP_LISTEN")
2516     dev[0].dump_monitor()
2517     dev[1].dump_monitor()
2518
2519 def test_sigma_dut_dpp_proto_responder_pkex(dev, apdev):
2520     """sigma_dut DPP protocol testing - Responder (PKEX)"""
2521     check_dpp_capab(dev[0])
2522     check_dpp_capab(dev[1])
2523     tests = [("InvalidValue", "PKEXCRResponse", "WrappedData",
2524               "BootstrapResult,Errorsent",
2525               None),
2526              ("MissingAttribute", "PKEXExchangeResponse", "DPPStatus",
2527               "BootstrapResult,Errorsent",
2528               "No DPP Status attribute"),
2529              ("MissingAttribute", "PKEXCRResponse", "BSKey",
2530               "BootstrapResult,Errorsent",
2531               "No valid peer bootstrapping key found")]
2532     for step, frame, attr, result, fail in tests:
2533         dev[0].request("FLUSH")
2534         dev[1].request("FLUSH")
2535         sigma = start_sigma_dut(dev[0].ifname)
2536         try:
2537             run_sigma_dut_dpp_proto_responder_pkex(dev, step, frame, attr,
2538                                                    result, fail)
2539         finally:
2540             stop_sigma_dut(sigma)
2541
2542 def run_sigma_dut_dpp_proto_responder_pkex(dev, step, frame, attr, result, fail):
2543     t = threading.Thread(target=dpp_proto_init_pkex, args=(dev[1],))
2544     t.start()
2545     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)
2546     t.join()
2547     if result not in res:
2548         raise Exception("Unexpected result: " + res)
2549     if fail:
2550         ev = dev[1].wait_event(["DPP-FAIL"], timeout=5)
2551         if ev is None or fail not in ev:
2552             raise Exception("Failure not reported correctly:" + str(ev))
2553
2554     dev[1].request("DPP_STOP_LISTEN")
2555     dev[0].dump_monitor()
2556     dev[1].dump_monitor()
2557
2558 def init_sigma_dut_dpp_proto_peer_disc_req(dev, apdev):
2559     check_dpp_capab(dev[0])
2560     check_dpp_capab(dev[1])
2561
2562     csign = "30770201010420768240a3fc89d6662d9782f120527fe7fb9edc6366ab0b9c7dde96125cfd250fa00a06082a8648ce3d030107a144034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
2563     csign_pub = "3059301306072a8648ce3d020106082a8648ce3d030107034200042908e1baf7bf413cc66f9e878a03e8bb1835ba94b033dbe3d6969fc8575d5eb5dfda1cb81c95cee21d0cd7d92ba30541ffa05cb6296f5dd808b0c1c2a83c0708"
2564     ap_connector = "eyJ0eXAiOiJkcHBDb24iLCJraWQiOiJwYWtZbXVzd1dCdWpSYTl5OEsweDViaTVrT3VNT3dzZHRlaml2UG55ZHZzIiwiYWxnIjoiRVMyNTYifQ.eyJncm91cHMiOlt7Imdyb3VwSWQiOiIqIiwibmV0Um9sZSI6ImFwIn1dLCJuZXRBY2Nlc3NLZXkiOnsia3R5IjoiRUMiLCJjcnYiOiJQLTI1NiIsIngiOiIybU5vNXZuRkI5bEw3d1VWb1hJbGVPYzBNSEE1QXZKbnpwZXZULVVTYzVNIiwieSI6IlhzS3dqVHJlLTg5WWdpU3pKaG9CN1haeUttTU05OTl3V2ZaSVl0bi01Q3MifX0.XhjFpZgcSa7G2lHy0OCYTvaZFRo5Hyx6b7g7oYyusLC7C_73AJ4_BxEZQVYJXAtDuGvb3dXSkHEKxREP9Q6Qeg"
2565     ap_netaccesskey = "30770201010420ceba752db2ad5200fa7bc565b9c05c69b7eb006751b0b329b0279de1c19ca67ca00a06082a8648ce3d030107a14403420004da6368e6f9c507d94bef0515a1722578e73430703902f267ce97af4fe51273935ec2b08d3adefbcf588224b3261a01ed76722a630cf7df7059f64862d9fee42b"
2566
2567     params = {"ssid": "DPPNET01",
2568               "wpa": "2",
2569               "ieee80211w": "2",
2570               "wpa_key_mgmt": "DPP",
2571               "rsn_pairwise": "CCMP",
2572               "dpp_connector": ap_connector,
2573               "dpp_csign": csign_pub,
2574               "dpp_netaccesskey": ap_netaccesskey}
2575     try:
2576         hapd = hostapd.add_ap(apdev[0], params)
2577     except:
2578         raise HwsimSkip("DPP not supported")
2579
2580     dev[0].set("dpp_config_processing", "2")
2581
2582     cmd = "DPP_CONFIGURATOR_ADD key=" + csign
2583     res = dev[1].request(cmd)
2584     if "FAIL" in res:
2585         raise Exception("Failed to add configurator")
2586     conf_id = int(res)
2587
2588     id0 = dev[1].dpp_bootstrap_gen(chan="81/6", mac=True)
2589     uri0 = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id0)
2590
2591     dev[1].set("dpp_configurator_params",
2592                " conf=sta-dpp ssid=%s configurator=%d" % (to_hex("DPPNET01"),
2593                                                           conf_id))
2594     cmd = "DPP_LISTEN 2437 role=configurator"
2595     if "OK" not in dev[1].request(cmd):
2596         raise Exception("Failed to start listen operation")
2597
2598     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri0))
2599     if "status,COMPLETE" not in res:
2600         raise Exception("dev_exec_action did not succeed: " + res)
2601
2602 def test_sigma_dut_dpp_proto_peer_disc_req(dev, apdev):
2603     """sigma_dut DPP protocol testing - Peer Discovery Request"""
2604     sigma = start_sigma_dut(dev[0].ifname)
2605     try:
2606         init_sigma_dut_dpp_proto_peer_disc_req(dev, apdev)
2607
2608         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)
2609         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,Errorsent" not in res:
2610             raise Exception("Unexpected result: " + res)
2611     finally:
2612         dev[0].set("dpp_config_processing", "0", allow_fail=True)
2613         stop_sigma_dut(sigma)
2614
2615 def test_sigma_dut_dpp_self_config(dev, apdev):
2616     """sigma_dut DPP Configurator enrolling an AP and using self-configuration"""
2617     check_dpp_capab(dev[0])
2618
2619     hapd = hostapd.add_ap(apdev[0], {"ssid": "unconfigured"})
2620     check_dpp_capab(hapd)
2621
2622     sigma = start_sigma_dut(dev[0].ifname)
2623     try:
2624         dev[0].set("dpp_config_processing", "2")
2625         id = hapd.dpp_bootstrap_gen(chan="81/1", mac=True)
2626         uri = hapd.request("DPP_BOOTSTRAP_GET_URI %d" % id)
2627
2628         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri))
2629         if "status,COMPLETE" not in res:
2630             raise Exception("dev_exec_action did not succeed: " + res)
2631
2632         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")
2633         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
2634             raise Exception("Unexpected result: " + res)
2635         update_hapd_config(hapd)
2636
2637         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"
2638         res = sigma_dut_cmd(cmd, timeout=10)
2639         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK,NetworkIntroResult,OK,NetworkConnectResult,OK" not in res:
2640             raise Exception("Unexpected result: " + res)
2641     finally:
2642         stop_sigma_dut(sigma)
2643         dev[0].set("dpp_config_processing", "0")
2644
2645 def test_sigma_dut_ap_dpp_self_config(dev, apdev, params):
2646     """sigma_dut DPP AP Configurator using self-configuration"""
2647     logdir = os.path.join(params['logdir'],
2648                           "sigma_dut_ap_dpp_self_config.sigma-hostapd")
2649     with HWSimRadio() as (radio, iface):
2650         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
2651         try:
2652             run_sigma_dut_ap_dpp_self_config(dev, apdev)
2653         finally:
2654             stop_sigma_dut(sigma)
2655             dev[0].set("dpp_config_processing", "0", allow_fail=True)
2656
2657 def run_sigma_dut_ap_dpp_self_config(dev, apdev):
2658     check_dpp_capab(dev[0])
2659
2660     sigma_dut_cmd_check("ap_reset_default,program,DPP")
2661
2662     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Single,DPPProvisioningRole,Configurator,DPPConfEnrolleeRole,AP,DPPBS,QR,DPPConfIndex,1,DPPSelfConfigure,Yes,DPPTimeout,6", timeout=10)
2663     if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
2664             raise Exception("Unexpected result: " + res)
2665
2666     dev[0].set("dpp_config_processing", "2")
2667
2668     id = dev[0].dpp_bootstrap_gen(chan="81/11", mac=True)
2669     uri = dev[0].request("DPP_BOOTSTRAP_GET_URI %d" % id)
2670     cmd = "DPP_LISTEN 2462 role=enrollee"
2671     if "OK" not in dev[0].request(cmd):
2672         raise Exception("Failed to start listen operation")
2673
2674     res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri))
2675     if "status,COMPLETE" not in res:
2676         raise Exception("dev_exec_action did not succeed: " + res)
2677     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"
2678     res = sigma_dut_cmd(cmd)
2679     if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
2680         raise Exception("Unexpected result: " + res)
2681     dev[0].wait_connected()
2682     dev[0].request("DISCONNECT")
2683     dev[0].wait_disconnected()
2684     sigma_dut_cmd_check("ap_reset_default")
2685
2686
2687 def test_sigma_dut_ap_dpp_relay(dev, apdev, params):
2688     """sigma_dut DPP AP as Relay to Controller"""
2689     logdir = os.path.join(params['logdir'],
2690                           "sigma_dut_ap_dpp_relay.sigma-hostapd")
2691     with HWSimRadio() as (radio, iface):
2692         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
2693         try:
2694             run_sigma_dut_ap_dpp_relay(dev, apdev)
2695         finally:
2696             stop_sigma_dut(sigma)
2697             dev[1].request("DPP_CONTROLLER_STOP")
2698
2699 def run_sigma_dut_ap_dpp_relay(dev, apdev):
2700     check_dpp_capab(dev[0])
2701     check_dpp_capab(dev[1])
2702
2703     # Controller
2704     conf_id = dev[1].dpp_configurator_add()
2705     dev[1].set("dpp_configurator_params",
2706                " conf=sta-dpp configurator=%d" % conf_id)
2707     id_c = dev[1].dpp_bootstrap_gen()
2708     uri_c = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id_c)
2709     res = dev[1].request("DPP_BOOTSTRAP_INFO %d" % id_c)
2710     pkhash = None
2711     for line in res.splitlines():
2712         name, value = line.split('=')
2713         if name == "pkhash":
2714             pkhash = value
2715             break
2716     if not pkhash:
2717         raise Exception("Could not fetch public key hash from Controller")
2718     if "OK" not in dev[1].request("DPP_CONTROLLER_START"):
2719         raise Exception("Failed to start Controller")
2720
2721     sigma_dut_cmd_check("ap_reset_default,program,DPP")
2722     sigma_dut_cmd_check("ap_preset_testparameters,program,DPP,DPPConfiguratorAddress,127.0.0.1,DPPConfiguratorPKHash," + pkhash)
2723     res = sigma_dut_cmd_check("dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR")
2724
2725     dev[0].dpp_auth_init(uri=uri_c, role="enrollee")
2726     wait_auth_success(dev[1], dev[0], configurator=dev[1], enrollee=dev[0])
2727
2728     sigma_dut_cmd_check("ap_reset_default")
2729
2730 def dpp_init_tcp_enrollee(dev, id1):
2731     logger.info("Starting DPP initiator/enrollee (TCP) in a thread")
2732     time.sleep(1)
2733     cmd = "DPP_AUTH_INIT peer=%d role=enrollee tcp_addr=127.0.0.1" % id1
2734     if "OK" not in dev.request(cmd):
2735         raise Exception("Failed to initiate DPP Authentication")
2736     ev = dev.wait_event(["DPP-CONF-RECEIVED"], timeout=5)
2737     if ev is None:
2738         raise Exception("DPP configuration not completed (Enrollee)")
2739     logger.info("DPP initiator/enrollee done")
2740
2741 def test_sigma_dut_dpp_tcp_conf_resp(dev, apdev):
2742     """sigma_dut DPP TCP Configurator (Controller) as responder"""
2743     run_sigma_dut_dpp_tcp_conf_resp(dev)
2744
2745 def run_sigma_dut_dpp_tcp_conf_resp(dev, status_query=False):
2746     check_dpp_capab(dev[0])
2747     check_dpp_capab(dev[1])
2748     sigma = start_sigma_dut(dev[0].ifname)
2749     try:
2750         cmd = "dev_exec_action,program,DPP,DPPActionType,GetLocalBootstrap,DPPCryptoIdentifier,P-256,DPPBS,QR"
2751         res = sigma_dut_cmd(cmd)
2752         if "status,COMPLETE" not in res:
2753             raise Exception("dev_exec_action did not succeed: " + res)
2754         hex = res.split(',')[3]
2755         uri = from_hex(hex)
2756         logger.info("URI from sigma_dut: " + uri)
2757
2758         id1 = dev[1].dpp_qr_code(uri)
2759
2760         t = threading.Thread(target=dpp_init_tcp_enrollee, args=(dev[1], id1))
2761         t.start()
2762         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Responder,DPPConfIndex,1,DPPAuthDirection,Single,DPPProvisioningRole,Configurator,DPPConfEnrolleeRole,STA,DPPSigningKeyECC,P-256,DPPBS,QR,DPPOverTCP,yes,DPPTimeout,6"
2763         if status_query:
2764             cmd += ",DPPStatusQuery,Yes"
2765         res = sigma_dut_cmd(cmd, timeout=10)
2766         t.join()
2767         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
2768             raise Exception("Unexpected result: " + res)
2769         if status_query and "StatusResult,0" not in res:
2770             raise Exception("Status query did not succeed: " + res)
2771     finally:
2772         stop_sigma_dut(sigma)
2773
2774 def test_sigma_dut_dpp_tcp_enrollee_init(dev, apdev):
2775     """sigma_dut DPP TCP Enrollee as initiator"""
2776     check_dpp_capab(dev[0])
2777     check_dpp_capab(dev[1])
2778     sigma = start_sigma_dut(dev[0].ifname)
2779     try:
2780         # Controller
2781         conf_id = dev[1].dpp_configurator_add()
2782         dev[1].set("dpp_configurator_params",
2783                    " conf=sta-dpp configurator=%d" % conf_id)
2784         id_c = dev[1].dpp_bootstrap_gen()
2785         uri_c = dev[1].request("DPP_BOOTSTRAP_GET_URI %d" % id_c)
2786         if "OK" not in dev[1].request("DPP_CONTROLLER_START"):
2787             raise Exception("Failed to start Controller")
2788
2789         res = sigma_dut_cmd("dev_exec_action,program,DPP,DPPActionType,SetPeerBootstrap,DPPBootstrappingdata,%s,DPPBS,QR" % to_hex(uri_c))
2790         if "status,COMPLETE" not in res:
2791             raise Exception("dev_exec_action did not succeed: " + res)
2792
2793         cmd = "dev_exec_action,program,DPP,DPPActionType,AutomaticDPP,DPPAuthRole,Initiator,DPPAuthDirection,Single,DPPProvisioningRole,Enrollee,DPPConfEnrolleeRole,STA,DPPBS,QR,DPPOverTCP,127.0.0.1,DPPTimeout,6"
2794         res = sigma_dut_cmd(cmd, timeout=10)
2795         if "BootstrapResult,OK,AuthResult,OK,ConfResult,OK" not in res:
2796             raise Exception("Unexpected result: " + res)
2797     finally:
2798         stop_sigma_dut(sigma)
2799         dev[1].request("DPP_CONTROLLER_STOP")
2800
2801 def test_sigma_dut_preconfigured_profile(dev, apdev):
2802     """sigma_dut controlled connection using preconfigured profile"""
2803     try:
2804         run_sigma_dut_preconfigured_profile(dev, apdev)
2805     finally:
2806         dev[0].set("ignore_old_scan_res", "0")
2807
2808 def run_sigma_dut_preconfigured_profile(dev, apdev):
2809     ifname = dev[0].ifname
2810     sigma = start_sigma_dut(ifname)
2811
2812     try:
2813         params = hostapd.wpa2_params(ssid="test-psk", passphrase="12345678")
2814         hapd = hostapd.add_ap(apdev[0], params)
2815         dev[0].connect("test-psk", psk="12345678", scan_freq="2412",
2816                        only_add_network=True)
2817
2818         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
2819         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s" % (ifname, "test-psk"),
2820                             timeout=10)
2821         sigma_dut_wait_connected(ifname)
2822         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
2823         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
2824         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
2825     finally:
2826         stop_sigma_dut(sigma)
2827
2828 def test_sigma_dut_wps_pbc(dev, apdev):
2829     """sigma_dut and WPS PBC Enrollee"""
2830     try:
2831         run_sigma_dut_wps_pbc(dev, apdev)
2832     finally:
2833         dev[0].set("ignore_old_scan_res", "0")
2834
2835 def run_sigma_dut_wps_pbc(dev, apdev):
2836     ssid = "test-wps-conf"
2837     hapd = hostapd.add_ap(apdev[0],
2838                           {"ssid": "wps", "eap_server": "1", "wps_state": "2",
2839                            "wpa_passphrase": "12345678", "wpa": "2",
2840                            "wpa_key_mgmt": "WPA-PSK", "rsn_pairwise": "CCMP"})
2841     hapd.request("WPS_PBC")
2842
2843     ifname = dev[0].ifname
2844     sigma = start_sigma_dut(ifname)
2845
2846     try:
2847         cmd = "start_wps_registration,interface,%s" % ifname
2848         cmd += ",WpsRole,Enrollee"
2849         cmd += ",WpsConfigMethod,PBC"
2850         sigma_dut_cmd_check(cmd, timeout=15)
2851
2852         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
2853         hapd.disable()
2854         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
2855     finally:
2856         stop_sigma_dut(sigma)
2857     dev[0].flush_scan_cache()
2858
2859 def test_sigma_dut_sta_scan_bss(dev, apdev):
2860     """sigma_dut sta_scan_bss"""
2861     hapd = hostapd.add_ap(apdev[0], {"ssid": "test"})
2862     sigma = start_sigma_dut(dev[0].ifname)
2863     try:
2864         cmd = "sta_scan_bss,Interface,%s,BSSID,%s" % (dev[0].ifname, \
2865                                                       hapd.own_addr())
2866         res = sigma_dut_cmd(cmd, timeout=10)
2867         if "ssid,test,bsschannel,1" not in res:
2868             raise Exception("Unexpected result: " + res)
2869     finally:
2870         stop_sigma_dut(sigma)
2871
2872 def test_sigma_dut_sta_scan_ssid_bssid(dev, apdev):
2873     """sigma_dut sta_scan GetParameter,SSID_BSSID"""
2874     hostapd.add_ap(apdev[0], {"ssid": "abcdef"})
2875     hostapd.add_ap(apdev[1], {"ssid": "qwerty"})
2876     sigma = start_sigma_dut(dev[0].ifname)
2877     try:
2878         cmd = "sta_scan,Interface,%s,GetParameter,SSID_BSSID" % dev[0].ifname
2879         res = sigma_dut_cmd(cmd, timeout=10)
2880         if "abcdef" not in res or "qwerty" not in res:
2881             raise Exception("Unexpected result: " + res)
2882     finally:
2883         stop_sigma_dut(sigma)
2884
2885 def test_sigma_dut_ap_osen(dev, apdev, params):
2886     """sigma_dut controlled AP with OSEN"""
2887     logdir = os.path.join(params['logdir'],
2888                           "sigma_dut_ap_osen.sigma-hostapd")
2889     with HWSimRadio() as (radio, iface):
2890         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
2891         try:
2892             sigma_dut_cmd_check("ap_reset_default")
2893             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-hs20,MODE,11ng")
2894             sigma_dut_cmd_check("ap_set_radius,NAME,AP,IPADDR,127.0.0.1,PORT,1812,PASSWORD,radius")
2895             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,OSEN,PMF,Optional")
2896             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
2897
2898             # RSN-OSEN (for OSU)
2899             dev[0].connect("test-hs20", proto="OSEN", key_mgmt="OSEN",
2900                            pairwise="CCMP", group="GTK_NOT_USED",
2901                            eap="WFA-UNAUTH-TLS", identity="osen@example.com",
2902                            ca_cert="auth_serv/ca.pem", scan_freq="2412")
2903
2904             sigma_dut_cmd_check("ap_reset_default")
2905         finally:
2906             stop_sigma_dut(sigma)
2907
2908 def test_sigma_dut_ap_eap_osen(dev, apdev, params):
2909     """sigma_dut controlled AP with EAP+OSEN"""
2910     logdir = os.path.join(params['logdir'],
2911                           "sigma_dut_ap_eap_osen.sigma-hostapd")
2912     with HWSimRadio() as (radio, iface):
2913         sigma = start_sigma_dut(iface, bridge="ap-br0", hostapd_logdir=logdir)
2914         try:
2915             sigma_dut_cmd_check("ap_reset_default")
2916             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-hs20,MODE,11ng")
2917             sigma_dut_cmd_check("ap_set_radius,NAME,AP,IPADDR,127.0.0.1,PORT,1812,PASSWORD,radius")
2918             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-ENT-OSEN,PMF,Optional")
2919             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
2920
2921             subprocess.call(['brctl', 'setfd', 'ap-br0', '0'])
2922             subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
2923
2924             # RSN-OSEN (for OSU)
2925             dev[0].connect("test-hs20", proto="OSEN", key_mgmt="OSEN",
2926                            pairwise="CCMP",
2927                            eap="WFA-UNAUTH-TLS", identity="osen@example.com",
2928                            ca_cert="auth_serv/ca.pem", ieee80211w='2',
2929                            scan_freq="2412")
2930             # RSN-EAP (for data connection)
2931             dev[1].connect("test-hs20", key_mgmt="WPA-EAP", eap="TTLS",
2932                            identity="hs20-test", password="password",
2933                            ca_cert="auth_serv/ca.pem", phase2="auth=MSCHAPV2",
2934                            ieee80211w='2', scan_freq="2412")
2935
2936             hwsim_utils.test_connectivity(dev[0], dev[1], broadcast=False,
2937                                           success_expected=False, timeout=1)
2938
2939             sigma_dut_cmd_check("ap_reset_default")
2940         finally:
2941             stop_sigma_dut(sigma)
2942             subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'down'],
2943                             stderr=open('/dev/null', 'w'))
2944             subprocess.call(['brctl', 'delbr', 'ap-br0'],
2945                             stderr=open('/dev/null', 'w'))
2946
2947 def test_sigma_dut_ap_eap(dev, apdev, params):
2948     """sigma_dut controlled AP WPA2-Enterprise"""
2949     logdir = os.path.join(params['logdir'], "sigma_dut_ap_eap.sigma-hostapd")
2950     with HWSimRadio() as (radio, iface):
2951         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
2952         try:
2953             sigma_dut_cmd_check("ap_reset_default")
2954             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-eap,MODE,11ng")
2955             sigma_dut_cmd_check("ap_set_radius,NAME,AP,IPADDR,127.0.0.1,PORT,1812,PASSWORD,radius")
2956             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-ENT")
2957             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
2958
2959             dev[0].connect("test-eap", key_mgmt="WPA-EAP", eap="GPSK",
2960                            identity="gpsk user",
2961                            password="abcdefghijklmnop0123456789abcdef",
2962                            scan_freq="2412")
2963
2964             sigma_dut_cmd_check("ap_reset_default")
2965         finally:
2966             stop_sigma_dut(sigma)
2967
2968 def test_sigma_dut_ap_eap_sha256(dev, apdev, params):
2969     """sigma_dut controlled AP WPA2-Enterprise SHA256"""
2970     logdir = os.path.join(params['logdir'],
2971                           "sigma_dut_ap_eap_sha256.sigma-hostapd")
2972     with HWSimRadio() as (radio, iface):
2973         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
2974         try:
2975             sigma_dut_cmd_check("ap_reset_default")
2976             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-eap,MODE,11ng")
2977             sigma_dut_cmd_check("ap_set_radius,NAME,AP,IPADDR,127.0.0.1,PORT,1812,PASSWORD,radius")
2978             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-ENT-256")
2979             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
2980
2981             dev[0].connect("test-eap", key_mgmt="WPA-EAP-SHA256", eap="GPSK",
2982                            identity="gpsk user",
2983                            password="abcdefghijklmnop0123456789abcdef",
2984                            scan_freq="2412")
2985
2986             sigma_dut_cmd_check("ap_reset_default")
2987         finally:
2988             stop_sigma_dut(sigma)
2989
2990 def test_sigma_dut_ap_ft_eap(dev, apdev, params):
2991     """sigma_dut controlled AP FT-EAP"""
2992     logdir = os.path.join(params['logdir'], "sigma_dut_ap_ft_eap.sigma-hostapd")
2993     with HWSimRadio() as (radio, iface):
2994         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
2995         try:
2996             sigma_dut_cmd_check("ap_reset_default")
2997             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-ft-eap,MODE,11ng,DOMAIN,0101,FT_OA,Enable")
2998             sigma_dut_cmd_check("ap_set_radius,NAME,AP,IPADDR,127.0.0.1,PORT,1812,PASSWORD,radius")
2999             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,FT-EAP")
3000             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
3001
3002             dev[0].connect("test-ft-eap", key_mgmt="FT-EAP", eap="GPSK",
3003                            identity="gpsk user",
3004                            password="abcdefghijklmnop0123456789abcdef",
3005                            scan_freq="2412")
3006
3007             sigma_dut_cmd_check("ap_reset_default")
3008         finally:
3009             stop_sigma_dut(sigma)
3010
3011 def test_sigma_dut_ap_ft_psk(dev, apdev, params):
3012     """sigma_dut controlled AP FT-PSK"""
3013     logdir = os.path.join(params['logdir'], "sigma_dut_ap_ft_psk.sigma-hostapd")
3014     with HWSimRadio() as (radio, iface):
3015         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
3016         try:
3017             sigma_dut_cmd_check("ap_reset_default")
3018             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-ft-psk,MODE,11ng,DOMAIN,0101,FT_OA,Enable")
3019             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,FT-PSK,PSK,12345678")
3020             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
3021
3022             dev[0].connect("test-ft-psk", key_mgmt="FT-PSK", psk="12345678",
3023                            scan_freq="2412")
3024
3025             sigma_dut_cmd_check("ap_reset_default")
3026         finally:
3027             stop_sigma_dut(sigma)
3028
3029 def test_sigma_dut_ap_ft_over_ds_psk(dev, apdev, params):
3030     """sigma_dut controlled AP FT-PSK (over-DS)"""
3031     logdir = os.path.join(params['logdir'],
3032                           "sigma_dut_ap_ft_over_ds_psk.sigma-hostapd")
3033     conffile = os.path.join(params['logdir'],
3034                             "sigma_dut_ap_ft_over_ds_psk.sigma-conf")
3035     with HWSimRadio() as (radio, iface):
3036         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
3037         try:
3038             sigma_dut_cmd_check("ap_reset_default")
3039             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-ft-psk,MODE,11ng,DOMAIN,0101,FT_DS,Enable")
3040             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,FT-PSK,PSK,12345678")
3041             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
3042
3043             with open("/tmp/sigma_dut-ap.conf", "rb") as f:
3044                 with open(conffile, "wb") as f2:
3045                     f2.write(f.read())
3046
3047             dev[0].connect("test-ft-psk", key_mgmt="FT-PSK", psk="12345678",
3048                            scan_freq="2412")
3049
3050             sigma_dut_cmd_check("ap_reset_default")
3051         finally:
3052             stop_sigma_dut(sigma)
3053
3054 def test_sigma_dut_ap_ent_ft_eap(dev, apdev, params):
3055     """sigma_dut controlled AP WPA-EAP and FT-EAP"""
3056     logdir = os.path.join(params['logdir'],
3057                           "sigma_dut_ap_ent_ft_eap.sigma-hostapd")
3058     with HWSimRadio() as (radio, iface):
3059         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
3060         try:
3061             sigma_dut_cmd_check("ap_reset_default")
3062             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-ent-ft-eap,MODE,11ng,DOMAIN,0101,FT_OA,Enable")
3063             sigma_dut_cmd_check("ap_set_radius,NAME,AP,IPADDR,127.0.0.1,PORT,1812,PASSWORD,radius")
3064             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-ENT-FT-EAP")
3065             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
3066
3067             dev[0].connect("test-ent-ft-eap", key_mgmt="FT-EAP", eap="GPSK",
3068                            identity="gpsk user",
3069                            password="abcdefghijklmnop0123456789abcdef",
3070                            scan_freq="2412")
3071             dev[1].connect("test-ent-ft-eap", key_mgmt="WPA-EAP", eap="GPSK",
3072                            identity="gpsk user",
3073                            password="abcdefghijklmnop0123456789abcdef",
3074                            scan_freq="2412")
3075
3076             sigma_dut_cmd_check("ap_reset_default")
3077         finally:
3078             stop_sigma_dut(sigma)
3079
3080 def test_sigma_dut_venue_url(dev, apdev):
3081     """sigma_dut controlled Venue URL fetch"""
3082     try:
3083         run_sigma_dut_venue_url(dev, apdev)
3084     finally:
3085         dev[0].set("ignore_old_scan_res", "0")
3086
3087 def run_sigma_dut_venue_url(dev, apdev):
3088     ifname = dev[0].ifname
3089     sigma = start_sigma_dut(ifname)
3090
3091     try:
3092         ssid = "venue"
3093         params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
3094         params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
3095         params["ieee80211w"] = "2"
3096
3097         venue_group = 1
3098         venue_type = 13
3099         venue_info = struct.pack('BB', venue_group, venue_type)
3100         lang1 = "eng"
3101         name1 = "Example venue"
3102         lang2 = "fin"
3103         name2 = "Esimerkkipaikka"
3104         venue1 = struct.pack('B', len(lang1 + name1)) + lang1.encode() + name1.encode()
3105         venue2 = struct.pack('B', len(lang2 + name2)) + lang2.encode() + name2.encode()
3106         venue_name = binascii.hexlify(venue_info + venue1 + venue2)
3107
3108         url1 = "http://example.com/venue"
3109         url2 = "https://example.org/venue-info/"
3110         params["venue_group"] = str(venue_group)
3111         params["venue_type"] = str(venue_type)
3112         params["venue_name"] = [lang1 + ":" + name1, lang2 + ":" + name2]
3113         params["venue_url"] = ["1:" + url1, "2:" + url2]
3114
3115         hapd = hostapd.add_ap(apdev[0], params)
3116
3117         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,PMF" % ifname)
3118         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
3119         sigma_dut_cmd_check("sta_set_psk,interface,%s,ssid,%s,passphrase,%s,encpType,aes-ccmp,keymgmttype,wpa2,PMF,Required" % (ifname, "venue", "12345678"))
3120         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "venue"),
3121                             timeout=10)
3122         sigma_dut_wait_connected(ifname)
3123         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
3124         sigma_dut_cmd_check("sta_hs2_venue_info,interface," + ifname + ",Display,Yes")
3125         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
3126         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
3127     finally:
3128         stop_sigma_dut(sigma)
3129
3130 def test_sigma_dut_hs20_assoc_24(dev, apdev):
3131     """sigma_dut controlled Hotspot 2.0 connection (2.4 GHz)"""
3132     run_sigma_dut_hs20_assoc(dev, apdev, True)
3133
3134 def test_sigma_dut_hs20_assoc_5(dev, apdev):
3135     """sigma_dut controlled Hotspot 2.0 connection (5 GHz)"""
3136     run_sigma_dut_hs20_assoc(dev, apdev, False)
3137
3138 def run_sigma_dut_hs20_assoc(dev, apdev, band24):
3139     hapd0 = None
3140     hapd1 = None
3141     try:
3142         bssid0 = apdev[0]['bssid']
3143         params = hs20_ap_params()
3144         params['hessid'] = bssid0
3145         hapd0 = hostapd.add_ap(apdev[0], params)
3146
3147         bssid1 = apdev[1]['bssid']
3148         params = hs20_ap_params()
3149         params['hessid'] = bssid0
3150         params["hw_mode"] = "a"
3151         params["channel"] = "36"
3152         params["country_code"] = "US"
3153         hapd1 = hostapd.add_ap(apdev[1], params)
3154
3155         band = "2.4" if band24 else "5"
3156         exp_bssid = bssid0 if band24 else bssid1
3157         run_sigma_dut_hs20_assoc_2(dev, apdev, band, exp_bssid)
3158     finally:
3159         dev[0].request("DISCONNECT")
3160         if hapd0:
3161             hapd0.request("DISABLE")
3162         if hapd1:
3163             hapd1.request("DISABLE")
3164         subprocess.call(['iw', 'reg', 'set', '00'])
3165         dev[0].flush_scan_cache()
3166
3167 def run_sigma_dut_hs20_assoc_2(dev, apdev, band, expect_bssid):
3168     check_eap_capa(dev[0], "MSCHAPV2")
3169     dev[0].flush_scan_cache()
3170
3171     ifname = dev[0].ifname
3172     sigma = start_sigma_dut(ifname)
3173
3174     try:
3175         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,HS2-R3" % ifname)
3176         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
3177         sigma_dut_cmd_check("sta_add_credential,interface,%s,type,uname_pwd,realm,example.com,username,hs20-test,password,password" % ifname)
3178         res = sigma_dut_cmd_check("sta_hs2_associate,interface,%s,band,%s" % (ifname, band),
3179                                   timeout=15)
3180         sigma_dut_wait_connected(ifname)
3181         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
3182         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
3183         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
3184     finally:
3185         stop_sigma_dut(sigma)
3186
3187     if "BSSID," + expect_bssid not in res:
3188         raise Exception("Unexpected BSSID: " + res)
3189
3190 def test_sigma_dut_ap_hs20(dev, apdev, params):
3191     """sigma_dut controlled AP with Hotspot 2.0 parameters"""
3192     logdir = os.path.join(params['logdir'],
3193                           "sigma_dut_ap_hs20.sigma-hostapd")
3194     conffile = os.path.join(params['logdir'],
3195                             "sigma_dut_ap_hs20.sigma-conf")
3196     with HWSimRadio() as (radio, iface):
3197         sigma = start_sigma_dut(iface, hostapd_logdir=logdir)
3198         try:
3199             sigma_dut_cmd_check("ap_reset_default,NAME,AP,program,HS2-R3")
3200             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,WLAN_TAG,1,CHANNEL,1,SSID,test-hs20,MODE,11ng")
3201             sigma_dut_cmd_check("ap_set_radius,NAME,AP,WLAN_TAG,1,IPADDR,127.0.0.1,PORT,1812,PASSWORD,radius")
3202             sigma_dut_cmd_check("ap_set_security,NAME,AP,WLAN_TAG,1,KEYMGNT,WPA2-ENT")
3203             sigma_dut_cmd_check("ap_set_hs2,NAME,AP,WLAN_TAG,1,HESSID,02:12:34:56:78:9a,NAI_REALM_LIST,1,OPER_NAME,1")
3204             sigma_dut_cmd_check("ap_set_hs2,NAME,AP,WLAN_TAG,1,OSU_SERVER_URI,https://example.com/ https://example.org/,OSU_SSID,test-osu,OSU_METHOD,SOAP SOAP,OSU_PROVIDER_LIST,10,OSU_PROVIDER_NAI_LIST,4")
3205             sigma_dut_cmd_check("ap_set_hs2,NAME,AP,WLAN_TAG,1,NET_AUTH_TYPE,2")
3206             sigma_dut_cmd_check("ap_set_hs2,NAME,AP,WLAN_TAG,1,VENUE_NAME,1")
3207             sigma_dut_cmd_check("ap_set_hs2,NAME,AP,WLAN_TAG,1,DOMAIN_LIST,example.com")
3208             sigma_dut_cmd_check("ap_set_hs2,NAME,AP,WLAN_TAG,1,OPERATOR_ICON_METADATA,1")
3209             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,WLAN_TAG,2,CHANNEL,1,SSID,test-osu,MODE,11ng")
3210             sigma_dut_cmd_check("ap_set_security,NAME,AP,WLAN_TAG,2,KEYMGNT,NONE")
3211             sigma_dut_cmd_check("ap_set_hs2,NAME,AP,WLAN_TAG,2,OSU,1")
3212             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
3213
3214             with open("/tmp/sigma_dut-ap.conf", "rb") as f:
3215                 with open(conffile, "wb") as f2:
3216                     f2.write(f.read())
3217
3218             sigma_dut_cmd_check("ap_reset_default")
3219         finally:
3220             stop_sigma_dut(sigma)
3221
3222 def test_sigma_dut_eap_ttls_uosc(dev, apdev, params):
3223     """sigma_dut controlled STA and EAP-TTLS with UOSC"""
3224     logdir = params['logdir']
3225
3226     with open("auth_serv/ca.pem", "r") as f:
3227         with open(os.path.join(logdir, "sigma_dut_eap_ttls_uosc.ca.pem"),
3228                   "w") as f2:
3229             f2.write(f.read())
3230
3231     src = "auth_serv/server.pem"
3232     dst = os.path.join(logdir, "sigma_dut_eap_ttls_uosc.server.der")
3233     hashdst = os.path.join(logdir, "sigma_dut_eap_ttls_uosc.server.pem.sha256")
3234     subprocess.check_call(["openssl", "x509", "-in", src, "-out", dst,
3235                            "-outform", "DER"],
3236                           stderr=open('/dev/null', 'w'))
3237     with open(dst, "rb") as f:
3238         der = f.read()
3239     hash = hashlib.sha256(der).digest()
3240     with open(hashdst, "w") as f:
3241         f.write(binascii.hexlify(hash).decode())
3242
3243     dst = os.path.join(logdir, "sigma_dut_eap_ttls_uosc.incorrect.pem.sha256")
3244     with open(dst, "w") as f:
3245         f.write(32*"00")
3246
3247     ssid = "test-wpa2-eap"
3248     params = hostapd.wpa2_eap_params(ssid=ssid)
3249     hapd = hostapd.add_ap(apdev[0], params)
3250
3251     ifname = dev[0].ifname
3252     sigma = start_sigma_dut(ifname, cert_path=logdir)
3253
3254     try:
3255         cmd = "sta_set_security,type,eapttls,interface,%s,ssid,%s,keymgmttype,wpa2,encType,AES-CCMP,PairwiseCipher,AES-CCMP-128,username,DOMAIN\mschapv2 user,password,password,ServerCert,sigma_dut_eap_ttls_uosc.incorrect.pem" % (ifname, ssid)
3256
3257         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
3258         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
3259         sigma_dut_cmd_check(cmd)
3260         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, ssid),
3261                             timeout=10)
3262         ev = dev[0].wait_event(["CTRL-EVENT-EAP-TLS-CERT-ERROR"], timeout=10)
3263         if ev is None:
3264             raise Exception("Server certificate error not reported")
3265
3266         res = sigma_dut_cmd_check("dev_exec_action,program,WPA3,interface,%s,ServerCertTrust,Accept" % ifname)
3267         if "ServerCertTrustResult,Accepted" not in res:
3268             raise Exception("Server certificate trust was not accepted")
3269         sigma_dut_wait_connected(ifname)
3270         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
3271         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
3272         dev[0].dump_monitor()
3273     finally:
3274         stop_sigma_dut(sigma)
3275
3276 def test_sigma_dut_eap_ttls_uosc_tod(dev, apdev, params):
3277     """sigma_dut controlled STA and EAP-TTLS with UOSC/TOD-STRICT"""
3278     run_sigma_dut_eap_ttls_uosc_tod(dev, apdev, params, False)
3279
3280 def test_sigma_dut_eap_ttls_uosc_tod_tofu(dev, apdev, params):
3281     """sigma_dut controlled STA and EAP-TTLS with UOSC/TOD-TOFU"""
3282     run_sigma_dut_eap_ttls_uosc_tod(dev, apdev, params, True)
3283
3284 def run_sigma_dut_eap_ttls_uosc_tod(dev, apdev, params, tofu):
3285     logdir = params['logdir']
3286
3287     name = "sigma_dut_eap_ttls_uosc_tod"
3288     if tofu:
3289         name += "_tofu"
3290     with open("auth_serv/ca.pem", "r") as f:
3291         with open(os.path.join(logdir, name + ".ca.pem"), "w") as f2:
3292             f2.write(f.read())
3293
3294     if tofu:
3295         src = "auth_serv/server-certpol2.pem"
3296     else:
3297         src = "auth_serv/server-certpol.pem"
3298     dst = os.path.join(logdir, name + ".server.der")
3299     hashdst = os.path.join(logdir, name + ".server.pem.sha256")
3300     subprocess.check_call(["openssl", "x509", "-in", src, "-out", dst,
3301                            "-outform", "DER"],
3302                           stderr=open('/dev/null', 'w'))
3303     with open(dst, "rb") as f:
3304         der = f.read()
3305     hash = hashlib.sha256(der).digest()
3306     with open(hashdst, "w") as f:
3307         f.write(binascii.hexlify(hash).decode())
3308
3309     ssid = "test-wpa2-eap"
3310     params = int_eap_server_params()
3311     params["ssid"] = ssid
3312     if tofu:
3313         params["server_cert"] = "auth_serv/server-certpol2.pem"
3314         params["private_key"] = "auth_serv/server-certpol2.key"
3315     else:
3316         params["server_cert"] = "auth_serv/server-certpol.pem"
3317         params["private_key"] = "auth_serv/server-certpol.key"
3318     hapd = hostapd.add_ap(apdev[0], params)
3319
3320     ifname = dev[0].ifname
3321     sigma = start_sigma_dut(ifname, cert_path=logdir)
3322
3323     try:
3324         cmd = ("sta_set_security,type,eapttls,interface,%s,ssid,%s,keymgmttype,wpa2,encType,AES-CCMP,PairwiseCipher,AES-CCMP-128,trustedRootCA," + name + ".ca.pem,username,DOMAIN\mschapv2 user,password,password,ServerCert," + name + ".server.pem") % (ifname, ssid)
3325         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
3326         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
3327         sigma_dut_cmd_check(cmd)
3328         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, ssid),
3329                             timeout=10)
3330         sigma_dut_wait_connected(ifname)
3331         sigma_dut_cmd_check("sta_get_ip_config,interface," + ifname)
3332         sigma_dut_cmd_check("sta_disconnect,interface," + ifname + ",maintain_profile,1")
3333         dev[0].wait_disconnected()
3334         dev[0].dump_monitor()
3335
3336         hapd.disable()
3337         params = hostapd.wpa2_eap_params(ssid=ssid)
3338         hapd = hostapd.add_ap(apdev[0], params)
3339
3340         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, ssid),
3341                             timeout=10)
3342         ev = dev[0].wait_event(["CTRL-EVENT-EAP-TLS-CERT-ERROR"], timeout=10)
3343         if ev is None:
3344             raise Exception("Server certificate error not reported")
3345
3346         res = sigma_dut_cmd_check("dev_exec_action,program,WPA3,interface,%s,ServerCertTrust,Accept" % ifname)
3347         if "ServerCertTrustResult,Accepted" in res:
3348             raise Exception("Server certificate trust override was accepted unexpectedly")
3349         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
3350         dev[0].dump_monitor()
3351     finally:
3352         stop_sigma_dut(sigma)
3353
3354 def test_sigma_dut_eap_ttls_uosc_initial_tod_strict(dev, apdev, params):
3355     """sigma_dut controlled STA and EAP-TTLS with initial UOSC/TOD-STRICT"""
3356     run_sigma_dut_eap_ttls_uosc_initial_tod(dev, apdev, params, False)
3357
3358 def test_sigma_dut_eap_ttls_uosc_initial_tod_tofu(dev, apdev, params):
3359     """sigma_dut controlled STA and EAP-TTLS with initial UOSC/TOD-TOFU"""
3360     run_sigma_dut_eap_ttls_uosc_initial_tod(dev, apdev, params, True)
3361
3362 def run_sigma_dut_eap_ttls_uosc_initial_tod(dev, apdev, params, tofu):
3363     logdir = params['logdir']
3364
3365     name = "sigma_dut_eap_ttls_uosc_initial_tod"
3366     if tofu:
3367         name += "_tofu"
3368     with open("auth_serv/rsa3072-ca.pem", "r") as f:
3369         with open(os.path.join(logdir, name + ".ca.pem"), "w") as f2:
3370             f2.write(f.read())
3371
3372     if tofu:
3373         src = "auth_serv/server-certpol2.pem"
3374     else:
3375         src = "auth_serv/server-certpol.pem"
3376     dst = os.path.join(logdir, name + ".server.der")
3377     hashdst = os.path.join(logdir, name + ".server.pem.sha256")
3378     subprocess.check_call(["openssl", "x509", "-in", src, "-out", dst,
3379                            "-outform", "DER"],
3380                           stderr=open('/dev/null', 'w'))
3381     with open(dst, "rb") as f:
3382         der = f.read()
3383     hash = hashlib.sha256(der).digest()
3384     with open(hashdst, "w") as f:
3385         f.write(binascii.hexlify(hash).decode())
3386
3387     ssid = "test-wpa2-eap"
3388     params = int_eap_server_params()
3389     params["ssid"] = ssid
3390     if tofu:
3391         params["server_cert"] = "auth_serv/server-certpol2.pem"
3392         params["private_key"] = "auth_serv/server-certpol2.key"
3393     else:
3394         params["server_cert"] = "auth_serv/server-certpol.pem"
3395         params["private_key"] = "auth_serv/server-certpol.key"
3396     hapd = hostapd.add_ap(apdev[0], params)
3397
3398     ifname = dev[0].ifname
3399     sigma = start_sigma_dut(ifname, cert_path=logdir)
3400
3401     try:
3402         cmd = ("sta_set_security,type,eapttls,interface,%s,ssid,%s,keymgmttype,wpa2,encType,AES-CCMP,PairwiseCipher,AES-CCMP-128,trustedRootCA," + name + ".ca.pem,username,DOMAIN\mschapv2 user,password,password") % (ifname, ssid)
3403         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
3404         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
3405         sigma_dut_cmd_check(cmd)
3406         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, ssid),
3407                             timeout=10)
3408         ev = dev[0].wait_event(["CTRL-EVENT-EAP-TLS-CERT-ERROR"], timeout=15)
3409         if ev is None:
3410             raise Exception("Server certificate validation failure not reported")
3411
3412         res = sigma_dut_cmd_check("dev_exec_action,program,WPA3,interface,%s,ServerCertTrust,Accept" % ifname)
3413         if not tofu and "ServerCertTrustResult,Accepted" in res:
3414             raise Exception("Server certificate trust override was accepted unexpectedly")
3415         if tofu and "ServerCertTrustResult,Accepted" not in res:
3416             raise Exception("Server certificate trust override was not accepted")
3417         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
3418         dev[0].dump_monitor()
3419     finally:
3420         stop_sigma_dut(sigma)
3421
3422 def test_sigma_dut_eap_ttls_uosc_ca_mistrust(dev, apdev, params):
3423     """sigma_dut controlled STA and EAP-TTLS with UOSC when CA is not trusted"""
3424     check_domain_suffix_match(dev[0])
3425     logdir = params['logdir']
3426
3427     with open("auth_serv/ca.pem", "r") as f:
3428         with open(os.path.join(logdir,
3429                                "sigma_dut_eap_ttls_uosc_ca_mistrust.ca.pem"),
3430                   "w") as f2:
3431             f2.write(f.read())
3432
3433     ssid = "test-wpa2-eap"
3434     params = int_eap_server_params()
3435     params["ssid"] = ssid
3436     params["ca_cert"] = "auth_serv/rsa3072-ca.pem"
3437     params["server_cert"] = "auth_serv/rsa3072-server.pem"
3438     params["private_key"] = "auth_serv/rsa3072-server.key"
3439     hapd = hostapd.add_ap(apdev[0], params)
3440
3441     ifname = dev[0].ifname
3442     sigma = start_sigma_dut(ifname, cert_path=logdir)
3443
3444     try:
3445         cmd = "sta_set_security,type,eapttls,interface,%s,ssid,%s,keymgmttype,wpa2,encType,AES-CCMP,PairwiseCipher,AES-CCMP-128,trustedRootCA,sigma_dut_eap_ttls_uosc_ca_mistrust.ca.pem,username,DOMAIN\mschapv2 user,password,password,domainSuffix,w1.fi" % (ifname, ssid)
3446         sigma_dut_cmd_check("sta_reset_default,interface,%s,prog,WPA3" % ifname)
3447         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
3448         sigma_dut_cmd_check(cmd)
3449         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, ssid),
3450                             timeout=10)
3451         ev = dev[0].wait_event(["CTRL-EVENT-EAP-TLS-CERT-ERROR"], timeout=10)
3452         if ev is None:
3453             raise Exception("Server certificate error not reported")
3454
3455         res = sigma_dut_cmd_check("dev_exec_action,program,WPA3,interface,%s,ServerCertTrust,Accept" % ifname)
3456         if "ServerCertTrustResult,Accepted" not in res:
3457             raise Exception("Server certificate trust was not accepted")
3458         sigma_dut_wait_connected(ifname)
3459         sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
3460         sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
3461         dev[0].dump_monitor()
3462     finally:
3463         stop_sigma_dut(sigma)
3464
3465 def start_sae_pwe_ap(apdev, sae_pwe):
3466     ssid = "test-sae"
3467     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
3468     params['wpa_key_mgmt'] = 'SAE'
3469     params["ieee80211w"] = "2"
3470     params['sae_groups'] = '19'
3471     params['sae_pwe'] = str(sae_pwe)
3472     return hostapd.add_ap(apdev, params)
3473
3474 def connect_sae_pwe_sta(dev, ifname, extra=None):
3475     dev.dump_monitor()
3476     sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
3477     sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
3478     cmd = "sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2" % (ifname, "test-sae", "12345678")
3479     if extra:
3480         cmd += "," + extra
3481     sigma_dut_cmd_check(cmd)
3482     sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
3483                         timeout=10)
3484     sigma_dut_wait_connected(ifname)
3485     sigma_dut_cmd_check("sta_disconnect,interface," + ifname)
3486     dev.wait_disconnected()
3487     sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
3488     dev.dump_monitor()
3489
3490 def no_connect_sae_pwe_sta(dev, ifname, extra=None):
3491     dev.dump_monitor()
3492     sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
3493     sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
3494     cmd = "sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2" % (ifname, "test-sae", "12345678")
3495     if extra:
3496         cmd += "," + extra
3497     sigma_dut_cmd_check(cmd)
3498     sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
3499                         timeout=10)
3500     ev = dev.wait_event(["CTRL-EVENT-CONNECTED",
3501                          "CTRL-EVENT-NETWORK-NOT-FOUND"], timeout=10)
3502     if ev is None or "CTRL-EVENT-CONNECTED" in ev:
3503         raise Exception("Unexpected connection result")
3504     sigma_dut_cmd_check("sta_reset_default,interface," + ifname)
3505     dev.dump_monitor()
3506
3507 def test_sigma_dut_sae_h2e(dev, apdev):
3508     """sigma_dut controlled SAE H2E association (AP using loop+H2E)"""
3509     if "SAE" not in dev[0].get_capability("auth_alg"):
3510         raise HwsimSkip("SAE not supported")
3511
3512     start_sae_pwe_ap(apdev[0], 2)
3513
3514     ifname = dev[0].ifname
3515     sigma = start_sigma_dut(ifname, sae_h2e=True)
3516     try:
3517         connect_sae_pwe_sta(dev[0], ifname)
3518         connect_sae_pwe_sta(dev[0], ifname, extra="sae_pwe,h2e")
3519         connect_sae_pwe_sta(dev[0], ifname, extra="sae_pwe,loop")
3520         res = sigma_dut_cmd("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2,sae_pwe,unknown" % (ifname, "test-sae", "12345678"))
3521         if res != "status,ERROR,errorCode,Unsupported sae_pwe value":
3522             raise Exception("Unexpected error result: " + res)
3523     finally:
3524         stop_sigma_dut(sigma)
3525         dev[0].set("sae_pwe", "0")
3526
3527 def test_sigma_dut_sae_h2e_ap_loop(dev, apdev):
3528     """sigma_dut controlled SAE H2E association (AP using loop-only)"""
3529     if "SAE" not in dev[0].get_capability("auth_alg"):
3530         raise HwsimSkip("SAE not supported")
3531
3532     start_sae_pwe_ap(apdev[0], 0)
3533
3534     ifname = dev[0].ifname
3535     sigma = start_sigma_dut(ifname, sae_h2e=True)
3536     try:
3537         connect_sae_pwe_sta(dev[0], ifname)
3538         connect_sae_pwe_sta(dev[0], ifname, extra="sae_pwe,loop")
3539         no_connect_sae_pwe_sta(dev[0], ifname, extra="sae_pwe,h2e")
3540     finally:
3541         stop_sigma_dut(sigma)
3542         dev[0].set("sae_pwe", "0")
3543
3544 def test_sigma_dut_sae_h2e_ap_h2e(dev, apdev):
3545     """sigma_dut controlled SAE H2E association (AP using H2E-only)"""
3546     if "SAE" not in dev[0].get_capability("auth_alg"):
3547         raise HwsimSkip("SAE not supported")
3548
3549     start_sae_pwe_ap(apdev[0], 1)
3550
3551     ifname = dev[0].ifname
3552     sigma = start_sigma_dut(ifname, sae_h2e=True)
3553     try:
3554         connect_sae_pwe_sta(dev[0], ifname)
3555         no_connect_sae_pwe_sta(dev[0], ifname, extra="sae_pwe,loop")
3556         connect_sae_pwe_sta(dev[0], ifname, extra="sae_pwe,h2e")
3557     finally:
3558         stop_sigma_dut(sigma)
3559         dev[0].set("sae_pwe", "0")
3560
3561 def test_sigma_dut_ap_sae_h2e(dev, apdev, params):
3562     """sigma_dut controlled AP with SAE H2E"""
3563     logdir = os.path.join(params['logdir'],
3564                           "sigma_dut_ap_sae_h2e.sigma-hostapd")
3565     if "SAE" not in dev[0].get_capability("auth_alg"):
3566         raise HwsimSkip("SAE not supported")
3567     with HWSimRadio() as (radio, iface):
3568         sigma = start_sigma_dut(iface, sae_h2e=True, hostapd_logdir=logdir)
3569         try:
3570             sigma_dut_cmd_check("ap_reset_default")
3571             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
3572             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK,12345678")
3573             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
3574
3575             for sae_pwe in [0, 1, 2]:
3576                 dev[0].request("SET sae_groups ")
3577                 dev[0].set("sae_pwe", str(sae_pwe))
3578                 dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
3579                                ieee80211w="2", scan_freq="2412")
3580                 dev[0].request("REMOVE_NETWORK all")
3581                 dev[0].wait_disconnected()
3582                 dev[0].dump_monitor()
3583
3584             sigma_dut_cmd_check("ap_reset_default")
3585         finally:
3586             stop_sigma_dut(sigma)
3587             dev[0].set("sae_pwe", "0")
3588
3589 def test_sigma_dut_ap_sae_h2e_only(dev, apdev, params):
3590     """sigma_dut controlled AP with SAE H2E-only"""
3591     logdir = os.path.join(params['logdir'],
3592                           "sigma_dut_ap_sae_h2e.sigma-hostapd")
3593     if "SAE" not in dev[0].get_capability("auth_alg"):
3594         raise HwsimSkip("SAE not supported")
3595     with HWSimRadio() as (radio, iface):
3596         sigma = start_sigma_dut(iface, sae_h2e=True, hostapd_logdir=logdir)
3597         try:
3598             sigma_dut_cmd_check("ap_reset_default")
3599             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
3600             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK,12345678,sae_pwe,h2e")
3601             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
3602
3603             dev[0].request("SET sae_groups ")
3604             dev[0].set("sae_pwe", "1")
3605             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
3606                            ieee80211w="2", scan_freq="2412")
3607             dev[0].request("REMOVE_NETWORK all")
3608             dev[0].wait_disconnected()
3609             dev[0].dump_monitor()
3610
3611             dev[0].set("sae_pwe", "0")
3612             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
3613                            ieee80211w="2", scan_freq="2412", wait_connect=False)
3614             ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED",
3615                                     "CTRL-EVENT-NETWORK-NOT-FOUND"], timeout=10)
3616             dev[0].request("DISCONNECT")
3617             if ev is None or "CTRL-EVENT-CONNECTED" in ev:
3618                 raise Exception("Unexpected connection result")
3619
3620             sigma_dut_cmd_check("ap_reset_default")
3621         finally:
3622             stop_sigma_dut(sigma)
3623             dev[0].set("sae_pwe", "0")
3624
3625 def test_sigma_dut_ap_sae_loop_only(dev, apdev, params):
3626     """sigma_dut controlled AP with SAE looping-only"""
3627     logdir = os.path.join(params['logdir'],
3628                           "sigma_dut_ap_sae_h2e.sigma-hostapd")
3629     if "SAE" not in dev[0].get_capability("auth_alg"):
3630         raise HwsimSkip("SAE not supported")
3631     with HWSimRadio() as (radio, iface):
3632         sigma = start_sigma_dut(iface, sae_h2e=True, hostapd_logdir=logdir)
3633         try:
3634             sigma_dut_cmd_check("ap_reset_default")
3635             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
3636             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK,12345678,sae_pwe,loop")
3637             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
3638
3639             dev[0].request("SET sae_groups ")
3640             dev[0].set("sae_pwe", "0")
3641             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
3642                            ieee80211w="2", scan_freq="2412")
3643             dev[0].request("REMOVE_NETWORK all")
3644             dev[0].wait_disconnected()
3645             dev[0].dump_monitor()
3646
3647             dev[0].set("sae_pwe", "1")
3648             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
3649                            ieee80211w="2", scan_freq="2412", wait_connect=False)
3650             ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED",
3651                                     "CTRL-EVENT-NETWORK-NOT-FOUND"], timeout=10)
3652             dev[0].request("DISCONNECT")
3653             if ev is None or "CTRL-EVENT-CONNECTED" in ev:
3654                 raise Exception("Unexpected connection result")
3655
3656             sigma_dut_cmd_check("ap_reset_default")
3657         finally:
3658             stop_sigma_dut(sigma)
3659             dev[0].set("sae_pwe", "0")
3660
3661 def test_sigma_dut_sae_h2e_loop_forcing(dev, apdev):
3662     """sigma_dut controlled SAE H2E misbehavior with looping forced"""
3663     if "SAE" not in dev[0].get_capability("auth_alg"):
3664         raise HwsimSkip("SAE not supported")
3665
3666     ssid = "test-sae"
3667     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
3668     params['wpa_key_mgmt'] = 'SAE'
3669     params["ieee80211w"] = "2"
3670     params['sae_pwe'] = '1'
3671     hapd = hostapd.add_ap(apdev[0], params)
3672
3673     ifname = dev[0].ifname
3674     sigma = start_sigma_dut(ifname)
3675     try:
3676         sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
3677         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
3678         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2,IgnoreH2E_RSNXE_BSSMemSel,1" % (ifname, "test-sae", "12345678"))
3679         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
3680                             timeout=10)
3681         ev = dev[0].wait_event(["SME: Trying to authenticate with"], timeout=10)
3682         if ev is None:
3683             raise Exception("No authentication attempt reported")
3684         ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=0.5)
3685         if ev is not None:
3686             raise Exception("Unexpected connection reported")
3687     finally:
3688         stop_sigma_dut(sigma)
3689
3690 def test_sigma_dut_sae_h2e_enabled_group_rejected(dev, apdev):
3691     """sigma_dut controlled SAE H2E misbehavior with rejected groups"""
3692     if "SAE" not in dev[0].get_capability("auth_alg"):
3693         raise HwsimSkip("SAE not supported")
3694
3695     ssid = "test-sae"
3696     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
3697     params['wpa_key_mgmt'] = 'SAE'
3698     params["ieee80211w"] = "2"
3699     params['sae_groups'] = "19 20"
3700     params['sae_pwe'] = '1'
3701     hapd = hostapd.add_ap(apdev[0], params)
3702
3703     ifname = dev[0].ifname
3704     sigma = start_sigma_dut(ifname, sae_h2e=True)
3705     try:
3706         sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
3707         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
3708         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2,ECGroupID_RGE,19 123" % (ifname, "test-sae", "12345678"))
3709         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
3710                             timeout=10)
3711         ev = dev[0].wait_event(["SME: Trying to authenticate with"], timeout=10)
3712         if ev is None:
3713             raise Exception("No authentication attempt reported")
3714         ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=0.5)
3715         if ev is not None:
3716             raise Exception("Unexpected connection reported")
3717     finally:
3718         stop_sigma_dut(sigma)
3719
3720 def test_sigma_dut_sae_h2e_rsnxe_mismatch(dev, apdev):
3721     """sigma_dut controlled SAE H2E misbehavior with RSNXE"""
3722     if "SAE" not in dev[0].get_capability("auth_alg"):
3723         raise HwsimSkip("SAE not supported")
3724
3725     ssid = "test-sae"
3726     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
3727     params['wpa_key_mgmt'] = 'SAE'
3728     params["ieee80211w"] = "2"
3729     params['sae_groups'] = "19"
3730     params['sae_pwe'] = '1'
3731     hapd = hostapd.add_ap(apdev[0], params)
3732
3733     ifname = dev[0].ifname
3734     sigma = start_sigma_dut(ifname, sae_h2e=True)
3735     try:
3736         sigma_dut_cmd_check("sta_reset_default,interface,%s" % ifname)
3737         sigma_dut_cmd_check("sta_set_ip_config,interface,%s,dhcp,0,ip,127.0.0.11,mask,255.255.255.0" % ifname)
3738         sigma_dut_cmd_check("sta_set_security,interface,%s,ssid,%s,passphrase,%s,type,SAE,encpType,aes-ccmp,keymgmttype,wpa2,RSNXE_Content,EapolM2:F40100" % (ifname, "test-sae", "12345678"))
3739         sigma_dut_cmd_check("sta_associate,interface,%s,ssid,%s,channel,1" % (ifname, "test-sae"),
3740                             timeout=10)
3741         ev = dev[0].wait_event(["SME: Trying to authenticate with"], timeout=10)
3742         if ev is None:
3743             raise Exception("No authentication attempt reported")
3744         ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=0.5)
3745         if ev is not None:
3746             raise Exception("Unexpected connection reported")
3747     finally:
3748         stop_sigma_dut(sigma)
3749         dev[0].set("sae_pwe", "0")
3750
3751 def test_sigma_dut_ap_sae_h2e_rsnxe_mismatch(dev, apdev, params):
3752     """sigma_dut controlled SAE H2E AP misbehavior with RSNXE"""
3753     logdir = os.path.join(params['logdir'],
3754                           "sigma_dut_ap_sae_h2e_rsnxe_mismatch.sigma-hostapd")
3755     if "SAE" not in dev[0].get_capability("auth_alg"):
3756         raise HwsimSkip("SAE not supported")
3757     with HWSimRadio() as (radio, iface):
3758         sigma = start_sigma_dut(iface, sae_h2e=True, hostapd_logdir=logdir)
3759         try:
3760             sigma_dut_cmd_check("ap_reset_default")
3761             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
3762             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK,12345678,sae_pwe,h2e,RSNXE_Content,EapolM3:F40100")
3763             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
3764
3765             dev[0].request("SET sae_groups ")
3766             dev[0].set("sae_pwe", "1")
3767             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
3768                            ieee80211w="2", scan_freq="2412", wait_connect=False)
3769             ev = dev[0].wait_event(["Associated with"], timeout=10)
3770             if ev is None:
3771                 raise Exception("No indication of association seen")
3772             ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED",
3773                                     "CTRL-EVENT-DISCONNECTED"], timeout=10)
3774             dev[0].request("DISCONNECT")
3775             if ev is None:
3776                 raise Exception("No disconnection seen")
3777             if "CTRL-EVENT-DISCONNECTED" not in ev:
3778                 raise Exception("Unexpected connection")
3779
3780             sigma_dut_cmd_check("ap_reset_default")
3781         finally:
3782             stop_sigma_dut(sigma)
3783             dev[0].set("sae_pwe", "0")
3784
3785 def test_sigma_dut_ap_sae_h2e_group_rejection(dev, apdev, params):
3786     """sigma_dut controlled AP with SAE H2E-only and group rejection"""
3787     logdir = os.path.join(params['logdir'],
3788                           "sigma_dut_ap_sae_h2e_group_rejection.sigma-hostapd")
3789     if "SAE" not in dev[0].get_capability("auth_alg"):
3790         raise HwsimSkip("SAE not supported")
3791     with HWSimRadio() as (radio, iface):
3792         sigma = start_sigma_dut(iface, sae_h2e=True, hostapd_logdir=logdir)
3793         try:
3794             sigma_dut_cmd_check("ap_reset_default")
3795             sigma_dut_cmd_check("ap_set_wireless,NAME,AP,CHANNEL,1,SSID,test-sae,MODE,11ng")
3796             sigma_dut_cmd_check("ap_set_security,NAME,AP,KEYMGNT,WPA2-SAE,PSK,12345678,sae_pwe,h2e")
3797             sigma_dut_cmd_check("ap_config_commit,NAME,AP")
3798
3799             dev[0].request("SET sae_groups 21 20 19")
3800             dev[0].set("sae_pwe", "1")
3801             dev[0].connect("test-sae", key_mgmt="SAE", psk="12345678",
3802                            ieee80211w="2", scan_freq="2412")
3803             addr = dev[0].own_addr()
3804             res = sigma_dut_cmd_check("dev_exec_action,program,WPA3,Dest_MAC,%s,Rejected_DH_Groups,1" % addr)
3805             if "DHGroupVerResult,21 20" not in res:
3806                 raise Exception("Unexpected dev_exec_action response: " + res)
3807
3808             sigma_dut_cmd_check("ap_reset_default")
3809         finally:
3810             stop_sigma_dut(sigma)
3811             dev[0].set("sae_pwe", "0")