+++ /dev/null
-# Python class for controlling wlantest
-# Copyright (c) 2013-2014, Jouni Malinen <j@w1.fi>
-#
-# This software may be distributed under the terms of the BSD license.
-# See README for more details.
-
-import re
-import os
-import posixpath
-import time
-import subprocess
-import logging
-import wpaspy
-
-logger = logging.getLogger()
-
-class Wlantest:
- remote_host = None
- setup_params = None
- exe_thread = None
- exe_res = []
- monitor_mod = None
- setup_done = False
-
- @classmethod
- def stop_remote_wlantest(cls):
- if cls.exe_thread is None:
- # Local flow - no need for remote operations
- return
-
- cls.remote_host.execute(["killall", "-9", "wlantest"])
- cls.remote_host.wait_execute_complete(cls.exe_thread, 5)
- cls.exe_thread = None
- cls.exe_res = []
-
- @classmethod
- def reset_remote_wlantest(cls):
- cls.stop_remote_wlantest()
- cls.remote_host = None
- cls.setup_params = None
- cls.exe_thread = None
- cls.exe_res = []
- cls.monitor_mod = None
- cls.setup_done = False
-
- @classmethod
- def start_remote_wlantest(cls):
- if cls.remote_host is None:
- # Local flow - no need for remote operations
- return
- if cls.exe_thread is not None:
- raise Exception("Cannot start wlantest twice")
-
- log_dir = cls.setup_params['log_dir']
- ifaces = re.split('; | |, ', cls.remote_host.ifname)
- ifname = ifaces[0]
- exe = cls.setup_params["wlantest"]
- tc_name = cls.setup_params["tc_name"]
- base_log_name = tc_name + "_wlantest_" + \
- cls.remote_host.name + "_" + ifname
- log_file = posixpath.join(log_dir, base_log_name + ".log")
- pcap_file = posixpath.join(log_dir, base_log_name + ".pcapng")
- cmd = "{} -i {} -n {} -c -dtN -L {}".format(exe, ifname,
- pcap_file, log_file)
- cls.remote_host.add_log(log_file)
- cls.remote_host.add_log(pcap_file)
- cls.exe_thread = cls.remote_host.execute_run(cmd.split(), cls.exe_res)
- # Give wlantest a chance to start working
- time.sleep(1)
-
- @classmethod
- def register_remote_wlantest(cls, host, setup_params, monitor_mod):
- if cls.remote_host is not None:
- raise Exception("Cannot register remote wlantest twice")
- cls.remote_host = host
- cls.setup_params = setup_params
- cls.monitor_mod = monitor_mod
- status, buf = host.execute(["which", setup_params['wlantest']])
- if status != 0:
- raise Exception(host.name + " - wlantest: " + buf)
- status, buf = host.execute(["which", setup_params['wlantest_cli']])
- if status != 0:
- raise Exception(host.name + " - wlantest_cli: " + buf)
-
- @classmethod
- def chan_from_wpa(cls, wpa, is_p2p=False):
- if cls.monitor_mod is None:
- return
- m = cls.monitor_mod
- return m.setup(cls.remote_host, [m.get_monitor_params(wpa, is_p2p)])
-
- @classmethod
- def setup(cls, wpa, is_p2p=False):
- cls.chan_from_wpa(wpa, is_p2p)
- cls.start_remote_wlantest()
- cls.setup_done = True
-
- def __init__(self):
- if not self.setup_done:
- raise Exception("Cannot create Wlantest instance before setup()")
- if os.path.isfile('../../wlantest/wlantest_cli'):
- self.wlantest_cli = '../../wlantest/wlantest_cli'
- else:
- self.wlantest_cli = 'wlantest_cli'
-
- def cli_cmd(self, params):
- if self.remote_host is not None:
- exe = self.setup_params["wlantest_cli"]
- ret = self.remote_host.execute([exe] + params)
- if ret[0] != 0:
- raise Exception("wlantest_cli failed")
- return ret[1]
- else:
- return subprocess.check_output([self.wlantest_cli] + params)
-
- def flush(self):
- res = self.cli_cmd(["flush"])
- if "FAIL" in res:
- raise Exception("wlantest_cli flush failed")
-
- def relog(self):
- res = self.cli_cmd(["relog"])
- if "FAIL" in res:
- raise Exception("wlantest_cli relog failed")
-
- def add_passphrase(self, passphrase):
- res = self.cli_cmd(["add_passphrase", passphrase])
- if "FAIL" in res:
- raise Exception("wlantest_cli add_passphrase failed")
-
- def add_wepkey(self, key):
- res = self.cli_cmd(["add_wepkey", key])
- if "FAIL" in res:
- raise Exception("wlantest_cli add_key failed")
-
- def info_bss(self, field, bssid):
- res = self.cli_cmd(["info_bss", field, bssid])
- if "FAIL" in res:
- raise Exception("Could not get BSS info from wlantest for " + bssid)
- return res
-
- def get_bss_counter(self, field, bssid):
- try:
- res = self.cli_cmd(["get_bss_counter", field, bssid])
- except Exception, e:
- return 0
- if "FAIL" in res:
- return 0
- return int(res)
-
- def clear_bss_counters(self, bssid):
- self.cli_cmd(["clear_bss_counters", bssid])
-
- def info_sta(self, field, bssid, addr):
- res = self.cli_cmd(["info_sta", field, bssid, addr])
- if "FAIL" in res:
- raise Exception("Could not get STA info from wlantest for " + addr)
- return res
-
- def get_sta_counter(self, field, bssid, addr):
- res = self.cli_cmd(["get_sta_counter", field, bssid, addr])
- if "FAIL" in res:
- raise Exception("wlantest_cli command failed")
- return int(res)
-
- def clear_sta_counters(self, bssid, addr):
- res = self.cli_cmd(["clear_sta_counters", bssid, addr])
- if "FAIL" in res:
- raise Exception("wlantest_cli command failed")
-
- def tdls_clear(self, bssid, addr1, addr2):
- self.cli_cmd(["clear_tdls_counters", bssid, addr1, addr2])
-
- def get_tdls_counter(self, field, bssid, addr1, addr2):
- res = self.cli_cmd(["get_tdls_counter", field, bssid, addr1, addr2])
- if "FAIL" in res:
- raise Exception("wlantest_cli command failed")
- return int(res)
-
- def require_ap_pmf_mandatory(self, bssid):
- res = self.info_bss("rsn_capab", bssid)
- if "MFPR" not in res:
- raise Exception("AP did not require PMF")
- if "MFPC" not in res:
- raise Exception("AP did not enable PMF")
- res = self.info_bss("key_mgmt", bssid)
- if "PSK-SHA256" not in res:
- raise Exception("AP did not enable SHA256-based AKM for PMF")
-
- def require_ap_pmf_optional(self, bssid):
- res = self.info_bss("rsn_capab", bssid)
- if "MFPR" in res:
- raise Exception("AP required PMF")
- if "MFPC" not in res:
- raise Exception("AP did not enable PMF")
-
- def require_ap_no_pmf(self, bssid):
- res = self.info_bss("rsn_capab", bssid)
- if "MFPR" in res:
- raise Exception("AP required PMF")
- if "MFPC" in res:
- raise Exception("AP enabled PMF")
-
- def require_sta_pmf_mandatory(self, bssid, addr):
- res = self.info_sta("rsn_capab", bssid, addr)
- if "MFPR" not in res:
- raise Exception("STA did not require PMF")
- if "MFPC" not in res:
- raise Exception("STA did not enable PMF")
-
- def require_sta_pmf(self, bssid, addr):
- res = self.info_sta("rsn_capab", bssid, addr)
- if "MFPC" not in res:
- raise Exception("STA did not enable PMF")
-
- def require_sta_no_pmf(self, bssid, addr):
- res = self.info_sta("rsn_capab", bssid, addr)
- if "MFPC" in res:
- raise Exception("STA enabled PMF")
-
- def require_sta_key_mgmt(self, bssid, addr, key_mgmt):
- res = self.info_sta("key_mgmt", bssid, addr)
- if key_mgmt not in res:
- raise Exception("Unexpected STA key_mgmt")
-
- def get_tx_tid(self, bssid, addr, tid):
- res = self.cli_cmd(["get_tx_tid", bssid, addr, str(tid)])
- if "FAIL" in res:
- raise Exception("wlantest_cli command failed")
- return int(res)
-
- def get_rx_tid(self, bssid, addr, tid):
- res = self.cli_cmd(["get_rx_tid", bssid, addr, str(tid)])
- if "FAIL" in res:
- raise Exception("wlantest_cli command failed")
- return int(res)
-
- def get_tid_counters(self, bssid, addr):
- tx = {}
- rx = {}
- for tid in range(0, 17):
- tx[tid] = self.get_tx_tid(bssid, addr, tid)
- rx[tid] = self.get_rx_tid(bssid, addr, tid)
- return [ tx, rx ]
projects / mech_eap.git / blobdiff