cloudeebus/cloudeebus.py

   1 #!/usr/bin/env python
   2
   3 # Cloudeebus
   4 #
   5 # Copyright 2012 Intel Corporation.
   6 #
   7 # Licensed under the Apache License, Version 2.0 (the "License");
   8 # you may not use this file except in compliance with the License.
   9 # You may obtain a copy of the License at
  10 #
  11 # http://www.apache.org/licenses/LICENSE-2.0
  12 #
  13 # Unless required by applicable law or agreed to in writing, software
  14 # distributed under the License is distributed on an "AS IS" BASIS,
  15 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16 # See the License for the specific language governing permissions and
  17 # limitations under the License.
  18 #
  19 # Luc Yriarte <luc.yriarte@intel.com>
  20 # Christophe Guiraud <christophe.guiraud@intel.com>
  21 #
  22
  23
  24 import argparse, dbus, json, sys
  25
  26 from twisted.internet import glib2reactor
  27 # Configure the twisted mainloop to be run inside the glib mainloop.
  28 # This must be done before importing the other twisted modules
  29 glib2reactor.install()
  30 from twisted.internet import reactor, defer
  31
  32 from autobahn.websocket import listenWS
  33 from autobahn.wamp import exportRpc, WampServerFactory, WampCraServerProtocol
  34
  35 from dbus.mainloop.glib import DBusGMainLoop
  36
  37 import gobject
  38 gobject.threads_init()
  39
  40 from dbus import glib
  41 glib.init_threads()
  42
  43 # enable debug log
  44 from twisted.python import log
  45
  46
  47
  48 ###############################################################################
  49
  50 VERSION = "0.2"
  51 OPENDOOR = False
  52 CREDENTIALS = {}
  53 WHITELIST = []
  54
  55 ###############################################################################
  56 class DbusCache:
  57     '''
  58     Global cache of DBus connexions and signal handlers
  59     '''
  60     def __init__(self):
  61         self.dbusConnexions = {}
  62         self.signalHandlers = {}
  63
  64
  65     def reset(self):
  66         '''
  67         Disconnect signal handlers before resetting cache.
  68         '''
  69         self.dbusConnexions = {}
  70         # disconnect signal handlers
  71         for key in self.signalHandlers:
  72             self.signalHandlers[key].disconnect()
  73         self.signalHandlers = {}
  74
  75
  76     def dbusConnexion(self, busName):
  77         if not self.dbusConnexions.has_key(busName):
  78             if busName == "session":
  79                 self.dbusConnexions[busName] = dbus.SessionBus()
  80             elif busName == "system":
  81                 self.dbusConnexions[busName] = dbus.SystemBus()
  82             else:
  83                 raise Exception("Error: invalid bus: %s" % busName)
  84         return self.dbusConnexions[busName]
  85
  86
  87
  88 ###############################################################################
  89 class DbusSignalHandler:
  90     '''
  91     signal hash id as busName#senderName#objectName#interfaceName#signalName
  92     '''
  93     def __init__(self, busName, senderName, objectName, interfaceName, signalName):
  94         self.id = "#".join([senderName, objectName, interfaceName, signalName])
  95         # connect handler to signal
  96         self.bus = cache.dbusConnexion(busName)
  97         self.bus.add_signal_receiver(self.handleSignal, signalName, interfaceName, senderName, objectName)
  98
  99
 100     def disconnect(self):
 101         names = self.id.split("#")
 102         self.bus.remove_signal_receiver(self.handleSignal, names[3], names[2], names[0], names[1])
 103
 104
 105     def handleSignal(self, *args):
 106         '''
 107         publish dbus args under topic hash id
 108         '''
 109         factory.dispatch(self.id, json.dumps(args))
 110
 111
 112
 113 ###############################################################################
 114 class DbusCallHandler:
 115     '''
 116     deferred reply to return dbus results
 117     '''
 118     def __init__(self, method, args):
 119         self.pending = False
 120         self.request = defer.Deferred()
 121         self.method = method
 122         self.args = args
 123
 124
 125     def callMethod(self):
 126         '''
 127         dbus method async call
 128         '''
 129         self.pending = True
 130         self.method(*self.args, reply_handler=self.dbusSuccess, error_handler=self.dbusError)
 131         return self.request
 132
 133
 134     def dbusSuccess(self, *result):
 135         '''
 136         return JSON string result array
 137         '''
 138         self.request.callback(json.dumps(result))
 139         self.pending = False
 140
 141
 142     def dbusError(self, error):
 143         '''
 144         return dbus error message
 145         '''
 146         self.request.errback(error.get_dbus_message())
 147         self.pending = False
 148
 149
 150
 151 ###############################################################################
 152 class CloudeebusService:
 153     '''
 154     support for sending DBus messages and registering for DBus signals
 155     '''
 156     def __init__(self, permissions):
 157         self.permissions = permissions;
 158         self.proxyObjects = {}
 159         self.proxyMethods = {}
 160         self.pendingCalls = []
 161
 162
 163     def proxyObject(self, busName, serviceName, objectName):
 164         '''
 165         object hash id as serviceName#objectName
 166         '''
 167         id = "#".join([serviceName, objectName])
 168         if not self.proxyObjects.has_key(id):
 169             if not OPENDOOR:
 170                 # check permissions, array.index throws exception
 171                 self.permissions.index(serviceName)
 172             bus = cache.dbusConnexion(busName)
 173             self.proxyObjects[id] = bus.get_object(serviceName, objectName)
 174         return self.proxyObjects[id]
 175
 176
 177     def proxyMethod(self, busName, serviceName, objectName, interfaceName, methodName):
 178         '''
 179         method hash id as serviceName#objectName#interfaceName#methodName
 180         '''
 181         id = "#".join([serviceName, objectName, interfaceName, methodName])
 182         if not self.proxyMethods.has_key(id):
 183             obj = self.proxyObject(busName, serviceName, objectName)
 184             self.proxyMethods[id] = obj.get_dbus_method(methodName, interfaceName)
 185         return self.proxyMethods[id]
 186
 187
 188     @exportRpc
 189     def dbusRegister(self, list):
 190         '''
 191         arguments: bus, sender, object, interface, signal
 192         '''
 193         if len(list) < 5:
 194             raise Exception("Error: expected arguments: bus, sender, object, interface, signal)")
 195
 196         if not OPENDOOR:
 197             # check permissions, array.index throws exception
 198             self.permissions.index(list[1])
 199
 200         # check if a handler exists
 201         sigId = "#".join(list[1:5])
 202         if cache.signalHandlers.has_key(sigId):
 203             return sigId
 204
 205         # create a handler that will publish the signal
 206         dbusSignalHandler = DbusSignalHandler(*list[0:5])
 207         cache.signalHandlers[sigId] = dbusSignalHandler
 208
 209         return dbusSignalHandler.id
 210
 211
 212     @exportRpc
 213     def dbusSend(self, list):
 214         '''
 215         arguments: bus, destination, object, interface, message, [args]
 216         '''
 217         # clear pending calls
 218         for call in self.pendingCalls:
 219             if not call.pending:
 220                 self.pendingCalls.remove(call)
 221
 222         if len(list) < 5:
 223             raise Exception("Error: expected arguments: bus, destination, object, interface, message, [args])")
 224
 225         # parse JSON arg list
 226         args = []
 227         if len(list) == 6:
 228             args = json.loads(list[5])
 229
 230         # get dbus proxy method
 231         method = self.proxyMethod(*list[0:5])
 232
 233         # use a deferred call handler to manage dbus results
 234         dbusCallHandler = DbusCallHandler(method, args)
 235         self.pendingCalls.append(dbusCallHandler)
 236         return dbusCallHandler.callMethod()
 237
 238
 239     @exportRpc
 240     def getVersion(self):
 241         '''
 242         return current version string
 243         '''
 244         return VERSION
 245
 246
 247
 248 ###############################################################################
 249 class CloudeebusServerProtocol(WampCraServerProtocol):
 250     '''
 251     connexion and session authentication management
 252     '''
 253
 254     def onSessionOpen(self):
 255         # CRA authentication options
 256         self.clientAuthTimeout = 0
 257         self.clientAuthAllowAnonymous = OPENDOOR
 258         # CRA authentication init
 259         WampCraServerProtocol.onSessionOpen(self)
 260
 261
 262     def getAuthPermissions(self, key, extra):
 263         return json.loads(extra.get("permissions", "[]"))
 264
 265
 266     def getAuthSecret(self, key):
 267         secret = CREDENTIALS.get(key, None)
 268         if secret is None:
 269             return None
 270         # secret must be of str type to be hashed
 271         return secret.encode('utf-8')
 272
 273
 274     def onAuthenticated(self, key, permissions):
 275         if not OPENDOOR:
 276             # check authentication key
 277             if key is None:
 278                 raise Exception("Authentication failed")
 279             # check permissions, array.index throws exception
 280             for req in permissions:
 281                 WHITELIST.index(req)
 282         # create cloudeebus service instance
 283         self.cloudeebusService = CloudeebusService(permissions)
 284         # register it for RPC
 285         self.registerForRpc(self.cloudeebusService)
 286         # register for Publish / Subscribe
 287         self.registerForPubSub("", True)
 288
 289
 290     def connectionLost(self, reason):
 291         WampCraServerProtocol.connectionLost(self, reason)
 292         if factory.getConnectionCount() == 0:
 293             cache.reset()
 294
 295
 296
 297 ###############################################################################
 298
 299 if __name__ == '__main__':
 300
 301     cache = DbusCache()
 302
 303     parser = argparse.ArgumentParser(description='Javascript DBus bridge.')
 304     parser.add_argument('-v', '--version', action='store_true',
 305         help='print version and exit')
 306     parser.add_argument('-d', '--debug', action='store_true',
 307         help='log debug info on standard output')
 308     parser.add_argument('-o', '--opendoor', action='store_true',
 309         help='allow anonymous access to all services')
 310     parser.add_argument('-p', '--port', default='9000',
 311         help='port number')
 312     parser.add_argument('-c', '--credentials',
 313         help='path to credentials file')
 314     parser.add_argument('-w', '--whitelist',
 315         help='path to whitelist file')
 316
 317     args = parser.parse_args(sys.argv[1:])
 318
 319     if args.version:
 320         print("Cloudeebus version " + VERSION)
 321         exit(0)
 322
 323     if args.debug:
 324         log.startLogging(sys.stdout)
 325
 326     OPENDOOR = args.opendoor
 327
 328     if args.credentials:
 329         jfile = open(args.credentials)
 330         CREDENTIALS = json.load(jfile)
 331         jfile.close()
 332
 333     if args.whitelist:
 334         jfile = open(args.whitelist)
 335         WHITELIST = json.load(jfile)
 336         jfile.close()
 337
 338     uri = "ws://localhost:" + args.port
 339
 340     factory = WampServerFactory(uri, debugWamp = args.debug)
 341     factory.protocol = CloudeebusServerProtocol
 342     factory.setProtocolOptions(allowHixie76 = True)
 343
 344     listenWS(factory)
 345
 346     DBusGMainLoop(set_as_default=True)
 347
 348     reactor.run()