scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-11-01 03:07:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Initial work on sniffer launcher script.

Browse Source 
Still need to launch wireshark...
This commit is contained in:
Ben Greear
2020-03-16 13:26:45 -07:00
parent b58cbdd23b
commit 6ec75958ed
2 changed files with 231 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

228
lf_sniff.py Executable file
View File

@@ -0,0 +1,228 @@
#!/usr/bin/python3
'''
Sniff stations on one set of radios using secondary radios.
make sure pexpect is installed:
$ sudo yum install python3-pexpect
$ sudo yum install python3-xlsxwriter
You might need to install pexpect-serial using pip:
$ pip3 install pexpect-serial
$ pip3 install XlsxWriter
The user is responsible for setting up the stations oustide of this script, however.
When specifying ports, if the port starts with [Number]., like 1.eth1, then the 1 specifies
the resource ID.
./lf_sniff.py --lfmgr 192.168.100.178 \
--station "1.wlan0 1.wlan1" --sniffer_radios "2.wiphy0 2.wiphy1" \
--duration_min 5
'''
# TODO: Maybe HTML output too?
# TODO: Allow selecting tabs or commas for output files
import sys
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit()
import re
import logging
import time
from time import sleep
import pprint
import argparse
import subprocess
import xlsxwriter
from subprocess import PIPE
NL = "\n"
CR = "\r\n"
Q = '"'
A = "'"
FORMAT = '%(asctime)s %(name)s %(levelname)s: %(message)s'
lfmgr = "127.0.0.1"
lfstation = "1.wlan0"
sniffer_radios = "2.wiphy0"
dur = 5 * 60
moni_flags = "0x100000000"; # 160Mhz mode enabled
# rssi_adjust = (current_nf - nf_at_calibration)
def usage():
print("$0 ")
print("--station: LANforge station names (1.wlan0 1.wlan1 ...)")
print("--sniffer_radios: LANforge radios to use as sniffers (2.wiphy0 2.wiphy1 ...)")
print("--lfmgr: LANforge manager IP address")
print("--duration: Duration to run traffic, in minutes")
print("--moni_flags: Monitor flags (see LANforge CLI help for set_wifi_monitor command) Default enables 160Mhz")
print("-h|--help")
def main():
global lfmgr
global lfstation
global sniffer_radios
global dur
global moni_flags
parser = argparse.ArgumentParser(description="Sniffer control Script")
parser.add_argument("--sniffer_radios", type=str, help="LANforge sniffer radios to use (2.wiphy0 2.wiphy1 ...)")
parser.add_argument("--station", type=str, help="LANforge stations to use (1.wlan0 1.wlan1 etc)")
parser.add_argument("--lfmgr", type=str, help="LANforge Manager IP address")
parser.add_argument("--duration", type=float, help="Duration to sniff, in minutes")
parser.add_argument("--moni_flags", type=str, help="Monitor port flags, see LANforge CLI help for set_wifi_monitor. Default enables 160Mhz")
args = None
try:
args = parser.parse_args()
if (args.station != None):
lfstation = args.station
if (args.sniffer_radios != None):
sniffer_radios = args.sniffer_radios
if (args.lfmgr != None):
lfmgr = args.lfmgr
if (args.duration != None):
dur = args.duration * 60
if (args.moni_flags != None):
moni_flags = args.moni_flags
filehandler = None
except Exception as e:
logging.exception(e);
usage()
exit(2);
# Use subprocess.check_output("Cmd") to utilize existing LF scripts.
lfstations = lfstation.split()
radios = sniffer_radios.split()
idx = 0
for sta in lfstations:
sta_resource = "1"
sta_name = sta;
if sta[0].isdigit():
tmpa = sta.split(".", 1);
sta_resource = tmpa[0];
sta_name = tmpa[1];
# Assume station is up and/or something else is bringing it up
channel = 36
bsssid = "00:00:00:00:00:00"
i = 0
wait_ip_print = False;
wait_assoc_print = False;
# Wait until LANforge station connects
while True:
port_stats = subprocess.run(["./lf_portmod.pl", "--manager", lfmgr, "--card", sta_resource, "--port_name", sta_name,
"--show_port", "AP,IP,Mode,NSS,Bandwidth,Probed-Channel,Signal,Noise,Status,RX-Rate"], stdout=PIPE, stderr=PIPE);
pss = port_stats.stdout.decode('utf-8', 'ignore');
_status = None
_ip = None
for line in pss.splitlines():
#print("line: %s\n"%line)
m = re.search('AP:\s+(.*)', line)
if (m != None):
bssid = m.group(1)
m = re.search('Status:\s+(.*)', line)
if (m != None):
_status = m.group(1)
m = re.search('Probed-Channel:\s+(.*)', line)
if (m != None):
channel = m.group(1)
m = re.search('IP:\s+(.*)', line)
if (m != None):
_ip = m.group(1)
#print("IP %s Status %s"%(_ip, _status))
if (_status == "Authorized"):
if ((_ip != None) and (_ip != "0.0.0.0")):
print("Station is associated with IP address.")
break
else:
if (not wait_ip_print):
print("Waiting for station %s.%s to get IP Address."%(sta_resource, sta_name))
wait_ip_print = True
else:
if (not wait_assoc_print):
print("Waiting up to 180s for station %s.%s to associate."%(sta_resource, sta_name))
wait_assoc_print = True
i = i + 1
# We wait a fairly long time since AP will take a long time to start on a CAC channel.
if (i > 180):
err = "ERROR: Station did not connect within 180 seconds."
print(err)
e_tot += err
e_tot += " "
if (args.wait_forever):
print("Will continue waiting, you may wish to debug the system...")
i = 0
else:
break
time.sleep(1)
# Get station AID and other info
port_stats = subprocess.run(["./lf_portmod.pl", "--manager", lfmgr,
"--cli_cmd", "probe_port 1 %s %s"%(sta_resource, sta_name)], stdout=PIPE, stderr=PIPE);
pss = port_stats.stdout.decode('utf-8', 'ignore');
aid = 0
for line in pss.splitlines():
m = re.search('Local-AID:\s+(.*)', line)
if (m != None):
aid = m.group(1)
break
# Create monitor on radio X
rad = radios[idx]
rad_resource = "1"
rad_name = rad;
#print("idx: %i moni: %s\n"%(idx, moni))
if rad[0].isdigit():
tmpa = rad.split(".", 1)
rad_resource = tmpa[0]
rad_name = tmpa[1]
# Set channel on the radio
subprocess.run(["./lf_portmod.pl", "--manager", lfmgr, "--card", rad_resource, "--port_name", rad_name,
"--set_channel", "%s"%channel]);
# Get radio index so we can name the monitor similar to how the system would auto-create them
port_stats = subprocess.run(["./lf_portmod.pl", "--manager", lfmgr, "--card", rad_resource, "--port_name", rad_name,
"--show_port", "Port"], stdout=PIPE, stderr=PIPE);
pss = port_stats.stdout.decode('utf-8', 'ignore');
moni_idx = "0"
for line in pss.splitlines():
m = re.search('Port:\s+(.*)', line)
if (m != None):
moni_idx = m.group(1)
# Create monitor interface
mname = "moni%sa"%(moni_idx);
subprocess.run(["./lf_portmod.pl", "--manager", lfmgr,
"--cli_cmd", "add_monitor 1 %s %s %s %s 0xFFFFFFFFFFFF %s %s"%(rad_resource, rad_name, mname, moni_flags, aid, bssid)]);
print("Created monitor interface: %s on resource %s\n"%(mname, rad_resource));
idx = idx + 1
# ----- ----- ----- ----- ----- ----- ----- ----- ----- ----- ----- -----
if __name__ == '__main__':
main()
####
####
####

6
lf_tos_test.py
View File

@@ -244,7 +244,7 @@ def main():
cxnames = []
for sta in lfstations:
e_tot = ""
sta_resource=1
sta_resource = "1"
sta_name = sta;
if sta[0].isdigit():
tmpa = sta.split(".", 1);
@@ -290,7 +290,7 @@ def main():
print("Waiting up to 180s for station %s.%s to associate."%(sta_resource, sta_name))
wait_assoc_print = True
i += 1
i = i + 1
# We wait a fairly long time since AP will take a long time to start on a CAC channel.
if (i > 180):
err = "ERROR: Station did not connect within 180 seconds."
@@ -350,7 +350,7 @@ def main():
# Gather probe results and record data, verify NSS, BW, Channel
count = 0
for sta in lfstations:
sta_resource=1
sta_resource = "1"
sta_name = sta;
if sta[0].isdigit():
tmpa = sta.split(".", 1);