libeap/tests/hwsim/test_fst_config.py

   1 # FST configuration tests
   2 # Copyright (c) 2015, Qualcomm Atheros, Inc.
   3 #
   4 # This software may be distributed under the terms of the BSD license.
   5 # See README for more details.
   6
   7 import logging
   8 logger = logging.getLogger()
   9 import subprocess
  10 import time
  11 import os
  12 import signal
  13 import hostapd
  14 import wpasupplicant
  15 import utils
  16
  17 import fst_test_common
  18
  19 class FstLauncherConfig:
  20     """FstLauncherConfig class represents configuration to be used for
  21     FST config tests related hostapd/wpa_supplicant instances"""
  22     def __init__(self, iface, fst_group, fst_pri, fst_llt=None):
  23         self.iface = iface
  24         self.fst_group = fst_group
  25         self.fst_pri = fst_pri
  26         self.fst_llt = fst_llt # None llt means no llt parameter will be set
  27
  28     def ifname(self):
  29         return self.iface
  30
  31     def is_ap(self):
  32         """Returns True if the configuration is for AP, otherwise - False"""
  33         raise Exception("Virtual is_ap() called!")
  34
  35     def to_file(self, pathname):
  36         """Creates configuration file to be used by FST config tests related
  37         hostapd/wpa_supplicant instances"""
  38         raise Exception("Virtual to_file() called!")
  39
  40 class FstLauncherConfigAP(FstLauncherConfig):
  41     """FstLauncherConfigAP class represents configuration to be used for
  42     FST config tests related hostapd instance"""
  43     def __init__(self, iface, ssid, mode, chan, fst_group, fst_pri,
  44                  fst_llt=None):
  45         self.ssid = ssid
  46         self.mode = mode
  47         self.chan = chan
  48         FstLauncherConfig.__init__(self, iface, fst_group, fst_pri, fst_llt)
  49
  50     def is_ap(self):
  51         return True
  52
  53     def get_channel(self):
  54         return self.chan
  55
  56     def to_file(self, pathname):
  57         """Creates configuration file to be used by FST config tests related
  58         hostapd instance"""
  59         with open(pathname, "w") as f:
  60             f.write("country_code=US\n"
  61                     "interface=%s\n"
  62                     "ctrl_interface=/var/run/hostapd\n"
  63                     "ssid=%s\n"
  64                     "channel=%s\n"
  65                     "hw_mode=%s\n"
  66                     "ieee80211n=1\n" % (self.iface, self.ssid, self.chan,
  67                                         self.mode))
  68             if len(self.fst_group) != 0:
  69                 f.write("fst_group_id=%s\n"
  70                         "fst_priority=%s\n" % (self.fst_group, self.fst_pri))
  71                 if self.fst_llt is not None:
  72                     f.write("fst_llt=%s\n" % self.fst_llt)
  73         with open(pathname, "r") as f:
  74             logger.debug("wrote hostapd config file %s:\n%s" % (pathname,
  75                                                                 f.read()))
  76
  77 class FstLauncherConfigSTA(FstLauncherConfig):
  78     """FstLauncherConfig class represents configuration to be used for
  79     FST config tests related wpa_supplicant instance"""
  80     def __init__(self, iface, fst_group, fst_pri, fst_llt=None):
  81         FstLauncherConfig.__init__(self, iface, fst_group, fst_pri, fst_llt)
  82
  83     def is_ap(self):
  84         return False
  85
  86     def to_file(self, pathname):
  87         """Creates configuration file to be used by FST config tests related
  88         wpa_supplicant instance"""
  89         with open(pathname, "w") as f:
  90             f.write("ctrl_interface=DIR=/var/run/wpa_supplicant\n"
  91                 "p2p_no_group_iface=1\n")
  92             if len(self.fst_group) != 0:
  93                 f.write("fst_group_id=%s\n"
  94                     "fst_priority=%s\n" % (self.fst_group, self.fst_pri))
  95                 if self.fst_llt is not None:
  96                     f.write("fst_llt=%s\n" % self.fst_llt)
  97         with open(pathname, "r") as f:
  98             logger.debug("wrote wpa_supplicant config file %s:\n%s" % (pathname, f.read()))
  99
 100 class FstLauncher:
 101     """FstLauncher class is responsible for launching and cleaning up of FST
 102     config tests related hostapd/wpa_supplicant instances"""
 103     def __init__(self, logpath):
 104         self.logger = logging.getLogger()
 105         self.fst_logpath = logpath
 106         self.cfgs_to_run = []
 107         self.hapd_fst_global = '/var/run/hostapd-fst-global'
 108         self.wsup_fst_global = '/tmp/fststa'
 109         self.nof_aps = 0
 110         self.nof_stas = 0
 111         self.reg_ctrl = fst_test_common.HapdRegCtrl()
 112         self.test_is_supported()
 113
 114     def __del__(self):
 115         self.cleanup()
 116
 117     @staticmethod
 118     def test_is_supported():
 119         h = hostapd.HostapdGlobal()
 120         resp = h.request("FST-MANAGER TEST_REQUEST IS_SUPPORTED")
 121         if not resp.startswith("OK"):
 122             raise utils.HwsimSkip("FST not supported")
 123         w = wpasupplicant.WpaSupplicant(global_iface='/tmp/wpas-wlan5')
 124         resp = w.global_request("FST-MANAGER TEST_REQUEST IS_SUPPORTED")
 125         if not resp.startswith("OK"):
 126             raise utils.HwsimSkip("FST not supported")
 127
 128     def get_cfg_pathname(self, cfg):
 129         """Returns pathname of ifname based configuration file"""
 130         return self.fst_logpath +'/'+ cfg.ifname() + '.conf'
 131
 132     def add_cfg(self, cfg):
 133         """Adds configuration to be used for launching hostapd/wpa_supplicant
 134         instances"""
 135         if cfg not in self.cfgs_to_run:
 136             self.cfgs_to_run.append(cfg)
 137         if cfg.is_ap() == True:
 138             self.nof_aps += 1
 139         else:
 140             self.nof_stas += 1
 141
 142     def remove_cfg(self, cfg):
 143         """Removes configuration previously added with add_cfg"""
 144         if cfg in self.cfgs_to_run:
 145             self.cfgs_to_run.remove(cfg)
 146         if cfg.is_ap() == True:
 147             self.nof_aps -= 1
 148         else:
 149             self.nof_stas -= 1
 150         config_file = self.get_cfg_pathname(cfg);
 151         if os.path.exists(config_file):
 152             os.remove(config_file)
 153
 154     def run_hostapd(self):
 155         """Lauches hostapd with interfaces configured according to
 156         FstLauncherConfigAP configurations added"""
 157         if self.nof_aps == 0:
 158             raise Exception("No FST APs to start")
 159         pidfile = self.fst_logpath + '/' + 'myhostapd.pid'
 160         mylogfile = self.fst_logpath + '/' + 'fst-hostapd'
 161         prg = os.path.join(self.fst_logpath,
 162                            'alt-hostapd/hostapd/hostapd')
 163         if not os.path.exists(prg):
 164             prg = '../../hostapd/hostapd'
 165         cmd = [ prg, '-B', '-ddd',
 166                 '-P', pidfile, '-f', mylogfile, '-g', self.hapd_fst_global]
 167         for i in range(0, len(self.cfgs_to_run)):
 168             cfg = self.cfgs_to_run[i]
 169             if cfg.is_ap() == True:
 170                 cfgfile = self.get_cfg_pathname(cfg)
 171                 cfg.to_file(cfgfile)
 172                 cmd.append(cfgfile)
 173                 self.reg_ctrl.add_ap(cfg.ifname(), cfg.get_channel())
 174         self.logger.debug("Starting fst hostapd: " + ' '.join(cmd))
 175         res = subprocess.call(cmd)
 176         self.logger.debug("fst hostapd start result: %d" % res)
 177         if res == 0:
 178             self.reg_ctrl.start()
 179         return res
 180
 181     def run_wpa_supplicant(self):
 182         """Lauches wpa_supplicant with interfaces configured according to
 183         FstLauncherConfigSTA configurations added"""
 184         if self.nof_stas == 0:
 185             raise Exception("No FST STAs to start")
 186         pidfile = self.fst_logpath + '/' + 'mywpa_supplicant.pid'
 187         mylogfile = self.fst_logpath + '/' + 'fst-wpa_supplicant'
 188         prg = os.path.join(self.fst_logpath,
 189                            'alt-wpa_supplicant/wpa_supplicant/wpa_supplicant')
 190         if not os.path.exists(prg):
 191             prg = '../../wpa_supplicant/wpa_supplicant'
 192         cmd = [ prg, '-B', '-ddd',
 193                 '-P' + pidfile, '-f', mylogfile, '-g', self.wsup_fst_global ]
 194         sta_no = 0
 195         for i in range(0, len(self.cfgs_to_run)):
 196             cfg = self.cfgs_to_run[i]
 197             if cfg.is_ap() == False:
 198                 cfgfile = self.get_cfg_pathname(cfg)
 199                 cfg.to_file(cfgfile)
 200                 cmd.append('-c' + cfgfile)
 201                 cmd.append('-i' + cfg.ifname())
 202                 cmd.append('-Dnl80211')
 203                 if sta_no != self.nof_stas -1:
 204                     cmd.append('-N')    # Next station configuration
 205                 sta_no += 1
 206         self.logger.debug("Starting fst supplicant: " + ' '.join(cmd))
 207         res = subprocess.call(cmd)
 208         self.logger.debug("fst supplicant start result: %d" % res)
 209         return res
 210
 211     def cleanup(self):
 212         """Terminates hostapd/wpa_supplicant processes previously launched with
 213         run_hostapd/run_wpa_supplicant"""
 214         pidfile = self.fst_logpath + '/' + 'myhostapd.pid'
 215         self.kill_pid(pidfile)
 216         pidfile = self.fst_logpath + '/' + 'mywpa_supplicant.pid'
 217         self.kill_pid(pidfile)
 218         self.reg_ctrl.stop()
 219         while len(self.cfgs_to_run) != 0:
 220             cfg = self.cfgs_to_run[0]
 221             self.remove_cfg(cfg)
 222
 223     def kill_pid(self, pidfile):
 224         """Kills process by PID file"""
 225         if not os.path.exists(pidfile):
 226             return
 227         pid = -1
 228         try:
 229             pf = file(pidfile, 'r')
 230             pid = int(pf.read().strip())
 231             pf.close()
 232             self.logger.debug("kill_pid %s --> pid %d" % (pidfile, pid))
 233             os.kill(pid, signal.SIGTERM)
 234             for i in range(10):
 235                 try:
 236                     # Poll the pid (Is the process still existing?)
 237                     os.kill(pid, 0)
 238                 except OSError:
 239                     # No, already done
 240                     break
 241                 # Wait and check again
 242                 time.sleep(1)
 243         except Exception, e:
 244             self.logger.debug("Didn't stop the pid=%d. Was it stopped already? (%s)" % (pid, str(e)))
 245
 246
 247 def parse_ies(iehex, el=-1):
 248     """Parses the information elements hex string 'iehex' in format
 249     "0a0b0c0d0e0f". If no 'el' defined just checks the IE string for integrity.
 250     If 'el' is defined returns the list of hex values of the specific IE (or
 251     empty list if the element is not in the string."""
 252     iel = [iehex[i:i + 2] for i in range(0, len(iehex), 2)]
 253     for i in range(0, len(iel)):
 254          iel[i] = int(iel[i], 16)
 255     # Sanity check
 256     i = 0
 257     res = []
 258     while i < len(iel):
 259         logger.debug("IE found: %x" % iel[i])
 260         if el != -1 and el == iel[i]:
 261             res = iel[i + 2:i + 2 + iel[i + 1]]
 262         i += 2 + iel[i + 1]
 263     if i != len(iel):
 264         logger.error("Bad IE string: " + iehex)
 265         res = []
 266     return res
 267
 268 def scan_and_get_bss(dev, frq):
 269     """Issues a scan on given device on given frequency, returns the bss info
 270     dictionary ('ssid','ie','flags', etc.) or None. Note, the function
 271     implies there is only one AP on the given channel. If not a case,
 272     the function must be changed to call dev.get_bss() till the AP with the
 273     [b]ssid that we need is found"""
 274     dev.scan(freq=frq)
 275     return dev.get_bss('0')
 276
 277
 278 # AP configuration tests
 279
 280 def run_test_ap_configuration(apdev, test_params,
 281                               fst_group = fst_test_common.fst_test_def_group,
 282                               fst_pri = fst_test_common.fst_test_def_prio_high,
 283                               fst_llt = fst_test_common.fst_test_def_llt):
 284     """Runs FST hostapd where the 1st AP configuration is fixed, the 2nd fst
 285     configuration is provided by the parameters. Returns the result of the run:
 286     0 - no errors discovered, an error otherwise. The function is used for
 287     simplek "bad configuration" tests."""
 288     logdir = test_params['logdir']
 289     fst_launcher = FstLauncher(logdir)
 290     ap1 = FstLauncherConfigAP(apdev[0]['ifname'], 'fst_goodconf', 'a',
 291                               fst_test_common.fst_test_def_chan_a,
 292                               fst_test_common.fst_test_def_group,
 293                               fst_test_common.fst_test_def_prio_low,
 294                               fst_test_common.fst_test_def_llt)
 295     ap2 = FstLauncherConfigAP(apdev[1]['ifname'], 'fst_badconf', 'b',
 296                               fst_test_common.fst_test_def_chan_g, fst_group,
 297                               fst_pri, fst_llt)
 298     fst_launcher.add_cfg(ap1)
 299     fst_launcher.add_cfg(ap2)
 300     res = fst_launcher.run_hostapd()
 301     return res
 302
 303 def run_test_sta_configuration(test_params,
 304                                fst_group = fst_test_common.fst_test_def_group,
 305                                fst_pri = fst_test_common.fst_test_def_prio_high,
 306                                fst_llt = fst_test_common.fst_test_def_llt):
 307     """Runs FST wpa_supplicant where the 1st STA configuration is fixed, the
 308     2nd fst configuration is provided by the parameters. Returns the result of
 309     the run: 0 - no errors discovered, an error otherwise. The function is used
 310     for simple "bad configuration" tests."""
 311     logdir = test_params['logdir']
 312     fst_launcher = FstLauncher(logdir)
 313     sta1 = FstLauncherConfigSTA('wlan5',
 314                                 fst_test_common.fst_test_def_group,
 315                                 fst_test_common.fst_test_def_prio_low,
 316                                 fst_test_common.fst_test_def_llt)
 317     sta2 = FstLauncherConfigSTA('wlan6', fst_group, fst_pri, fst_llt)
 318     fst_launcher.add_cfg(sta1)
 319     fst_launcher.add_cfg(sta2)
 320     res = fst_launcher.run_wpa_supplicant()
 321     return res
 322
 323 def test_fst_ap_config_llt_neg(dev, apdev, test_params):
 324     """FST AP configuration negative LLT"""
 325     res = run_test_ap_configuration(apdev, test_params, fst_llt = '-1')
 326     if res == 0:
 327         raise Exception("hostapd started with a negative llt")
 328
 329 def test_fst_ap_config_llt_zero(dev, apdev, test_params):
 330     """FST AP configuration zero LLT"""
 331     res = run_test_ap_configuration(apdev, test_params, fst_llt = '0')
 332     if res == 0:
 333         raise Exception("hostapd started with a zero llt")
 334
 335 def test_fst_ap_config_llt_too_big(dev, apdev, test_params):
 336     """FST AP configuration LLT is too big"""
 337     res = run_test_ap_configuration(apdev, test_params,
 338                                     fst_llt = '4294967296') #0x100000000
 339     if res == 0:
 340         raise Exception("hostapd started with llt that is too big")
 341
 342 def test_fst_ap_config_llt_nan(dev, apdev, test_params):
 343     """FST AP configuration LLT is not a number"""
 344     res = run_test_ap_configuration(apdev, test_params, fst_llt = 'nan')
 345     if res == 0:
 346         raise Exception("hostapd started with llt not a number")
 347
 348 def test_fst_ap_config_pri_neg(dev, apdev, test_params):
 349     """FST AP configuration Priority negative"""
 350     res = run_test_ap_configuration(apdev, test_params, fst_pri = '-1')
 351     if res == 0:
 352         raise Exception("hostapd started with a negative fst priority")
 353
 354 def test_fst_ap_config_pri_zero(dev, apdev, test_params):
 355     """FST AP configuration Priority zero"""
 356     res = run_test_ap_configuration(apdev, test_params, fst_pri = '0')
 357     if res == 0:
 358         raise Exception("hostapd started with a zero fst priority")
 359
 360 def test_fst_ap_config_pri_large(dev, apdev, test_params):
 361     """FST AP configuration Priority too large"""
 362     res = run_test_ap_configuration(apdev, test_params, fst_pri = '256')
 363     if res == 0:
 364         raise Exception("hostapd started with too large fst priority")
 365
 366 def test_fst_ap_config_pri_nan(dev, apdev, test_params):
 367     """FST AP configuration Priority not a number"""
 368     res = run_test_ap_configuration(apdev, test_params, fst_pri = 'nan')
 369     if res == 0:
 370         raise Exception("hostapd started with fst priority not a number")
 371
 372 def test_fst_ap_config_group_len(dev, apdev, test_params):
 373     """FST AP configuration Group max length"""
 374     res = run_test_ap_configuration(apdev, test_params,
 375                                     fst_group = 'fstg5678abcd34567')
 376     if res == 0:
 377         raise Exception("hostapd started with fst_group length too big")
 378
 379 def test_fst_ap_config_good(dev, apdev, test_params):
 380     """FST AP configuration good parameters"""
 381     res = run_test_ap_configuration(apdev, test_params)
 382     if res != 0:
 383         raise Exception("hostapd didn't start with valid config parameters")
 384
 385 def test_fst_ap_config_default(dev, apdev, test_params):
 386     """FST AP configuration default parameters"""
 387     res = run_test_ap_configuration(apdev, test_params, fst_llt = None)
 388     if res != 0:
 389         raise Exception("hostapd didn't start with valid config parameters")
 390
 391
 392 # STA configuration tests
 393
 394 def test_fst_sta_config_llt_neg(dev, apdev, test_params):
 395     """FST STA configuration negative LLT"""
 396     res = run_test_sta_configuration(test_params, fst_llt = '-1')
 397     if res == 0:
 398         raise Exception("wpa_supplicant started with a negative llt")
 399
 400 def test_fst_sta_config_llt_zero(dev, apdev, test_params):
 401     """FST STA configuration zero LLT"""
 402     res = run_test_sta_configuration(test_params, fst_llt = '0')
 403     if res == 0:
 404         raise Exception("wpa_supplicant started with a zero llt")
 405
 406 def test_fst_sta_config_llt_large(dev, apdev, test_params):
 407     """FST STA configuration LLT is too large"""
 408     res = run_test_sta_configuration(test_params,
 409                                      fst_llt = '4294967296') #0x100000000
 410     if res == 0:
 411         raise Exception("wpa_supplicant started with llt that is too large")
 412
 413 def test_fst_sta_config_llt_nan(dev, apdev, test_params):
 414     """FST STA configuration LLT is not a number"""
 415     res = run_test_sta_configuration(test_params, fst_llt = 'nan')
 416     if res == 0:
 417         raise Exception("wpa_supplicant started with llt not a number")
 418
 419 def test_fst_sta_config_pri_neg(dev, apdev, test_params):
 420     """FST STA configuration Priority negative"""
 421     res = run_test_sta_configuration(test_params, fst_pri = '-1')
 422     if res == 0:
 423         raise Exception("wpa_supplicant started with a negative fst priority")
 424
 425 def test_fst_sta_config_pri_zero(dev, apdev, test_params):
 426     """FST STA configuration Priority zero"""
 427     res = run_test_sta_configuration(test_params, fst_pri = '0')
 428     if res == 0:
 429         raise Exception("wpa_supplicant started with a zero fst priority")
 430
 431 def test_fst_sta_config_pri_big(dev, apdev, test_params):
 432     """FST STA configuration Priority too large"""
 433     res = run_test_sta_configuration(test_params, fst_pri = '256')
 434     if res == 0:
 435         raise Exception("wpa_supplicant started with too large fst priority")
 436
 437 def test_fst_sta_config_pri_nan(dev, apdev, test_params):
 438     """FST STA configuration Priority not a number"""
 439     res = run_test_sta_configuration(test_params, fst_pri = 'nan')
 440     if res == 0:
 441         raise Exception("wpa_supplicant started with fst priority not a number")
 442
 443 def test_fst_sta_config_group_len(dev, apdev, test_params):
 444     """FST STA configuration Group max length"""
 445     res = run_test_sta_configuration(test_params,
 446                                      fst_group = 'fstg5678abcd34567')
 447     if res == 0:
 448         raise Exception("wpa_supplicant started with fst_group length too big")
 449
 450 def test_fst_sta_config_good(dev, apdev, test_params):
 451     """FST STA configuration good parameters"""
 452     res = run_test_sta_configuration(test_params)
 453     if res != 0:
 454         raise Exception("wpa_supplicant didn't start with valid config parameters")
 455
 456 def test_fst_sta_config_default(dev, apdev, test_params):
 457     """FST STA configuration default parameters"""
 458     res = run_test_sta_configuration(test_params, fst_llt = None)
 459     if res != 0:
 460         raise Exception("wpa_supplicant didn't start with valid config parameters")
 461
 462 def test_fst_scan_mb(dev, apdev, test_params):
 463     """FST scan valid MB IE presence with normal start"""
 464     logdir = test_params['logdir']
 465
 466     # Test valid MB IE in scan results
 467     fst_launcher = FstLauncher(logdir)
 468     ap1 = FstLauncherConfigAP(apdev[0]['ifname'], 'fst_11a', 'a',
 469                               fst_test_common.fst_test_def_chan_a,
 470                               fst_test_common.fst_test_def_group,
 471                               fst_test_common.fst_test_def_prio_high)
 472     ap2 = FstLauncherConfigAP(apdev[1]['ifname'], 'fst_11g', 'b',
 473                               fst_test_common.fst_test_def_chan_g,
 474                               fst_test_common.fst_test_def_group,
 475                               fst_test_common.fst_test_def_prio_low)
 476     fst_launcher.add_cfg(ap1)
 477     fst_launcher.add_cfg(ap2)
 478     res = fst_launcher.run_hostapd()
 479     if res != 0:
 480         raise Exception("hostapd didn't start properly")
 481     try:
 482         mbie1=[]
 483         flags1 = ''
 484         mbie2=[]
 485         flags2 = ''
 486         # Scan 1st AP
 487         vals1 = scan_and_get_bss(dev[0], fst_test_common.fst_test_def_freq_a)
 488         if vals1 != None:
 489             if 'ie' in vals1:
 490                 mbie1 = parse_ies(vals1['ie'], 0x9e)
 491             if 'flags' in vals1:
 492                 flags1 = vals1['flags']
 493         # Scan 2nd AP
 494         vals2 = scan_and_get_bss(dev[2], fst_test_common.fst_test_def_freq_g)
 495         if vals2 != None:
 496             if 'ie' in vals2:
 497                 mbie2 = parse_ies(vals2['ie'],0x9e)
 498             if 'flags' in vals2:
 499                 flags2 = vals2['flags']
 500     finally:
 501          fst_launcher.cleanup()
 502
 503     if len(mbie1) == 0:
 504         raise Exception("No MB IE created by 1st AP")
 505     if len(mbie2) == 0:
 506         raise Exception("No MB IE created by 2nd AP")
 507
 508 def test_fst_scan_nomb(dev, apdev, test_params):
 509     """FST scan no MB IE presence with 1 AP start"""
 510     logdir = test_params['logdir']
 511
 512     # Test valid MB IE in scan results
 513     fst_launcher = FstLauncher(logdir)
 514     ap1 = FstLauncherConfigAP(apdev[0]['ifname'], 'fst_11a', 'a',
 515                               fst_test_common.fst_test_def_chan_a,
 516                               fst_test_common.fst_test_def_group,
 517                               fst_test_common.fst_test_def_prio_high)
 518     fst_launcher.add_cfg(ap1)
 519     res = fst_launcher.run_hostapd()
 520     if res != 0:
 521         raise Exception("Hostapd didn't start properly")
 522     try:
 523         time.sleep(2)
 524         mbie1=[]
 525         flags1 = ''
 526         vals1 = scan_and_get_bss(dev[0], fst_test_common.fst_test_def_freq_a)
 527         if vals1 != None:
 528             if 'ie' in vals1:
 529                 mbie1 = parse_ies(vals1['ie'], 0x9e)
 530             if 'flags' in vals1:
 531                 flags1 = vals1['flags']
 532     finally:
 533         fst_launcher.cleanup()
 534
 535     if len(mbie1) != 0:
 536         raise Exception("MB IE exists with 1 AP")