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