1#!/usr/bin/env python3 2# 3# Copyright 2017 - The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17import fcntl 18import os 19import selenium 20import splinter 21import time 22from acts import logger 23from acts.controllers import access_point 24from acts.controllers.ap_lib import bridge_interface 25from acts.controllers.ap_lib import hostapd_security 26from acts.controllers.ap_lib import hostapd_ap_preset 27 28BROWSER_WAIT_SHORT = 1 29BROWSER_WAIT_MED = 3 30BROWSER_WAIT_LONG = 30 31BROWSER_WAIT_EXTRA_LONG = 60 32 33 34def create(configs): 35 """Factory method for retail AP class. 36 37 Args: 38 configs: list of dicts containing ap settings. ap settings must contain 39 the following: brand, model, ip_address, username and password 40 """ 41 SUPPORTED_APS = { 42 ("Netgear", "R7000"): "NetgearR7000AP", 43 ("Netgear", "R7000NA"): "NetgearR7000NAAP", 44 ("Netgear", "R7500"): "NetgearR7500AP", 45 ("Netgear", "R7800"): "NetgearR7800AP", 46 ("Netgear", "R8000"): "NetgearR8000AP", 47 ("Netgear", "R8500"): "NetgearR8500AP", 48 ("Google", "Wifi"): "GoogleWifiAP" 49 } 50 objs = [] 51 for config in configs: 52 try: 53 ap_class_name = SUPPORTED_APS[(config["brand"], config["model"])] 54 ap_class = globals()[ap_class_name] 55 except KeyError: 56 raise KeyError("Invalid retail AP brand and model combination.") 57 objs.append(ap_class(config)) 58 return objs 59 60 61def detroy(objs): 62 return 63 64 65class BlockingBrowser(splinter.driver.webdriver.chrome.WebDriver): 66 """Class that implements a blocking browser session on top of selenium. 67 68 The class inherits from and builds upon splinter/selenium's webdriver class 69 and makes sure that only one such webdriver is active on a machine at any 70 single time. The class ensures single session operation using a lock file. 71 The class is to be used within context managers (e.g. with statements) to 72 ensure locks are always properly released. 73 """ 74 def __init__(self, headless, timeout): 75 """Constructor for BlockingBrowser class. 76 77 Args: 78 headless: boolean to control visible/headless browser operation 79 timeout: maximum time allowed to launch browser 80 """ 81 self.log = logger.create_tagged_trace_logger("ChromeDriver") 82 self.chrome_options = splinter.driver.webdriver.chrome.Options() 83 self.chrome_options.add_argument("--no-proxy-server") 84 self.chrome_options.add_argument("--no-sandbox") 85 self.chrome_options.add_argument("--allow-running-insecure-content") 86 self.chrome_options.add_argument("--ignore-certificate-errors") 87 self.chrome_capabilities = selenium.webdriver.common.desired_capabilities.DesiredCapabilities.CHROME.copy( 88 ) 89 self.chrome_capabilities["acceptSslCerts"] = True 90 self.chrome_capabilities["acceptInsecureCerts"] = True 91 if headless: 92 self.chrome_options.add_argument("--headless") 93 self.chrome_options.add_argument("--disable-gpu") 94 self.lock_file_path = "/usr/local/bin/chromedriver" 95 self.timeout = timeout 96 97 def __enter__(self): 98 """Entry context manager for BlockingBrowser. 99 100 The enter context manager for BlockingBrowser attempts to lock the 101 browser file. If successful, it launches and returns a chromedriver 102 session. If an exception occurs while starting the browser, the lock 103 file is released. 104 """ 105 self.lock_file = open(self.lock_file_path, "r") 106 start_time = time.time() 107 while time.time() < start_time + self.timeout: 108 try: 109 fcntl.flock(self.lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB) 110 except BlockingIOError: 111 time.sleep(BROWSER_WAIT_SHORT) 112 continue 113 try: 114 self.driver = selenium.webdriver.Chrome( 115 options=self.chrome_options, 116 desired_capabilities=self.chrome_capabilities) 117 self.element_class = splinter.driver.webdriver.WebDriverElement 118 self._cookie_manager = splinter.driver.webdriver.cookie_manager.CookieManager( 119 self.driver) 120 super(splinter.driver.webdriver.chrome.WebDriver, 121 self).__init__(2) 122 return super(BlockingBrowser, self).__enter__() 123 except: 124 fcntl.flock(self.lock_file, fcntl.LOCK_UN) 125 self.lock_file.close() 126 raise RuntimeError("Error starting browser. " 127 "Releasing lock file.") 128 raise TimeoutError("Could not start chrome browser in time.") 129 130 def __exit__(self, exc_type, exc_value, traceback): 131 """Exit context manager for BlockingBrowser. 132 133 The exit context manager simply calls the parent class exit and 134 releases the lock file. 135 """ 136 try: 137 super(BlockingBrowser, self).__exit__(exc_type, exc_value, 138 traceback) 139 except: 140 raise RuntimeError("Failed to quit browser. Releasing lock file.") 141 finally: 142 fcntl.flock(self.lock_file, fcntl.LOCK_UN) 143 self.lock_file.close() 144 145 def restart(self): 146 """Method to restart browser session without releasing lock file.""" 147 self.quit() 148 self.__enter__() 149 150 def visit_persistent(self, 151 url, 152 page_load_timeout, 153 num_tries, 154 backup_url="about:blank", 155 check_for_element=None): 156 """Method to visit webpages and retry upon failure. 157 158 The function visits a web page and checks the the resulting URL matches 159 the intended URL, i.e. no redirects have happened 160 161 Args: 162 url: the intended url 163 page_load_timeout: timeout for page visits 164 num_tries: number of tries before url is declared unreachable 165 backup_url: url to visit if first url is not reachable. This can be 166 used to simply refresh the browser and try again or to re-login to 167 the AP 168 check_for_element: element id to check for existence on page 169 """ 170 self.driver.set_page_load_timeout(page_load_timeout) 171 for idx in range(num_tries): 172 try: 173 self.visit(url) 174 except: 175 self.restart() 176 177 page_reached = self.url.split("/")[-1] == url.split("/")[-1] 178 if check_for_element: 179 time.sleep(BROWSER_WAIT_MED) 180 element = self.find_by_id(check_for_element) 181 if not element: 182 page_reached = 0 183 if page_reached: 184 break 185 else: 186 try: 187 self.visit(backup_url) 188 except: 189 self.restart() 190 191 if idx == num_tries - 1: 192 self.log.error("URL unreachable. Current URL: {}".format( 193 self.url)) 194 raise RuntimeError("URL unreachable.") 195 196 197class WifiRetailAP(object): 198 """Base class implementation for retail ap. 199 200 Base class provides functions whose implementation is shared by all aps. 201 If some functions such as set_power not supported by ap, checks will raise 202 exceptions. 203 """ 204 def __init__(self, ap_settings): 205 self.ap_settings = ap_settings.copy() 206 self.log = logger.create_tagged_trace_logger("AccessPoint|{}".format( 207 self._get_control_ip_address())) 208 # Lock AP 209 if self.ap_settings.get('lock_ap', 0): 210 self.lock_timeout = self.ap_settings.get('lock_timeout', 3600) 211 self._lock_ap() 212 213 def read_ap_settings(self): 214 """Function that reads current ap settings. 215 216 Function implementation is AP dependent and thus base class raises exception 217 if function not implemented in child class. 218 """ 219 raise NotImplementedError 220 221 def validate_ap_settings(self): 222 """Function to validate ap settings. 223 224 This function compares the actual ap settings read from the web GUI 225 with the assumed settings saved in the AP object. When called after AP 226 configuration, this method helps ensure that our configuration was 227 successful. 228 Note: Calling this function updates the stored ap_settings 229 230 Raises: 231 ValueError: If read AP settings do not match stored settings. 232 """ 233 assumed_ap_settings = self.ap_settings.copy() 234 actual_ap_settings = self.read_ap_settings() 235 if assumed_ap_settings != actual_ap_settings: 236 self.log.warning( 237 "Discrepancy in AP settings. Some settings may have been overwritten." 238 ) 239 240 def configure_ap(self, **config_flags): 241 """Function that configures ap based on values of ap_settings. 242 243 Function implementation is AP dependent and thus base class raises exception 244 if function not implemented in child class. 245 246 Args: 247 config_flags: optional configuration flags 248 """ 249 raise NotImplementedError 250 251 def set_region(self, region): 252 """Function that sets AP region. 253 254 This function sets the region for the AP. Note that this may overwrite 255 channel and bandwidth settings in cases where the new region does not 256 support the current wireless configuration. 257 258 Args: 259 region: string indicating AP region 260 """ 261 self.log.warning("Updating region may overwrite wireless settings.") 262 setting_to_update = {"region": region} 263 self.update_ap_settings(setting_to_update) 264 265 def set_radio_on_off(self, network, status): 266 """Function that turns the radio on or off. 267 268 Args: 269 network: string containing network identifier (2G, 5G_1, 5G_2) 270 status: boolean indicating on or off (0: off, 1: on) 271 """ 272 setting_to_update = {"status_{}".format(network): int(status)} 273 self.update_ap_settings(setting_to_update) 274 275 def set_ssid(self, network, ssid): 276 """Function that sets network SSID. 277 278 Args: 279 network: string containing network identifier (2G, 5G_1, 5G_2) 280 ssid: string containing ssid 281 """ 282 setting_to_update = {"ssid_{}".format(network): str(ssid)} 283 self.update_ap_settings(setting_to_update) 284 285 def set_channel(self, network, channel): 286 """Function that sets network channel. 287 288 Args: 289 network: string containing network identifier (2G, 5G_1, 5G_2) 290 channel: string or int containing channel 291 """ 292 setting_to_update = {"channel_{}".format(network): str(channel)} 293 self.update_ap_settings(setting_to_update) 294 295 def set_bandwidth(self, network, bandwidth): 296 """Function that sets network bandwidth/mode. 297 298 Args: 299 network: string containing network identifier (2G, 5G_1, 5G_2) 300 bandwidth: string containing mode, e.g. 11g, VHT20, VHT40, VHT80. 301 """ 302 setting_to_update = {"bandwidth_{}".format(network): str(bandwidth)} 303 self.update_ap_settings(setting_to_update) 304 305 def set_power(self, network, power): 306 """Function that sets network transmit power. 307 308 Args: 309 network: string containing network identifier (2G, 5G_1, 5G_2) 310 power: string containing power level, e.g., 25%, 100% 311 """ 312 setting_to_update = {"power_{}".format(network): str(power)} 313 self.update_ap_settings(setting_to_update) 314 315 def set_security(self, network, security_type, *password): 316 """Function that sets network security setting and password. 317 318 Args: 319 network: string containing network identifier (2G, 5G_1, 5G_2) 320 security: string containing security setting, e.g., WPA2-PSK 321 password: optional argument containing password 322 """ 323 if (len(password) == 1) and (type(password[0]) == str): 324 setting_to_update = { 325 "security_type_{}".format(network): str(security_type), 326 "password_{}".format(network): str(password[0]) 327 } 328 else: 329 setting_to_update = { 330 "security_type_{}".format(network): str(security_type) 331 } 332 self.update_ap_settings(setting_to_update) 333 334 def set_rate(self): 335 """Function that configures rate used by AP. 336 337 Function implementation is not supported by most APs and thus base 338 class raises exception if function not implemented in child class. 339 """ 340 raise NotImplementedError 341 342 def update_ap_settings(self, dict_settings={}, **named_settings): 343 """Function to update settings of existing AP. 344 345 Function copies arguments into ap_settings and calls configure_retail_ap 346 to apply them. 347 348 Args: 349 *dict_settings accepts single dictionary of settings to update 350 **named_settings accepts named settings to update 351 Note: dict and named_settings cannot contain the same settings. 352 """ 353 settings_to_update = dict(dict_settings, **named_settings) 354 if len(settings_to_update) != len(dict_settings) + len(named_settings): 355 raise KeyError("The following keys were passed twice: {}".format( 356 (set(dict_settings.keys()).intersection( 357 set(named_settings.keys()))))) 358 if not set(settings_to_update.keys()).issubset( 359 set(self.ap_settings.keys())): 360 raise KeyError( 361 "The following settings are invalid for this AP: {}".format( 362 set(settings_to_update.keys()).difference( 363 set(self.ap_settings.keys())))) 364 365 updates_requested = False 366 status_toggle_flag = False 367 for setting, value in settings_to_update.items(): 368 if self.ap_settings[setting] != value: 369 self.ap_settings[setting] = value 370 if "status" in setting: 371 status_toggle_flag = True 372 updates_requested = True 373 374 if updates_requested: 375 self.configure_ap(status_toggled=status_toggle_flag) 376 377 def band_lookup_by_channel(self, channel): 378 """Function that gives band name by channel number. 379 380 Args: 381 channel: channel number to lookup 382 Returns: 383 band: name of band which this channel belongs to on this ap 384 """ 385 for key, value in self.channel_band_map.items(): 386 if channel in value: 387 return key 388 raise ValueError("Invalid channel passed in argument.") 389 390 def _get_control_ip_address(self): 391 """Function to get AP's Control Interface IP address.""" 392 if "ssh_config" in self.ap_settings.keys(): 393 return self.ap_settings["ssh_config"]["host"] 394 else: 395 return self.ap_settings["ip_address"] 396 397 def _lock_ap(self): 398 """Function to lock the ap while tests are running.""" 399 self.lock_file_path = "/tmp/{}_{}_{}.lock".format( 400 self.ap_settings['brand'], self.ap_settings['model'], 401 self._get_control_ip_address()) 402 if not os.path.exists(self.lock_file_path): 403 with open(self.lock_file_path, 'w'): 404 pass 405 self.lock_file = open(self.lock_file_path, "r") 406 start_time = time.time() 407 self.log.info('Trying to acquire AP lock.') 408 while time.time() < start_time + self.lock_timeout: 409 try: 410 fcntl.flock(self.lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB) 411 except BlockingIOError: 412 time.sleep(BROWSER_WAIT_SHORT) 413 continue 414 self.log.info('AP lock acquired.') 415 return 416 raise RuntimeError("Could not lock AP in time.") 417 418 def _unlock_ap(self): 419 """Function to unlock the AP when tests are done.""" 420 self.log.info('Releasing AP lock.') 421 if hasattr(self, "lock_file"): 422 fcntl.flock(self.lock_file, fcntl.LOCK_UN) 423 self.lock_file.close() 424 425 426class NetgearR7000AP(WifiRetailAP): 427 """Class that implements Netgear R7500 AP.""" 428 def __init__(self, ap_settings): 429 super().__init__(ap_settings) 430 self.init_gui_data() 431 # Read and update AP settings 432 self.read_ap_settings() 433 if not set(ap_settings.items()).issubset(self.ap_settings.items()): 434 self.update_ap_settings(ap_settings) 435 436 def init_gui_data(self): 437 """Function to initialize data used while interacting with web GUI""" 438 self.config_page = ( 439 "{protocol}://{username}:{password}@" 440 "{ip_address}:{port}/WLG_wireless_dual_band_r10.htm").format( 441 protocol=self.ap_settings["protocol"], 442 username=self.ap_settings["admin_username"], 443 password=self.ap_settings["admin_password"], 444 ip_address=self.ap_settings["ip_address"], 445 port=self.ap_settings["port"]) 446 self.config_page_nologin = ( 447 "{protocol}://{ip_address}:{port}/" 448 "WLG_wireless_dual_band_r10.htm").format( 449 protocol=self.ap_settings["protocol"], 450 ip_address=self.ap_settings["ip_address"], 451 port=self.ap_settings["port"]) 452 self.config_page_advanced = ( 453 "{protocol}://{username}:{password}@" 454 "{ip_address}:{port}/WLG_adv_dual_band2.htm").format( 455 protocol=self.ap_settings["protocol"], 456 username=self.ap_settings["admin_username"], 457 password=self.ap_settings["admin_password"], 458 ip_address=self.ap_settings["ip_address"], 459 port=self.ap_settings["port"]) 460 self.networks = ["2G", "5G_1"] 461 self.channel_band_map = { 462 "2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11], 463 "5G_1": [ 464 36, 40, 44, 48, 52, 56, 60, 64, 100, 104, 108, 112, 116, 120, 465 124, 128, 132, 136, 140, 149, 153, 157, 161, 165 466 ] 467 } 468 self.region_map = { 469 "1": "Africa", 470 "2": "Asia", 471 "3": "Australia", 472 "4": "Canada", 473 "5": "Europe", 474 "6": "Israel", 475 "7": "Japan", 476 "8": "Korea", 477 "9": "Mexico", 478 "10": "South America", 479 "11": "United States", 480 "12": "Middle East(Algeria/Syria/Yemen)", 481 "14": "Russia", 482 "16": "China", 483 "17": "India", 484 "18": "Malaysia", 485 "19": "Middle East(Iran/Labanon/Qatar)", 486 "20": "Middle East(Turkey/Egypt/Tunisia/Kuwait)", 487 "21": "Middle East(Saudi Arabia)", 488 "22": "Middle East(United Arab Emirates)", 489 "23": "Singapore", 490 "24": "Taiwan" 491 } 492 self.config_page_fields = { 493 "region": "WRegion", 494 ("2G", "status"): "enable_ap", 495 ("5G_1", "status"): "enable_ap_an", 496 ("2G", "ssid"): "ssid", 497 ("5G_1", "ssid"): "ssid_an", 498 ("2G", "channel"): "w_channel", 499 ("5G_1", "channel"): "w_channel_an", 500 ("2G", "bandwidth"): "opmode", 501 ("5G_1", "bandwidth"): "opmode_an", 502 ("2G", "power"): "enable_tpc", 503 ("5G_1", "power"): "enable_tpc_an", 504 ("2G", "security_type"): "security_type", 505 ("5G_1", "security_type"): "security_type_an", 506 ("2G", "password"): "passphrase", 507 ("5G_1", "password"): "passphrase_an" 508 } 509 self.bw_mode_values = { 510 "g and b": "11g", 511 "145Mbps": "VHT20", 512 "300Mbps": "VHT40", 513 "HT80": "VHT80" 514 } 515 self.power_mode_values = { 516 "1": "100%", 517 "2": "75%", 518 "3": "50%", 519 "4": "25%" 520 } 521 self.bw_mode_text = { 522 "11g": "Up to 54 Mbps", 523 "VHT20": "Up to 289 Mbps", 524 "VHT40": "Up to 600 Mbps", 525 "VHT80": "Up to 1300 Mbps" 526 } 527 528 def read_ap_settings(self): 529 """Function to read ap settings.""" 530 with BlockingBrowser(self.ap_settings["headless_browser"], 531 900) as browser: 532 # Visit URL 533 browser.visit_persistent(self.config_page, BROWSER_WAIT_MED, 10) 534 535 for key, value in self.config_page_fields.items(): 536 if "status" in key: 537 browser.visit_persistent(self.config_page_advanced, 538 BROWSER_WAIT_MED, 10) 539 config_item = browser.find_by_name(value) 540 self.ap_settings["{}_{}".format(key[1], key[0])] = int( 541 config_item.first.checked) 542 browser.visit_persistent(self.config_page, 543 BROWSER_WAIT_MED, 10) 544 else: 545 config_item = browser.find_by_name(value) 546 if "bandwidth" in key: 547 self.ap_settings["{}_{}".format( 548 key[1], key[0])] = self.bw_mode_values[ 549 config_item.first.value] 550 elif "power" in key: 551 self.ap_settings["{}_{}".format( 552 key[1], key[0])] = self.power_mode_values[ 553 config_item.first.value] 554 elif "region" in key: 555 self.ap_settings["region"] = self.region_map[ 556 config_item.first.value] 557 elif "security_type" in key: 558 for item in config_item: 559 if item.checked: 560 self.ap_settings["{}_{}".format( 561 key[1], key[0])] = item.value 562 else: 563 config_item = browser.find_by_name(value) 564 self.ap_settings["{}_{}".format( 565 key[1], key[0])] = config_item.first.value 566 return self.ap_settings.copy() 567 568 def configure_ap(self, **config_flags): 569 """Function to configure ap wireless settings.""" 570 # Turn radios on or off 571 if config_flags["status_toggled"]: 572 self.configure_radio_on_off() 573 # Configure radios 574 with BlockingBrowser(self.ap_settings["headless_browser"], 575 900) as browser: 576 # Visit URL 577 browser.visit_persistent(self.config_page, BROWSER_WAIT_MED, 10) 578 browser.visit_persistent(self.config_page_nologin, 579 BROWSER_WAIT_MED, 10, self.config_page) 580 581 # Update region, and power/bandwidth for each network 582 config_item = browser.find_by_name( 583 self.config_page_fields["region"]).first 584 config_item.select_by_text(self.ap_settings["region"]) 585 for key, value in self.config_page_fields.items(): 586 if "power" in key: 587 config_item = browser.find_by_name(value).first 588 config_item.select_by_text(self.ap_settings["{}_{}".format( 589 key[1], key[0])]) 590 elif "bandwidth" in key: 591 config_item = browser.find_by_name(value).first 592 try: 593 config_item.select_by_text( 594 self.bw_mode_text[self.ap_settings["{}_{}".format( 595 key[1], key[0])]]) 596 except AttributeError: 597 self.log.warning( 598 "Cannot select bandwidth. Keeping AP default.") 599 600 # Update security settings (passwords updated only if applicable) 601 for key, value in self.config_page_fields.items(): 602 if "security_type" in key: 603 browser.choose( 604 value, self.ap_settings["{}_{}".format(key[1], 605 key[0])]) 606 if self.ap_settings["{}_{}".format(key[1], 607 key[0])] == "WPA2-PSK": 608 config_item = browser.find_by_name( 609 self.config_page_fields[(key[0], 610 "password")]).first 611 config_item.fill(self.ap_settings["{}_{}".format( 612 "password", key[0])]) 613 614 # Update SSID and channel for each network 615 # NOTE: Update ordering done as such as workaround for R8000 616 # wherein channel and SSID get overwritten when some other 617 # variables are changed. However, region does have to be set before 618 # channel in all cases. 619 for key, value in self.config_page_fields.items(): 620 if "ssid" in key: 621 config_item = browser.find_by_name(value).first 622 config_item.fill(self.ap_settings["{}_{}".format( 623 key[1], key[0])]) 624 elif "channel" in key: 625 config_item = browser.find_by_name(value).first 626 try: 627 config_item.select(self.ap_settings["{}_{}".format( 628 key[1], key[0])]) 629 time.sleep(BROWSER_WAIT_SHORT) 630 except AttributeError: 631 self.log.warning( 632 "Cannot select channel. Keeping AP default.") 633 try: 634 alert = browser.get_alert() 635 alert.accept() 636 except: 637 pass 638 639 time.sleep(BROWSER_WAIT_SHORT) 640 browser.find_by_name("Apply").first.click() 641 time.sleep(BROWSER_WAIT_SHORT) 642 try: 643 alert = browser.get_alert() 644 alert.accept() 645 time.sleep(BROWSER_WAIT_SHORT) 646 except: 647 time.sleep(BROWSER_WAIT_SHORT) 648 browser.visit_persistent(self.config_page, BROWSER_WAIT_EXTRA_LONG, 649 10) 650 651 def configure_radio_on_off(self): 652 """Helper configuration function to turn radios on/off.""" 653 with BlockingBrowser(self.ap_settings["headless_browser"], 654 900) as browser: 655 # Visit URL 656 browser.visit_persistent(self.config_page, BROWSER_WAIT_MED, 10) 657 browser.visit_persistent(self.config_page_advanced, 658 BROWSER_WAIT_MED, 10) 659 660 # Turn radios on or off 661 for key, value in self.config_page_fields.items(): 662 if "status" in key: 663 config_item = browser.find_by_name(value).first 664 if self.ap_settings["{}_{}".format(key[1], key[0])]: 665 config_item.check() 666 else: 667 config_item.uncheck() 668 669 time.sleep(BROWSER_WAIT_SHORT) 670 browser.find_by_name("Apply").first.click() 671 time.sleep(BROWSER_WAIT_EXTRA_LONG) 672 browser.visit_persistent(self.config_page, BROWSER_WAIT_EXTRA_LONG, 673 10) 674 675 676class NetgearR7000NAAP(NetgearR7000AP): 677 """Class that implements Netgear R7000 NA AP.""" 678 def init_gui_data(self): 679 """Function to initialize data used while interacting with web GUI""" 680 super.init_gui_data() 681 self.region_map["11"] = "North America" 682 683 684class NetgearR7500AP(WifiRetailAP): 685 """Class that implements Netgear R7500 AP.""" 686 def __init__(self, ap_settings): 687 super().__init__(ap_settings) 688 self.init_gui_data() 689 # Read and update AP settings 690 self.read_ap_settings() 691 if not set(ap_settings.items()).issubset(self.ap_settings.items()): 692 self.update_ap_settings(ap_settings) 693 694 def init_gui_data(self): 695 """Function to initialize data used while interacting with web GUI""" 696 self.config_page = ("{protocol}://{username}:{password}@" 697 "{ip_address}:{port}/index.htm").format( 698 protocol=self.ap_settings["protocol"], 699 username=self.ap_settings["admin_username"], 700 password=self.ap_settings["admin_password"], 701 ip_address=self.ap_settings["ip_address"], 702 port=self.ap_settings["port"]) 703 self.config_page_advanced = ( 704 "{protocol}://{username}:{password}@" 705 "{ip_address}:{port}/adv_index.htm").format( 706 protocol=self.ap_settings["protocol"], 707 username=self.ap_settings["admin_username"], 708 password=self.ap_settings["admin_password"], 709 ip_address=self.ap_settings["ip_address"], 710 port=self.ap_settings["port"]) 711 self.networks = ["2G", "5G_1"] 712 self.channel_band_map = { 713 "2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11], 714 "5G_1": [ 715 36, 40, 44, 48, 52, 56, 60, 64, 100, 104, 108, 112, 116, 120, 716 124, 128, 132, 136, 140, 149, 153, 157, 161, 165 717 ] 718 } 719 self.config_page_fields = { 720 "region": "WRegion", 721 ("2G", "status"): "enable_ap", 722 ("5G_1", "status"): "enable_ap_an", 723 ("2G", "ssid"): "ssid", 724 ("5G_1", "ssid"): "ssid_an", 725 ("2G", "channel"): "w_channel", 726 ("5G_1", "channel"): "w_channel_an", 727 ("2G", "bandwidth"): "opmode", 728 ("5G_1", "bandwidth"): "opmode_an", 729 ("2G", "security_type"): "security_type", 730 ("5G_1", "security_type"): "security_type_an", 731 ("2G", "password"): "passphrase", 732 ("5G_1", "password"): "passphrase_an" 733 } 734 self.region_map = { 735 "0": "Africa", 736 "1": "Asia", 737 "2": "Australia", 738 "3": "Canada", 739 "4": "Europe", 740 "5": "Israel", 741 "6": "Japan", 742 "7": "Korea", 743 "8": "Mexico", 744 "9": "South America", 745 "10": "United States", 746 "11": "China", 747 "12": "India", 748 "13": "Malaysia", 749 "14": "Middle East(Algeria/Syria/Yemen)", 750 "15": "Middle East(Iran/Labanon/Qatar)", 751 "16": "Middle East(Turkey/Egypt/Tunisia/Kuwait)", 752 "17": "Middle East(Saudi Arabia)", 753 "18": "Middle East(United Arab Emirates)", 754 "19": "Russia", 755 "20": "Singapore", 756 "21": "Taiwan" 757 } 758 self.bw_mode_text_2g = { 759 "11g": "Up to 54 Mbps", 760 "VHT20": "Up to 289 Mbps", 761 "VHT40": "Up to 600 Mbps" 762 } 763 self.bw_mode_text_5g = { 764 "VHT20": "Up to 347 Mbps", 765 "VHT40": "Up to 800 Mbps", 766 "VHT80": "Up to 1733 Mbps" 767 } 768 self.bw_mode_values = { 769 "1": "11g", 770 "2": "VHT20", 771 "3": "VHT40", 772 "7": "VHT20", 773 "8": "VHT40", 774 "9": "VHT80" 775 } 776 777 def read_ap_settings(self): 778 """Function to read ap wireless settings.""" 779 # Get radio status (on/off) 780 self.read_radio_on_off() 781 # Get radio configuration. Note that if both radios are off, the below 782 # code will result in an error 783 with BlockingBrowser(self.ap_settings["headless_browser"], 784 900) as browser: 785 browser.visit_persistent(self.config_page, 786 BROWSER_WAIT_MED, 787 10, 788 check_for_element="wireless") 789 wireless_button = browser.find_by_id("wireless").first 790 wireless_button.click() 791 time.sleep(BROWSER_WAIT_MED) 792 793 with browser.get_iframe("formframe") as iframe: 794 for key, value in self.config_page_fields.items(): 795 if "bandwidth" in key: 796 config_item = iframe.find_by_name(value).first 797 self.ap_settings["{}_{}".format( 798 key[1], 799 key[0])] = self.bw_mode_values[config_item.value] 800 elif "region" in key: 801 config_item = iframe.find_by_name(value).first 802 self.ap_settings["region"] = self.region_map[ 803 config_item.value] 804 elif "password" in key: 805 try: 806 config_item = iframe.find_by_name(value).first 807 self.ap_settings["{}_{}".format( 808 key[1], key[0])] = config_item.value 809 self.ap_settings["{}_{}".format( 810 "security_type", key[0])] = "WPA2-PSK" 811 except: 812 self.ap_settings["{}_{}".format( 813 key[1], key[0])] = "defaultpassword" 814 self.ap_settings["{}_{}".format( 815 "security_type", key[0])] = "Disable" 816 elif ("channel" in key) or ("ssid" in key): 817 config_item = iframe.find_by_name(value).first 818 self.ap_settings["{}_{}".format( 819 key[1], key[0])] = config_item.value 820 else: 821 pass 822 return self.ap_settings.copy() 823 824 def configure_ap(self, **config_flags): 825 """Function to configure ap wireless settings.""" 826 # Turn radios on or off 827 if config_flags["status_toggled"]: 828 self.configure_radio_on_off() 829 # Configure radios 830 with BlockingBrowser(self.ap_settings["headless_browser"], 831 900) as browser: 832 browser.visit_persistent(self.config_page, 833 BROWSER_WAIT_MED, 834 10, 835 check_for_element="wireless") 836 wireless_button = browser.find_by_id("wireless").first 837 wireless_button.click() 838 time.sleep(BROWSER_WAIT_MED) 839 840 with browser.get_iframe("formframe") as iframe: 841 # Update AP region. Must be done before channel setting 842 config_item = iframe.find_by_name( 843 self.config_page_fields["region"]).first 844 config_item.select_by_text(self.ap_settings["region"]) 845 # Update wireless settings for each network 846 for key, value in self.config_page_fields.items(): 847 if "ssid" in key: 848 config_item = iframe.find_by_name(value).first 849 config_item.fill(self.ap_settings["{}_{}".format( 850 key[1], key[0])]) 851 elif "channel" in key: 852 channel_string = "0" * (int(self.ap_settings[ 853 "{}_{}".format(key[1], key[0])]) < 10) + str( 854 self.ap_settings["{}_{}".format( 855 key[1], key[0])]) + "(DFS)" * (48 < int( 856 self.ap_settings["{}_{}".format( 857 key[1], key[0])]) < 149) 858 config_item = iframe.find_by_name(value).first 859 try: 860 config_item.select_by_text(channel_string) 861 except AttributeError: 862 self.log.warning( 863 "Cannot select channel. Keeping AP default.") 864 elif key == ("2G", "bandwidth"): 865 config_item = iframe.find_by_name(value).first 866 try: 867 config_item.select_by_text( 868 str(self.bw_mode_text_2g[self.ap_settings[ 869 "{}_{}".format(key[1], key[0])]])) 870 except AttributeError: 871 self.log.warning( 872 "Cannot select bandwidth. Keeping AP default.") 873 elif key == ("5G_1", "bandwidth"): 874 config_item = iframe.find_by_name(value).first 875 try: 876 config_item.select_by_text( 877 str(self.bw_mode_text_5g[self.ap_settings[ 878 "{}_{}".format(key[1], key[0])]])) 879 except AttributeError: 880 self.log.warning( 881 "Cannot select bandwidth. Keeping AP default.") 882 # Update passwords for WPA2-PSK protected networks 883 # (Must be done after security type is selected) 884 for key, value in self.config_page_fields.items(): 885 if "security_type" in key: 886 iframe.choose( 887 value, 888 self.ap_settings["{}_{}".format(key[1], key[0])]) 889 if self.ap_settings["{}_{}".format( 890 key[1], key[0])] == "WPA2-PSK": 891 config_item = iframe.find_by_name( 892 self.config_page_fields[(key[0], 893 "password")]).first 894 config_item.fill(self.ap_settings["{}_{}".format( 895 "password", key[0])]) 896 897 apply_button = iframe.find_by_name("Apply") 898 apply_button[0].click() 899 time.sleep(BROWSER_WAIT_SHORT) 900 try: 901 alert = browser.get_alert() 902 alert.accept() 903 except: 904 pass 905 time.sleep(BROWSER_WAIT_SHORT) 906 try: 907 alert = browser.get_alert() 908 alert.accept() 909 except: 910 pass 911 time.sleep(BROWSER_WAIT_SHORT) 912 time.sleep(BROWSER_WAIT_EXTRA_LONG) 913 browser.visit_persistent(self.config_page, BROWSER_WAIT_EXTRA_LONG, 914 10) 915 916 def configure_radio_on_off(self): 917 """Helper configuration function to turn radios on/off.""" 918 with BlockingBrowser(self.ap_settings["headless_browser"], 919 900) as browser: 920 browser.visit_persistent(self.config_page, BROWSER_WAIT_MED, 10) 921 browser.visit_persistent(self.config_page_advanced, 922 BROWSER_WAIT_MED, 923 10, 924 check_for_element="advanced_bt") 925 advanced_button = browser.find_by_id("advanced_bt").first 926 advanced_button.click() 927 time.sleep(BROWSER_WAIT_MED) 928 wireless_button = browser.find_by_id("wladv").first 929 wireless_button.click() 930 time.sleep(BROWSER_WAIT_MED) 931 932 with browser.get_iframe("formframe") as iframe: 933 # Turn radios on or off 934 for key, value in self.config_page_fields.items(): 935 if "status" in key: 936 config_item = iframe.find_by_name(value).first 937 if self.ap_settings["{}_{}".format(key[1], key[0])]: 938 config_item.check() 939 else: 940 config_item.uncheck() 941 942 time.sleep(BROWSER_WAIT_SHORT) 943 browser.find_by_name("Apply").first.click() 944 time.sleep(BROWSER_WAIT_EXTRA_LONG) 945 browser.visit_persistent(self.config_page, 946 BROWSER_WAIT_EXTRA_LONG, 10) 947 948 def read_radio_on_off(self): 949 """Helper configuration function to read radio status.""" 950 with BlockingBrowser(self.ap_settings["headless_browser"], 951 900) as browser: 952 browser.visit_persistent(self.config_page, BROWSER_WAIT_MED, 10) 953 browser.visit_persistent(self.config_page_advanced, 954 BROWSER_WAIT_MED, 955 10, 956 check_for_element="advanced_bt") 957 advanced_button = browser.find_by_id("advanced_bt").first 958 advanced_button.click() 959 time.sleep(BROWSER_WAIT_SHORT) 960 wireless_button = browser.find_by_id("wladv").first 961 wireless_button.click() 962 time.sleep(BROWSER_WAIT_MED) 963 964 with browser.get_iframe("formframe") as iframe: 965 # Turn radios on or off 966 for key, value in self.config_page_fields.items(): 967 if "status" in key: 968 config_item = iframe.find_by_name(value).first 969 self.ap_settings["{}_{}".format(key[1], key[0])] = int( 970 config_item.checked) 971 972 973class NetgearR7800AP(NetgearR7500AP): 974 """Class that implements Netgear R7800 AP. 975 976 Since most of the class' implementation is shared with the R7500, this 977 class inherits from NetgearR7500AP and simply redifines config parameters 978 """ 979 def __init__(self, ap_settings): 980 super().__init__(ap_settings) 981 self.init_gui_data() 982 # Overwrite minor differences from R7500 AP 983 self.bw_mode_text_2g["VHT20"] = "Up to 347 Mbps" 984 # Read and update AP settings 985 self.read_ap_settings() 986 if not set(ap_settings.items()).issubset(self.ap_settings.items()): 987 self.update_ap_settings(ap_settings) 988 989 990class NetgearR8000AP(NetgearR7000AP): 991 """Class that implements Netgear R8000 AP. 992 993 Since most of the class' implementation is shared with the R7000, this 994 class inherits from NetgearR7000AP and simply redifines config parameters 995 """ 996 def __init__(self, ap_settings): 997 super().__init__(ap_settings) 998 self.init_gui_data() 999 # Overwrite minor differences from R7000 AP 1000 self.config_page = ( 1001 "{protocol}://{username}:{password}@" 1002 "{ip_address}:{port}/WLG_wireless_dual_band_r8000.htm").format( 1003 protocol=self.ap_settings["protocol"], 1004 username=self.ap_settings["admin_username"], 1005 password=self.ap_settings["admin_password"], 1006 ip_address=self.ap_settings["ip_address"], 1007 port=self.ap_settings["port"]) 1008 self.config_page_nologin = ( 1009 "{protocol}://{ip_address}:{port}/" 1010 "WLG_wireless_dual_band_r8000.htm").format( 1011 protocol=self.ap_settings["protocol"], 1012 ip_address=self.ap_settings["ip_address"], 1013 port=self.ap_settings["port"]) 1014 self.config_page_advanced = ( 1015 "{protocol}://{username}:{password}@" 1016 "{ip_address}:{port}/WLG_adv_dual_band2_r8000.htm").format( 1017 protocol=self.ap_settings["protocol"], 1018 username=self.ap_settings["admin_username"], 1019 password=self.ap_settings["admin_password"], 1020 ip_address=self.ap_settings["ip_address"], 1021 port=self.ap_settings["port"]) 1022 self.networks = ["2G", "5G_1", "5G_2"] 1023 self.channel_band_map = { 1024 "2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11], 1025 "5G_1": [36, 40, 44, 48], 1026 "5G_2": [149, 153, 157, 161, 165] 1027 } 1028 self.config_page_fields = { 1029 "region": "WRegion", 1030 ("2G", "status"): "enable_ap", 1031 ("5G_1", "status"): "enable_ap_an", 1032 ("5G_2", "status"): "enable_ap_an_2", 1033 ("2G", "ssid"): "ssid", 1034 ("5G_1", "ssid"): "ssid_an", 1035 ("5G_2", "ssid"): "ssid_an_2", 1036 ("2G", "channel"): "w_channel", 1037 ("5G_1", "channel"): "w_channel_an", 1038 ("5G_2", "channel"): "w_channel_an_2", 1039 ("2G", "bandwidth"): "opmode", 1040 ("5G_1", "bandwidth"): "opmode_an", 1041 ("5G_2", "bandwidth"): "opmode_an_2", 1042 ("2G", "security_type"): "security_type", 1043 ("5G_1", "security_type"): "security_type_an", 1044 ("5G_2", "security_type"): "security_type_an_2", 1045 ("2G", "password"): "passphrase", 1046 ("5G_1", "password"): "passphrase_an", 1047 ("5G_2", "password"): "passphrase_an_2" 1048 } 1049 # Read and update AP settings 1050 self.read_ap_settings() 1051 if not set(ap_settings.items()).issubset(self.ap_settings.items()): 1052 self.update_ap_settings(ap_settings) 1053 1054 1055class NetgearR8500AP(NetgearR7000AP): 1056 """Class that implements Netgear R8500 AP. 1057 1058 Since most of the class' implementation is shared with the R7000, this 1059 class inherits from NetgearR7000AP and simply redifines config parameters 1060 """ 1061 def __init__(self, ap_settings): 1062 super().__init__(ap_settings) 1063 self.init_gui_data() 1064 # Overwrite minor differences from R8000 AP 1065 self.config_page = ( 1066 "{protocol}://{username}:{password}@" 1067 "{ip_address}:{port}/WLG_wireless_tri_band.htm").format( 1068 protocol=self.ap_settings["protocol"], 1069 username=self.ap_settings["admin_username"], 1070 password=self.ap_settings["admin_password"], 1071 ip_address=self.ap_settings["ip_address"], 1072 port=self.ap_settings["port"]) 1073 self.config_page_nologin = ( 1074 "{protocol}://{ip_address}:{port}/" 1075 "WLG_wireless_tri_band.htm").format( 1076 protocol=self.ap_settings["protocol"], 1077 ip_address=self.ap_settings["ip_address"], 1078 port=self.ap_settings["port"]) 1079 self.config_page_advanced = ( 1080 "{protocol}://{username}:{password}@" 1081 "{ip_address}:{port}/WLG_adv_tri_band2.htm").format( 1082 protocol=self.ap_settings["protocol"], 1083 username=self.ap_settings["admin_username"], 1084 password=self.ap_settings["admin_password"], 1085 ip_address=self.ap_settings["ip_address"], 1086 port=self.ap_settings["port"]) 1087 self.networks = ["2G", "5G_1", "5G_2"] 1088 self.channel_band_map = { 1089 "2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11], 1090 "5G_1": [36, 40, 44, 48], 1091 "5G_2": [149, 153, 157, 161, 165] 1092 } 1093 self.config_page_fields = { 1094 "region": "WRegion", 1095 ("2G", "status"): "enable_ap", 1096 ("5G_1", "status"): "enable_ap_an", 1097 ("5G_2", "status"): "enable_ap_an_2", 1098 ("2G", "ssid"): "ssid", 1099 ("5G_1", "ssid"): "ssid_an", 1100 ("5G_2", "ssid"): "ssid_an_2", 1101 ("2G", "channel"): "w_channel", 1102 ("5G_1", "channel"): "w_channel_an", 1103 ("5G_2", "channel"): "w_channel_an_2", 1104 ("2G", "bandwidth"): "opmode", 1105 ("5G_1", "bandwidth"): "opmode_an", 1106 ("5G_2", "bandwidth"): "opmode_an_2", 1107 ("2G", "security_type"): "security_type", 1108 ("5G_1", "security_type"): "security_type_an", 1109 ("5G_2", "security_type"): "security_type_an_2", 1110 ("2G", "password"): "passphrase", 1111 ("5G_1", "password"): "passphrase_an", 1112 ("5G_2", "password"): "passphrase_an_2" 1113 } 1114 self.bw_mode_text = { 1115 "11g": "Up to 54 Mbps", 1116 "VHT20": "Up to 433 Mbps", 1117 "VHT40": "Up to 1000 Mbps", 1118 "VHT80": "Up to 2165 Mbps" 1119 } 1120 # Read and update AP settings 1121 self.read_ap_settings() 1122 if not set(ap_settings.items()).issubset(self.ap_settings.items()): 1123 self.update_ap_settings(ap_settings) 1124 1125 1126class GoogleWifiAP(WifiRetailAP): 1127 """ Class that implements Google Wifi AP. 1128 1129 This class is a work in progress 1130 """ 1131 def __init__(self, ap_settings): 1132 super().__init__(ap_settings) 1133 # Initialize AP 1134 if self.ap_settings["status_2G"] and self.ap_settings["status_5G_1"]: 1135 raise ValueError("Error initializing Google Wifi AP. " 1136 "Only one interface can be enabled at a time.") 1137 self.channel_band_map = { 1138 "2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11], 1139 "5G_1": [36, 40, 44, 48, 149, 153, 157, 161, 165] 1140 } 1141 self.BW_MODE_MAP = { 1142 "legacy": 20, 1143 "VHT20": 20, 1144 "VHT40": 40, 1145 "VHT80": 80 1146 } 1147 self.default_settings = { 1148 "region": "United States", 1149 "brand": "Google", 1150 "model": "Wifi", 1151 "hostapd_profile": "whirlwind", 1152 "status_2G": 0, 1153 "status_5G_1": 0, 1154 "ssid_2G": "GoogleWifi_2G", 1155 "ssid_5G_1": "GoogleWifi_5G", 1156 "channel_2G": 11, 1157 "channel_5G_1": 149, 1158 "bandwidth_2G": "VHT20", 1159 "bandwidth_5G_1": "VHT20", 1160 "power_2G": "auto", 1161 "power_5G_1": "auto", 1162 "security_type_2G": "Open", 1163 "security_type_5G_1": "Open", 1164 "subnet_2G": "192.168.1.0/24", 1165 "subnet_5G_1": "192.168.9.0/24", 1166 "password_2G": "password", 1167 "password_5G_1": "password" 1168 } 1169 1170 for setting in self.default_settings.keys(): 1171 if setting not in self.ap_settings: 1172 self.log.warning( 1173 "{0} not found during init. Setting {0} = {1}".format( 1174 setting, self.default_settings[setting])) 1175 self.ap_settings[setting] = self.default_settings[setting] 1176 init_settings = self.ap_settings.copy() 1177 init_settings["ap_subnet"] = { 1178 "2g": self.ap_settings["subnet_2G"], 1179 "5g": self.ap_settings["subnet_5G_1"] 1180 } 1181 self.access_point = access_point.AccessPoint(init_settings) 1182 self.configure_ap() 1183 1184 def read_ap_settings(self): 1185 """Function that reads current ap settings.""" 1186 return self.ap_settings.copy() 1187 1188 def update_ap_settings(self, dict_settings={}, **named_settings): 1189 """Function to update settings of existing AP. 1190 1191 Function copies arguments into ap_settings and calls configure_ap 1192 to apply them. 1193 1194 Args: 1195 dict_settings: single dictionary of settings to update 1196 **named_settings: named settings to update 1197 Note: dict and named_settings cannot contain the same settings. 1198 """ 1199 settings_to_update = dict(dict_settings, **named_settings) 1200 if len(settings_to_update) != len(dict_settings) + len(named_settings): 1201 raise KeyError("The following keys were passed twice: {}".format( 1202 (set(dict_settings.keys()).intersection( 1203 set(named_settings.keys()))))) 1204 if not set(settings_to_update.keys()).issubset( 1205 set(self.ap_settings.keys())): 1206 raise KeyError( 1207 "The following settings are invalid for this AP: {}".format( 1208 set(settings_to_update.keys()).difference( 1209 set(self.ap_settings.keys())))) 1210 1211 updating_2G = any(["2G" in x for x in settings_to_update.keys()]) 1212 updating_5G_1 = any(["5G_1" in x for x in settings_to_update.keys()]) 1213 if updating_2G and updating_5G_1: 1214 raise ValueError( 1215 "Error updating Google WiFi AP. " 1216 "One interface can be activated and updated at a time") 1217 elif updating_2G: 1218 # If updating an interface and not explicitly setting its status, 1219 # it is assumed that the interface is to be ENABLED and updated 1220 if "status_2G" not in settings_to_update: 1221 settings_to_update["status_2G"] = 1 1222 settings_to_update["status_5G_1"] = 0 1223 elif updating_5G_1: 1224 if "status_5G_1" not in settings_to_update: 1225 settings_to_update["status_2G"] = 0 1226 settings_to_update["status_5G_1"] = 1 1227 1228 updates_requested = False 1229 for setting, value in settings_to_update.items(): 1230 if self.ap_settings[setting] != value: 1231 self.ap_settings[setting] = value 1232 updates_requested = True 1233 1234 if updates_requested: 1235 self.configure_ap() 1236 1237 def configure_ap(self): 1238 """Function to configure Google Wifi.""" 1239 self.log.info("Stopping Google Wifi interfaces.") 1240 self.access_point.stop_all_aps() 1241 1242 if self.ap_settings["status_2G"] == 1: 1243 network = "2G" 1244 self.log.info("Bringing up 2.4 GHz network.") 1245 elif self.ap_settings["status_5G_1"] == 1: 1246 network = "5G_1" 1247 self.log.info("Bringing up 5 GHz network.") 1248 else: 1249 return 1250 1251 bss_settings = [] 1252 ssid = self.ap_settings["ssid_{}".format(network)] 1253 security_mode = self.ap_settings["security_type_{}".format( 1254 network)].lower() 1255 if "wpa" in security_mode: 1256 password = self.ap_settings["password_{}".format(network)] 1257 security = hostapd_security.Security(security_mode=security_mode, 1258 password=password) 1259 else: 1260 security = hostapd_security.Security(security_mode=None, 1261 password=None) 1262 channel = int(self.ap_settings["channel_{}".format(network)]) 1263 bandwidth = self.BW_MODE_MAP[self.ap_settings["bandwidth_{}".format( 1264 network)]] 1265 config = hostapd_ap_preset.create_ap_preset( 1266 channel=channel, 1267 ssid=ssid, 1268 security=security, 1269 bss_settings=bss_settings, 1270 vht_bandwidth=bandwidth, 1271 profile_name=self.ap_settings["hostapd_profile"], 1272 iface_wlan_2g=self.access_point.wlan_2g, 1273 iface_wlan_5g=self.access_point.wlan_5g) 1274 config_bridge = self.access_point.generate_bridge_configs(channel) 1275 brconfigs = bridge_interface.BridgeInterfaceConfigs( 1276 config_bridge[0], "lan0", config_bridge[2]) 1277 self.access_point.bridge.startup(brconfigs) 1278 self.access_point.start_ap(config) 1279 self.set_power(network, self.ap_settings["power_{}".format(network)]) 1280 self.log.info("AP started on channel {} with SSID {}".format( 1281 channel, ssid)) 1282 1283 def set_power(self, network, power): 1284 """Function that sets network transmit power. 1285 1286 Args: 1287 network: string containing network identifier (2G, 5G_1, 5G_2) 1288 power: power level in dBm 1289 """ 1290 if power == "auto": 1291 power_string = "auto" 1292 else: 1293 if not float(power).is_integer(): 1294 self.log.info( 1295 "Power in dBm must be an integer. Setting to {}".format( 1296 int(power))) 1297 power = int(power) 1298 power_string = "fixed {}".format(int(power) * 100) 1299 1300 if "2G" in network: 1301 interface = self.access_point.wlan_2g 1302 self.ap_settings["power_2G"] = power 1303 elif "5G_1" in network: 1304 interface = self.access_point.wlan_5g 1305 self.ap_settings["power_5G_1"] = power 1306 self.access_point.ssh.run("iw dev {} set txpower {}".format( 1307 interface, power_string)) 1308 1309 def set_rate(self, 1310 network, 1311 mode=None, 1312 num_streams=None, 1313 rate=None, 1314 short_gi=0): 1315 """Function that sets rate. 1316 1317 Args: 1318 network: string containing network identifier (2G, 5G_1, 5G_2) 1319 mode: string indicating the WiFi standard to use 1320 num_streams: number of MIMO streams. used only for VHT 1321 rate: data rate of MCS index to use 1322 short_gi: boolean controlling the use of short guard interval 1323 """ 1324 if "2G" in network: 1325 interface = self.access_point.wlan_2g 1326 interface_short = "2.4" 1327 elif "5G_1" in network: 1328 interface = self.access_point.wlan_5g 1329 interface_short = "5" 1330 1331 if "legacy" in mode.lower(): 1332 cmd_string = "iw dev {0} set bitrates legacy-{1} {2} ht-mcs-{1} vht-mcs-{1}".format( 1333 interface, interface_short, rate) 1334 elif "vht" in mode.lower(): 1335 cmd_string = "iw dev {0} set bitrates legacy-{1} ht-mcs-{1} vht-mcs-{1} {2}:{3}".format( 1336 interface, interface_short, num_streams, rate) 1337 if short_gi: 1338 cmd_string = cmd_string + " sgi-{}".format(interface_short) 1339 elif "ht" in mode.lower(): 1340 cmd_string = "iw dev {0} set bitrates legacy-{1} ht-mcs-{1} {2} vht-mcs-{1}".format( 1341 interface, interface_short, rate) 1342 if short_gi: 1343 cmd_string = cmd_string + " sgi-{}".format(interface_short) 1344 self.access_point.ssh.run(cmd_string) 1345