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