Edit File: p2p_connect.py
#!/usr/bin/python # Tests p2p_connect # Will try to connect to another peer # and form a group ######### MAY NEED TO RUN AS SUDO ############# import dbus import sys, os import time import gobject import getopt from dbus.mainloop.glib import DBusGMainLoop def usage(): print "Usage:" print " %s -i <interface_name> -m <wps_method> \ " \ % sys.argv[0] print " -a <addr> [-p <pin>] [-g <go_intent>] \ " print " [-w <wpas_dbus_interface>]" print "Options:" print " -i = interface name" print " -m = wps method" print " -a = peer address" print " -p = pin number (8 digits)" print " -g = group owner intent" print " -w = wpas dbus interface = fi.w1.wpa_supplicant1" print "Example:" print " %s -i wlan0 -a 0015008352c0 -m display -p 12345670" % sys.argv[0] # Required Signals def GONegotiationSuccess(status): print "Go Negotiation Success" def GONegotiationFailure(status): print 'Go Negotiation Failed. Status:' print format(status) os._exit(0) def GroupStarted(properties): if properties.has_key("group_object"): print 'Group Formation Complete %s' \ % properties["group_object"] os._exit(0) def WpsFailure(status, etc): print "WPS Authentication Failure".format(status) print etc os._exit(0) class P2P_Connect(): # Needed Variables global bus global wpas_object global interface_object global p2p_interface global ifname global wpas global wpas_dbus_interface global timeout global path global wps_method global go_intent global addr global pin # Dbus Paths global wpas_dbus_opath global wpas_dbus_interfaces_opath global wpas_dbus_interfaces_interface global wpas_dbus_interfaces_p2pdevice # Dictionary of Arguements global p2p_connect_arguements # Constructor def __init__(self,ifname,wpas_dbus_interface,addr, pin,wps_method,go_intent): # Initializes variables and threads self.ifname = ifname self.wpas_dbus_interface = wpas_dbus_interface self.wps_method = wps_method self.go_intent = go_intent self.addr = addr self.pin = pin # Generating interface/object paths self.wpas_dbus_opath = \ "/" + self.wpas_dbus_interface.replace(".","/") self.wpas_wpas_dbus_interfaces_opath = \ self.wpas_dbus_opath + "/Interfaces" self.wpas_dbus_interfaces_interface = \ self.wpas_dbus_interface + ".Interface" self.wpas_dbus_interfaces_p2pdevice = \ self.wpas_dbus_interfaces_interface + ".P2PDevice" # Getting interfaces and objects DBusGMainLoop(set_as_default=True) self.bus = dbus.SystemBus() self.wpas_object = self.bus.get_object( self.wpas_dbus_interface, self.wpas_dbus_opath) self.wpas = dbus.Interface( self.wpas_object, self.wpas_dbus_interface) # See if wpa_supplicant already knows about this interface self.path = None try: self.path = self.wpas.GetInterface(ifname) except: if not str(exc).startswith( self.wpas_dbus_interface + \ ".InterfaceUnknown:"): raise exc try: path = self.wpas.CreateInterface( {'Ifname': ifname, 'Driver': 'test'}) time.sleep(1) except dbus.DBusException, exc: if not str(exc).startswith( self.wpas_dbus_interface + \ ".InterfaceExists:"): raise exc # Get Interface and objects self.interface_object = self.bus.get_object( self.wpas_dbus_interface,self.path) self.p2p_interface = dbus.Interface( self.interface_object, self.wpas_dbus_interfaces_p2pdevice) # Add signals self.bus.add_signal_receiver(GONegotiationSuccess, dbus_interface=self.wpas_dbus_interfaces_p2pdevice, signal_name="GONegotiationSuccess") self.bus.add_signal_receiver(GONegotiationFailure, dbus_interface=self.wpas_dbus_interfaces_p2pdevice, signal_name="GONegotiationFailure") self.bus.add_signal_receiver(GroupStarted, dbus_interface=self.wpas_dbus_interfaces_p2pdevice, signal_name="GroupStarted") self.bus.add_signal_receiver(WpsFailure, dbus_interface=self.wpas_dbus_interfaces_p2pdevice, signal_name="WpsFailed") #Constructing all the arguements needed to connect def constructArguements(self): # Adding required arguements self.p2p_connect_arguements = {'wps_method':self.wps_method, 'peer':dbus.ObjectPath(self.path+'/Peers/'+self.addr)} # Display requires a pin, and a go intent of 15 if (self.wps_method == 'display'): if (self.pin != None): self.p2p_connect_arguements.update({'pin':self.pin}) else: print "Error:\n Pin required for wps_method=display" usage() quit() if (self.go_intent != None and int(self.go_intent) != 15): print "go_intent overwritten to 15" self.go_intent = '15' # Keypad requires a pin, and a go intent of less than 15 elif (self.wps_method == 'keypad'): if (self.pin != None): self.p2p_connect_arguements.update({'pin':self.pin}) else: print "Error:\n Pin required for wps_method=keypad" usage() quit() if (self.go_intent != None and int(self.go_intent) == 15): error = "Error :\n Group Owner intent cannot be" + \ " 15 for wps_method=keypad" print error usage() quit() # Doesn't require pin # for ./wpa_cli, p2p_connect [mac] [pin#], wps_method=keypad elif (self.wps_method == 'pin'): if (self.pin != None): print "pin ignored" # No pin is required for pbc so it is ignored elif (self.wps_method == 'pbc'): if (self.pin != None): print "pin ignored" else: print "Error:\n wps_method not supported or does not exist" usage() quit() # Go_intent is optional for all arguements if (self.go_intent != None): self.p2p_connect_arguements.update( {'go_intent':dbus.Int32(self.go_intent)}) # Running p2p_connect def run(self): try: result_pin = self.p2p_interface.Connect( self.p2p_connect_arguements) except dbus.DBusException, exc: raise exc if (self.wps_method == 'pin' and \ not self.p2p_connect_arguements.has_key('pin') ): print "Connect return with pin value of %d " % int(result_pin) gobject.MainLoop().run() if __name__ == "__main__": # Required interface_name = None wps_method = None addr = None # Conditionally optional pin = None # Optional wpas_dbus_interface = 'fi.w1.wpa_supplicant1' go_intent = None # Using getopts to handle options try: options, args = getopt.getopt(sys.argv[1:],"hi:m:a:p:g:w:") except getopt.GetoptError: usage() quit() # If theres a switch, override default option for key, value in options: # Help if (key == "-h"): usage() quit() # Interface Name elif (key == "-i"): interface_name = value # WPS Method elif (key == "-m"): wps_method = value # Address elif (key == "-a"): addr = value # Pin elif (key == "-p"): pin = value # Group Owner Intent elif (key == "-g"): go_intent = value # Dbus interface elif (key == "-w"): wpas_dbus_interface = value else: assert False, "unhandled option" # Required Arguements check if (interface_name == None or wps_method == None or addr == None): print "Error:\n Required arguements not specified" usage() quit() # Group Owner Intent Check if (go_intent != None and (int(go_intent) > 15 or int(go_intent) < 0) ): print "Error:\n Group Owner Intent must be between 0 and 15 inclusive" usage() quit() # Pin Check if (pin != None and len(pin) != 8): print "Error:\n Pin is not 8 digits" usage() quit() try: p2p_connect_test = P2P_Connect(interface_name,wpas_dbus_interface, addr,pin,wps_method,go_intent) except: print "Error:\n Invalid Arguements" usage() quit() p2p_connect_test.constructArguements() p2p_connect_test.run() os._exit(0)