mirror of
https://github.com/vanhoefm/fragattacks.git
synced 2024-11-25 16:58:41 -05:00
89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
|
# FST tests related definitions
|
||
|
|
# Copyright (c) 2015, Qualcomm Atheros, Inc.
|
||
|
|
#
|
||
|
|
# This software may be distributed under the terms of the BSD license.
|
||
|
|
# See README for more details.
|
||
|
|
|
||
|
|
import subprocess
|
||
|
|
import logging
|
||
|
|
|
||
|
|
import hostapd
|
||
|
|
|
||
|
|
logger = logging.getLogger()
|
||
|
|
|
||
|
|
fst_test_def_group='fstg0'
|
||
|
|
fst_test_def_freq_g='2412' # Channel 1
|
||
|
|
fst_test_def_freq_a='5180' # Channel 36
|
||
|
|
fst_test_def_chan_g='1'
|
||
|
|
fst_test_def_chan_a='36'
|
||
|
|
fst_test_def_prio_low='100'
|
||
|
|
fst_test_def_prio_high='110'
|
||
|
|
fst_test_def_llt='100'
|
||
|
|
fst_test_def_reg_domain='00'
|
||
|
|
|
||
|
|
class HapdRegCtrl:
|
||
|
|
def __init__(self):
|
||
|
|
self.refcnt = 0
|
||
|
|
self.ifname = None
|
||
|
|
self.changed = False
|
||
|
|
|
||
|
|
def __del__(self):
|
||
|
|
if self.refcnt != 0 and self.changed == True:
|
||
|
|
self.restore_reg_domain()
|
||
|
|
|
||
|
|
def start(self):
|
||
|
|
if self.ifname != None:
|
||
|
|
hapd = hostapd.Hostapd(self.ifname)
|
||
|
|
self.changed = self.wait_hapd_reg_change(hapd)
|
||
|
|
|
||
|
|
def stop(self):
|
||
|
|
if self.changed == True:
|
||
|
|
self.restore_reg_domain()
|
||
|
|
self.changed = False
|
||
|
|
|
||
|
|
def add_ap(self, ifname, chan):
|
||
|
|
if self.changed == False and self.channel_may_require_reg_change(chan):
|
||
|
|
self.ifname = ifname
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def channel_may_require_reg_change(chan):
|
||
|
|
if int(chan) > 14:
|
||
|
|
return True
|
||
|
|
return False
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def wait_hapd_reg_change(hapd):
|
||
|
|
state = hapd.get_status_field("state")
|
||
|
|
if state != "COUNTRY_UPDATE":
|
||
|
|
state = hapd.get_status_field("state")
|
||
|
|
if state != "ENABLED":
|
||
|
|
raise Exception("Unexpected interface state - expected COUNTRY_UPDATE")
|
||
|
|
else:
|
||
|
|
logger.debug("fst hostapd: regulatory domain already set")
|
||
|
|
return True
|
||
|
|
|
||
|
|
logger.debug("fst hostapd: waiting for regulatory domain to be set...")
|
||
|
|
|
||
|
|
ev = hapd.wait_event(["AP-ENABLED"], timeout=10)
|
||
|
|
if not ev:
|
||
|
|
raise Exception("AP setup timed out")
|
||
|
|
|
||
|
|
logger.debug("fst hostapd: regulatory domain set")
|
||
|
|
|
||
|
|
state = hapd.get_status_field("state")
|
||
|
|
if state != "ENABLED":
|
||
|
|
raise Exception("Unexpected interface state - expected ENABLED")
|
||
|
|
|
||
|
|
logger.debug("fst hostapd: regulatory domain ready")
|
||
|
|
return True
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def restore_reg_domain():
|
||
|
|
logger.debug("fst hostapd: waiting for regulatory domain to be restored...")
|
||
|
|
|
||
|
|
res = subprocess.call(['iw', 'reg', 'set', fst_test_def_reg_domain])
|
||
|
|
if res != 0:
|
||
|
|
raise Exception("Cannot restore regulatory domain")
|
||
|
|
|
||
|
|
logger.debug("fst hostapd: regulatory domain ready")
|