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, io, 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 OPENDOOR = False
  51 CREDENTIALS = {}
  52 WHITELIST = []
  53
  54 ###############################################################################
  55
  56 def hashId(list):
  57         str = list[0]
  58         for item in list[1:len(list)]:
  59                 str += "#" + item
  60         return str
  61
  62 ###############################################################################
  63 class DbusCache:
  64         def __init__(self):
  65                 self.reset()
  66
  67
  68         def reset(self):
  69                 # dbus connexions
  70                 self.dbusConnexions = {}
  71                 # signal handlers
  72                 self.signalHandlers = {}
  73
  74
  75         def dbusConnexion(self, busName):
  76                 if not self.dbusConnexions.has_key(busName):
  77                         if busName == "session":
  78                                 self.dbusConnexions[busName] = dbus.SessionBus()
  79                         elif busName == "system":
  80                                 self.dbusConnexions[busName] = dbus.SystemBus()
  81                         else:
  82                                 raise Exception("Error: invalid bus: %s" % busName)
  83                 return self.dbusConnexions[busName]
  84
  85
  86
  87 ###############################################################################
  88 class DbusSignalHandler:
  89         def __init__(self, object, senderName, objectName, interfaceName, signalName):
  90                 # publish hash id
  91                 self.id = hashId([senderName, objectName, interfaceName, signalName])
  92                 # connect dbus proxy object to signal
  93                 self.proxyObject = object
  94                 self.proxyObject.connect_to_signal(signalName, self.handleSignal, interfaceName)
  95
  96
  97         def handleSignal(self, *args):
  98                 # publish dbus args under topic hash id
  99                 factory.dispatch(self.id, json.dumps(args))
 100
 101
 102
 103 ###############################################################################
 104 class DbusCallHandler:
 105         def __init__(self, method, args):
 106                 # deferred reply to return dbus results
 107                 self.pending = False
 108                 self.request = defer.Deferred()
 109                 self.method = method
 110                 self.args = args
 111
 112
 113         def callMethod(self):
 114                 # dbus method async call
 115                 self.pending = True
 116                 self.method(*self.args, reply_handler=self.dbusSuccess, error_handler=self.dbusError)
 117                 return self.request
 118
 119
 120         def dbusSuccess(self, *result):
 121                 # return JSON string result array
 122                 self.request.callback(json.dumps(result))
 123                 self.pending = False
 124
 125
 126         def dbusError(self, error):
 127                 # return dbus error message
 128                 self.request.errback(error.get_dbus_message())
 129                 self.pending = False
 130
 131
 132
 133 ###############################################################################
 134 class CloudeebusService:
 135         def __init__(self, permissions):
 136                 self.permissions = permissions;
 137                 # proxy objects
 138                 self.proxyObjects = {}
 139                 # proxy methods
 140                 self.proxyMethods = {}
 141                 # pending dbus calls
 142                 self.pendingCalls = []
 143
 144
 145         def proxyObject(self, busName, serviceName, objectName):
 146                 id = hashId([serviceName, objectName])
 147                 if not self.proxyObjects.has_key(id):
 148                         if not OPENDOOR:
 149                                 # check permissions, array.index throws exception
 150                                 self.permissions.index(serviceName)
 151                         bus = cache.dbusConnexion(busName)
 152                         self.proxyObjects[id] = bus.get_object(serviceName, objectName)
 153                 return self.proxyObjects[id]
 154
 155
 156         def proxyMethod(self, busName, serviceName, objectName, interfaceName, methodName):
 157                 id = hashId([serviceName, objectName, interfaceName, methodName])
 158                 if not self.proxyMethods.has_key(id):
 159                         obj = self.proxyObject(busName, serviceName, objectName)
 160                         self.proxyMethods[id] = obj.get_dbus_method(methodName, interfaceName)
 161                 return self.proxyMethods[id]
 162
 163
 164         @exportRpc
 165         def dbusRegister(self, list):
 166                 # read arguments list by position
 167                 if len(list) < 5:
 168                         raise Exception("Error: expected arguments: bus, sender, object, interface, signal)")
 169
 170                 # check if a handler exists
 171                 sigId = hashId(list[1:5])
 172                 if cache.signalHandlers.has_key(sigId):
 173                         return sigId
 174
 175                 # get dbus proxy object
 176                 object = self.proxyObject(list[0], list[1], list[2])
 177
 178                 # create a handler that will publish the signal
 179                 dbusSignalHandler = DbusSignalHandler(object, *list[1:5])
 180                 cache.signalHandlers[sigId] = dbusSignalHandler
 181
 182                 return dbusSignalHandler.id
 183
 184
 185         @exportRpc
 186         def dbusSend(self, list):
 187                 # clear pending calls
 188                 for call in self.pendingCalls:
 189                         if not call.pending:
 190                                 self.pendingCalls.remove(call)
 191
 192                 # read arguments list by position
 193                 if len(list) < 5:
 194                         raise Exception("Error: expected arguments: bus, destination, object, interface, message, [args])")
 195
 196                 # parse JSON arg list
 197                 args = []
 198                 if len(list) == 6:
 199                         args = json.loads(list[5])
 200
 201                 # get dbus proxy method
 202                 method = self.proxyMethod(list[0], *list[1:5])
 203
 204                 # use a deferred call handler to manage dbus results
 205                 dbusCallHandler = DbusCallHandler(method, args)
 206                 self.pendingCalls.append(dbusCallHandler)
 207                 return dbusCallHandler.callMethod()
 208
 209
 210
 211 ###############################################################################
 212 class CloudeebusServerProtocol(WampCraServerProtocol):
 213
 214         def onSessionOpen(self):
 215                 # CRA authentication options
 216                 self.clientAuthTimeout = 0
 217                 self.clientAuthAllowAnonymous = OPENDOOR
 218                 # CRA authentication init
 219                 WampCraServerProtocol.onSessionOpen(self)
 220
 221
 222         def getAuthPermissions(self, key, extra):
 223                 return json.loads(extra.get("permissions", "[]"))
 224
 225
 226         def getAuthSecret(self, key):
 227                 secret = CREDENTIALS.get(key, None)
 228                 if secret is None:
 229                         return None
 230                 # secret must be of str type to be hashed
 231                 return secret.encode('utf-8')
 232
 233
 234         def onAuthenticated(self, key, permissions):
 235                 if not OPENDOOR:
 236                         # check authentication key
 237                         if key is None:
 238                                 raise Exception("Authentication failed")
 239                         # check permissions, array.index throws exception
 240                         for req in permissions:
 241                                 WHITELIST.index(req)
 242                 # create cloudeebus service instance
 243                 self.cloudeebusService = CloudeebusService(permissions)
 244                 # register it for RPC
 245                 self.registerForRpc(self.cloudeebusService)
 246                 # register for Publish / Subscribe
 247                 self.registerForPubSub("", True)
 248
 249
 250         def connectionLost(self, reason):
 251                 WampCraServerProtocol.connectionLost(self, reason)
 252                 if factory.getConnectionCount() == 0:
 253                         cache.reset()
 254
 255
 256
 257 ###############################################################################
 258
 259 if __name__ == '__main__':
 260
 261         cache = DbusCache()
 262
 263         parser = argparse.ArgumentParser(description='Javascript DBus bridge.')
 264         parser.add_argument('-d', '--debug', action='store_true')
 265         parser.add_argument('-o', '--opendoor', action='store_true')
 266         parser.add_argument('-p', '--port', default='9000')
 267         parser.add_argument('-c', '--credentials')
 268         parser.add_argument('-w', '--whitelist')
 269
 270         args = parser.parse_args(sys.argv[1:])
 271
 272         if args.debug:
 273                 log.startLogging(sys.stdout)
 274
 275         OPENDOOR = args.opendoor
 276
 277         if args.credentials:
 278                 jfile = open(args.credentials)
 279                 CREDENTIALS = json.load(jfile)
 280                 jfile.close()
 281
 282         if args.whitelist:
 283                 jfile = open(args.whitelist)
 284                 WHITELIST = json.load(jfile)
 285                 jfile.close()
 286
 287         uri = "ws://localhost:" + args.port
 288
 289         factory = WampServerFactory(uri, debugWamp = args.debug)
 290         factory.protocol = CloudeebusServerProtocol
 291         factory.setProtocolOptions(allowHixie76 = True)
 292
 293         listenWS(factory)
 294
 295         DBusGMainLoop(set_as_default=True)
 296
 297         reactor.run()