xref: /aosp_15_r20/external/autotest/server/cros/ap_configurators/static_ap_configurator.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)
1# Lint as: python2, python3
2# Copyright (c) 2013 The Chromium Authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6import collections
7import logging
8import pprint
9import re
10
11from six.moves import xmlrpc_client as xmlrpclib
12from autotest_lib.client.common_lib import global_config
13from autotest_lib.client.common_lib.cros.network import ap_constants
14from autotest_lib.client.common_lib.cros.network import xmlrpc_datatypes
15from autotest_lib.client.common_lib.cros.network import xmlrpc_security_types
16from autotest_lib.server.cros.ap_configurators import ap_configurator
17from autotest_lib.server.cros.ap_configurators import ap_spec
18
19CartridgeCmd = collections.namedtuple('CartridgeCmd', ['method', 'args'])
20RPM_FRONTEND_SERVER = global_config.global_config.get_config_value(
21        'CROS', 'rpm_frontend_uri')
22
23# DHCP delayed devices.  Some APs need additional time for the DHCP
24# server to come on-line.  These are AP based, so the BSS is used
25# since that is unique.
26DHCP_DELAY_DEVICES=['44:94:fc:71:88:9b', # Netgear wndr4300
27                    '10:0d:7f:4d:68:3c', # Netgear wndr3700v4
28                    '14:35:8b:0b:6c:80', # Medialink mwn_wapr150nv2
29                    '20:4e:7f:49:86:8f'] # Netgear wpn824n
30
31class StaticAPConfigurator(ap_configurator.APConfiguratorAbstract):
32    """Derived class to supply AP configuration information."""
33
34
35    def __init__(self, ap_config):
36        """
37        Initialize instance
38
39        @param ap_config: AP object to configure this instance
40
41        """
42        super(StaticAPConfigurator, self).__init__()
43        self._command_list = list()
44
45        # This allows the ability to build a generic configurator
46        # which can be used to get access to the members above.
47        self.class_name = ap_config.get_class()
48        self._short_name = ap_config.get_model()
49        self.mac_address = ap_config.get_wan_mac()
50        self.host_name = ap_config.get_wan_host()
51        # Get corresponding PDU from host name.
52        self.pdu = re.sub('host\\d+', 'rpm1', self.host_name) + '.cros'
53        self.channel = ap_config.get_channel()
54        self.band = ap_config.get_band()
55        self.current_band = ap_config.get_band()
56        self.security = ap_config.get_security()
57        if self.security == ap_spec.SECURITY_TYPE_MIXED:
58           self.security = [ap_spec.SECURITY_TYPE_WPA2PSK,
59                            ap_spec.SECURITY_TYPE_WPAPSK]
60        else:
61           self.security = [self.security]
62        self.psk = ap_config.get_psk()
63        self._ssid = ap_config.get_ssid()
64        self.rpm_unit = ap_config.get_rpm_unit()
65
66        self._configuration_success = ap_constants.CONFIG_SUCCESS
67        self.config_data = ap_config
68
69        name_dict = {'Router name': self._short_name,
70                     'Controller class': self.class_name,
71                     '2.4 GHz MAC Address': ap_config.get_bss(),
72                     '5 GHz MAC Address': ap_config.get_bss5(),
73                     'Hostname': ap_config.get_wan_host()}
74
75        self._name = pprint.pformat(name_dict)
76
77        # Check if a delay needs to be added for this AP.
78        if (ap_config.get_bss() in DHCP_DELAY_DEVICES or
79            ap_config.get_bss5() in DHCP_DELAY_DEVICES):
80            self._dhcp_delay = 60
81
82        self.rpm_client = xmlrpclib.ServerProxy(RPM_FRONTEND_SERVER,
83                                                verbose=False,
84                                                allow_none=True)
85
86
87    def __str__(self):
88        """Prettier display of the object"""
89        return('AP Name: %s\n'
90               'BSS: %s\n'
91               'SSID: %s\n'
92               'Short name: %s' % (self.name,
93                   self.config_data.get_bss(), self._ssid,
94                   self.short_name))
95
96
97    @property
98    def ssid(self):
99        """Returns the SSID."""
100        return self._ssid
101
102
103    def power_down_router(self):
104        """power down via rpm"""
105        self._append_rpm_command('OFF')
106
107
108    def power_up_router(self):
109        """power up via rpm"""
110        self._append_rpm_command('ON')
111
112
113    def _append_rpm_command(self, command):
114        if self.rpm_unit is None:
115            return
116
117        self._command_list.append(CartridgeCmd(
118                self.rpm_client.set_power_via_rpm,
119                [
120                        self.host_name,
121                        self.rpm_unit.hostname,
122                        self.rpm_unit.outlet,
123                        None,
124                        command,
125                ],
126        ))
127
128
129    def set_using_ap_spec(self, set_ap_spec, power_up=True):
130        """
131        Sets all configurator options.
132
133        Note: for StaticAPs there is no config required, so the only action
134        here is to power up if needed
135
136        @param set_ap_spec: APSpec object
137
138        """
139        if power_up:
140            self.power_up_router()
141
142
143    def apply_settings(self):
144        """Allow cartridge to run commands in _command_list"""
145        self.check_pdu_status()
146        for command in self._command_list:
147            logging.debug("Command to run method: %s", command.method.__name__)
148            logging.debug("Command to run with args: %s", str(command.args))
149            command.method(*command.args)
150
151
152    def reset_command_list(self):
153        """Resets all internal command state."""
154        self._command_list = list()
155
156
157    @property
158    def name(self):
159        """Returns a string to describe the router."""
160        return self._name
161
162
163    @property
164    def short_name(self):
165        """Returns a short string to describe the router."""
166        return self._short_name
167
168
169    def get_supported_bands(self):
170        """Returns a list of dictionaries describing the supported bands.
171
172        Example: returned is a dictionary of band and a list of channels. The
173                 band object returned must be one of those defined in the
174                 __init___ of this class.
175
176        supported_bands = [{'band' : self.band_2GHz,
177                            'channels' : [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]},
178                           {'band' : self.band_5ghz,
179                            'channels' : [26, 40, 44, 48, 149, 153, 165]}]
180
181        @return a list of dictionaries as described above
182
183        """
184        supported_bands = [{'band' : self.band,
185                            'channels' : [self.channel]}]
186
187        return supported_bands
188
189
190    def get_supported_modes(self):
191        """
192        Returns a list of dictionaries describing the supported modes.
193
194        Example: returned is a dictionary of band and a list of modes. The band
195                 and modes objects returned must be one of those defined in the
196                 __init___ of this class.
197
198        supported_modes = [{'band' : ap_spec.BAND_2GHZ,
199                            'modes' : [mode_b, mode_b | mode_g]},
200                           {'band' : ap_spec.BAND_5GHZ,
201                            'modes' : [mode_a, mode_n, mode_a | mode_n]}]
202
203        @return a list of dictionaries as described above
204
205        """
206        supported_modes = [{'band' : self.band,
207                            'modes' : [ap_spec.DEFAULT_5GHZ_MODE
208                    if self.channel in ap_spec.VALID_5GHZ_CHANNELS
209                    else ap_spec.DEFAULT_2GHZ_MODE]}]
210
211        return supported_modes
212
213
214    def is_visibility_supported(self):
215        """
216        Returns if AP supports setting the visibility (SSID broadcast).
217
218        @return False
219
220        """
221        return False
222
223
224    def is_band_and_channel_supported(self, band, channel):
225        """
226        Returns if a given band and channel are supported.
227
228        @param band: the band to check if supported
229        @param channel: the channel to check if supported
230
231        @return True if combination is supported; False otherwise.
232
233        """
234        bands = self.get_supported_bands()
235        for current_band in bands:
236            if (current_band['band'] == band and
237                channel in current_band['channels']):
238                return True
239        return False
240
241
242    def is_security_mode_supported(self, security_mode):
243        """
244        Returns if a given security_type is supported.
245
246        @param security_mode: one of the following modes:
247                         self.security_disabled,
248                         self.security_wep,
249                         self.security_wpapsk,
250                         self.security_wpa2psk
251
252        @return True if the security mode is supported; False otherwise.
253
254        """
255        return security_mode in self.security
256
257
258    def get_association_parameters(self):
259        """
260        Creates an AssociationParameters from the configured AP.
261
262        @returns AssociationParameters for the configured AP.
263
264        """
265        security_config = None
266        if (ap_spec.SECURITY_TYPE_WPAPSK in self.security or
267            ap_spec.SECURITY_TYPE_WPA2PSK in self.security):
268            # Not all of this is required but doing it just in case.
269            security_config = xmlrpc_security_types.WPAConfig(
270                    psk=self.psk,
271                    wpa_mode=xmlrpc_security_types.WPAConfig.MODE_MIXED_WPA,
272                    wpa_ciphers=[xmlrpc_security_types.WPAConfig.CIPHER_CCMP,
273                                 xmlrpc_security_types.WPAConfig.CIPHER_TKIP],
274                    wpa2_ciphers=[xmlrpc_security_types.WPAConfig.CIPHER_CCMP])
275        # TODO(jabele) Allow StaticAPs configured as hidden
276        #              by way of the ap_config file
277        return xmlrpc_datatypes.AssociationParameters(
278                ssid=self._ssid, security_config=security_config,
279                discovery_timeout=45, association_timeout=30,
280                configuration_timeout=30, is_hidden=False)
281