#!/usr/bin/env python3 # # Copyright 2019 - The Android Open Source Project # # 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 csv import os import posixpath import time import zipfile import acts_contrib.test_utils.wifi.wifi_test_utils as wutils from acts import context from acts import logger from acts import utils from acts.controllers.utils_lib import ssh WifiEnums = wutils.WifiEnums SNIFFER_TIMEOUT = 6 MEDIUM_SLEEP = 3 SHORT_SLEEP = 1 def create(configs): """Factory method for sniffer. Args: configs: list of dicts with sniffer settings. Settings must contain the following : ssh_settings, type, OS, interface. Returns: objs: list of sniffer class objects. """ objs = [] for config in configs: try: if config['type'] == 'tshark': if config['os'] == 'unix': objs.append(TsharkSnifferOnUnix(config)) elif config['os'] == 'linux': objs.append(TsharkSnifferOnLinux(config)) elif config['os'] == 'android': objs.append(TsharkSnifferOnAndroid(config)) else: raise RuntimeError('Wrong sniffer config') elif config['type'] == 'mock': objs.append(MockSniffer(config)) except KeyError: raise KeyError('Invalid sniffer configurations') return objs def destroy(objs): return class OtaSnifferBase(object): """Base class defining common sniffers functions.""" _log_file_counter = 0 @property def started(self): raise NotImplementedError('started must be specified.') def start_capture(self, network, duration=30): """Starts the sniffer Capture. Args: network: dict containing network information such as SSID, etc. duration: duration of sniffer capture in seconds. """ raise NotImplementedError('start_capture must be specified.') def stop_capture(self, tag=''): """Stops the sniffer Capture. Args: tag: string to tag sniffer capture file name with. """ raise NotImplementedError('stop_capture must be specified.') def _get_remote_dump_path(self): """Returns name of the sniffer dump file.""" remote_file_name = 'sniffer_dump.{}'.format( self.sniffer_output_file_type) remote_dump_path = posixpath.join(posixpath.sep, 'tmp', remote_file_name) return remote_dump_path def _get_full_file_path(self, tag=None): """Returns the full file path for the sniffer capture dump file. Returns the full file path (on test machine) for the sniffer capture dump file. Args: tag: The tag appended to the sniffer capture dump file . """ tags = [tag, 'count', OtaSnifferBase._log_file_counter] out_file_name = 'Sniffer_Capture_%s.%s' % ('_'.join([ str(x) for x in tags if x != '' and x is not None ]), self.sniffer_output_file_type) OtaSnifferBase._log_file_counter += 1 file_path = os.path.join(self.log_path, out_file_name) return file_path @property def log_path(self): current_context = context.get_current_context() full_out_dir = os.path.join(current_context.get_full_output_path(), 'sniffer_captures') # Ensure the directory exists. os.makedirs(full_out_dir, exist_ok=True) return full_out_dir class MockSniffer(OtaSnifferBase): """Class that implements mock sniffer for test development and debug.""" def __init__(self, config): self.log = logger.create_tagged_trace_logger('Mock Sniffer') def start_capture(self, network, duration=30): """Starts sniffer capture on the specified machine. Args: network: dict of network credentials. duration: duration of the sniff. """ self.log.debug('Starting sniffer.') def stop_capture(self): """Stops the sniffer. Returns: log_file: name of processed sniffer. """ self.log.debug('Stopping sniffer.') log_file = self._get_full_file_path() with open(log_file, 'w') as file: file.write('this is a sniffer dump.') return log_file class TsharkSnifferBase(OtaSnifferBase): """Class that implements Tshark based sniffer controller. """ TYPE_SUBTYPE_DICT = { '0': 'Association Requests', '1': 'Association Responses', '2': 'Reassociation Requests', '3': 'Resssociation Responses', '4': 'Probe Requests', '5': 'Probe Responses', '8': 'Beacon', '9': 'ATIM', '10': 'Disassociations', '11': 'Authentications', '12': 'Deauthentications', '13': 'Actions', '24': 'Block ACK Requests', '25': 'Block ACKs', '26': 'PS-Polls', '27': 'RTS', '28': 'CTS', '29': 'ACK', '30': 'CF-Ends', '31': 'CF-Ends/CF-Acks', '32': 'Data', '33': 'Data+CF-Ack', '34': 'Data+CF-Poll', '35': 'Data+CF-Ack+CF-Poll', '36': 'Null', '37': 'CF-Ack', '38': 'CF-Poll', '39': 'CF-Ack+CF-Poll', '40': 'QoS Data', '41': 'QoS Data+CF-Ack', '42': 'QoS Data+CF-Poll', '43': 'QoS Data+CF-Ack+CF-Poll', '44': 'QoS Null', '46': 'QoS CF-Poll (Null)', '47': 'QoS CF-Ack+CF-Poll (Null)' } TSHARK_COLUMNS = [ 'frame_number', 'frame_time_relative', 'mactime', 'frame_len', 'rssi', 'channel', 'ta', 'ra', 'bssid', 'type', 'subtype', 'duration', 'seq', 'retry', 'pwrmgmt', 'moredata', 'ds', 'phy', 'radio_datarate', 'vht_datarate', 'radiotap_mcs_index', 'vht_mcs', 'wlan_data_rate', '11n_mcs_index', '11ac_mcs', '11n_bw', '11ac_bw', 'vht_nss', 'mcs_gi', 'vht_gi', 'vht_coding', 'ba_bm', 'fc_status', 'bf_report' ] TSHARK_OUTPUT_COLUMNS = [ 'frame_number', 'frame_time_relative', 'mactime', 'ta', 'ra', 'bssid', 'rssi', 'channel', 'frame_len', 'Info', 'radio_datarate', 'radiotap_mcs_index', 'pwrmgmt', 'phy', 'vht_nss', 'vht_mcs', 'vht_datarate', '11ac_mcs', '11ac_bw', 'vht_gi', 'vht_coding', 'wlan_data_rate', '11n_mcs_index', '11n_bw', 'mcs_gi', 'type', 'subtype', 'duration', 'seq', 'retry', 'moredata', 'ds', 'ba_bm', 'fc_status', 'bf_report' ] TSHARK_FIELDS_LIST = [ 'frame.number', 'frame.time_relative', 'radiotap.mactime', 'frame.len', 'radiotap.dbm_antsignal', 'wlan_radio.channel', 'wlan.ta', 'wlan.ra', 'wlan.bssid', 'wlan.fc.type', 'wlan.fc.type_subtype', 'wlan.duration', 'wlan.seq', 'wlan.fc.retry', 'wlan.fc.pwrmgt', 'wlan.fc.moredata', 'wlan.fc.ds', 'wlan_radio.phy', 'radiotap.datarate', 'radiotap.vht.datarate.0', 'radiotap.mcs.index', 'radiotap.vht.mcs.0', 'wlan_radio.data_rate', 'wlan_radio.11n.mcs_index', 'wlan_radio.11ac.mcs', 'wlan_radio.11n.bandwidth', 'wlan_radio.11ac.bandwidth', 'radiotap.vht.nss.0', 'radiotap.mcs.gi', 'radiotap.vht.gi', 'radiotap.vht.coding.0', 'wlan.ba.bm', 'wlan.fcs.status', 'wlan.vht.compressed_beamforming_report.snr' ] def __init__(self, config): self.sniffer_proc_pid = None self.log = logger.create_tagged_trace_logger('Tshark Sniffer') self.ssh_config = config['ssh_config'] self.sniffer_os = config['os'] self.run_as_sudo = config.get('run_as_sudo', False) self.sniffer_output_file_type = config['output_file_type'] self.sniffer_snap_length = config['snap_length'] self.sniffer_interface = config['interface'] self.sniffer_disabled = False #Logging into sniffer self.log.info('Logging into sniffer.') self._sniffer_server = ssh.connection.SshConnection( ssh.settings.from_config(self.ssh_config)) # Get tshark params self.tshark_fields = self._generate_tshark_fields( self.TSHARK_FIELDS_LIST) self.tshark_path = self._sniffer_server.run('which tshark').stdout @property def _started(self): return self.sniffer_proc_pid is not None def _scan_for_networks(self): """Scans for wireless networks on the sniffer.""" raise NotImplementedError def _get_tshark_command(self, duration): """Frames the appropriate tshark command. Args: duration: duration to sniff for. Returns: tshark_command : appropriate tshark command. """ tshark_command = '{} -l -i {} -I -t u -a duration:{}'.format( self.tshark_path, self.sniffer_interface, int(duration)) if self.run_as_sudo: tshark_command = 'sudo {}'.format(tshark_command) return tshark_command def _get_sniffer_command(self, tshark_command): """ Frames the appropriate sniffer command. Args: tshark_command: framed tshark command Returns: sniffer_command: appropriate sniffer command """ if self.sniffer_output_file_type in ['pcap', 'pcapng']: sniffer_command = ' {tshark} -s {snaplength} -w {log_file} '.format( tshark=tshark_command, snaplength=self.sniffer_snap_length, log_file=self._get_remote_dump_path()) elif self.sniffer_output_file_type == 'csv': sniffer_command = '{tshark} {fields} > {log_file}'.format( tshark=tshark_command, fields=self.tshark_fields, log_file=self._get_remote_dump_path()) else: raise KeyError('Sniffer output file type not configured correctly') return sniffer_command def _generate_tshark_fields(self, fields): """Generates tshark fields to be appended to the tshark command. Args: fields: list of tshark fields to be appended to the tshark command. Returns: tshark_fields: string of tshark fields to be appended to the tshark command. """ tshark_fields = "-T fields -y IEEE802_11_RADIO -E separator='^'" for field in fields: tshark_fields = tshark_fields + ' -e {}'.format(field) return tshark_fields def _configure_sniffer(self, network, chan, bw): """ Connects to a wireless network using networksetup utility. Args: network: dictionary of network credentials; SSID and password. """ raise NotImplementedError def _run_tshark(self, sniffer_command): """Starts the sniffer. Args: sniffer_command: sniffer command to execute. """ self.log.debug('Starting sniffer.') sniffer_job = self._sniffer_server.run_async(sniffer_command) self.sniffer_proc_pid = sniffer_job.stdout def _stop_tshark(self): """ Stops the sniffer.""" self.log.debug('Stopping sniffer') # while loop to kill the sniffer process stop_time = time.time() + SNIFFER_TIMEOUT while time.time() < stop_time: # Wait before sending more kill signals time.sleep(0.1) try: # Returns 1 if process was killed self._sniffer_server.run( 'ps aux| grep {} | grep -v grep'.format( self.sniffer_proc_pid)) except: return try: # Returns error if process was killed already self._sniffer_server.run('sudo kill -15 {}'.format( str(self.sniffer_proc_pid))) except: # Except is hit when tshark is already dead but we will break # out of the loop when confirming process is dead using ps aux pass self.log.warning('Could not stop sniffer. Trying with SIGKILL.') try: self.log.debug('Killing sniffer with SIGKILL.') self._sniffer_server.run('sudo kill -9 {}'.format( str(self.sniffer_proc_pid))) except: self.log.debug('Sniffer process may have stopped succesfully.') def _process_tshark_dump(self, log_file): """ Process tshark dump for better readability. Processes tshark dump for better readability and saves it to a file. Adds an info column at the end of each row. Format of the info columns: subtype of the frame, sequence no and retry status. Args: log_file : unprocessed sniffer output Returns: log_file : processed sniffer output """ temp_dump_file = os.path.join(self.log_path, 'sniffer_temp_dump.csv') utils.exe_cmd('cp {} {}'.format(log_file, temp_dump_file)) with open(temp_dump_file, 'r') as input_csv, open(log_file, 'w') as output_csv: reader = csv.DictReader(input_csv, fieldnames=self.TSHARK_COLUMNS, delimiter='^') writer = csv.DictWriter(output_csv, fieldnames=self.TSHARK_OUTPUT_COLUMNS, delimiter='\t') writer.writeheader() for row in reader: if row['subtype'] in self.TYPE_SUBTYPE_DICT: row['Info'] = '{sub} S={seq} retry={retry_status}'.format( sub=self.TYPE_SUBTYPE_DICT[row['subtype']], seq=row['seq'], retry_status=row['retry']) else: row['Info'] = '{} S={} retry={}\n'.format( row['subtype'], row['seq'], row['retry']) writer.writerow(row) utils.exe_cmd('rm -f {}'.format(temp_dump_file)) return log_file def start_capture(self, network, chan, bw, duration=60): """Starts sniffer capture on the specified machine. Args: network: dict describing network to sniff on. duration: duration of sniff. """ # Checking for existing sniffer processes if self._started: self.log.debug('Sniffer already running') return # Configure sniffer self._configure_sniffer(network, chan, bw) tshark_command = self._get_tshark_command(duration) sniffer_command = self._get_sniffer_command(tshark_command) # Starting sniffer capture by executing tshark command self._run_tshark(sniffer_command) def stop_capture(self, tag=''): """Stops the sniffer. Args: tag: tag to be appended to the sniffer output file. Returns: log_file: path to sniffer dump. """ # Checking if there is an ongoing sniffer capture if not self._started: self.log.debug('No sniffer process running') return # Killing sniffer process self._stop_tshark() # Processing writing capture output to file log_file = self._get_full_file_path(tag) self._sniffer_server.run('sudo chmod 777 {}'.format( self._get_remote_dump_path())) self._sniffer_server.pull_file(log_file, self._get_remote_dump_path()) if self.sniffer_output_file_type == 'csv': log_file = self._process_tshark_dump(log_file) if self.sniffer_output_file_type == 'pcap': zip_file_path = log_file[:-4] + "zip" zipfile.ZipFile(zip_file_path, 'w', zipfile.ZIP_DEFLATED).write( log_file, arcname=log_file.split('/')[-1]) os.remove(log_file) self.sniffer_proc_pid = None return log_file class TsharkSnifferOnUnix(TsharkSnifferBase): """Class that implements Tshark based sniffer controller on Unix systems.""" def _scan_for_networks(self): """Scans the wireless networks on the sniffer. Returns: scan_results : output of the scan command. """ scan_command = '/usr/local/bin/airport -s' scan_result = self._sniffer_server.run(scan_command).stdout return scan_result def _configure_sniffer(self, network, chan, bw): """Connects to a wireless network using networksetup utility. Args: network: dictionary of network credentials; SSID and password. """ self.log.debug('Connecting to network {}'.format(network['SSID'])) if 'password' not in network: network['password'] = '' connect_command = 'networksetup -setairportnetwork en0 {} {}'.format( network['SSID'], network['password']) self._sniffer_server.run(connect_command) class TsharkSnifferOnLinux(TsharkSnifferBase): """Class that implements Tshark based sniffer controller on Linux.""" def __init__(self, config): super().__init__(config) self._init_sniffer() self.channel = None self.bandwidth = None def _init_sniffer(self): """Function to configure interface for the first time""" self._sniffer_server.run('sudo modprobe -r iwlwifi') self._sniffer_server.run('sudo dmesg -C') self._sniffer_server.run('cat /dev/null | sudo tee /var/log/syslog') self._sniffer_server.run('sudo modprobe iwlwifi debug=0x1') # Wait for wifi config changes before trying to further configuration # e.g. setting monitor mode (which will fail if above is not complete) time.sleep(SHORT_SLEEP) def start_capture(self, network, chan, bw, duration=60): """Starts sniffer capture on the specified machine. Args: network: dict describing network to sniff on. duration: duration of sniff. """ # If sniffer doesnt support the channel, return if '6g' in str(chan): self.log.debug('Channel not supported on sniffer') return # Checking for existing sniffer processes if self._started: self.log.debug('Sniffer already running') return # Configure sniffer self._configure_sniffer(network, chan, bw) tshark_command = self._get_tshark_command(duration) sniffer_command = self._get_sniffer_command(tshark_command) # Starting sniffer capture by executing tshark command self._run_tshark(sniffer_command) def set_monitor_mode(self, chan, bw): """Function to configure interface to monitor mode Brings up the sniffer wireless interface in monitor mode and tunes it to the appropriate channel and bandwidth Args: chan: primary channel (int) to tune the sniffer to bw: bandwidth (int) to tune the sniffer to """ if chan == self.channel and bw == self.bandwidth: return self.channel = chan self.bandwidth = bw channel_map = { 80: { tuple(range(36, 50, 2)): 42, tuple(range(52, 66, 2)): 58, tuple(range(100, 114, 2)): 106, tuple(range(116, 130, 2)): 122, tuple(range(132, 146, 2)): 138, tuple(range(149, 163, 2)): 155 }, 40: { (36, 38, 40): 38, (44, 46, 48): 46, (52, 54, 56): 54, (60, 62, 64): 62, (100, 102, 104): 102, (108, 110, 112): 108, (116, 118, 120): 118, (124, 126, 128): 126, (132, 134, 136): 134, (140, 142, 144): 142, (149, 151, 153): 151, (157, 159, 161): 159 }, 160: { (36, 38, 40): 50 } } if chan <= 13: primary_freq = WifiEnums.channel_2G_to_freq[chan] else: primary_freq = WifiEnums.channel_5G_to_freq[chan] self._sniffer_server.run('sudo ifconfig {} down'.format( self.sniffer_interface)) self._sniffer_server.run('sudo iwconfig {} mode monitor'.format( self.sniffer_interface)) self._sniffer_server.run('sudo ifconfig {} up'.format( self.sniffer_interface)) if bw in channel_map: for tuple_chan in channel_map[bw]: if chan in tuple_chan: center_freq = WifiEnums.channel_5G_to_freq[channel_map[bw] [tuple_chan]] self._sniffer_server.run( 'sudo iw dev {} set freq {} {} {}'.format( self.sniffer_interface, primary_freq, bw, center_freq)) else: self._sniffer_server.run('sudo iw dev {} set freq {}'.format( self.sniffer_interface, primary_freq)) def _configure_sniffer(self, network, chan, bw): """ Connects to a wireless network using networksetup utility. Args: network: dictionary of network credentials; SSID and password. """ self.log.debug('Setting monitor mode on Ch {}, bw {}'.format(chan, bw)) self.set_monitor_mode(chan, bw) class TsharkSnifferOnAndroid(TsharkSnifferBase): """Class that implements Tshark based sniffer controller on Linux.""" def __init__(self, config): super().__init__(config) self._init_sniffer(config) self.channel = None self.bandwidth = None def _init_sniffer(self, config): """Function to configure interface for the first time""" tshark_D_output = self._sniffer_server.run('tshark -D').stdout self.sniffer_sn = config['serial'] time.sleep(MEDIUM_SLEEP) if config['interface'] in tshark_D_output: self.log.info("Target sniffer interface {} detected".format( config['interface'])) else: self.log.error('Target sniffer interface {} NOT detected'.format( config['interface'])) # Wait for wifi config changes before trying to further configuration # e.g. setting monitor mode (which will fail if above is not complete) time.sleep(SHORT_SLEEP) def start_capture(self, network, chan, bw, duration=60): """Starts sniffer capture on the specified machine. Args: network: dict describing network to sniff on. duration: duration of sniff. """ # If sniffer doesnt support the channel, return # Checking for existing sniffer processes if self._started: return # Configure sniffer self._configure_sniffer(network, chan, bw) tshark_command = self._get_tshark_command(duration) sniffer_command = self._get_sniffer_command(tshark_command) # Starting sniffer capture by executing tshark command self._run_tshark(sniffer_command) def set_monitor_mode(self, chan, bw): """Function to configure interface to monitor mode Brings up the sniffer wireless interface in monitor mode and tunes it to the appropriate channel and bandwidth Args: chan: primary channel (int) to tune the sniffer to bw: bandwidth (int) to tune the sniffer to """ if chan == self.channel and bw == self.bandwidth: return self.channel = chan self.bandwidth = bw self._sniffer_server.run('adb -s {} shell wl chanspec {}/{}'.format( self.sniffer_sn, chan, bw)) time.sleep(SHORT_SLEEP) sniffer_chanspec = self._sniffer_server.run( 'adb -s {} shell wl chanspec'.format(self.sniffer_sn)).stdout time.sleep(SHORT_SLEEP) self.log.info("Sniffer channel: {}".format(sniffer_chanspec)) def _configure_sniffer(self, network, chan, bw): """ Connects to a wireless network using networksetup utility. Args: network: dictionary of network credentials; SSID and password. """ self.log.debug('Setting monitor mode on Ch {}, bw {}'.format(chan, bw)) self.set_monitor_mode(chan, bw)