blob: 8cc76e818379ea51e6320b6ff25077261271437e [file] [log] [blame]
#!/usr/bin/python
from os import O_NONBLOCK
from sys import stdin, stdout, exit, version_info, argv
from fcntl import fcntl, F_GETFL, F_SETFL
import glib
import dbus
import dbus.mainloop.glib
import gobject
import argparse
WPA_NAME='fi.w1.wpa_supplicant1'
WPA_INTF='fi.w1.wpa_supplicant1'
WPA_PATH='/fi/w1/wpa_supplicant1'
WPA_IF_INTF = WPA_INTF + '.Interface'
WPA_P2P_INTF = WPA_IF_INTF + '.P2PDevice'
WPA_GROUP_INTF = WPA_INTF + '.Group'
WPA_PEER_INTF = WPA_INTF + '.Peer'
DBUS_PROPERTIES_INTF = 'org.freedesktop.DBus.Properties'
P2P_GROUP_CAPAB_GROUP_OWNER = 1 << 0
class ArgFields:
for field in ('help', 'metavar'):
exec('{}="{}"'.format(field, field))
class InputLine:
def __init__(self, handler):
self.line = ''
self.handler = handler
flags = fcntl(stdin.fileno(), F_GETFL)
flags |= O_NONBLOCK
fcntl(stdin.fileno(), F_SETFL, flags)
glib.io_add_watch(stdin, glib.IO_IN, self.input_cb)
self.prompt()
def prompt(self):
self.line = ''
print '> ',
stdout.flush()
def input_cb(self, fd, event):
if event != glib.IO_IN:
return
self.line += fd.read();
for line in self.line.split('\n'):
line = line.strip()
if len(line) == 0:
break
self.handler(line.strip())
self.prompt()
return True
def error_print(ex):
print 'Command Error: %s' % ex
def checkarg(nb_args = 0, min_args = False):
def under(function):
def wrapper(*args, **kwargs):
resuls = True
if min_args:
result = len(args[1]) < nb_args
else:
result = len(args[1]) != nb_args
if result:
raise Exception('Command %s takes %s arguments' %
(function.__name__, nb_args))
return function(*args, **kwargs)
return wrapper
return under
def print_dict(d):
for k in d:
try:
if type(d[k]) is dbus.Byte:
print 'Key %s --> 0x%x' % (k, d[k])
else:
print 'Key %s --> %s' % (k, d[k])
except:
print "Error: Key %s content cannot be printed" % k
pass
def print_tuple(t):
for e in t:
if type(e) is dbus.Dictionary:
print_dict(e)
else:
print 'Element: %s' % e
class Wpa_s:
def __init__(self, bus, iface_name, command):
self.wpa = dbus.Interface(bus.get_object(WPA_NAME, WPA_PATH), WPA_INTF)
bus.add_signal_receiver(self.__wpa_property_changed, path=WPA_PATH,
member_keyword='signal')
bus.add_signal_receiver(self.__InterfaceAdded, path=WPA_PATH,
signal_name='InterfaceAdded')
bus.add_signal_receiver(self.__InterfaceRemoved, path=WPA_PATH,
signal_name='InterfaceRemoved')
self.__reset()
self.bus = bus
self.debug = False
self.line_in = InputLine(self.__command)
if iface_name:
try:
self.create_if([iface_name])
except:
print "Error creating interface: %s" % iface_name
if len(command.strip(' ')):
self.__command(command)
def help(self, args):
list = self.command_list.keys()
list.sort()
for key in list:
help = ''
if (self.command_list[key].has_key(ArgFields.help)):
help = self.command_list[key][ArgFields.help]
print "%s\t%s" % (key.rjust(25), help.ljust(50))
def __command(self, cmd_line):
cmd = cmd_line.split(' ')
try:
func = getattr(self, cmd[0])
except Exception, e:
print 'Error: command unknown - %s' % e
return
try:
func(cmd[1:])
except Exception, e:
error_print(e)
def __wpa_property_changed(*args, **kwargs):
print 'WPA - Signal: %s' % kwargs.get('signal')
def __if_property_changed(*args, **kwargs):
signal = kwargs.get('signal')
print 'IF - Signal: %s' % signal
if signal == 'BSSAdded':
return
if args[0].debug:
print_tuple(args[1:])
def __p2p_property_changed(*args, **kwargs):
print 'IF P2P - Signal: %s' % kwargs.get('signal')
if args[0].debug:
print_tuple(args[1:])
def __peer_if_p2p_property_changed(*args, **kwargs):
print 'Peer - ',
args[0].__p2p_property_changed(*args, **kwargs)
def __DeviceFound(self, object_path):
self.peers[object_path] = None
peer = self.bus.get_object(WPA_INTF, object_path)
peer_if = dbus.Interface(peer, DBUS_PROPERTIES_INTF)
self.bus.add_signal_receiver(self.__peer_if_p2p_property_changed,
dbus_interface=WPA_PEER_INTF,
path=object_path, member_keyword='signal')
self.peers[object_path] = peer_if.GetAll(WPA_PEER_INTF)
if self.debug:
print_dict(self.peers[object_path])
def __DeviceLost(self, object_path):
if object_path in self.peers:
del self.peers[object_path]
def __PeerJoined(self, object_path):
print 'Peer %s joined' % object_path
def __PeerDisconnected(self, object_path):
print 'Peer %s disconnected' % object_path
def __group_if_property_changed(*args, **kwargs):
print 'Group - ',
args[0].__if_property_changed(*args, **kwargs)
def __group_if_p2p_property_changed(*args, **kwargs):
print 'Group - ',
args[0].__p2p_property_changed(*args, **kwargs)
def __GroupFinished(self, properties):
print 'Group running on %s is being removed' % ifname
self.group_obj = self.group_if = self.group_iface_path = None
if self.debug:
print_dict(properties)
def __InvitationResult(self, response):
print 'Invitation result status: %d ' % response['status']
if response.has_key('bssid'):
print 'bssid: %s' % response['bssid']
if self.debug:
print_dict(response)
def __GroupStarted(self, properties):
self.group_obj = properties['group_object']
self.bus.add_signal_receiver(self.__PeerJoined,
dbus_interface=WPA_GROUP_INTF,
path=self.group_obj,
signal_name='PeerJoined')
self.bus.add_signal_receiver(self.__PeerDisconnected,
dbus_interface=WPA_GROUP_INTF,
path=self.group_obj,
signal_name='PeerDisconnected')
self.group_iface_path = properties['interface_object']
self.group_if = dbus.Interface(self.bus.get_object(WPA_INTF,
self.group_iface_path),
WPA_P2P_INTF)
self.bus.add_signal_receiver(self.__group_if_property_changed,
dbus_interface=WPA_IF_INTF,
path=self.group_iface_path,
member_keyword='signal')
self.bus.add_signal_receiver(self.__group_if_p2p_property_changed,
dbus_interface=WPA_P2P_INTF,
path=self.group_iface_path,
member_keyword='signal')
self.bus.add_signal_receiver(self.__GroupFinished,
dbus_interface=WPA_P2P_INTF,
path=self.group_iface_path,
signal_name='GroupFinished')
self.bus.add_signal_receiver(self.__InvitationResult,
dbus_interface=WPA_P2P_INTF,
path=self.iface_path,
signal_name='InvitationResult')
if self.debug:
group = dbus.Interface(self.bus.get_object(WPA_INTF,
self.group_obj),
DBUS_PROPERTIES_INTF)
print_dict(group.GetAll(WPA_GROUP_INTF))
def __ServiceDiscoveryResponse(self, response):
peer = response['peer_object']
if peer in self.peers:
print 'Peer %s has this TLVs:' % (self.peers[peer]['DeviceName'])
print response['tlvs']
def __InterfaceAdded(self, path, properties):
print 'Interface %s Added (%s)' % (properties['Ifname'], path)
if self.debug:
print_dict(properties)
p2p = dbus.Interface(self.bus.get_object(WPA_INTF,
path), DBUS_PROPERTIES_INTF)
print_dict(p2p.GetAll(WPA_P2P_INTF))
def __InterfaceRemoved(self, path):
print 'Interface Removed (%s)' % (path)
def __listen_if_signals(self):
self.bus.add_signal_receiver(self.__if_property_changed,
dbus_interface=WPA_IF_INTF,
path=self.iface_path,
member_keyword='signal')
self.bus.add_signal_receiver(self.__p2p_property_changed,
dbus_interface=WPA_P2P_INTF,
path=self.iface_path,
member_keyword='signal')
self.bus.add_signal_receiver(self.__GroupStarted,
dbus_interface=WPA_P2P_INTF,
path=self.iface_path,
signal_name='GroupStarted')
self.bus.add_signal_receiver(self.__DeviceFound,
dbus_interface=WPA_P2P_INTF,
path=self.iface_path,
signal_name='DeviceFound')
self.bus.add_signal_receiver(self.__DeviceLost,
dbus_interface=WPA_P2P_INTF,
path=self.iface_path,
signal_name='DeviceLost')
self.bus.add_signal_receiver(self.__ServiceDiscoveryResponse,
dbus_interface=WPA_P2P_INTF,
path=self.iface_path,
signal_name='ServiceDiscoveryResponse')
def __reset(self):
self.iface_path = self.iface_name = self.iface = None
self.p2p = self.group_if = self.group_obj = None
self.peers = {}
def __set_if(self, iface_name):
self.iface = dbus.Interface(self.bus.get_object(WPA_INTF,
self.iface_path), WPA_IF_INTF)
self.p2p = dbus.Interface(self.bus.get_object(WPA_INTF,
self.iface_path), WPA_P2P_INTF)
p2p_if = dbus.Interface(self.p2p, DBUS_PROPERTIES_INTF)
p2p_if.Set(WPA_P2P_INTF, 'P2PDeviceConfig',
dbus.Dictionary({ 'DeviceName' : 'ConnManP2P' },
signature='sv'))
print 'Interface %s: %s' % (iface_name, self.iface_path)
self.iface_name = iface_name
self.__listen_if_signals()
@checkarg()
def enable_debug(self, args):
self.debug = True
@checkarg()
def disable_debug(self, args):
self.debug = False
@checkarg(nb_args=1)
def create_if(self, args):
self.__reset()
self.iface_path = self.wpa.CreateInterface(({ 'Ifname' : args[0]} ))
self.__set_if(args[0])
@checkarg(nb_args=1)
def get_if(self, args):
self.__reset()
self.iface_path = self.wpa.GetInterface(args[0])
self.__set_if(args[0])
@checkarg()
def del_if(self, args = None):
if not self.iface_path:
return
self.wpa.RemoveInterface(self.iface_path)
print 'Interface %s removed' % self.iface_name
self.__reset()
@checkarg()
def scan(self, args = None):
if not self.iface:
return
self.iface.Scan(({ 'Type': 'passive' }))
print 'Scan started'
@checkarg()
def quit(self, args = None):
self.del_if(args)
exit(0)
@checkarg(nb_args=1)
def set_command_list(self, command_list):
self.command_list = command_list[0]
@checkarg()
def p2p_find(self, args = None):
if not self.p2p:
return
self.p2p.Find(({}))
@checkarg()
def p2p_stop_find(self, args = None):
if not self.p2p:
return
self.p2p.StopFind()
@checkarg()
def p2p_peers(self, args = None):
if not self.iface:
return
for p in self.peers:
print 'Peer Name=%s' % (self.peers[p]['DeviceName'])
def __find_peer(self, peer_name, ret_object_path = False):
if len(self.peers) == 0:
return None
peer = None
for p in self.peers:
if self.peers[p]['DeviceName'] == peer_name:
peer = self.peers[p]
break
if not peer:
print 'No peer found under the name: %s' % peer_name
p = None
if ret_object_path:
return p
else:
return peer
@checkarg(nb_args = 1)
def p2p_peer(self, args):
peer_path = self.__find_peer(args[0], True)
if peer_path:
peer = self.bus.get_object(WPA_INTF, peer_path)
peer_if = dbus.Interface(peer, DBUS_PROPERTIES_INTF)
self.peers[peer_path] = peer_if.GetAll(WPA_PEER_INTF)
print_dict(self.peers[peer_path])
@checkarg(nb_args = 1)
def p2p_connect(self, args):
if not self.p2p:
return
peer = self.__find_peer(args[0])
if not peer:
return
peer_path = self.__find_peer(args[0], True)
if (peer['groupcapability'] & P2P_GROUP_CAPAB_GROUP_OWNER ==
P2P_GROUP_CAPAB_GROUP_OWNER):
print 'Joining an existing P2P group'
pin = self.p2p.Connect(({ 'peer' : peer_path,
'wps_method' : 'pbc',
'join' : True,
'go_intent' : 0 }))
else:
print 'Associating with another P2P device'
pin = self.p2p.Connect(({ 'peer' : peer_path,
'wps_method' : 'pbc',
'join' : False,
'go_intent' : 7 }))
if not pin:
print 'WPS PIN in use: %s' % pin
@checkarg(nb_args = 1)
def p2p_disconnect(self, args):
if not self.p2p:
return
peer = self.__find_peer(args[0])
if not peer:
return
if not self.group_if:
print 'Peer %s is not connected' % (peer['DeviceName'])
return
self.group_if.Disconnect()
@checkarg()
def p2p_group_add(self, args):
if not self.p2p:
return
self.p2p.GroupAdd(({ 'persistent' : dbus.Boolean(1) }))
@checkarg()
def p2p_group_remove(self, args):
if not self.group_if:
return
self.group_if.Disconnect()
@checkarg()
def p2p_group(self, args):
if not self.group_obj:
return
group = dbus.Interface(self.bus.get_object(WPA_INTF,
self.group_obj), DBUS_PROPERTIES_INTF)
print_dict(group.GetAll(WPA_GROUP_INTF))
@checkarg()
def p2p_flush(self, args):
if not self.p2p:
return
self.p2p.Flush()
@checkarg()
def p2p_serv_disc_req(self, args = None):
if not self.p2p:
return
""" We request all kind of services """
sd_req = dbus.Array(signature='y', variant_level=1)
for a in [2,0,0,1]:
sd_req.append(dbus.Byte(a))
ref = self.p2p.ServiceDiscoveryRequest(({ 'tlv' : sd_req }))
print 'Service discovery reference: %s' % ref
@checkarg(nb_args = 1)
def p2p_serv_disc_cancel_req(self, args):
if not self.p2p:
return
self.p2p.ServiceDiscoveryCancelRequest(int(args[0]))
@checkarg(nb_args = 3)
def p2p_service_add(self, args):
if not self.p2p:
return
service = { 'service_type' : args[0] }
if args[0] == 'upnp':
service['version'] = args[1]
service['service'] = args[2]
elif args[0] == 'bonjour':
service['query'] = args[1]
service['response'] = args[2]
else:
print 'Unknown service: %s' % args[0]
return
self.p2p.AddService((service))
@checkarg(nb_args = 2, min_args = True)
def p2p_service_del(self, args):
if not self.p2p:
return
service = { 'service_type' : args[0] }
if args[0] == 'upnp':
service['version'] = args[1]
service['service'] = args[2]
elif args[0] == 'bonjour':
service['query'] = args[1]
else:
print 'Unknown service: %s' % args[0]
return
self.p2p.DeleteService((service))
@checkarg()
def p2p_service_flush(self, args = None):
if not self.p2p:
return
self.p2p.FlushService()
@checkarg(nb_args = 1)
def p2p_invite(self, args):
if not self.p2p or not self.group_if:
return
peer_path = self.__find_peer(args[0], True)
if not peer_path:
return
self.group_if.Invite({ 'peer' : peer_path})
def build_args(parser):
parser.add_argument('-d', default=False, action='store_true',
dest='debug', help='enable debug')
parser.add_argument('-i', metavar='<interface>', dest='ifname',
help='interface name')
command = {}
command['quit'] = {}
command['enable_debug'] = {}
command['disable_debug'] = {}
command['create_if'] = {ArgFields.help:'<iface_name> - create interface'}
command['get_if'] = {ArgFields.help:'<iface_name> - get interface'}
command['del_if'] = {ArgFields.help:'removes current interface'}
command['scan'] = {}
command['p2p_find'] = {}
command['p2p_stop_find'] = {}
command['p2p_flush'] = {}
command['p2p_group_add'] = {ArgFields.help:'adds an autonomous group'}
command['p2p_group_remove'] = {}
command['p2p_group'] = {}
command['p2p_peers'] = {}
command['p2p_peer'] = {ArgFields.help:'<p2p device name> - get info for a '
'peer'}
command['p2p_connect'] = {ArgFields.help:'<p2p device name>'}
command['p2p_disconnect'] = {ArgFields.help:'<p2p device name>'}
command['p2p_serv_disc_req'] = {}
command['p2p_serv_disc_cancel_req'] = {ArgFields.help:'<identifier>'}
command['p2p_service_add'] = {ArgFields.help:'<service type> '
'<version/query> <service/response>'}
command['p2p_service_del'] = {ArgFields.help:'<service type> '
'<version/query> [<service>]'}
command['p2p_service_flush'] = {}
command['p2p_invite'] = {ArgFields.help:'<p2p device name>'}
command_list = command.keys()
command_list.sort()
subparsers = parser.add_subparsers(help='commands', dest='command')
subparsers.add_parser('')
for key in command_list:
help=None
metavar=None
if command[key].has_key(ArgFields.help):
help = command[key][ArgFields.help]
if command[key].has_key(ArgFields.metavar):
metavar = command[key][ArgFields.metavar]
command_parser = subparsers.add_parser(key, help=help)
command_parser.add_argument(key, nargs='*', metavar=metavar, help=help)
return command
def main():
if version_info.major != 2:
print 'You need to run this under Python 2.x'
exit(1)
parser = argparse.ArgumentParser(description='Connman P2P Test')
command_list = build_args(parser)
argv[1:] += ['']
args = parser.parse_args(argv[1:])
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
opts = []
if args.command:
opts = getattr(args, args.command)
params = ''
if opts and len(opts[0]):
params = ' ' + ''.join(opts)
bus = dbus.SystemBus()
mainloop = gobject.MainLoop()
wpa_s = Wpa_s(bus, args.ifname, args.command + params)
if (args.debug):
wpa_s.enable_debug([])
wpa_s.set_command_list([command_list])
mainloop.run()
if __name__ == '__main__':
main()