tests/hwsim/test_sigma_dut.py

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