test/simple-agent

   1 #!/usr/bin/python
   2
   3 from __future__ import absolute_import, print_function, unicode_literals
   4
   5 from optparse import OptionParser
   6 import sys
   7 import dbus
   8 import dbus.service
   9 import dbus.mainloop.glib
  10 try:
  11   from gi.repository import GObject
  12 except ImportError:
  13   import gobject as GObject
  14 import bluezutils
  15
  16 BUS_NAME = 'org.bluez'
  17 AGENT_INTERFACE = 'org.bluez.Agent1'
  18 AGENT_PATH = "/test/agent"
  19
  20 bus = None
  21 device_obj = None
  22 dev_path = None
  23
  24 def ask(prompt):
  25         try:
  26                 return raw_input(prompt)
  27         except:
  28                 return input(prompt)
  29
  30 def set_trusted(path):
  31         props = dbus.Interface(bus.get_object("org.bluez", path),
  32                                         "org.freedesktop.DBus.Properties")
  33         props.Set("org.bluez.Device1", "Trusted", True)
  34
  35 def dev_connect(path):
  36         dev = dbus.Interface(bus.get_object("org.bluez", path),
  37                                                         "org.bluez.Device1")
  38         dev.Connect()
  39
  40 class Rejected(dbus.DBusException):
  41         _dbus_error_name = "org.bluez.Error.Rejected"
  42
  43 class Agent(dbus.service.Object):
  44         exit_on_release = True
  45
  46         def set_exit_on_release(self, exit_on_release):
  47                 self.exit_on_release = exit_on_release
  48
  49         @dbus.service.method(AGENT_INTERFACE,
  50                                         in_signature="", out_signature="")
  51         def Release(self):
  52                 print("Release")
  53                 if self.exit_on_release:
  54                         mainloop.quit()
  55
  56         @dbus.service.method(AGENT_INTERFACE,
  57                                         in_signature="os", out_signature="")
  58         def AuthorizeService(self, device, uuid):
  59                 print("AuthorizeService (%s, %s)" % (device, uuid))
  60                 authorize = ask("Authorize connection (yes/no): ")
  61                 if (authorize == "yes"):
  62                         return
  63                 raise Rejected("Connection rejected by user")
  64
  65         @dbus.service.method(AGENT_INTERFACE,
  66                                         in_signature="o", out_signature="s")
  67         def RequestPinCode(self, device):
  68                 print("RequestPinCode (%s)" % (device))
  69                 set_trusted(device)
  70                 return ask("Enter PIN Code: ")
  71
  72         @dbus.service.method(AGENT_INTERFACE,
  73                                         in_signature="o", out_signature="u")
  74         def RequestPasskey(self, device):
  75                 print("RequestPasskey (%s)" % (device))
  76                 set_trusted(device)
  77                 passkey = ask("Enter passkey: ")
  78                 return dbus.UInt32(passkey)
  79
  80         @dbus.service.method(AGENT_INTERFACE,
  81                                         in_signature="ouq", out_signature="")
  82         def DisplayPasskey(self, device, passkey, entered):
  83                 print("DisplayPasskey (%s, %06u entered %u)" %
  84                                                 (device, passkey, entered))
  85
  86         @dbus.service.method(AGENT_INTERFACE,
  87                                         in_signature="os", out_signature="")
  88         def DisplayPinCode(self, device, pincode):
  89                 print("DisplayPinCode (%s, %s)" % (device, pincode))
  90
  91         @dbus.service.method(AGENT_INTERFACE,
  92                                         in_signature="ou", out_signature="")
  93         def RequestConfirmation(self, device, passkey):
  94                 print("RequestConfirmation (%s, %06d)" % (device, passkey))
  95                 confirm = ask("Confirm passkey (yes/no): ")
  96                 if (confirm == "yes"):
  97                         set_trusted(device)
  98                         return
  99                 raise Rejected("Passkey doesn't match")
 100
 101         @dbus.service.method(AGENT_INTERFACE,
 102                                         in_signature="o", out_signature="")
 103         def RequestAuthorization(self, device):
 104                 print("RequestAuthorization (%s)" % (device))
 105                 auth = ask("Authorize? (yes/no): ")
 106                 if (auth == "yes"):
 107                         return
 108                 raise Rejected("Pairing rejected")
 109
 110         @dbus.service.method(AGENT_INTERFACE,
 111                                         in_signature="", out_signature="")
 112         def Cancel(self):
 113                 print("Cancel")
 114
 115 def pair_reply():
 116         print("Device paired")
 117         set_trusted(dev_path)
 118         dev_connect(dev_path)
 119         mainloop.quit()
 120
 121 def pair_error(error):
 122         err_name = error.get_dbus_name()
 123         if err_name == "org.freedesktop.DBus.Error.NoReply" and device_obj:
 124                 print("Timed out. Cancelling pairing")
 125                 device_obj.CancelPairing()
 126         else:
 127                 print("Creating device failed: %s" % (error))
 128
 129
 130         mainloop.quit()
 131
 132 if __name__ == '__main__':
 133         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
 134
 135         bus = dbus.SystemBus()
 136
 137         capability = "KeyboardDisplay"
 138
 139         parser = OptionParser()
 140         parser.add_option("-i", "--adapter", action="store",
 141                                         type="string",
 142                                         dest="adapter_pattern",
 143                                         default=None)
 144         parser.add_option("-c", "--capability", action="store",
 145                                         type="string", dest="capability")
 146         parser.add_option("-t", "--timeout", action="store",
 147                                         type="int", dest="timeout",
 148                                         default=60000)
 149         (options, args) = parser.parse_args()
 150         if options.capability:
 151                 capability  = options.capability
 152
 153         path = "/test/agent"
 154         agent = Agent(bus, path)
 155
 156         mainloop = GObject.MainLoop()
 157
 158         obj = bus.get_object(BUS_NAME, "/org/bluez");
 159         manager = dbus.Interface(obj, "org.bluez.AgentManager1")
 160         manager.RegisterAgent(path, capability)
 161
 162         print("Agent registered")
 163
 164         # Fix-up old style invocation (BlueZ 4)
 165         if len(args) > 0 and args[0].startswith("hci"):
 166                 options.adapter_pattern = args[0]
 167                 del args[:1]
 168
 169         if len(args) > 0:
 170                 device = bluezutils.find_device(args[0],
 171                                                 options.adapter_pattern)
 172                 dev_path = device.object_path
 173                 agent.set_exit_on_release(False)
 174                 device.Pair(reply_handler=pair_reply, error_handler=pair_error,
 175                                                                 timeout=60000)
 176                 device_obj = device
 177         else:
 178                 manager.RequestDefaultAgent(path)
 179
 180         mainloop.run()
 181
 182         #adapter.UnregisterAgent(path)
 183         #print("Agent unregistered")