blob: 676a20610b30817043f13beb83deeeb8977e7e3e [file] [log] [blame]
#!/usr/bin/env python3
#
# Copyright 2022 The Fuchsia Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import multiprocessing
import time
from dataclasses import dataclass
from typing import Any, Tuple
from antlion import logger, signals
from antlion.controllers.ap_lib.hostapd_security import FuchsiaSecurityType
from antlion.controllers.fuchsia_lib.sl4f import SL4F
from antlion.controllers.fuchsia_lib.ssh import FuchsiaSSHProvider
from antlion.controllers.fuchsia_lib.wlan_lib import FuchsiaWlanLib
SAVED_NETWORKS = "saved_networks"
CLIENT_STATE = "client_connections_state"
CONNECTIONS_ENABLED = "ConnectionsEnabled"
CONNECTIONS_DISABLED = "ConnectionsDisabled"
STATE_CONNECTED = "Connected"
STATE_CONNECTING = "Connecting"
STATE_DISCONNECTED = "Disconnected"
STATE_CONNECTION_STOPPED = "ConnectionStopped"
SESSION_MANAGER_TIMEOUT_SEC = 10
FUCHSIA_DEFAULT_WLAN_CONFIGURE_TIMEOUT = 30
DEFAULT_GET_UPDATE_TIMEOUT = 60
class WlanPolicyControllerError(signals.ControllerError):
pass
@dataclass
class PreservedState:
saved_networks: list[dict[str, Any]] | None
client_connections_state: str | None
@dataclass
class GetCountryCodeFromPhyParams:
wlan_lib: FuchsiaWlanLib
id: str
def get_country_code_from_phy(params: GetCountryCodeFromPhyParams) -> Tuple[str, str]:
get_country_response = params.wlan_lib.wlanGetCountry(params.id)
if get_country_response.get("error"):
raise ConnectionError(
f"Failed to query PHY ID ({params.id}) for country. "
f'Error: {get_country_response["error"]}'
)
country_code = "".join(
[chr(ascii_char) for ascii_char in get_country_response["result"]]
)
return (params.id, country_code)
class WlanPolicyController:
"""Contains methods related to the wlan policy layer, to be used in the
FuchsiaDevice object.
Attributes:
sl4f: sl4f module for communicating to the WLAN policy controller.
ssh: transport to fuchsia device to stop component processes.
"""
def __init__(self, sl4f: SL4F, ssh: FuchsiaSSHProvider):
"""
Args:
sl4f: sl4f module for communicating to the WLAN policy controller.
ssh: transport to fuchsia device to stop component processes.
"""
self.preserved_networks_and_client_state: PreservedState | None = None
self.policy_configured = False
self.sl4f = sl4f
self.ssh = ssh
self.log = logger.create_tagged_trace_logger(
f"WlanPolicyController | {self.ssh.config.host_name}"
)
def configure_wlan(
self,
preserve_saved_networks: bool,
timeout_sec: int = FUCHSIA_DEFAULT_WLAN_CONFIGURE_TIMEOUT,
) -> None:
"""Sets up wlan policy layer.
Args:
preserve_saved_networks: whether to clear existing saved
networks and client state, to be restored at test close.
timeout_sec: time to wait for device to configure WLAN.
"""
# We need to stop session manager to free control of
# fuchsia.wlan.policy.ClientController, which can only be used by a
# single caller at a time. SL4F needs the ClientController to trigger
# WLAN policy state changes. On eng builds the session_manager can be
# restarted after being stopped during reboot so we attempt killing the
# session manager process for 10 seconds.
# See https://cs.opensource.google/fuchsia/fuchsia/+/main:sdk/fidl/fuchsia.wlan.policy/client_provider.fidl
if "cast_agent.cm" in self.ssh.run("ps").stdout:
end_time_session_manager_sec = time.time() + SESSION_MANAGER_TIMEOUT_SEC
while time.time() < end_time_session_manager_sec:
self.ssh.stop_component("session_manager", is_cfv2_component=True)
# Acquire control of policy layer
end_time_config_sec = time.time() + timeout_sec
controller_errors = []
while time.time() < end_time_config_sec:
# Create a client controller
response = self.sl4f.wlan_policy_lib.wlanCreateClientController()
if response.get("error"):
controller_errors.append(response["error"])
self.log.debug(response["error"])
time.sleep(1)
continue
break
else:
self.log.warning(
"Failed to create and use a WLAN policy client controller. Errors: ["
+ "; ".join(controller_errors)
+ "]"
)
raise WlanPolicyControllerError(
"Failed to create and use a WLAN policy client controller."
)
self.log.info("ACTS tests now have control of the WLAN policy layer.")
if preserve_saved_networks and not self.preserved_networks_and_client_state:
self.preserved_networks_and_client_state = (
self.remove_and_preserve_networks_and_client_state()
)
if not self.start_client_connections():
raise WlanPolicyControllerError(
"Failed to start client connections during configuration."
)
self.policy_configured = True
def _deconfigure_wlan(self) -> None:
if not self.stop_client_connections():
raise WlanPolicyControllerError(
"Failed to stop client connections during deconfiguration."
)
self.policy_configured = False
def clean_up(self) -> None:
if self.preserved_networks_and_client_state is not None:
# It is possible for policy to have been configured before, but
# deconfigured before test end. In this case, in must be setup
# before restoring networks
if not self.policy_configured:
self.configure_wlan(False)
self.restore_preserved_networks_and_client_state()
def start_client_connections(self) -> bool:
"""Allow device to connect to networks via policy layer (including
autoconnecting to saved networks).
Returns:
True, if successful. False otherwise."""
start_response = self.sl4f.wlan_policy_lib.wlanStartClientConnections()
if start_response.get("error"):
self.log.error(
f"Failed to start client connections. Err: {start_response['error']}"
)
return False
return True
def stop_client_connections(self) -> bool:
"""Prevent device from connecting and autoconnecting to networks via the
policy layer.
Returns:
True, if successful. False otherwise."""
stop_response = self.sl4f.wlan_policy_lib.wlanStopClientConnections()
if stop_response.get("error"):
self.log.error(
f"Failed to stop client connections. Err: {stop_response['error']}"
)
return False
return True
def save_and_connect(
self,
ssid: str,
security: FuchsiaSecurityType,
password: str | None = None,
timeout: int = 30,
):
"""Saves and connects to the network. This is the policy version of
connect and check_connect_response because the policy layer
requires a saved network and the policy connect does not return
success or failure
Args:
ssid: string, the network name
security: string, security type of network (see sl4f.wlan_policy_lib)
password: string, the credential of the network if applicable
timeout: int, time in seconds to wait for connection
Returns:
True, if successful. False otherwise.
"""
# Save network and check response
if not self.save_network(ssid, security, password=password):
return False
# Make connect call and check response
self.sl4f.wlan_policy_lib.wlanSetNewListener()
if not self.send_connect_command(ssid, security):
return False
return self.wait_for_connect(ssid, security, timeout=timeout)
def save_and_wait_for_autoconnect(
self,
ssid: str,
security: FuchsiaSecurityType,
password: str | None = None,
timeout: int = 30,
) -> bool:
"""Saves a network and waits, expecting an autoconnection to the newly
saved network. This differes from save_and_connect, as it doesn't
expressly trigger a connection first. There are cases in which an
autoconnect won't occur after a save (like if the device is connected
already), so this should be used with caution to test very specific
situations.
Args:
ssid: string, the network name
security: string, security type of network (see sl4f.wlan_policy_lib)
password: string, the credential of the network if applicable
timeout: int, time in seconds to wait for connection
Returns:
True, if successful. False otherwise.
"""
if not self.save_network(ssid, security, password=password):
return False
return self.wait_for_connect(ssid, security, timeout=timeout)
def remove_and_wait_for_disconnect(
self,
ssid: str,
security_type: FuchsiaSecurityType,
password: str | None = None,
state: str | None = None,
status: str | None = None,
timeout: int = 30,
) -> bool:
"""Removes a single network and waits for a disconnect. It is not
guaranteed the device will stay disconnected, as it may autoconnect
to a different saved network.
Args:
ssid: string, the network name
security: string, security type of network (see sl4f.wlan_policy_lib)
password: string, the credential of the network if applicable
state: string, The connection state we are expecting, ie "Disconnected" or
"Failed"
status: string, The disconnect status we expect, it "ConnectionStopped" or
"ConnectionFailed"
timeout: int, time in seconds to wait for connection
Returns:
True, if successful. False otherwise.
"""
self.sl4f.wlan_policy_lib.wlanSetNewListener()
if not self.remove_network(ssid, security_type, password=password):
return False
return self.wait_for_disconnect(
ssid, security_type, state=state, status=status, timeout=timeout
)
def remove_all_networks_and_wait_for_no_connections(
self, timeout_sec: int = DEFAULT_GET_UPDATE_TIMEOUT
) -> bool:
"""Removes all networks and waits until device is not connected to any
networks. This should be used as the policy version of disconnect.
Args:
timeout_sec: The time to wait to see no connections.
Returns:
True, if successful. False otherwise.
"""
self.sl4f.wlan_policy_lib.wlanSetNewListener()
if not self.remove_all_networks():
self.log.error(
"Failed to remove all networks. Cannot continue to "
"wait_for_no_connections."
)
return False
return self.wait_for_no_connections(timeout_sec=timeout_sec)
def save_network(
self,
ssid: str,
security_type: FuchsiaSecurityType,
password: str | None = None,
) -> bool:
"""Save a network via the policy layer.
Args:
ssid: string, the network name
security: string, security type of network (see sl4f.wlan_policy_lib)
password: string, the credential of the network if applicable
Returns:
True, if successful. False otherwise.
"""
save_response = self.sl4f.wlan_policy_lib.wlanSaveNetwork(
ssid, security_type, target_pwd=password
)
if save_response.get("error"):
self.log.error(
f"Failed to save network {ssid} with error: {save_response['error']}"
)
return False
return True
def remove_network(
self,
ssid: str,
security_type: FuchsiaSecurityType,
password: str | None = None,
) -> bool:
"""Remove a saved network via the policy layer.
Args:
ssid: string, the network name
security: string, security type of network (see sl4f.wlan_policy_lib)
password: string, the credential of the network if applicable
Returns:
True, if successful. False otherwise.
"""
remove_response = self.sl4f.wlan_policy_lib.wlanRemoveNetwork(
ssid, security_type, target_pwd=password
)
if remove_response.get("error"):
self.log.error(
"Failed to remove network %s with error: %s"
% (ssid, remove_response["error"])
)
return False
return True
def remove_all_networks(self) -> bool:
"""Removes all saved networks from device.
Returns:
True, if successful. False otherwise.
"""
remove_all_response = self.sl4f.wlan_policy_lib.wlanRemoveAllNetworks()
if remove_all_response.get("error"):
self.log.error(
f"Error occurred removing all networks: {remove_all_response['error']}"
)
return False
return True
def get_saved_networks(self) -> list[Any]:
"""Retrieves saved networks from device.
Returns:
list of saved networks
Raises:
WlanPolicyControllerError, if retrieval fails.
"""
saved_networks_response = self.sl4f.wlan_policy_lib.wlanGetSavedNetworks()
if saved_networks_response.get("error"):
raise WlanPolicyControllerError(
f"Failed to retrieve saved networks: {saved_networks_response['error']}"
)
return saved_networks_response["result"]
def send_connect_command(
self, ssid: str, security_type: FuchsiaSecurityType
) -> bool:
"""Sends a connect command to a network that is already saved. This does
not wait to guarantee the connection is successful (for that, use
save_and_connect).
Args:
ssid: string, the network name
security: string, security type of network (see sl4f.wlan_policy_lib)
password: string, the credential of the network if applicable
Returns:
True, if command send successfully. False otherwise.
"""
connect_response = self.sl4f.wlan_policy_lib.wlanConnect(ssid, security_type)
if connect_response.get("error"):
self.log.error(
"Error occurred when sending policy connect command: %s"
% connect_response["error"]
)
return False
return True
def wait_for_connect(
self, ssid: str, security_type: FuchsiaSecurityType, timeout: int = 30
) -> bool:
"""Wait until the device has connected to the specified network.
Args:
ssid: string, the network name
security: string, security type of network (see sl4f.wlan_policy_lib)
timeout: int, seconds to wait for a update showing connection
Returns:
True if we see a connect to the network, False otherwise.
"""
# Wait until we've connected.
end_time = time.time() + timeout
while time.time() < end_time:
time_left = max(1, int(end_time - time.time()))
try:
update = self.sl4f.wlan_policy_lib.wlanGetUpdate(timeout=time_left)
except TimeoutError:
self.log.error(
"Timed out waiting for response from device "
'while waiting for network with SSID "%s" to '
"connect. Device took too long to connect or "
"the request timed out for another reason." % ssid
)
self.sl4f.wlan_policy_lib.wlanSetNewListener()
return False
if update.get("error"):
# This can occur for many reasons, so it is not necessarily a
# failure.
self.log.debug(
f"Error occurred getting status update: {update['error']}"
)
continue
for network in update["result"]["networks"]:
if network["id"]["ssid"] == ssid or network["id"][
"type_"
].lower() == str(security_type):
if "state" not in network:
raise WlanPolicyControllerError(
"WLAN status missing state field."
)
elif network["state"].lower() == STATE_CONNECTED.lower():
return True
# Wait a bit before requesting another status update
time.sleep(1)
# Stopped getting updates because out timeout
self.log.error(f'Timed out waiting for network with SSID "{ssid}" to connect')
return False
def wait_for_disconnect(
self,
ssid: str,
security_type: FuchsiaSecurityType,
state: str | None = None,
status: str | None = None,
timeout: int = 30,
) -> bool:
"""Wait for a disconnect of the specified network on the given device. This
will check that the correct connection state and disconnect status are
given in update. If we do not see a disconnect after some time,
return false.
Args:
ssid: string, the network name
security: string, security type of network (see sl4f.wlan_policy_lib)
state: string, The connection state we are expecting, ie "Disconnected" or
"Failed"
status: string, The disconnect status we expect, it "ConnectionStopped" or
"ConnectionFailed"
timeout: int, seconds to wait before giving up
Returns: True if we saw a disconnect as specified, or False otherwise.
"""
if not state:
state = STATE_DISCONNECTED
if not status:
status = STATE_CONNECTION_STOPPED
end_time = time.time() + timeout
while time.time() < end_time:
time_left = max(1, int(end_time - time.time()))
try:
update = self.sl4f.wlan_policy_lib.wlanGetUpdate(timeout=time_left)
except TimeoutError:
self.log.error(
"Timed out waiting for response from device "
'while waiting for network with SSID "%s" to '
"disconnect. Device took too long to disconnect "
"or the request timed out for another reason." % ssid
)
self.sl4f.wlan_policy_lib.wlanSetNewListener()
return False
if update.get("error"):
# This can occur for many reasons, so it is not necessarily a
# failure.
self.log.debug(
f"Error occurred getting status update: {update['error']}"
)
continue
# Update should include network, either connected to or recently disconnected.
if len(update["result"]["networks"]) == 0:
raise WlanPolicyControllerError("WLAN state update is missing network.")
for network in update["result"]["networks"]:
if network["id"]["ssid"] == ssid or network["id"][
"type_"
].lower() == str(security_type):
if "state" not in network or "status" not in network:
raise WlanPolicyControllerError(
"Client state summary's network is missing fields"
)
# If still connected, we will wait for another update and check again
elif network["state"].lower() == STATE_CONNECTED.lower():
continue
elif network["state"].lower() == STATE_CONNECTING.lower():
self.log.error(
'Update is "Connecting", but device should already be '
"connected; expected disconnect"
)
return False
# Check that the network state and disconnect status are expected, ie
# that it isn't ConnectionFailed when we expect ConnectionStopped
elif (
network["state"].lower() != state.lower()
or network["status"].lower() != status.lower()
):
self.log.error(
"Connection failed: a network failure occurred that is unrelated"
"to remove network or incorrect status update. \nExpected state: "
f"{state}, Status: {status},\nActual update: {network}"
)
return False
else:
return True
# Wait a bit before requesting another status update
time.sleep(1)
# Stopped getting updates because out timeout
self.log.error(f'Timed out waiting for network with SSID "{ssid}" to connect')
return False
def wait_for_no_connections(
self, timeout_sec: int = DEFAULT_GET_UPDATE_TIMEOUT
) -> bool:
"""Waits to see that there are no existing connections the device. This
is the simplest way to watch for disconnections when only a single
network is saved/present.
Args:
timeout_sec: The time to wait to see no connections.
Returns:
True, if successful. False, if still connected after timeout.
"""
# If there are already no existing connections when this function is called,
# then an update won't be generated by the device, and we'll time out.
# Force an update by getting a new listener.
self.sl4f.wlan_policy_lib.wlanSetNewListener()
end_time = time.time() + timeout_sec
while time.time() < end_time:
time_left = max(1, int(end_time - time.time()))
try:
update = self.sl4f.wlan_policy_lib.wlanGetUpdate(timeout=time_left)
except TimeoutError:
self.log.info(
"Timed out getting status update while waiting for all"
" connections to end."
)
self.sl4f.wlan_policy_lib.wlanSetNewListener()
return False
if update["error"] != None:
self.log.info("Failed to get status update")
return False
# If any network is connected or being connected to, wait for them
# to disconnect.
if any(
network["state"].lower()
in {STATE_CONNECTED.lower(), STATE_CONNECTING.lower()}
for network in update["result"]["networks"]
):
continue
else:
return True
return False
def remove_and_preserve_networks_and_client_state(self) -> PreservedState:
"""Preserves networks already saved on devices before removing them to
setup up for a clean test environment. Records the state of client
connections before tests.
Raises:
WlanPolicyControllerError, if the network removal is unsuccessful
"""
# Save preexisting saved networks
state = PreservedState(saved_networks=None, client_connections_state=None)
saved_networks_response = self.sl4f.wlan_policy_lib.wlanGetSavedNetworks()
if saved_networks_response.get("error"):
raise WlanPolicyControllerError(
"Failed to get preexisting saved networks: "
f'{saved_networks_response["error"]}'
)
state.saved_networks = saved_networks_response.get("result")
# Remove preexisting saved networks
if not self.remove_all_networks():
raise WlanPolicyControllerError(
"Failed to clear networks and disconnect at FuchsiaDevice creation."
)
self.sl4f.wlan_policy_lib.wlanSetNewListener()
update_response = self.sl4f.wlan_policy_lib.wlanGetUpdate()
update_result = update_response.get("result", {})
state.client_connections_state = update_result.get("state")
if update_result.get("state") is None:
self.log.warn(
"Failed to get update; test will not start or "
"stop client connections at the end of the test."
)
self.log.info("Saved networks cleared and preserved.")
return state
def restore_preserved_networks_and_client_state(self) -> None:
"""Restore saved networks and client state onto device if they have
been preserved.
"""
if self.preserved_networks_and_client_state is None:
self.log.info("No preserved networks or client state to restore")
return
if not self.remove_all_networks():
self.log.warn("Failed to remove saved networks before restore.")
restore_success = True
saved_networks = self.preserved_networks_and_client_state.saved_networks
if saved_networks is not None:
for network in saved_networks:
if not self.save_network(
network["ssid"],
network["security_type"],
network["credential_value"],
):
self.log.warn(f'Failed to restore network "{network["ssid"]}"')
restore_success = False
client_state = self.preserved_networks_and_client_state.client_connections_state
if client_state is not None:
state_restored = (
self.start_client_connections()
if client_state == CONNECTIONS_ENABLED
else self.stop_client_connections()
)
if not state_restored:
self.log.warn("Failed to restore client connections state.")
restore_success = False
if restore_success:
self.log.info("Preserved networks and client state restored.")
self.preserved_networks_and_client_state = None
def _get_phy_ids(self) -> list[str]:
"""Get a list of WLAN physical interfaces."""
phy_ids_response = self.sl4f.wlan_lib.wlanPhyIdList()
if phy_ids_response.get("error"):
raise ConnectionError(
f'Failed to get phy ids from DUT. Error: {phy_ids_response["error"]}'
)
return phy_ids_response["result"]
def _get_phy_country_codes(self) -> dict[str, str]:
"""Get mapping of WLAN interfaces to the country code they are set to."""
phy_ids = self._get_phy_ids()
args = [GetCountryCodeFromPhyParams(self.sl4f.wlan_lib, id) for id in phy_ids]
with multiprocessing.Pool() as p:
items = p.map(get_country_code_from_phy, args)
return dict(items)
def set_country_code(self, country_code: str, timeout_sec: int = 10) -> None:
"""Set and verify the country code on the device.
Args:
country_code: 2 character country code to set
timeout_sec: seconds to wait until timing out
Raises:
TimeoutError: the device did not set all interfaces to country_code within
timeout_sec
"""
unique_country_codes = set(self._get_phy_country_codes().values())
if len(unique_country_codes) == 1 and country_code in unique_country_codes:
# The country code is already set on all WLAN phys; skip setting the country
# code again.
self.log.debug(f"Country code already set to {country_code}")
return
self.log.debug(f"Setting DUT country code to {country_code}")
country_code_response = self.sl4f.regulatory_region_lib.setRegion(country_code)
if country_code_response.get("error"):
raise EnvironmentError(
f"Failed to set country code ({country_code}) on DUT. "
f'Error: {country_code_response["error"]}'
)
self.log.debug(
f"Verifying DUT country code was correctly set to {country_code}."
)
end_time = time.time() + timeout_sec
while time.time() < end_time:
for phy_id, code in self._get_phy_country_codes().items():
if code != country_code:
# Interface has incorrect country code; continue to wait.
self.log.debug(
f"WLAN interface {phy_id} has incorrect country code set: "
f"got {code}, wanted {country_code}"
)
break
else:
# All interfaces have the expected country code.
self.log.info(f"Successfully set DUT country code to {country_code}.")
return
time.sleep(0.5)
raise TimeoutError(
f"Failed to set country code to {country_code} within {timeout_sec}s"
)