import struct
import wpaspy
import remotehost
+import subprocess
logger = logging.getLogger()
wpas_ctrl = '/var/run/wpa_supplicant'
self.group_ifname = None
self.gctrl_mon = None
self.host = remotehost.Host(hostname, ifname)
+ self._group_dbg = None
if ifname:
self.set_ifname(ifname, hostname, port)
+ res = self.get_driver_status()
+ if 'capa.flags' in res and int(res['capa.flags'], 0) & 0x20000000:
+ self.p2p_dev_ifname = 'p2p-dev-' + self.ifname
+ else:
+ self.p2p_dev_ifname = ifname
else:
self.ifname = None
else:
self.global_mon = None
+ def cmd_execute(self, cmd_array, shell=False):
+ if self.hostname is None:
+ if shell:
+ cmd = ' '.join(cmd_array)
+ else:
+ cmd = cmd_array
+ proc = subprocess.Popen(cmd, stderr=subprocess.STDOUT,
+ stdout=subprocess.PIPE, shell=shell)
+ out = proc.communicate()[0]
+ ret = proc.returncode
+ return ret, out
+ else:
+ return self.host.execute(cmd_array)
+
def terminate(self):
if self.global_mon:
self.global_mon.detach()
def interface_add(self, ifname, config="", driver="nl80211",
drv_params=None, br_ifname=None, create=False,
set_ifname=True, all_params=False, if_type=None):
- status, groups = self.host.execute("id")
+ status, groups = self.host.execute(["id"])
if status != 0:
group = "admin"
group = "admin" if "(admin)" in groups else "adm"
if not create and set_ifname:
port = self.get_ctrl_iface_port(ifname)
self.set_ifname(ifname, self.hostname, port)
+ res = self.get_driver_status()
+ if 'capa.flags' in res and int(res['capa.flags'], 0) & 0x20000000:
+ self.p2p_dev_ifname = 'p2p-dev-' + self.ifname
+ else:
+ self.p2p_dev_ifname = ifname
def interface_remove(self, ifname):
self.remove_ifname()
logger.debug(self.global_dbg + ifname + ": CTRL(global): " + cmd)
return self.global_ctrl.request(cmd)
+ @property
+ def group_dbg(self):
+ if self._group_dbg is not None:
+ return self._group_dbg
+ if self.group_ifname is None:
+ raise Exception("Cannot have group_dbg without group_ifname")
+ if self.hostname is None:
+ self._group_dbg = self.group_ifname
+ else:
+ self._group_dbg = self.hostname + "/" + self.group_ifname
+ return self._group_dbg
+
def group_request(self, cmd):
if self.group_ifname and self.group_ifname != self.ifname:
- logger.debug(self.group_ifname + ": CTRL: " + cmd)
- gctrl = wpaspy.Ctrl(os.path.join(wpas_ctrl, self.group_ifname))
+ if self.hostname is None:
+ gctrl = wpaspy.Ctrl(os.path.join(wpas_ctrl, self.group_ifname))
+ else:
+ port = self.get_ctrl_iface_port(self.group_ifname)
+ gctrl = wpaspy.Ctrl(self.hostname, port)
+ logger.debug(self.group_dbg + ": CTRL(group): " + cmd)
return gctrl.request(cmd)
return self.request(cmd)
if iter == 60:
logger.error(self.ifname + ": Driver scan state did not clear")
print "Trying to clear cfg80211/mac80211 scan state"
- status, buf = self.host.execute("ifconfig " + self.ifname + " down")
+ status, buf = self.host.execute(["ifconfig", self.ifname, "down"])
if status != 0:
logger.info("ifconfig failed: " + buf)
logger.info(status)
- status, buf = self.host.execute("ifconfig " + self.ifname + " up")
+ status, buf = self.host.execute(["ifconfig", self.ifname, "up"])
if status != 0:
logger.info("ifconfig failed: " + buf)
logger.info(status)
raise Exception("SET_NETWORK failed")
return None
+ def p2pdev_request(self, cmd):
+ return self.global_request("IFNAME=" + self.p2p_dev_ifname + " " + cmd)
+
+ def p2pdev_add_network(self):
+ id = self.p2pdev_request("ADD_NETWORK")
+ if "FAIL" in id:
+ raise Exception("p2pdev ADD_NETWORK failed")
+ return int(id)
+
+ def p2pdev_set_network(self, id, field, value):
+ res = self.p2pdev_request("SET_NETWORK " + str(id) + " " + field + " " + value)
+ if "FAIL" in res:
+ raise Exception("p2pdev SET_NETWORK failed")
+ return None
+
+ def p2pdev_set_network_quoted(self, id, field, value):
+ res = self.p2pdev_request("SET_NETWORK " + str(id) + " " + field + ' "' + value + '"')
+ if "FAIL" in res:
+ raise Exception("p2pdev SET_NETWORK failed")
+ return None
+
def list_networks(self, p2p=False):
if p2p:
res = self.global_request("LIST_NETWORKS")
quoted = [ "realm", "username", "password", "domain", "imsi",
"excluded_ssid", "milenage", "ca_cert", "client_cert",
"private_key", "domain_suffix_match", "provisioning_sp",
- "roaming_partner", "phase1", "phase2" ]
+ "roaming_partner", "phase1", "phase2", "private_key_passwd" ]
for field in quoted:
if field in params:
self.set_cred_quoted(id, field, params[field])
if field in params:
self.set_cred(id, field, params[field])
- return id;
+ return id
def select_network(self, id, freq=None):
if freq:
raise Exception("MESH_GROUP_REMOVE failed")
return None
- def connect_network(self, id, timeout=10):
+ def connect_network(self, id, timeout=None):
+ if timeout is None:
+ timeout = 10 if self.hostname is None else 60
self.dump_monitor()
self.select_network(id)
self.wait_connected(timeout=timeout)
return None
def get_mcc(self):
- mcc = int(self.get_driver_status_field('capa.num_multichan_concurrent'))
- return 1 if mcc < 2 else mcc
+ mcc = int(self.get_driver_status_field('capa.num_multichan_concurrent'))
+ return 1 if mcc < 2 else mcc
def get_mib(self):
res = self.request("MIB")
res['ifname'] = s[2]
self.group_ifname = s[2]
try:
- self.gctrl_mon = wpaspy.Ctrl(os.path.join(wpas_ctrl, self.group_ifname))
+ if self.hostname is None:
+ self.gctrl_mon = wpaspy.Ctrl(os.path.join(wpas_ctrl,
+ self.group_ifname))
+ else:
+ port = self.get_ctrl_iface_port(self.group_ifname)
+ self.gctrl_mon = wpaspy.Ctrl(self.hostname, port)
self.gctrl_mon.attach()
except:
logger.debug("Could not open monitor socket for group interface")
def p2p_go_neg_auth_result(self, timeout=1, expect_failure=False):
go_neg_res = None
ev = self.wait_global_event(["P2P-GO-NEG-SUCCESS",
- "P2P-GO-NEG-FAILURE"], timeout);
+ "P2P-GO-NEG-FAILURE"], timeout)
if ev is None:
if expect_failure:
return None
raise Exception("Group formation timed out")
if "P2P-GO-NEG-SUCCESS" in ev:
go_neg_res = ev
- ev = self.wait_global_event(["P2P-GROUP-STARTED"], timeout);
+ ev = self.wait_global_event(["P2P-GROUP-STARTED"], timeout)
if ev is None:
if expect_failure:
return None
cmd = "P2P_CONNECT " + peer + " " + pin + " " + method
else:
cmd = "P2P_CONNECT " + peer + " " + method
- if go_intent:
+ if go_intent is not None:
cmd = cmd + ' go_intent=' + str(go_intent)
if freq:
cmd = cmd + ' freq=' + str(freq)
while True:
while self.gctrl_mon.pending():
ev = self.gctrl_mon.recv()
- logger.debug(self.group_ifname + ": " + ev)
+ logger.debug(self.group_dbg + "(group): " + ev)
for event in events:
if event in ev:
return ev
except:
pass
self.gctrl_mon = None
- ev = self.wait_global_event(["P2P-GROUP-REMOVED"], timeout=3)
+ timeout = 3 if self.hostname is None else 10
+ ev = self.wait_global_event(["P2P-GROUP-REMOVED"], timeout=timeout)
if ev is None:
raise Exception("Group removal event timed out")
if "reason=GO_ENDING_SESSION" not in ev:
return res.split(' ')
def get_bss(self, bssid, ifname=None):
- if not ifname or ifname == self.ifname:
+ if not ifname or ifname == self.ifname:
res = self.request("BSS " + bssid)
elif ifname == self.group_ifname:
res = self.group_request("BSS " + bssid)
if field != "freq":
raise Exception("Unexpected MGMT-RX event format: " + ev)
msg['freq'] = val
+
+ field,val = items[2].split('=')
+ if field != "datarate":
+ raise Exception("Unexpected MGMT-RX event format: " + ev)
+ msg['datarate'] = val
+
+ field,val = items[3].split('=')
+ if field != "ssi_signal":
+ raise Exception("Unexpected MGMT-RX event format: " + ev)
+ msg['ssi_signal'] = val
+
frame = binascii.unhexlify(items[4])
msg['frame'] = frame
raise Exception(error)
return ev
- def wait_disconnected(self, timeout=10, error="Disconnection timed out"):
+ def wait_disconnected(self, timeout=None, error="Disconnection timed out"):
+ if timeout is None:
+ timeout = 10 if self.hostname is None else 30
ev = self.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=timeout)
if ev is None:
raise Exception(error)
cmd = "P2P_ASP_PROVISION_RESP"
params = "status=%d" % status
- if role is not None:
- params += " role=" + role
+ if role is not None:
+ params += " role=" + role
if cpt is not None:
params += " cpt=" + cpt
projects / mech_eap.git / blobdiff