tests/hwsim/test_p2p_autogo.py

   1 # P2P autonomous GO test cases
   2 # Copyright (c) 2013-2014, Jouni Malinen <j@w1.fi>
   3 #
   4 # This software may be distributed under the terms of the BSD license.
   5 # See README for more details.
   6
   7 import time
   8 import subprocess
   9 import logging
  10 logger = logging.getLogger()
  11
  12 import hwsim_utils
  13 import utils
  14 from wlantest import Wlantest
  15 from wpasupplicant import WpaSupplicant
  16
  17 def autogo(go, freq=None, persistent=None):
  18     logger.info("Start autonomous GO " + go.ifname)
  19     res = go.p2p_start_go(freq=freq, persistent=persistent)
  20     logger.debug("res: " + str(res))
  21     return res
  22
  23 def connect_cli(go, client, social=False, freq=None):
  24     logger.info("Try to connect the client to the GO")
  25     pin = client.wps_read_pin()
  26     go.p2p_go_authorize_client(pin)
  27     res = client.p2p_connect_group(go.p2p_dev_addr(), pin, timeout=60,
  28                                    social=social, freq=freq)
  29     logger.info("Client connected")
  30     hwsim_utils.test_connectivity_p2p(go, client)
  31     return res
  32
  33 def test_autogo(dev):
  34     """P2P autonomous GO and client joining group"""
  35     res = autogo(dev[0])
  36     if "p2p-wlan" in res['ifname']:
  37         raise Exception("Unexpected group interface name on GO")
  38     res = connect_cli(dev[0], dev[1])
  39     if "p2p-wlan" in res['ifname']:
  40         raise Exception("Unexpected group interface name on client")
  41     bss = dev[1].get_bss("p2p_dev_addr=" + dev[0].p2p_dev_addr())
  42     if bss['bssid'] != dev[0].p2p_interface_addr():
  43         raise Exception("Unexpected BSSID in the BSS entry for the GO")
  44     id = bss['id']
  45     bss = dev[1].get_bss("ID-" + id)
  46     if bss['id'] != id:
  47         raise Exception("Could not find BSS entry based on id")
  48     res = dev[1].request("BSS RANGE=" + id + "- MASK=0x1")
  49     if "id=" + id not in res:
  50         raise Exception("Could not find BSS entry based on id range")
  51
  52     res = dev[1].request("SCAN_RESULTS")
  53     if "[P2P]" not in res:
  54         raise Exception("P2P flag missing from scan results: " + res)
  55
  56     # Presence request to increase testing coverage
  57     if "FAIL" in dev[1].group_request("P2P_PRESENCE_REQ 30000 102400"):
  58         raise Exception("Could not send presence request")
  59     ev = dev[1].wait_event(["P2P-PRESENCE-RESPONSE"])
  60     if ev is None:
  61         raise Exception("Timeout while waiting for Presence Response")
  62
  63     dev[0].remove_group()
  64     dev[1].wait_go_ending_session()
  65
  66 def test_autogo2(dev):
  67     """P2P autonomous GO with a separate group interface and client joining group"""
  68     dev[0].request("SET p2p_no_group_iface 0")
  69     res = autogo(dev[0], freq=2437)
  70     if "p2p-wlan" not in res['ifname']:
  71         raise Exception("Unexpected group interface name on GO")
  72     if res['ifname'] not in utils.get_ifnames():
  73         raise Exception("Could not find group interface netdev")
  74     connect_cli(dev[0], dev[1], social=True, freq=2437)
  75     dev[0].remove_group()
  76     dev[1].wait_go_ending_session()
  77     if res['ifname'] in utils.get_ifnames():
  78         raise Exception("Group interface netdev was not removed")
  79
  80 def test_autogo3(dev):
  81     """P2P autonomous GO and client with a separate group interface joining group"""
  82     dev[1].request("SET p2p_no_group_iface 0")
  83     autogo(dev[0], freq=2462)
  84     res = connect_cli(dev[0], dev[1], social=True, freq=2462)
  85     if "p2p-wlan" not in res['ifname']:
  86         raise Exception("Unexpected group interface name on client")
  87     if res['ifname'] not in utils.get_ifnames():
  88         raise Exception("Could not find group interface netdev")
  89     dev[0].remove_group()
  90     dev[1].wait_go_ending_session()
  91     dev[1].ping()
  92     if res['ifname'] in utils.get_ifnames():
  93         raise Exception("Group interface netdev was not removed")
  94
  95 def test_autogo4(dev):
  96     """P2P autonomous GO and client joining group (both with a separate group interface)"""
  97     dev[0].request("SET p2p_no_group_iface 0")
  98     dev[1].request("SET p2p_no_group_iface 0")
  99     res1 = autogo(dev[0], freq=2412)
 100     res2 = connect_cli(dev[0], dev[1], social=True, freq=2412)
 101     if "p2p-wlan" not in res1['ifname']:
 102         raise Exception("Unexpected group interface name on GO")
 103     if "p2p-wlan" not in res2['ifname']:
 104         raise Exception("Unexpected group interface name on client")
 105     ifnames = utils.get_ifnames()
 106     if res1['ifname'] not in ifnames:
 107         raise Exception("Could not find GO group interface netdev")
 108     if res2['ifname'] not in ifnames:
 109         raise Exception("Could not find client group interface netdev")
 110     dev[0].remove_group()
 111     dev[1].wait_go_ending_session()
 112     dev[1].ping()
 113     ifnames = utils.get_ifnames()
 114     if res1['ifname'] in ifnames:
 115         raise Exception("GO group interface netdev was not removed")
 116     if res2['ifname'] in ifnames:
 117         raise Exception("Client group interface netdev was not removed")
 118
 119 def test_autogo_m2d(dev):
 120     """P2P autonomous GO and clients not authorized"""
 121     autogo(dev[0], freq=2412)
 122     go_addr = dev[0].p2p_dev_addr()
 123
 124     dev[1].request("SET p2p_no_group_iface 0")
 125     if not dev[1].discover_peer(go_addr, social=True):
 126         raise Exception("GO " + go_addr + " not found")
 127     dev[1].dump_monitor()
 128
 129     if not dev[2].discover_peer(go_addr, social=True):
 130         raise Exception("GO " + go_addr + " not found")
 131     dev[2].dump_monitor()
 132
 133     logger.info("Trying to join the group when GO has not authorized the client")
 134     pin = dev[1].wps_read_pin()
 135     cmd = "P2P_CONNECT " + go_addr + " " + pin + " join"
 136     if "OK" not in dev[1].global_request(cmd):
 137         raise Exception("P2P_CONNECT join failed")
 138
 139     pin = dev[2].wps_read_pin()
 140     cmd = "P2P_CONNECT " + go_addr + " " + pin + " join"
 141     if "OK" not in dev[2].global_request(cmd):
 142         raise Exception("P2P_CONNECT join failed")
 143
 144     ev = dev[1].wait_global_event(["WPS-M2D"], timeout=10)
 145     if ev is None:
 146         raise Exception("No global M2D event")
 147     ifaces = dev[1].request("INTERFACES").splitlines()
 148     iface = ifaces[0] if "p2p-wlan" in ifaces[0] else ifaces[1]
 149     wpas = WpaSupplicant(ifname=iface)
 150     ev = wpas.wait_event(["WPS-M2D"], timeout=10)
 151     if ev is None:
 152         raise Exception("No M2D event on group interface")
 153
 154     ev = dev[2].wait_global_event(["WPS-M2D"], timeout=10)
 155     if ev is None:
 156         raise Exception("No global M2D event (2)")
 157     ev = dev[2].wait_event(["WPS-M2D"], timeout=10)
 158     if ev is None:
 159         raise Exception("No M2D event on group interface (2)")
 160
 161 def test_autogo_fail(dev):
 162     """P2P autonomous GO and incorrect PIN"""
 163     autogo(dev[0], freq=2412)
 164     go_addr = dev[0].p2p_dev_addr()
 165     dev[0].p2p_go_authorize_client("00000000")
 166
 167     dev[1].request("SET p2p_no_group_iface 0")
 168     if not dev[1].discover_peer(go_addr, social=True):
 169         raise Exception("GO " + go_addr + " not found")
 170     dev[1].dump_monitor()
 171
 172     logger.info("Trying to join the group when GO has not authorized the client")
 173     pin = dev[1].wps_read_pin()
 174     cmd = "P2P_CONNECT " + go_addr + " " + pin + " join"
 175     if "OK" not in dev[1].global_request(cmd):
 176         raise Exception("P2P_CONNECT join failed")
 177
 178     ev = dev[1].wait_global_event(["WPS-FAIL"], timeout=10)
 179     if ev is None:
 180         raise Exception("No global WPS-FAIL event")
 181
 182 def test_autogo_2cli(dev):
 183     """P2P autonomous GO and two clients joining group"""
 184     autogo(dev[0], freq=2412)
 185     connect_cli(dev[0], dev[1], social=True, freq=2412)
 186     connect_cli(dev[0], dev[2], social=True, freq=2412)
 187     hwsim_utils.test_connectivity_p2p(dev[1], dev[2])
 188     dev[0].global_request("P2P_REMOVE_CLIENT " + dev[1].p2p_dev_addr())
 189     dev[1].wait_go_ending_session()
 190     dev[0].remove_group()
 191     dev[2].wait_go_ending_session()
 192
 193 def test_autogo_pbc(dev):
 194     """P2P autonomous GO and PBC"""
 195     dev[1].request("SET p2p_no_group_iface 0")
 196     autogo(dev[0], freq=2412)
 197     if "FAIL" not in dev[0].group_request("WPS_PBC p2p_dev_addr=00:11:22:33:44"):
 198         raise Exception("Invalid WPS_PBC succeeded")
 199     if "OK" not in dev[0].group_request("WPS_PBC p2p_dev_addr=" + dev[1].p2p_dev_addr()):
 200         raise Exception("WPS_PBC failed")
 201     dev[2].p2p_connect_group(dev[0].p2p_dev_addr(), "pbc", timeout=0,
 202                              social=True)
 203     ev = dev[2].wait_event(["WPS-M2D"], timeout=15)
 204     if ev is None:
 205         raise Exception("WPS-M2D not reported")
 206     if "config_error=12" not in ev:
 207         raise Exception("Unexpected config_error: " + ev)
 208     dev[1].p2p_connect_group(dev[0].p2p_dev_addr(), "pbc", timeout=15,
 209                              social=True)
 210
 211 def test_autogo_tdls(dev):
 212     """P2P autonomous GO and two clients using TDLS"""
 213     wt = Wlantest()
 214     go = dev[0]
 215     logger.info("Start autonomous GO with fixed parameters " + go.ifname)
 216     id = go.add_network()
 217     go.set_network_quoted(id, "ssid", "DIRECT-tdls")
 218     go.set_network_quoted(id, "psk", "12345678")
 219     go.set_network(id, "mode", "3")
 220     go.set_network(id, "disabled", "2")
 221     res = go.p2p_start_go(persistent=id, freq="2462")
 222     logger.debug("res: " + str(res))
 223     wt.flush()
 224     wt.add_passphrase("12345678")
 225     connect_cli(go, dev[1], social=True, freq=2462)
 226     connect_cli(go, dev[2], social=True, freq=2462)
 227     hwsim_utils.test_connectivity_p2p(dev[1], dev[2])
 228     bssid = dev[0].p2p_interface_addr()
 229     addr1 = dev[1].p2p_interface_addr()
 230     addr2 = dev[2].p2p_interface_addr()
 231     dev[1].tdls_setup(addr2)
 232     time.sleep(1)
 233     hwsim_utils.test_connectivity_p2p(dev[1], dev[2])
 234     conf = wt.get_tdls_counter("setup_conf_ok", bssid, addr1, addr2);
 235     if conf == 0:
 236         raise Exception("No TDLS Setup Confirm (success) seen")
 237     dl = wt.get_tdls_counter("valid_direct_link", bssid, addr1, addr2);
 238     if dl == 0:
 239         raise Exception("No valid frames through direct link")
 240     wt.tdls_clear(bssid, addr1, addr2);
 241     dev[1].tdls_teardown(addr2)
 242     time.sleep(1)
 243     teardown = wt.get_tdls_counter("teardown", bssid, addr1, addr2);
 244     if teardown == 0:
 245         raise Exception("No TDLS Setup Teardown seen")
 246     wt.tdls_clear(bssid, addr1, addr2);
 247     hwsim_utils.test_connectivity_p2p(dev[1], dev[2])
 248     ap_path = wt.get_tdls_counter("valid_ap_path", bssid, addr1, addr2);
 249     if ap_path == 0:
 250         raise Exception("No valid frames via AP path")
 251     direct_link = wt.get_tdls_counter("valid_direct_link", bssid, addr1, addr2);
 252     if direct_link > 0:
 253         raise Exception("Unexpected frames through direct link")
 254     idirect_link = wt.get_tdls_counter("invalid_direct_link", bssid, addr1,
 255                                        addr2);
 256     if idirect_link > 0:
 257         raise Exception("Unexpected frames through direct link (invalid)")
 258     dev[2].remove_group()
 259     dev[1].remove_group()
 260     dev[0].remove_group()
 261
 262 def test_autogo_legacy(dev):
 263     """P2P autonomous GO and legacy clients"""
 264     res = autogo(dev[0], freq=2462)
 265     if dev[0].get_group_status_field("passphrase", extra="WPS") != res['passphrase']:
 266         raise Exception("passphrase mismatch")
 267
 268     logger.info("Connect P2P client")
 269     connect_cli(dev[0], dev[1], social=True, freq=2462)
 270
 271     logger.info("Connect legacy WPS client")
 272     pin = dev[2].wps_read_pin()
 273     dev[0].p2p_go_authorize_client(pin)
 274     dev[2].request("P2P_SET disabled 1")
 275     dev[2].dump_monitor()
 276     dev[2].request("WPS_PIN any " + pin)
 277     ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
 278     if ev is None:
 279         raise Exception("Association with the GO timed out")
 280     status = dev[2].get_status()
 281     if status['wpa_state'] != 'COMPLETED':
 282         raise Exception("Not fully connected")
 283     hwsim_utils.test_connectivity_p2p_sta(dev[1], dev[2])
 284     dev[2].request("DISCONNECT")
 285
 286     logger.info("Connect legacy non-WPS client")
 287     dev[2].request("FLUSH")
 288     dev[2].request("P2P_SET disabled 1")
 289     dev[2].connect(ssid=res['ssid'], psk=res['passphrase'], proto='RSN',
 290                    key_mgmt='WPA-PSK', pairwise='CCMP', group='CCMP',
 291                    scan_freq=res['freq'])
 292     hwsim_utils.test_connectivity_p2p_sta(dev[1], dev[2])
 293     dev[2].request("DISCONNECT")
 294
 295     dev[0].remove_group()
 296     dev[1].wait_go_ending_session()
 297
 298 def test_autogo_chan_switch(dev):
 299     """P2P autonomous GO switching channels"""
 300     autogo(dev[0], freq=2417)
 301     connect_cli(dev[0], dev[1])
 302     res = dev[0].request("CHAN_SWITCH 5 2422")
 303     if "FAIL" in res:
 304         # for now, skip test since mac80211_hwsim support is not yet widely
 305         # deployed
 306         return 'skip'
 307     ev = dev[0].wait_event(["AP-CSA-FINISHED"], timeout=10)
 308     if ev is None:
 309         raise Exception("CSA finished event timed out")
 310     if "freq=2422" not in ev:
 311         raise Exception("Unexpected cahnnel in CSA finished event")
 312     dev[0].dump_monitor()
 313     dev[1].dump_monitor()
 314     time.sleep(0.1)
 315     hwsim_utils.test_connectivity_p2p(dev[0], dev[1])
 316
 317 def test_autogo_extra_cred(dev):
 318     """P2P autonomous GO sending two WPS credentials"""
 319     if "FAIL" in dev[0].request("SET wps_testing_dummy_cred 1"):
 320         raise Exception("Failed to enable test mode")
 321     autogo(dev[0], freq=2412)
 322     connect_cli(dev[0], dev[1], social=True, freq=2412)
 323     dev[0].remove_group()
 324     dev[1].wait_go_ending_session()
 325
 326 def test_autogo_ifdown(dev):
 327     """P2P autonomous GO and external ifdown"""
 328     wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
 329     wpas.interface_add("wlan5")
 330     res = autogo(wpas)
 331     wpas.dump_monitor()
 332     wpas.interface_remove("wlan5")
 333     wpas.interface_add("wlan5")
 334     res = autogo(wpas)
 335     wpas.dump_monitor()
 336     subprocess.call(['sudo', 'ifconfig', res['ifname'], 'down'])
 337     ev = wpas.wait_global_event(["P2P-GROUP-REMOVED"], timeout=10)
 338     if ev is None:
 339         raise Exception("Group removal not reported")
 340     if res['ifname'] not in ev:
 341         raise Exception("Unexpected group removal event: " + ev)
 342
 343 def test_autogo_start_during_scan(dev):
 344     """P2P autonomous GO started during ongoing manual scan"""
 345     try:
 346         # use autoscan to set scan_req = MANUAL_SCAN_REQ
 347         if "OK" not in dev[0].request("AUTOSCAN periodic:1"):
 348             raise Exception("Failed to set autoscan")
 349         autogo(dev[0], freq=2462)
 350         connect_cli(dev[0], dev[1], social=True, freq=2462)
 351         dev[0].remove_group()
 352         dev[1].wait_go_ending_session()
 353     finally:
 354         dev[0].request("AUTOSCAN ")
 355
 356 def test_autogo_passphrase_len(dev):
 357     """P2P autonomous GO and longer passphrase"""
 358     try:
 359         if "OK" not in dev[0].request("SET p2p_passphrase_len 13"):
 360             raise Exception("Failed to set passphrase length")
 361         res = autogo(dev[0], freq=2412)
 362         if len(res['passphrase']) != 13:
 363             raise Exception("Unexpected passphrase length")
 364         if dev[0].get_group_status_field("passphrase", extra="WPS") != res['passphrase']:
 365             raise Exception("passphrase mismatch")
 366
 367         logger.info("Connect P2P client")
 368         connect_cli(dev[0], dev[1], social=True, freq=2412)
 369
 370         logger.info("Connect legacy WPS client")
 371         pin = dev[2].wps_read_pin()
 372         dev[0].p2p_go_authorize_client(pin)
 373         dev[2].request("P2P_SET disabled 1")
 374         dev[2].dump_monitor()
 375         dev[2].request("WPS_PIN any " + pin)
 376         ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
 377         if ev is None:
 378             raise Exception("Association with the GO timed out")
 379         status = dev[2].get_status()
 380         if status['wpa_state'] != 'COMPLETED':
 381             raise Exception("Not fully connected")
 382         dev[2].request("DISCONNECT")
 383
 384         logger.info("Connect legacy non-WPS client")
 385         dev[2].request("FLUSH")
 386         dev[2].request("P2P_SET disabled 1")
 387         dev[2].connect(ssid=res['ssid'], psk=res['passphrase'], proto='RSN',
 388                        key_mgmt='WPA-PSK', pairwise='CCMP', group='CCMP',
 389                        scan_freq=res['freq'])
 390         hwsim_utils.test_connectivity_p2p_sta(dev[1], dev[2])
 391         dev[2].request("DISCONNECT")
 392
 393         dev[0].remove_group()
 394         dev[1].wait_go_ending_session()
 395     finally:
 396         dev[0].request("SET p2p_passphrase_len 8")
 397
 398 def test_autogo_bridge(dev):
 399     """P2P autonomous GO in a bridge"""
 400     try:
 401         # use autoscan to set scan_req = MANUAL_SCAN_REQ
 402         if "OK" not in dev[0].request("AUTOSCAN periodic:1"):
 403             raise Exception("Failed to set autoscan")
 404         autogo(dev[0])
 405         subprocess.call(['sudo', 'brctl', 'addbr', 'p2p-br0'])
 406         subprocess.call(['sudo', 'brctl', 'setfd', 'p2p-br0', '0'])
 407         subprocess.call(['sudo', 'brctl', 'addif', 'p2p-br0', dev[0].ifname])
 408         subprocess.call(['sudo', 'ip', 'link', 'set', 'dev', 'p2p-br0', 'up'])
 409         time.sleep(0.1)
 410         subprocess.call(['sudo', 'brctl', 'delif', 'p2p-br0', dev[0].ifname])
 411         time.sleep(0.1)
 412         subprocess.call(['sudo', 'ip', 'link', 'set', 'dev', 'p2p-br0', 'down'])
 413         time.sleep(0.1)
 414         subprocess.call(['sudo', 'brctl', 'delbr', 'p2p-br0'])
 415         ev = dev[0].wait_global_event(["P2P-GROUP-REMOVED"], timeout=1)
 416         if ev is not None:
 417             raise Exception("P2P group removed unexpectedly")
 418         if dev[0].get_status_field('wpa_state') != "COMPLETED":
 419             raise Exception("Unexpected wpa_state")
 420         dev[0].remove_group()
 421     finally:
 422         dev[0].request("AUTOSCAN ")
 423         subprocess.Popen(['sudo', 'brctl', 'delif', 'p2p-br0', dev[0].ifname],
 424                          stderr=open('/dev/null', 'w'))
 425         subprocess.Popen(['sudo', 'ip', 'link', 'set', 'dev', 'p2p-br0', 'down'],
 426                          stderr=open('/dev/null', 'w'))
 427         subprocess.Popen(['sudo', 'brctl', 'delbr', 'p2p-br0'],
 428                          stderr=open('/dev/null', 'w'))