DGNum/hostapd
6
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions
hostapd/tests/hwsim/test_p2p_autogo.py
Jouni Malinen 4b9d79b66e tests: Make it less likely to overflow wlan5 control iface socket 
Number of test cases did not read all control interface socket events
from the dynamically added wlan5 interface. This could result in hitting
maximum socket TX queue length and failures in the following test cases.

Signed-off-by: Jouni Malinen <j@w1.fi>
2015-10-04 01:42:42 +03:00

708 lines
28 KiB
Python
Raw Blame History

# P2P autonomous GO test cases
# Copyright (c) 2013-2015, Jouni Malinen <j@w1.fi>
#
# This software may be distributed under the terms of the BSD license.
# See README for more details.
import time
import subprocess
import logging
logger = logging.getLogger()
import hwsim_utils
import utils
from utils import HwsimSkip
from wlantest import Wlantest
from wpasupplicant import WpaSupplicant
def autogo(go, freq=None, persistent=None):
logger.info("Start autonomous GO " + go.ifname)
res = go.p2p_start_go(freq=freq, persistent=persistent)
logger.debug("res: " + str(res))
return res
def connect_cli(go, client, social=False, freq=None):
logger.info("Try to connect the client to the GO")
pin = client.wps_read_pin()
go.p2p_go_authorize_client(pin)
res = client.p2p_connect_group(go.p2p_dev_addr(), pin, timeout=60,
social=social, freq=freq)
logger.info("Client connected")
hwsim_utils.test_connectivity_p2p(go, client)
return res
def test_autogo(dev):
"""P2P autonomous GO and client joining group"""
addr0 = dev[0].p2p_dev_addr()
addr2 = dev[2].p2p_dev_addr()
res = autogo(dev[0])
if "p2p-wlan" in res['ifname']:
raise Exception("Unexpected group interface name on GO")
res = connect_cli(dev[0], dev[1])
if "p2p-wlan" in res['ifname']:
raise Exception("Unexpected group interface name on client")
bss = dev[1].get_bss("p2p_dev_addr=" + addr0, res['ifname'])
if not bss or bss['bssid'] != dev[0].p2p_interface_addr():
raise Exception("Unexpected BSSID in the BSS entry for the GO")
id = bss['id']
bss = dev[1].get_bss("ID-" + id, res['ifname'])
if not bss or bss['id'] != id:
raise Exception("Could not find BSS entry based on id")
res = dev[1].group_request("BSS RANGE=" + id + "- MASK=0x1")
if "id=" + id not in res:
raise Exception("Could not find BSS entry based on id range")
res = dev[1].request("SCAN_RESULTS")
if "[P2P]" not in res:
raise Exception("P2P flag missing from scan results: " + res)
# Presence request to increase testing coverage
if "FAIL" not in dev[1].group_request("P2P_PRESENCE_REQ 30000"):
raise Exception("Invald P2P_PRESENCE_REQ accepted")
if "FAIL" not in dev[1].group_request("P2P_PRESENCE_REQ 30000 102400 30001"):
raise Exception("Invald P2P_PRESENCE_REQ accepted")
if "FAIL" in dev[1].group_request("P2P_PRESENCE_REQ 30000 102400"):
raise Exception("Could not send presence request")
ev = dev[1].wait_group_event(["P2P-PRESENCE-RESPONSE"], 10)
if ev is None:
raise Exception("Timeout while waiting for Presence Response")
if "FAIL" in dev[1].group_request("P2P_PRESENCE_REQ 30000 102400 20000 102400"):
raise Exception("Could not send presence request")
ev = dev[1].wait_group_event(["P2P-PRESENCE-RESPONSE"])
if ev is None:
raise Exception("Timeout while waiting for Presence Response")
if "FAIL" in dev[1].group_request("P2P_PRESENCE_REQ"):
raise Exception("Could not send presence request")
ev = dev[1].wait_group_event(["P2P-PRESENCE-RESPONSE"])
if ev is None:
raise Exception("Timeout while waiting for Presence Response")
if not dev[2].discover_peer(addr0):
raise Exception("Could not discover GO")
dev[0].dump_monitor()
dev[2].global_request("P2P_PROV_DISC " + addr0 + " display join")
ev = dev[0].wait_global_event(["P2P-PROV-DISC-SHOW-PIN"], timeout=10)
if ev is None:
raise Exception("GO did not report P2P-PROV-DISC-SHOW-PIN")
if "p2p_dev_addr=" + addr2 not in ev:
raise Exception("Unexpected P2P Device Address in event: " + ev)
if "group=" + dev[0].group_ifname not in ev:
raise Exception("Unexpected group interface in event: " + ev)
ev = dev[2].wait_global_event(["P2P-PROV-DISC-ENTER-PIN"], timeout=10)
if ev is None:
raise Exception("P2P-PROV-DISC-ENTER-PIN not reported")
dev[0].remove_group()
dev[1].wait_go_ending_session()
def test_autogo2(dev):
"""P2P autonomous GO with a separate group interface and client joining group"""
dev[0].request("SET p2p_no_group_iface 0")
res = autogo(dev[0], freq=2437)
if "p2p-wlan" not in res['ifname']:
raise Exception("Unexpected group interface name on GO")
if res['ifname'] not in utils.get_ifnames():
raise Exception("Could not find group interface netdev")
connect_cli(dev[0], dev[1], social=True, freq=2437)
dev[0].remove_group()
dev[1].wait_go_ending_session()
if res['ifname'] in utils.get_ifnames():
raise Exception("Group interface netdev was not removed")
def test_autogo3(dev):
"""P2P autonomous GO and client with a separate group interface joining group"""
dev[1].request("SET p2p_no_group_iface 0")
autogo(dev[0], freq=2462)
res = connect_cli(dev[0], dev[1], social=True, freq=2462)
if "p2p-wlan" not in res['ifname']:
raise Exception("Unexpected group interface name on client")
if res['ifname'] not in utils.get_ifnames():
raise Exception("Could not find group interface netdev")
dev[0].remove_group()
dev[1].wait_go_ending_session()
dev[1].ping()
if res['ifname'] in utils.get_ifnames():
raise Exception("Group interface netdev was not removed")
def test_autogo4(dev):
"""P2P autonomous GO and client joining group (both with a separate group interface)"""
dev[0].request("SET p2p_no_group_iface 0")
dev[1].request("SET p2p_no_group_iface 0")
res1 = autogo(dev[0], freq=2412)
res2 = connect_cli(dev[0], dev[1], social=True, freq=2412)
if "p2p-wlan" not in res1['ifname']:
raise Exception("Unexpected group interface name on GO")
if "p2p-wlan" not in res2['ifname']:
raise Exception("Unexpected group interface name on client")
ifnames = utils.get_ifnames()
if res1['ifname'] not in ifnames:
raise Exception("Could not find GO group interface netdev")
if res2['ifname'] not in ifnames:
raise Exception("Could not find client group interface netdev")
dev[0].remove_group()
dev[1].wait_go_ending_session()
dev[1].ping()
ifnames = utils.get_ifnames()
if res1['ifname'] in ifnames:
raise Exception("GO group interface netdev was not removed")
if res2['ifname'] in ifnames:
raise Exception("Client group interface netdev was not removed")
def test_autogo_m2d(dev):
"""P2P autonomous GO and clients not authorized"""
autogo(dev[0], freq=2412)
go_addr = dev[0].p2p_dev_addr()
dev[1].request("SET p2p_no_group_iface 0")
if not dev[1].discover_peer(go_addr, social=True):
raise Exception("GO " + go_addr + " not found")
dev[1].dump_monitor()
if not dev[2].discover_peer(go_addr, social=True):
raise Exception("GO " + go_addr + " not found")
dev[2].dump_monitor()
logger.info("Trying to join the group when GO has not authorized the client")
pin = dev[1].wps_read_pin()
cmd = "P2P_CONNECT " + go_addr + " " + pin + " join"
if "OK" not in dev[1].global_request(cmd):
raise Exception("P2P_CONNECT join failed")
pin = dev[2].wps_read_pin()
cmd = "P2P_CONNECT " + go_addr + " " + pin + " join"
if "OK" not in dev[2].global_request(cmd):
raise Exception("P2P_CONNECT join failed")
ev = dev[1].wait_global_event(["WPS-M2D"], timeout=16)
if ev is None:
raise Exception("No global M2D event")
ifaces = dev[1].request("INTERFACES").splitlines()
iface = ifaces[0] if "p2p-wlan" in ifaces[0] else ifaces[1]
wpas = WpaSupplicant(ifname=iface)
ev = wpas.wait_event(["WPS-M2D"], timeout=10)
if ev is None:
raise Exception("No M2D event on group interface")
ev = dev[2].wait_global_event(["WPS-M2D"], timeout=10)
if ev is None:
raise Exception("No global M2D event (2)")
ev = dev[2].wait_event(["WPS-M2D"], timeout=10)
if ev is None:
raise Exception("No M2D event on group interface (2)")
def test_autogo_fail(dev):
"""P2P autonomous GO and incorrect PIN"""
autogo(dev[0], freq=2412)
go_addr = dev[0].p2p_dev_addr()
dev[0].p2p_go_authorize_client("00000000")
dev[1].request("SET p2p_no_group_iface 0")
if not dev[1].discover_peer(go_addr, social=True):
raise Exception("GO " + go_addr + " not found")
dev[1].dump_monitor()
logger.info("Trying to join the group when GO has not authorized the client")
pin = dev[1].wps_read_pin()
cmd = "P2P_CONNECT " + go_addr + " " + pin + " join"
if "OK" not in dev[1].global_request(cmd):
raise Exception("P2P_CONNECT join failed")
ev = dev[1].wait_global_event(["WPS-FAIL"], timeout=10)
if ev is None:
raise Exception("No global WPS-FAIL event")
def test_autogo_2cli(dev):
"""P2P autonomous GO and two clients joining group"""
autogo(dev[0], freq=2412)
connect_cli(dev[0], dev[1], social=True, freq=2412)
connect_cli(dev[0], dev[2], social=True, freq=2412)
hwsim_utils.test_connectivity_p2p(dev[1], dev[2])
dev[0].global_request("P2P_REMOVE_CLIENT " + dev[1].p2p_dev_addr())
dev[1].wait_go_ending_session()
dev[0].global_request("P2P_REMOVE_CLIENT iface=" + dev[2].p2p_interface_addr())
dev[2].wait_go_ending_session()
if "FAIL" not in dev[0].global_request("P2P_REMOVE_CLIENT foo"):
raise Exception("Invalid P2P_REMOVE_CLIENT command accepted")
dev[0].remove_group()
def test_autogo_pbc(dev):
"""P2P autonomous GO and PBC"""
dev[1].global_request("SET p2p_no_group_iface 0")
autogo(dev[0], freq=2412)
if "FAIL" not in dev[0].group_request("WPS_PBC p2p_dev_addr=00:11:22:33:44"):
raise Exception("Invalid WPS_PBC succeeded")
if "OK" not in dev[0].group_request("WPS_PBC p2p_dev_addr=" + dev[1].p2p_dev_addr()):
raise Exception("WPS_PBC failed")
dev[2].p2p_connect_group(dev[0].p2p_dev_addr(), "pbc", timeout=0,
social=True)
ev = dev[2].wait_global_event(["WPS-M2D"], timeout=15)
if ev is None:
raise Exception("WPS-M2D not reported")
if "config_error=12" not in ev:
raise Exception("Unexpected config_error: " + ev)
dev[1].p2p_connect_group(dev[0].p2p_dev_addr(), "pbc", timeout=15,
social=True)
def test_autogo_tdls(dev):
"""P2P autonomous GO and two clients using TDLS"""
wt = Wlantest()
go = dev[0]
logger.info("Start autonomous GO with fixed parameters " + go.ifname)
id = go.add_network()
go.set_network_quoted(id, "ssid", "DIRECT-tdls")
go.set_network_quoted(id, "psk", "12345678")
go.set_network(id, "mode", "3")
go.set_network(id, "disabled", "2")
res = go.p2p_start_go(persistent=id, freq="2462")
logger.debug("res: " + str(res))
wt.flush()
wt.add_passphrase("12345678")
connect_cli(go, dev[1], social=True, freq=2462)
connect_cli(go, dev[2], social=True, freq=2462)
hwsim_utils.test_connectivity_p2p(dev[1], dev[2])
bssid = dev[0].p2p_interface_addr()
addr1 = dev[1].p2p_interface_addr()
addr2 = dev[2].p2p_interface_addr()
dev[1].tdls_setup(addr2)
time.sleep(1)
hwsim_utils.test_connectivity_p2p(dev[1], dev[2])
conf = wt.get_tdls_counter("setup_conf_ok", bssid, addr1, addr2);
if conf == 0:
raise Exception("No TDLS Setup Confirm (success) seen")
dl = wt.get_tdls_counter("valid_direct_link", bssid, addr1, addr2);
if dl == 0:
raise Exception("No valid frames through direct link")
wt.tdls_clear(bssid, addr1, addr2);
dev[1].tdls_teardown(addr2)
time.sleep(1)
teardown = wt.get_tdls_counter("teardown", bssid, addr1, addr2);
if teardown == 0:
raise Exception("No TDLS Setup Teardown seen")
wt.tdls_clear(bssid, addr1, addr2);
hwsim_utils.test_connectivity_p2p(dev[1], dev[2])
ap_path = wt.get_tdls_counter("valid_ap_path", bssid, addr1, addr2);
if ap_path == 0:
raise Exception("No valid frames via AP path")
direct_link = wt.get_tdls_counter("valid_direct_link", bssid, addr1, addr2);
if direct_link > 0:
raise Exception("Unexpected frames through direct link")
idirect_link = wt.get_tdls_counter("invalid_direct_link", bssid, addr1,
addr2);
if idirect_link > 0:
raise Exception("Unexpected frames through direct link (invalid)")
dev[2].remove_group()
dev[1].remove_group()
dev[0].remove_group()
def test_autogo_legacy(dev):
"""P2P autonomous GO and legacy clients"""
res = autogo(dev[0], freq=2462)
if dev[0].get_group_status_field("passphrase", extra="WPS") != res['passphrase']:
raise Exception("passphrase mismatch")
if dev[0].group_request("P2P_GET_PASSPHRASE") != res['passphrase']:
raise Exception("passphrase mismatch(2)")
logger.info("Connect P2P client")
connect_cli(dev[0], dev[1], social=True, freq=2462)
if "FAIL" not in dev[1].request("P2P_GET_PASSPHRASE"):
raise Exception("P2P_GET_PASSPHRASE succeeded on P2P Client")
logger.info("Connect legacy WPS client")
pin = dev[2].wps_read_pin()
dev[0].p2p_go_authorize_client(pin)
dev[2].request("P2P_SET disabled 1")
dev[2].dump_monitor()
dev[2].request("WPS_PIN any " + pin)
dev[2].wait_connected(timeout=30)
status = dev[2].get_status()
if status['wpa_state'] != 'COMPLETED':
raise Exception("Not fully connected")
hwsim_utils.test_connectivity_p2p_sta(dev[1], dev[2])
dev[2].request("DISCONNECT")
logger.info("Connect legacy non-WPS client")
dev[2].request("FLUSH")
dev[2].request("P2P_SET disabled 1")
dev[2].connect(ssid=res['ssid'], psk=res['passphrase'], proto='RSN',
key_mgmt='WPA-PSK', pairwise='CCMP', group='CCMP',
scan_freq=res['freq'])
hwsim_utils.test_connectivity_p2p_sta(dev[1], dev[2])
dev[2].request("DISCONNECT")
dev[0].remove_group()
dev[1].wait_go_ending_session()
def test_autogo_chan_switch(dev):
"""P2P autonomous GO switching channels"""
autogo(dev[0], freq=2417)
connect_cli(dev[0], dev[1])
res = dev[0].request("CHAN_SWITCH 5 2422")
if "FAIL" in res:
# for now, skip test since mac80211_hwsim support is not yet widely
# deployed
raise HwsimSkip("Assume mac80211_hwsim did not support channel switching")
ev = dev[0].wait_event(["AP-CSA-FINISHED"], timeout=10)
if ev is None:
raise Exception("CSA finished event timed out")
if "freq=2422" not in ev:
raise Exception("Unexpected cahnnel in CSA finished event")
dev[0].dump_monitor()
dev[1].dump_monitor()
time.sleep(0.1)
hwsim_utils.test_connectivity_p2p(dev[0], dev[1])
def test_autogo_extra_cred(dev):
"""P2P autonomous GO sending two WPS credentials"""
if "FAIL" in dev[0].request("SET wps_testing_dummy_cred 1"):
raise Exception("Failed to enable test mode")
autogo(dev[0], freq=2412)
connect_cli(dev[0], dev[1], social=True, freq=2412)
dev[0].remove_group()
dev[1].wait_go_ending_session()
def test_autogo_ifdown(dev):
"""P2P autonomous GO and external ifdown"""
wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
wpas.interface_add("wlan5")
res = autogo(wpas)
wpas.dump_monitor()
wpas.interface_remove("wlan5")
wpas.interface_add("wlan5")
res = autogo(wpas)
wpas.dump_monitor()
subprocess.call(['ifconfig', res['ifname'], 'down'])
ev = wpas.wait_global_event(["P2P-GROUP-REMOVED"], timeout=10)
if ev is None:
raise Exception("Group removal not reported")
if res['ifname'] not in ev:
raise Exception("Unexpected group removal event: " + ev)
def test_autogo_start_during_scan(dev):
"""P2P autonomous GO started during ongoing manual scan"""
try:
# use autoscan to set scan_req = MANUAL_SCAN_REQ
if "OK" not in dev[0].request("AUTOSCAN periodic:1"):
raise Exception("Failed to set autoscan")
autogo(dev[0], freq=2462)
connect_cli(dev[0], dev[1], social=True, freq=2462)
dev[0].remove_group()
dev[1].wait_go_ending_session()
finally:
dev[0].request("AUTOSCAN ")
def test_autogo_passphrase_len(dev):
"""P2P autonomous GO and longer passphrase"""
try:
if "OK" not in dev[0].request("SET p2p_passphrase_len 13"):
raise Exception("Failed to set passphrase length")
res = autogo(dev[0], freq=2412)
if len(res['passphrase']) != 13:
raise Exception("Unexpected passphrase length")
if dev[0].get_group_status_field("passphrase", extra="WPS") != res['passphrase']:
raise Exception("passphrase mismatch")
logger.info("Connect P2P client")
connect_cli(dev[0], dev[1], social=True, freq=2412)
logger.info("Connect legacy WPS client")
pin = dev[2].wps_read_pin()
dev[0].p2p_go_authorize_client(pin)
dev[2].request("P2P_SET disabled 1")
dev[2].dump_monitor()
dev[2].request("WPS_PIN any " + pin)
dev[2].wait_connected(timeout=30)
status = dev[2].get_status()
if status['wpa_state'] != 'COMPLETED':
raise Exception("Not fully connected")
dev[2].request("DISCONNECT")
logger.info("Connect legacy non-WPS client")
dev[2].request("FLUSH")
dev[2].request("P2P_SET disabled 1")
dev[2].connect(ssid=res['ssid'], psk=res['passphrase'], proto='RSN',
key_mgmt='WPA-PSK', pairwise='CCMP', group='CCMP',
scan_freq=res['freq'])
hwsim_utils.test_connectivity_p2p_sta(dev[1], dev[2])
dev[2].request("DISCONNECT")
dev[0].remove_group()
dev[1].wait_go_ending_session()
finally:
dev[0].request("SET p2p_passphrase_len 8")
def test_autogo_bridge(dev):
"""P2P autonomous GO in a bridge"""
try:
# use autoscan to set scan_req = MANUAL_SCAN_REQ
if "OK" not in dev[0].request("AUTOSCAN periodic:1"):
raise Exception("Failed to set autoscan")
autogo(dev[0])
ifname = dev[0].get_group_ifname()
subprocess.call(['brctl', 'addbr', 'p2p-br0'])
subprocess.call(['brctl', 'setfd', 'p2p-br0', '0'])
subprocess.call(['brctl', 'addif', 'p2p-br0', ifname])
subprocess.call(['ip', 'link', 'set', 'dev', 'p2p-br0', 'up'])
time.sleep(0.1)
subprocess.call(['brctl', 'delif', 'p2p-br0', ifname])
time.sleep(0.1)
subprocess.call(['ip', 'link', 'set', 'dev', 'p2p-br0', 'down'])
time.sleep(0.1)
subprocess.call(['brctl', 'delbr', 'p2p-br0'])
ev = dev[0].wait_global_event(["P2P-GROUP-REMOVED"], timeout=1)
if ev is not None:
raise Exception("P2P group removed unexpectedly")
if dev[0].get_group_status_field('wpa_state') != "COMPLETED":
raise Exception("Unexpected wpa_state")
dev[0].remove_group()
finally:
dev[0].request("AUTOSCAN ")
subprocess.Popen(['brctl', 'delif', 'p2p-br0', ifname],
stderr=open('/dev/null', 'w'))
subprocess.Popen(['ip', 'link', 'set', 'dev', 'p2p-br0', 'down'],
stderr=open('/dev/null', 'w'))
subprocess.Popen(['brctl', 'delbr', 'p2p-br0'],
stderr=open('/dev/null', 'w'))
def test_presence_req_on_group_interface(dev):
"""P2P_PRESENCE_REQ on group interface"""
dev[1].request("SET p2p_no_group_iface 0")
res = autogo(dev[0], freq=2437)
res = connect_cli(dev[0], dev[1], social=True, freq=2437)
if "FAIL" in dev[1].group_request("P2P_PRESENCE_REQ 30000 102400"):
raise Exception("Could not send presence request")
ev = dev[1].wait_group_event(["P2P-PRESENCE-RESPONSE"])
if ev is None:
raise Exception("Timeout while waiting for Presence Response")
dev[0].remove_group()
dev[1].wait_go_ending_session()
def test_autogo_join_auto_go_not_found(dev):
"""P2P_CONNECT-auto not finding GO"""
wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
wpas.interface_add("wlan5")
wpas.request("P2P_SET listen_channel 1")
wpas.global_request("SET p2p_no_group_iface 0")
autogo(wpas, freq=2412)
addr = wpas.p2p_dev_addr()
bssid = wpas.p2p_interface_addr()
wpas.dump_monitor()
dev[1].global_request("SET p2p_no_group_iface 0")
dev[1].scan_for_bss(bssid, freq=2412)
# This makes the GO not show up in the scan iteration following the
# P2P_CONNECT command by stopping beaconing and handling Probe Request
# frames externally (but not really replying to them). P2P listen mode is
# needed to keep the GO listening on the operating channel for the PD
# exchange.
if "OK" not in wpas.group_request("STOP_AP"):
raise Exception("STOP_AP failed")
wpas.dump_monitor()
wpas.group_request("SET ext_mgmt_frame_handling 1")
wpas.p2p_listen()
wpas.dump_monitor()
time.sleep(0.02)
dev[1].global_request("P2P_CONNECT " + addr + " pbc auto")
ev = dev[1].wait_global_event(["P2P-FALLBACK-TO-GO-NEG-ENABLED"], 15)
wpas.dump_monitor()
if ev is None:
raise Exception("Could not trigger old-scan-only case")
return
ev = dev[1].wait_global_event(["P2P-FALLBACK-TO-GO-NEG"], 15)
wpas.remove_group()
if ev is None:
raise Exception("Fallback to GO Negotiation not seen")
if "reason=GO-not-found" not in ev:
raise Exception("Unexpected reason for fallback: " + ev)
wpas.dump_monitor()
def test_autogo_join_auto(dev):
"""P2P_CONNECT-auto joining a group"""
autogo(dev[0])
addr = dev[0].p2p_dev_addr()
if "OK" not in dev[1].global_request("P2P_CONNECT " + addr + " pbc auto"):
raise Exception("P2P_CONNECT failed")
ev = dev[0].wait_global_event(["P2P-PROV-DISC-PBC-REQ"], timeout=15)
if ev is None:
raise Exception("Timeout on P2P-PROV-DISC-PBC-REQ")
if "group=" + dev[0].group_ifname not in ev:
raise Exception("Unexpected PD event contents: " + ev)
dev[0].group_request("WPS_PBC")
ev = dev[1].wait_global_event(["P2P-GROUP-STARTED"], timeout=15)
if ev is None:
raise Exception("Joining the group timed out")
dev[1].group_form_result(ev)
dev[0].remove_group()
dev[1].wait_go_ending_session()
dev[1].flush_scan_cache()
def test_autogo_join_auto_go_neg(dev):
"""P2P_CONNECT-auto fallback to GO Neg"""
dev[1].flush_scan_cache()
dev[0].p2p_listen()
addr = dev[0].p2p_dev_addr()
if "OK" not in dev[1].global_request("P2P_CONNECT " + addr + " pbc auto"):
raise Exception("P2P_CONNECT failed")
ev = dev[0].wait_global_event(["P2P-GO-NEG-REQUEST"], timeout=15)
if ev is None:
raise Exception("Timeout on P2P-GO-NEG-REQUEST")
peer = ev.split(' ')[1]
dev[0].p2p_go_neg_init(peer, None, "pbc", timeout=15, go_intent=15)
ev = dev[1].wait_global_event(["P2P-FALLBACK-TO-GO-NEG"], timeout=1)
if ev is None:
raise Exception("No P2P-FALLBACK-TO-GO-NEG event seen")
if "P2P-FALLBACK-TO-GO-NEG-ENABLED" in ev:
ev = dev[1].wait_global_event(["P2P-FALLBACK-TO-GO-NEG"], timeout=1)
if ev is None:
raise Exception("No P2P-FALLBACK-TO-GO-NEG event seen")
if "reason=peer-not-running-GO" not in ev:
raise Exception("Unexpected reason: " + ev)
ev = dev[1].wait_global_event(["P2P-GROUP-STARTED"], timeout=15)
if ev is None:
raise Exception("Joining the group timed out")
dev[1].group_form_result(ev)
dev[0].remove_group()
dev[1].wait_go_ending_session()
dev[1].flush_scan_cache()
def test_autogo_join_auto_go_neg_after_seeing_go(dev):
"""P2P_CONNECT-auto fallback to GO Neg after seeing GO"""
autogo(dev[0], freq=2412)
addr = dev[0].p2p_dev_addr()
bssid = dev[0].p2p_interface_addr()
dev[1].scan_for_bss(bssid, freq=2412)
dev[0].remove_group()
dev[0].p2p_listen()
if "OK" not in dev[1].global_request("P2P_CONNECT " + addr + " pbc auto"):
raise Exception("P2P_CONNECT failed")
ev = dev[1].wait_global_event(["P2P-FALLBACK-TO-GO-NEG-ENABLED"],
timeout=15)
if ev is None:
raise Exception("No P2P-FALLBACK-TO-GO-NEG-ENABLED event seen")
ev = dev[0].wait_global_event(["P2P-GO-NEG-REQUEST"], timeout=15)
if ev is None:
raise Exception("Timeout on P2P-GO-NEG-REQUEST")
peer = ev.split(' ')[1]
dev[0].p2p_go_neg_init(peer, None, "pbc", timeout=15, go_intent=15)
ev = dev[1].wait_global_event(["P2P-FALLBACK-TO-GO-NEG"], timeout=1)
if ev is None:
raise Exception("No P2P-FALLBACK-TO-GO-NEG event seen")
if "reason=no-ACK-to-PD-Req" not in ev and "reason=PD-failed" not in ev:
raise Exception("Unexpected reason: " + ev)
ev = dev[1].wait_global_event(["P2P-GROUP-STARTED"], timeout=15)
if ev is None:
raise Exception("Joining the group timed out")
dev[1].group_form_result(ev)
dev[0].remove_group()
dev[1].wait_go_ending_session()
dev[1].flush_scan_cache()
def test_go_search_non_social(dev):
"""P2P_FIND with freq parameter to scan a single channel"""
addr0 = dev[0].p2p_dev_addr()
autogo(dev[0], freq=2422)
dev[1].p2p_find(freq=2422)
ev = dev[1].wait_global_event(["P2P-DEVICE-FOUND"], timeout=3.5)
if ev is None:
raise Exception("Did not find GO quickly enough")
dev[2].p2p_listen()
ev = dev[1].wait_global_event(["P2P-DEVICE-FOUND"], timeout=5)
if ev is None:
raise Exception("Did not find peer")
dev[2].p2p_stop_find()
dev[1].p2p_stop_find()
dev[0].remove_group()
def test_autogo_many(dev):
"""P2P autonomous GO with large number of GO instances"""
dev[0].request("SET p2p_no_group_iface 0")
for i in range(100):
if "OK" not in dev[0].global_request("P2P_GROUP_ADD freq=2412"):
logger.info("Was able to add %d groups" % i)
if i < 5:
raise Exception("P2P_GROUP_ADD failed")
stop_ev = dev[0].wait_global_event(["P2P-GROUP-REMOVE"], timeout=1)
if stop_ev is not None:
raise Exception("Unexpected P2P-GROUP-REMOVE event")
break
ev = dev[0].wait_global_event(["P2P-GROUP-STARTED"], timeout=5)
if ev is None:
raise Exception("GO start up timed out")
dev[0].group_form_result(ev)
for i in dev[0].global_request("INTERFACES").splitlines():
dev[0].request("P2P_GROUP_REMOVE " + i)
dev[0].dump_monitor()
dev[0].request("P2P_GROUP_REMOVE *")
def test_autogo_many_clients(dev):
"""P2P autonomous GO and many clients (P2P IE fragmentation)"""
try:
_test_autogo_many_clients(dev)
finally:
dev[0].global_request("SET device_name Device A")
dev[1].global_request("SET device_name Device B")
dev[2].global_request("SET device_name Device C")
def _test_autogo_many_clients(dev):
# These long device names will push the P2P IE contents beyond the limit
# that requires fragmentation.
name0 = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
name1 = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"
name2 = "CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC"
name3 = "DDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD"
dev[0].global_request("SET device_name " + name0)
dev[1].global_request("SET device_name " + name1)
dev[2].global_request("SET device_name " + name2)
addr0 = dev[0].p2p_dev_addr()
res = autogo(dev[0], freq=2412)
bssid = dev[0].p2p_interface_addr()
connect_cli(dev[0], dev[1], social=True, freq=2412)
dev[0].dump_monitor()
connect_cli(dev[0], dev[2], social=True, freq=2412)
dev[0].dump_monitor()
wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
wpas.interface_add("wlan5")
wpas.global_request("SET device_name " + name3)
wpas.global_request("SET sec_device_type 1-11111111-1")
wpas.global_request("SET sec_device_type 2-22222222-2")
wpas.global_request("SET sec_device_type 3-33333333-3")
wpas.global_request("SET sec_device_type 4-44444444-4")
wpas.global_request("SET sec_device_type 5-55555555-5")
connect_cli(dev[0], wpas, social=True, freq=2412)
dev[0].dump_monitor()
dev[1].dump_monitor()
dev[1].p2p_find(freq=2412)
ev1 = dev[1].wait_global_event(["P2P-DEVICE-FOUND"], timeout=10)
if ev1 is None:
raise Exception("Could not find peer (1)")
ev2 = dev[1].wait_global_event(["P2P-DEVICE-FOUND"], timeout=10)
if ev2 is None:
raise Exception("Could not find peer (2)")
ev3 = dev[1].wait_global_event(["P2P-DEVICE-FOUND"], timeout=10)
if ev3 is None:
raise Exception("Could not find peer (3)")
dev[1].p2p_stop_find()
for i in [ name0, name2, name3 ]:
if i not in ev1 and i not in ev2 and i not in ev3:
raise Exception('name "%s" not found' % i)
Reference in a new issue View git blame Copy permalink