003 File Manager
Current Path:
/usr/src/contrib/wpa/wpa_supplicant/examples/p2p
usr
/
src
/
contrib
/
wpa
/
wpa_supplicant
/
examples
/
p2p
/
📁
..
📄
p2p_connect.py
(7.47 KB)
📄
p2p_disconnect.py
(4.02 KB)
📄
p2p_find.py
(4.62 KB)
📄
p2p_flush.py
(4.01 KB)
📄
p2p_group_add.py
(5.7 KB)
📄
p2p_invite.py
(5.13 KB)
📄
p2p_listen.py
(4.3 KB)
📄
p2p_stop_find.py
(4.19 KB)
Editing: 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 as 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 as 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)
Upload File
Create Folder