This python script will scan for WiFi networks in sectors. It is meant to be used with an antenna like the ALFACARD attached to a stepmotor. Once a full scan has been done a choice of the network is offered and the motor will position the antenna for best signal, at this point it if also possible to establish a full connection in cli mode :)
#############################################################################################################
# #
# AUTHOR: SHELL0CK - https://twitter.com/shell_ock #
# THIS SCRIPT IS MEANT TO BE USED TO WITH A WIFI DIRECTIONAL ANTENNA ATTACCHED TO A STEPMOTOR WITH A #
# RASPBERRYPY ALLOWS TO SCAN ALL WIFI NETWORKS NEARBY AND "HOME IN" WITH THE BEST SIGNAL STRENGTH #
# IT ALSO ALLOWS TO CONNECT TO THE SELECTED WIFI NETWORK IN CLI MODE :) #
# CREDITS TO [email protected] / @glennzw - FOR HIS SCRIPT EVIL.PY UPON WHICH THIS WORK IS BASED #
# #
#############################################################################################################
import time
import RPi.GPIO as GPIO
from shell import ex
import os
import shlex
import sqlite3
from time import gmtime, strftime
from subprocess import Popen, call, PIPE
import errno
from types import *
import logging
import sys
import argparse
import re
def usage():
print "\n" + "#"*100 + "\n"
print "usage: sudo python BestSignal.py "
print "i.e. sudo python BestSignal.py L 50 3 \n"
print "#"*100 + "\n"
sys.exit(-1)
if len(sys.argv) < 4:
usage()
else:
full_circle = 510.0
conversion_factor = 1.44444444444444444444444444444
direction_input = sys.argv[1] # i.e direction of turn, L or R
arch_width = int(sys.argv[2]) # the sector width to scan
rotation_step = int(sys.argv[3]) # sets the step for the scan
class bcolors:
HEADER = '\033[95m'
BLUE = '\033[94m'
GREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
WHITE ='\033[0;37m'
RED='\033[0;31m'
SUPPLICANT_LOG_FILE = "wpa_supplicant.log"
GPIO.setmode(GPIO.BOARD)
GPIO.setwarnings(False)
A = 7
B = 11
C = 13
D = 15
GPIO.setup(A, GPIO.OUT)
GPIO.setup(B, GPIO.OUT)
GPIO.setup(C, GPIO.OUT)
GPIO.setup(D, GPIO.OUT)
def GPIO_SETUP(a,b,c,d):
GPIO.output(A, a)
GPIO.output(B, b)
GPIO.output(C, c)
GPIO.output(D, d)
time.sleep(0.001)
print "Calling initial GPIO_SETUP"
GPIO_SETUP(0,0,0,0)
workingDirectory = os.getcwd()
def DataBaseCreation(DataBaseName):
DBName = DataBaseName + ".db" # name of the sqlite database file
TableName = "Scan" # name of the table to be created
First_field = 'ID' # name of the column
Integer_type = 'integer' # column data type
Position = 'Position' # name of the column
TEXT_type = 'TEXT' # column data type
BSSID = 'BSSID' # name of the column
SSID = 'SSID' # name of the column
Flags = 'Flags' # name of the column
Freq = 'Freq' # name of the column
Signal = 'Signal'
# Connecting to the database file
conn = sqlite3.connect(DBName)
c = conn.cursor()
print "Creating Table" + DataBaseName + ".db" + "...."
# Creating a second table with 1 column and set it as PRIMARY KEY
# note that PRIMARY KEY column must consist of unique values!
c.execute('CREATE TABLE {tn} ({nf} {ft} PRIMARY KEY AUTOINCREMENT)'\
.format(tn=TableName, nf=First_field, ft=Integer_type))
#add degree
c.execute("ALTER TABLE {tn} ADD COLUMN '{cn}' {ct}"\
.format(tn=TableName, cn=Position, ct=TEXT_type))
#add BSSID
c.execute("ALTER TABLE {tn} ADD COLUMN '{cn}' {ct}"\
.format(tn=TableName, cn=BSSID, ct=TEXT_type))
#add SSID
c.execute("ALTER TABLE {tn} ADD COLUMN '{cn}' {ct}"\
.format(tn=TableName, cn=SSID, ct=TEXT_type))
#add Flags
c.execute("ALTER TABLE {tn} ADD COLUMN '{cn}' {ct}"\
.format(tn=TableName, cn=Flags, ct=TEXT_type))
#add Freq
c.execute("ALTER TABLE {tn} ADD COLUMN '{cn}' {ct}"\
.format(tn=TableName, cn=Freq, ct=TEXT_type))
#add Signal
c.execute("ALTER TABLE {tn} ADD COLUMN '{cn}' {ct}"\
.format(tn=TableName, cn=Signal, ct=TEXT_type))
# Committing changes and closing the connection to the database file
print "Committing...."
conn.commit()
print "Exiting..."
conn.close()
print "DONE!"
"""
#################################################
"""
global DataBaseName
DataBaseName = strftime("%d-%m-%Y %H:%M:%S")
print "Creating database " + DataBaseName + "..."
DataBaseCreation(DataBaseName)
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s %(levelname)s %(filename)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
filename='evil.log',
filemode='w')
"""
#################################################
"""
def start_wpa(_iface):
"""
Terminates any running wpa_supplicant process, and then starts a new one.
"""
run_program("wpa_cli terminate")
time.sleep(1)
run_program("wpa_supplicant -B -Dwext -i %s -C /var/run/wpa_supplicant -f %s" %(_iface, SUPPLICANT_LOG_FILE))
"""
#################################################
"""
def run_program(rcmd):
"""
Runs a program, and it's paramters (e.g. rcmd="ls -lh /var/www")
Returns output if successful, or None and logs error if not.
"""
cmd = shlex.split(rcmd)
executable = cmd[0]
executable_options=cmd[1:]
try:
proc = Popen(([executable] + executable_options), stdout=PIPE, stderr=PIPE)
response = proc.communicate()
response_stdout, response_stderr = response[0], response[1]
except OSError, e:
if e.errno == errno.ENOENT:
logging.debug( "Unable to locate '%s' program. Is it in your path?" % executable )
else:
logging.error( "O/S error occured when trying to run '%s': \"%s\"" % (executable, str(e)) )
except ValueError, e:
logging.debug( "Value error occured. Check your parameters." )
else:
if proc.wait() != 0:
logging.debug( "Executable '%s' returned with the error: \"%s\"" %(executable,response_stderr) )
return response
else:
logging.debug( "Executable '%s' returned successfully. First line of response was \"%s\"" %(executable, response_stdout.split('\n')[0] ))
return response_stdout
"""
#################################################
"""
def chooseInterface():
listDevices = (ex('ifconfig -a')
| 'grep HW'
|'awk \'{print $1}\'')
listDevices2 = listDevices.stdout()
listDevices3 = {} #initializing dictionary
for l in range(len(listDevices2.splitlines())):
devNum1 = listDevices2.splitlines()[l]
if devNum1 == "wlan0":
devNum1 = "PI-WLAN0-wlan0"
elif devNum1 == "put_your_wlan1_card":
devNum1 = "Name_your_card1"
elif devNum1 == "put_your_wlan2_card":
devNum1 = "Name_your_card2"
else:
devNum1 = "UNKNOWN DEVICE" + "-" + devNum1
listDevices3.update({l: devNum1})
print "[" + str(l) + "] " + listDevices3.get(l)
print "Please make a choice"
intAttacks2 = raw_input()
intAttacks2 = int(intAttacks2)
intAttacks1 = listDevices3.get(intAttacks2)
toShowIntAttacks = intAttacks1.split('-')[0]
chosenInterface = intAttacks1.split('-')[1]
print "Interface chosen is " + toShowIntAttacks
return chosenInterface
"""
#################################################
"""
global scan_interface
scan_interface = chooseInterface()
cm = "sudo ifconfig " + scan_interface + " up"
os.system(cm)
"""
#################################################
"""
def get_networks(iface, retry=10):
while retry > 0:
if "OK" in run_program("wpa_cli -i %s scan" % iface):
networks=[]
r = run_program("wpa_cli -i %s scan_result" % iface).strip()
if "bssid" in r and len ( r.split("\n") ) >1:
for line in r.split("\n")[1:]:
b, fr, s, f = line.split()[:4]
ss = " ".join(line.split()[4:]) #Hmm, dirty
if ss == "":
ss = "Hidden"
networks.append( {"bssid":b, "freq":fr, "sig":s, "ssid":ss, "flag":f} )
return networks
retry-=1
logging.debug("Couldn't retrieve networks, retrying")
time.sleep(0.5)
logging.error("Failed to list networks")
"""
#################################################
"""
def expandtabs(string, n, f):
result1 = ""
pos1 = 0
for char in string:
if char == "\t":
# instead of the tab character, append the
# number of spaces to the next tab stop
char = " " * (n - pos1 % n)
if char == "\r":
char = " " * (f - pos1 % f)
#pos = 60
if char == "\n":
pos1 = 0
else:
pos1 += len(char)
result1 += char
return result1
def showNetworks(pos):
print "[+] Searching for available networks..." #+ scan_interface
start_wpa(scan_interface)
networks = get_networks(scan_interface)
if networks:
DataBase = workingDirectory + "/" + DataBaseName +".db"
connection = sqlite3.connect(DataBase)
networks = sorted(networks, key=lambda k: k['sig'])
print "[+] There are "+ bcolors.GREEN + "" + str(len(networks)) + "" + bcolors.WHITE + " Networks in range:"
for network in networks:
ssid1 = network['ssid']
sig1 = network['sig']
bssid1 = network['bssid']
flag1 = network['flag']
freq1 = network['freq']
result = bcolors.BLUE + " SSID: " + network['ssid'] + "\t" + bcolors.GREEN + " Signal: " + network['sig'] + bcolors.RED + " BSSID: " + network['bssid'] + bcolors.WARNING + " Flags: " + network['flag'] + bcolors.BLUE + "\r" + " Freq: " + str(network['freq']) + bcolors.WHITE
#print " Sig:\t%s" % network['sig']
#print " BSSID:\t%s" % network['bssid']
#print " Flags:\t%s" % network['flag']
#print " Freq:\t%s\n" % network['freq']
#print "Trying to commit to db..."
print(expandtabs(result, 36
,78))
try:
connection.execute("insert into Scan (Position, BSSID, SSID, Flags, Freq ,Signal) values (?,?,?,?,?,?)", (pos, bssid1, ssid1, flag1, freq1, sig1))
connection.commit()
except:
print "Committing FAILED :("
connection.close()
else:
print "[W] No wireless networks detected :-("
"""
#################################################
"""
def _disconnect_all(_iface):
"""
Disconnect all wireless networks.
"""
lines = run_program("wpa_cli -i %s list_networks" % _iface).split("\n")
if lines:
for line in lines[1:-1]:
run_program("wpa_cli -i %s remove_network %s" % (_iface, line.split()[0]))
"""
#################################################
"""
def connect_to_network(_iface, _ssid, _type, _pass=None):
"""
Associate to a wireless network. Support _type options:
*WPA[2], WEP, OPEN
"""
_disconnect_all(_iface)
time.sleep(1)
if run_program("wpa_cli -i %s add_network" % _iface) == "0\n":
if run_program('wpa_cli -i %s set_network 0 ssid \'"%s"\'' % (_iface,_ssid)) == "OK\n":
if _type == "OPEN":
run_program("wpa_cli -i %s set_network 0 auth_alg OPEN" % _iface)
run_program("wpa_cli -i %s set_network 0 key_mgmt NONE" % _iface)
elif _type == "WPA" or _type == "WPA2":
run_program('wpa_cli -i %s set_network 0 psk \'"%s"\'' % (_iface,_pass))
elif _type == "WEP":
run_program('wpa_cli -i %s set_network 0 wep_key \'"%s"\'' % (_iface,_pass))
else:
logging.error("Unsupported type")
run_program("wpa_cli -i %s select_network 0" % _iface)
"""
#################################################
"""
def is_associated(_iface):
"""
Check if we're associated to a network.
"""
if "wpa_state=COMPLETED" in run_program("wpa_cli -i %s status" % _iface):
return True
return False
"""
#################################################
"""
def has_ip(_iface):
"""
Check if we have an IP address assigned
"""
status = run_program("wpa_cli -i %s status" % _iface)
r = re.search("ip_address=(.*)", status)
if r:
return r.group(1)
return False
"""
#################################################
"""
def do_dhcp(_iface):
"""
Request a DHCP lease.
"""
run_program("dhclient %s" % _iface)
"""
#################################################
"""
def RIGHT_spin(step):
step = int(step*conversion_factor)
while step > 0:
GPIO_SETUP(1,0,0,0)
GPIO_SETUP(1,1,0,0)
GPIO_SETUP(0,1,0,0)
GPIO_SETUP(0,1,1,0)
GPIO_SETUP(0,0,1,0)
GPIO_SETUP(0,0,1,1)
GPIO_SETUP(0,0,0,1)
GPIO_SETUP(1,0,0,1)
step -= 1
def RIGHT_TURN(arch_width,scn):
#degree = full_circle/360*arch_width
degree = int(arch_width)
while degree > 0.0:
RIGHT_spin(rotation_step)
degree -= rotation_step
if scn == "y":
time.sleep(2)
print "Degree is " + str(degree)
showNetworks(degree)
def LEFT_spin(step):
step = int(step*conversion_factor)
while step > 0:
GPIO_SETUP(1,0,0,1)
GPIO_SETUP(0,0,0,1)
GPIO_SETUP(0,0,1,1)
GPIO_SETUP(0,0,1,0)
GPIO_SETUP(0,1,1,0)
GPIO_SETUP(0,1,0,0)
GPIO_SETUP(1,1,0,0)
GPIO_SETUP(1,0,0,0)
step -= 1
def LEFT_TURN(arch_width,scn):
#degree = full_circle/360*arch_width
degree = int(arch_width)
while degree > 0.0:
LEFT_spin(rotation_step)
degree -= rotation_step
if scn == "y":
time.sleep(2)
print "Degree is " + str(degree)
showNetworks(degree)
"""
#################################################
"""
def connection(direction):
DataBase = workingDirectory + "/" + DataBaseName +".db"
connection = sqlite3.connect(DataBase)
c = connection.cursor()
#query = "select SSID from Scan group by SSID;"
#query = "select SSID,Flags,Min(Signal) from ( select * from Scan order by Signal) group by SSID ;"
query = "select SSID,Flags,Min(Signal) from Scan group by SSID ;"
c.execute(query)
r = c.fetchall()
listAPs = {}
for i in range(len(r)):
result2 = "[" + str(i) + "] - " + bcolors.BLUE + r[i][0] + bcolors.RED + "\t" + " Encryption: " + r[i][1] + bcolors.GREEN + "\r" + " Best Signal: " + r[i][2] + bcolors.WHITE
print(expandtabs(result2, 38, 78))
listAPs.update({i: r[i][0]})
selAP = raw_input("Please make a choice:")
selAP = int(selAP)
selAP1 = listAPs.get(selAP)
print "Going with " + selAP1
query1 = "select * from Scan where SSID is '" + selAP1 + "' order by Signal limit 1;"
c.execute(query1)
r = c.fetchall()
Aim = r[0][1]
Aim = float(Aim)
Enc = r[0][4]
if "WPA" in Enc:
Enc1 = "WPA"
elif "WPA2" in Enc:
Enc1 = "WPA"
elif "WEP" in Enc:
Enc1 = "WEP"
else:
Enc1 = "OPEN"
print "Aim " + str(Aim)
deg1 = int(arch_width)-int(Aim)
#deg1 = int(Aim*conversion_factor)
print "Getting in position"
if direction == "R":
RIGHT_spin(deg1)
else:
LEFT_spin(deg1)
print "deg1 " + str(deg1)
connection.close()
GPIO.cleanup()
if Enc1 != "OPEN":
PASSWD = raw_input("Please provide the password required:")
else:
PASSWD = ""
connect_to_network(scan_interface, selAP1, Enc1, PASSWD)
while not is_associated(scan_interface):
time.sleep(1)
print "Success."
sys.stdout.write("[+] Requesting DHCP lease... ")
sys.stdout.flush()
do_dhcp(scan_interface)
while not has_ip(scan_interface):
time.sleep(1)
print "Success. (%s)" % has_ip(scan_interface)
print "[+] Associated and got lease. Good to go..."
if direction_input == "R":
RIGHT_TURN(arch_width,"y")
print "Returning to initial position"
LEFT_TURN(arch_width,"n")
print "DONE"
connection("R")
elif direction_input == "L":
LEFT_TURN(arch_width,"y")
print "Returning to initial position"
RIGHT_TURN(arch_width,"n")
print "DONE"
connection("L")
else:
GPIO_SETUP(0,0,0,0)
GPIO.cleanup()
© Scattiscattanti 2014-2018
