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