Layer 2 scan using wpa_supplicant cli

pscheduler-test-wifibssid/wifibssid/result-format

@@ -46,13 +46,13 @@ if format == 'text/plain':
         print("")
         print(entry["ssid"] + ":")
         print(f'   Signal: {entry["signal"]}')
-        print(f'   Address: {entry["address"]}') 
-        print(f'   Frequency: {entry["frequency"]}') 
-        print(f'   Quality: {entry["quality"]}')
-        print(f'   Bitrates: {entry["bitrates"]}')
-        print(f'   Encrypted: {entry["encrypted"]}')
-        print(f'   Channel: {entry["channel"]}')
-        print(f'   Mode: {entry["mode"]}') 
+        print(f'   Address: {entry["bssid"]}') 
+        print(f'   Frequency: {entry["freq"]}') 
+        # print(f'   Quality: {entry["quality"]}')
+        # print(f'   Bitrates: {entry["bitrates"]}')
+        # print(f'   Encrypted: {entry["encrypted"]}')
+        # print(f'   Channel: {entry["channel"]}')
+        print(f'   Mode: {entry["flags"]}') 
         print("") 
 
 elif format == 'text/html':
@@ -68,13 +68,13 @@ elif format == 'text/html':
     for entry in result['ssid_list']:
         print(f'    <tr><td><b>SSID</b></td> <td>{entry["ssid"]}</td></tr>')
         print(f'    <tr><td><b>Signal</b></td> <td>{entry["signal"]}</td></tr>') 
-        print(f'    <tr><td><b>Address</b></td> <td>{entry["address"]}</td></tr>')
-        print(f'    <tr><td><b>Frequency</b></td> <td>{entry["frequency"]}</td></tr>')
-        print(f'    <tr><td><b>Quality</b></td> <td>{entry["quality"]}</td></tr>')
-        print(f'    <tr><td><b>Bitrates</b></td> <td>{entry["bitrates"]}</td></tr>')
-        print(f'    <tr><td><b>Encrypted</b></td> <td>{entry["signal"]}</td></tr>')
-        print(f'    <tr><td><b>Channel</b></td> <td>{entry["channel"]}</td></tr>')
-        print(f'    <tr><td><b>Mode</b></td> <td>{entry["mode"]}</td></tr>')
+        print(f'    <tr><td><b>Address</b></td> <td>{entry["bssid"]}</td></tr>')
+        print(f'    <tr><td><b>Frequency</b></td> <td>{entry["freq"]}</td></tr>')
+        # print(f'    <tr><td><b>Quality</b></td> <td>{entry["quality"]}</td></tr>')
+        # print(f'    <tr><td><b>Bitrates</b></td> <td>{entry["bitrates"]}</td></tr>')
+        # print(f'    <tr><td><b>Encrypted</b></td> <td>{entry["signal"]}</td></tr>')
+        # print(f'    <tr><td><b>Channel</b></td> <td>{entry["channel"]}</td></tr>')
+        print(f'    <tr><td><b>Mode</b></td> <td>{entry["flags"]}</td></tr>')
     print('</table>')
 
 

pscheduler-test-wifibssid/wifibssid/validate.py

@@ -51,8 +51,7 @@
             },
             # If listed here, these parameters MUST be in the test spec.
             "required": [
-                "interface",
-		        "ssid",
+                "interface"
             ],
             # Treat other properties as acceptable.  This should
             # ALWAYS be false.

pscheduler-tool-bssidscanner/bssidscanner/duration

@@ -19,7 +19,7 @@ json = pscheduler.json_load(exit_on_error=True);
 
 
 timeout_iso = json.get("timeout", "PT10S")
-timeout = pscheduler.iso8601_as_timedelta(timeout_iso)+datetime.timedelta(seconds=2)
+timeout = pscheduler.iso8601_as_timedelta(timeout_iso)+datetime.timedelta(seconds=1)
 
 pscheduler.succeed_json({
     "duration": pscheduler.timedelta_as_iso8601( timeout )

pscheduler-tool-bssidscanner/bssidscanner/run

@@ -9,111 +9,115 @@
 # back to the test. Both system and api are able to be used here.
 #
 
+import select
+from os import getpid, makedirs, unlink, access, R_OK, system, listdir, path
 import datetime
-import subprocess
-import json
-import sys
-import time
-import argparse
-import time
-from wifi import Cell #make sure that you have pip installed wifi
-
 import pscheduler
+# import subprocess
+import time
 
-# from stdin
-input = pscheduler.json_load(exit_on_error=True)
-
-# Take input from test spec
-try:
-    interface = input['test']['spec']['interface']
-    ssid = input['test']['spec']['ssid']
-except KeyError:
-    pscheduler.fail('Missing data in input')
-
-# convert the comma separated ssids to a list
-ssid = list(ssid.split(','))
-
-duration = input['test']['spec'].get('duration', 'PT5S')
-duration = pscheduler.timedelta_as_seconds( pscheduler.iso8601_as_timedelta(duration) ) 
-timeout_iso = input['test']['spec'].get('timeout', 'PT10S')
-timeout = pscheduler.timedelta_as_seconds( pscheduler.iso8601_as_timedelta(timeout_iso) )
+#temp file preserved between consecutive runs but not reboots
+WPA_CONFIG_PATH = '/tmp/wpa_supplicant.conf'
+#initializes only when wpa_supplicant starts
+WPA_CTRL_IFACE_BASE = '/var/run/wpa_supplicant'
+ssid_set = set()
+
+# parse config
+pscheduler_input = pscheduler.json_load(exit_on_error=True)
+interface = pscheduler_input['test']['spec'].get('interface')
+get_ssid = pscheduler_input['test']['spec'].get('ssid', None)
+ssid_set = set(get_ssid.split(',')) if get_ssid else None
+duration_iso = pscheduler_input['test']['spec'].get('duration', 'PT5S')
+timeout_iso = pscheduler_input['test']['spec'].get('timeout', 'PT10S')
+timeout = pscheduler.timedelta_as_seconds(pscheduler.iso8601_as_timedelta(timeout_iso))
+duration = pscheduler.timedelta_as_seconds(pscheduler.iso8601_as_timedelta(duration_iso))
 start_time = datetime.datetime.now()
-succeeded = False
 error = ''
 diags = ''
-
-# Run the actual task here:
-def get_all_bssids(interface):
-    """
-    Scan the given interface for all bssids
-    Return a list of all bssids
-    """
-    start_time = time.time()
-    cells = Cell.all(interface) # Specify interface to scan on
-    wifi_list = []
-    for cell in cells:
-        bssid = {}
-        bssid['ssid'] = cell.ssid
-        bssid['signal'] = cell.signal
-        bssid['address'] = cell.address
-        bssid['frequency'] = pscheduler.si_as_number(cell.frequency[:-2])
-        quality_num,quality_denom = cell.quality.split('/')
-        bssid['quality'] = float(quality_num) / float(quality_denom)
-        bssid['bitrates'] = sorted(map(lambda v : pscheduler.si_as_number(v[:-3]), cell.bitrates))
-        bssid['encrypted'] = cell.encrypted
-        bssid['channel'] = cell.channel
-        bssid['mode'] = cell.mode
-        wifi_list.append(bssid)
-
-    end_time = time.time()
-    elapsed_time = end_time - start_time
-    log_msg = "Scan finished in " + str(elapsed_time)
-    return wifi_list, elapsed_time
-
-"""
-Scan on the given interface
-Output a list of all bssids in json format with the given ssid
-"""
-all_bssids, elapsed_time = get_all_bssids(interface)
 ssid_list = []
-
-# Check complete list for matching ssids
-for bssid in all_bssids:
-    #if no ssids were given then append all the ssids
-    if not ssid:
-        ssid_list.append(bssid)
-    else:
-    #if ssid/ssids were given then only append those 
-        if bssid['ssid'] in ssid:
-            ssid_list.append(bssid)
-
-succeeded = True
-
-# IMPORTANT NOTE: This code puts the process to sleep until the
-# scheduled start time has arrived.  It should be placed after all
-# preparatory code has been executed and immediately before the tool
-# is invoked (for plugins that run other programs) or any activity
-# that does a measurement (for those that don't).
-
-try:
-    pscheduler.sleep_until(input['schedule']['start'])
-except KeyError:
-    pscheduler.fail("Unable to find start time in input")
-
-
+dir_path = path.dirname(WPA_CONFIG_PATH)
+
+# check if interface already exists
+status, res, err = pscheduler.run_program(['wpa_cli', 'status'])
+if status:
+   # create config file and start wpa_supplicant
+   with open(WPA_CONFIG_PATH, 'w') as f:
+       f.write('ctrl_interface='+WPA_CTRL_IFACE_BASE+'\n')
+       f.write('update_config=1\n')
+       f.write('p2p_disabled=1\n')
+   status, out, err = pscheduler.run_program(['wpa_supplicant','-D','nl80211','-B','-i',interface,'-c',WPA_CONFIG_PATH])
+   if status:
+       pscheduler.succeed_json( {
+           'succeeded': False,
+           'diags': out + ' ' + str(status) + ' ' + err,
+           'error': 'failed to initialize interface',
+           'result': None
+    })
+
+# run wpa_cli
+start  = datetime.datetime.now()
+status, res, err = pscheduler.run_program(['wpa_cli', 'scan'])
+if not status:
+    status, res, err = pscheduler.run_program(['wpa_cli', 'scan_result'])
+else:
+    pscheduler.succeed_json( {
+        'succeeded': False,
+        'diags': res,
+        'error': 'wpa_cli scan failed',
+        'result': None
+        } )
+
+result = res.split('\n')
+
+# account for automatic scanning interval of cli
+while len(result) <= 3:
+    status, res, err = pscheduler.run_program(['wpa_cli', 'scan_result'])
+    result = res.split('\n')
+    time.sleep(0.3)
+
+for i in range(2, len(result)):
+    line = result[i].split('\t')
+    if len(line) == 1:
+        break
+    elif len(line) == 4:
+        bssid, freq, signal, flags = line
+    elif len(line) == 5:
+        bssid, freq, signal, flags, ssid = line
+
+    if not ssid_set or ssid in ssid_set:
+        # retrieve prev ssid if ssid is empty
+        if ssid == '':
+            ssid = ssid_list[-1]['ssid']
+        ssid_list.append({
+            'ssid': ssid,
+            'bssid': bssid,
+            'freq': int(freq),
+            'signal': int(signal),
+            'flags': flags
+            })
 end_time = datetime.datetime.now()
 
+# Check for empty result
+if len(ssid_list) == 0:
+    pscheduler.succeed_json( {
+        'succeeded': False,
+        'diags': '',
+        'error': 'scan completed but returned empty result',
+        'result': None
+        } )
+
 # Organize results into json data
 results = {
-    'succeeded': succeeded,
+    'succeeded': True,
     'result': {
         'schema': 1,
-        'time': pscheduler.timedelta_as_iso8601( end_time - start_time),
-        'succeeded': succeeded,
+        'time': pscheduler.timedelta_as_iso8601(end_time - start_time),
+        'succeeded': True,
         'ssid_list': ssid_list
     },
     'error': error,
-    'diags': diags }
+    'diags': diags
+}
 
-pscheduler.succeed_json(results)
 
+pscheduler.succeed_json(results)

pscheduler-tool-bssidscanner/bssidscanner/unibuild-packaging/deb/sudoers

@@ -0,0 +1,10 @@
+#
+# pscheduler-tool-bssidscanner
+#
+Cmnd_Alias PSCHEDULER_WPA_SUPPLICANT = /usr/sbin/wpa_supplicant
+pscheduler ALL = (root) NOPASSWD: PSCHEDULER_WPA_SUPPLICANT
+Defaults!PSCHEDULER_WPA_SUPPLICANT !requiretty
+
+Cmnd_Alias PSCHEDULER_WPA_CLI = /usr/sbin/wpa_cli
+pscheduler ALL = (root) NOPASSWD: PSCHEDULER_WPA_CLI
+Defaults!PSCHEDULER_WPA_CLI !requiretty

pscheduler-tool-bssidscanner/bssidscanner/unibuild-packaging/rpm/pscheduler-tool-bssidscanner.spec

@@ -44,6 +44,24 @@ make \
      DESTDIR=$RPM_BUILD_ROOT/%{dest} \
      install
 
+# Enable sudo for this tool
+WPA_SUPP=$(command -v wpa_supplicant)
+WPA_CLI=$(command -v wpa_cli)
+
+mkdir -p $RPM_BUILD_ROOT/%{_pscheduler_sudoersdir}
+cat > $RPM_BUILD_ROOT/%{_pscheduler_sudoersdir}/%{name} <<EOF
+
+#
+# %{name}
+#
+Cmnd_Alias PSCHEDULER_WPA_SUPP = ${WPA_SUPP}
+%{_pscheduler_user} ALL = (root) NOPASSWD: ${WPA_SUPP}
+Defaults!PSCHEDULER_WPA_SUPP !requiretty
+
+Cmnd_Alias PSCHEDULER_WPA_CLI = ${WPA_CLI}
+%{_pscheduler_user} ALL = (root) NOPASSWD: ${WPA_CLI}
+Defaults!PSCHEDULER_WPA_CLI !requiretty
+
 %post
 pscheduler internal warmboot