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 import re
  39 import dbus.service
  40 gobject.threads_init()
  41
  42 from dbus import glib
  43 glib.init_threads()
  44
  45 # enable debug log
  46 from twisted.python import log
  47
  48 # XML parser module
  49 from xml.etree.ElementTree import XMLParser
  50
  51 # For debug only
  52 import os
  53
  54 ###############################################################################
  55
  56 VERSION = "0.2.1"
  57 OPENDOOR = False
  58 CREDENTIALS = {}
  59 WHITELIST = []
  60
  61 ###############################################################################
  62 class DbusCache:
  63     '''
  64     Global cache of DBus connexions and signal handlers
  65     '''
  66     def __init__(self):
  67         self.dbusConnexions = {}
  68         self.signalHandlers = {}
  69
  70
  71     def reset(self):
  72         '''
  73         Disconnect signal handlers before resetting cache.
  74         '''
  75         self.dbusConnexions = {}
  76         # disconnect signal handlers
  77         for key in self.signalHandlers:
  78             self.signalHandlers[key].disconnect()
  79         self.signalHandlers = {}
  80
  81
  82     def dbusConnexion(self, busName):
  83         if not self.dbusConnexions.has_key(busName):
  84             if busName == "session":
  85                 self.dbusConnexions[busName] = dbus.SessionBus()
  86             elif busName == "system":
  87                 self.dbusConnexions[busName] = dbus.SystemBus()
  88             else:
  89                 raise Exception("Error: invalid bus: %s" % busName)
  90         return self.dbusConnexions[busName]
  91
  92
  93
  94 ###############################################################################
  95 class DbusSignalHandler:
  96     '''
  97     signal hash id as busName#senderName#objectName#interfaceName#signalName
  98     '''
  99     def __init__(self, busName, senderName, objectName, interfaceName, signalName):
 100         self.id = "#".join([busName, senderName, objectName, interfaceName, signalName])
 101         # connect handler to signal
 102         self.bus = cache.dbusConnexion(busName)
 103         self.bus.add_signal_receiver(self.handleSignal, signalName, interfaceName, senderName, objectName)
 104
 105
 106     def disconnect(self):
 107         names = self.id.split("#")
 108         self.bus.remove_signal_receiver(self.handleSignal, names[4], names[3], names[1], names[2])
 109
 110
 111     def handleSignal(self, *args):
 112         '''
 113         publish dbus args under topic hash id
 114         '''
 115         factory.dispatch(self.id, json.dumps(args))
 116
 117
 118
 119 ###############################################################################
 120 class DbusCallHandler:
 121     '''
 122     deferred reply to return dbus results
 123     '''
 124     def __init__(self, method, args):
 125         self.pending = False
 126         self.request = defer.Deferred()
 127         self.method = method
 128         self.args = args
 129
 130
 131     def callMethod(self):
 132         '''
 133         dbus method async call
 134         '''
 135         self.pending = True
 136         self.method(*self.args, reply_handler=self.dbusSuccess, error_handler=self.dbusError)
 137         return self.request
 138
 139
 140     def dbusSuccess(self, *result):
 141         '''
 142         return JSON string result array
 143         '''
 144         self.request.callback(json.dumps(result))
 145         self.pending = False
 146
 147
 148     def dbusError(self, error):
 149         '''
 150         return dbus error message
 151         '''
 152         self.request.errback(error.get_dbus_message())
 153         self.pending = False
 154
 155
 156
 157 ################################################################################
 158 class exec_code:
 159     def __init__(self) :
 160         self.exec_string = ""
 161         self.exec_code = None
 162         self.exec_code_valid = 1
 163         self.indent_level = 0
 164         self.indent_increment = 1
 165         self.line = 0
 166
 167     # __str__ : Return a string representation of the object, for
 168     # nice printing.
 169     def __str__(self) :
 170         return self.exec_string
 171
 172     def p(self) :
 173         print str(self)
 174
 175     def append_stmt(self, stmt) :
 176         self.exec_code_valid = 0
 177         self.line += 1
 178         if (stmt != "\n"):
 179             for x in range(0,self.indent_level):
 180                 self.exec_string = self.exec_string + ' '
 181             self.exec_string = self.exec_string + stmt + "\t\t# l:" + str(self.line) + '\n'
 182         else:
 183             if (stmt == "\n"):
 184                 self.exec_string = self.exec_string + "# l:" + str(self.line) + '\n'
 185             else:
 186                 self.exec_string = self.exec_string + stmt + "\t\t# l:" + str(self.line) + '\n'
 187
 188     def indent(self) :
 189         self.indent_level = self.indent_level + self.indent_increment
 190
 191     def dedent(self) :
 192         self.indent_level = self.indent_level - self.indent_increment
 193
 194     # compile : Compile exec_string into exec_code using the builtin
 195     # compile function. Skip if already in sync.
 196     def compile(self) :
 197         if not self.exec_code_valid :
 198             self.exec_code = compile(self.exec_string, "<string>", "exec")
 199         self.exec_code_valid = 1
 200
 201     def execute(self) :
 202         if not self.exec_code_valid :
 203             self.compile()
 204         exec self.exec_code
 205
 206
 207
 208 ################################################################################
 209 class XmlCb_Parser: # The target object of the parser
 210     maxDepth = 0
 211     depth = 0
 212     def __init__(self, dynDBusClass):
 213         self.dynDBusClass = dynDBusClass
 214
 215     def start(self, tag, attrib):   # Called for each opening tag.
 216         if (tag == 'node'):
 217             return
 218         # Set interface name
 219         if (tag == 'interface'):
 220             self.dynDBusClass.set_interface(attrib['name'])
 221             return
 222         # Set method name
 223         if (tag == 'method'):
 224             self.current = tag
 225             self.dynDBusClass.def_method(attrib['name'])
 226             return
 227         if (tag == 'signal'):
 228             self.current = tag
 229             self.dynDBusClass.def_signal(attrib['name'])
 230             return
 231
 232         # Set signature (in/out & name) for method
 233         if (tag == 'arg'):
 234             if (self.current == 'method'):
 235                 self.dynDBusClass.add_signature(attrib['name'],
 236                                                 attrib['direction'],
 237                                                 attrib['type'])
 238                 return
 239             if (self.current == 'signal'):
 240                 self.dynDBusClass.add_signature(attrib['name'], 'in',
 241                                                 attrib['type'])
 242                 return
 243     def end(self, tag):             # Called for each closing tag.
 244         if (tag == 'method'):
 245             self.dynDBusClass.add_dbus_method()
 246             self.dynDBusClass.add_body_method()
 247             self.dynDBusClass.end_method()
 248         if (tag == 'signal'):
 249             self.dynDBusClass.add_dbus_signal()
 250             self.dynDBusClass.add_body_signal()
 251             self.dynDBusClass.end_method()
 252
 253     def data(self, data):
 254         pass            # We do not need to do anything with data.
 255     def close(self):    # Called when all data has been parsed.
 256         return self.maxDepth
 257
 258
 259
 260 ################################################################################
 261 class dynDBusClass():
 262     def __init__(self, className, globalCtx, localCtx):
 263         self.className = className
 264         self.xmlCB = XmlCb_Parser(self)
 265         self.localCtx = localCtx
 266         self.globalCtx = globalCtx
 267         self.signature = {}
 268         self.class_code = exec_code()
 269         self.class_code.indent_increment = 4
 270         self.class_code.append_stmt("import dbus")
 271         self.class_code.append_stmt("\n")
 272         self.class_code.append_stmt("\n")
 273         self.class_code.append_stmt("class " + self.className + "(dbus.service.Object):")
 274         self.class_code.indent()
 275
 276         ## Overload of __init__ method
 277         self.def_method("__init__")
 278         self.add_method("bus, callback=None, objName='/sample', busName='org.cloudeebus'")
 279         self.add_stmt("self.bus = bus")
 280         self.add_stmt("self.objName = objName")
 281         self.add_stmt("self.callback = callback")
 282         self.add_stmt("dbus.service.Object.__init__(self, conn=bus, object_path=objName, bus_name=busName)")
 283         self.end_method()
 284
 285     def createDBusServiceFromXML(self, xml):
 286         self.parser = XMLParser(target=self.xmlCB)
 287         self.parser.feed(xml)
 288         self.parser.close()
 289
 290     def set_interface(self, ifName):
 291         self.ifName = ifName
 292
 293     def def_method(self, methodName):
 294         self.methodToAdd = methodName
 295         self.signalToAdd = None
 296         self.args_str = str()
 297         self.signature = {}
 298         self.signature['name'] = str()
 299         self.signature['in'] = str()
 300         self.signature['out'] = str()
 301
 302     def def_signal(self, signalName):
 303         self.methodToAdd = None
 304         self.signalToAdd = signalName
 305         self.args_str = str()
 306         self.signature = {}
 307         self.signature['name'] = str()
 308         self.signature['in'] = str()
 309         self.signature['out'] = str()
 310
 311     def add_signature(self, name, direction, signature):
 312         if (direction == 'in'):
 313             self.signature['in'] += signature
 314             if (self.signature['name'] != str()):
 315                 self.signature['name'] += ", "
 316             self.signature['name'] += name
 317         if (direction == 'out'):
 318             self.signature['out'] = signature
 319
 320     def add_method(self, args = None, async_success_cb = None, async_err_cb = None):
 321         async_cb_str = str()
 322         if (self.methodToAdd != None):
 323             name = self.methodToAdd
 324         else:
 325             name = self.signalToAdd
 326         if (args != None):
 327             self.args_str = args
 328         if (async_success_cb != None):
 329             async_cb_str = async_success_cb
 330         if (async_err_cb != None):
 331             if (async_cb_str != str()):
 332                 async_cb_str += ", "
 333             async_cb_str += async_err_cb
 334
 335         parameters = self.args_str
 336         if (async_cb_str != str()):
 337             if (parameters != str()):
 338                 parameters += ", "
 339             parameters +=async_cb_str
 340
 341         if (parameters != str()):
 342             self.class_code.append_stmt("def " + name + "(self, %s):" % parameters)
 343         else:
 344             self.class_code.append_stmt("def " + name + "(self):")
 345         self.class_code.indent()
 346
 347     def end_method(self):
 348         self.class_code.append_stmt("\n")
 349         self.class_code.append_stmt("\n")
 350         self.class_code.dedent()
 351
 352     def add_dbus_method(self):
 353         decorator = '@dbus.service.method("' + self.ifName + '"'
 354         if (self.signature.has_key('in') and self.signature['in'] != str()):
 355                 decorator += ", in_signature='" + self.signature['in'] + "'"
 356         if (self.signature.has_key('out') and self.signature['out'] != str()):
 357                 decorator += ", out_signature='" + self.signature['out'] + "'"
 358         decorator += ", async_callbacks=('dbus_async_cb', 'dbus_async_err_cb')"
 359         decorator += ")"
 360         self.class_code.append_stmt(decorator)
 361         if (self.signature.has_key('name') and self.signature['name'] != str()):
 362             self.add_method(self.signature['name'], async_success_cb='dbus_async_cb', async_err_cb='dbus_async_err_cb')
 363         else:
 364             self.add_method(async_success_cb='dbus_async_cb', async_err_cb='dbus_async_err_cb')
 365
 366     def add_dbus_signal(self):
 367         decorator = '@dbus.service.signal("' + self.ifName + '"'
 368         if (self.signature.has_key('in') and self.signature['in'] != str()):
 369                 decorator += ", signature='" + self.signature['in'] + "'"
 370         decorator += ")"
 371         self.class_code.append_stmt(decorator)
 372         if (self.signature.has_key('name') and self.signature['name'] != str()):
 373             self.add_method(self.signature['name'])
 374         else:
 375             self.add_method()
 376
 377     def add_body_method(self):
 378         if (self.methodToAdd != None):
 379             self.class_code.append_stmt("print 'In " + self.methodToAdd + "()'")
 380             if (self.args_str != str()):
 381                 self.class_code.append_stmt("self.callback('" + self.methodToAdd + "', dbus_async_cb, dbus_async_err_cb, %s)" % self.args_str)
 382             else:
 383                 self.class_code.append_stmt("self.callback('" + self.methodToAdd + "', dbus_async_cb, dbus_async_err_cb)")
 384
 385     def add_body_signal(self):
 386         self.class_code.append_stmt("return") ## TODO: Remove and fix with code ad hoc
 387         self.class_code.append_stmt("\n")
 388
 389     def add_stmt(self, stmt) :
 390         self.class_code.append_stmt(stmt)
 391
 392     def declare(self) :
 393         self.class_code.compile()
 394         exec(self.class_code.exec_string, self.globalCtx, self.localCtx)
 395
 396     def __str__(self) :
 397         return self.class_code.exec_string
 398
 399     # p : Since it is often useful to be able to look at the code
 400     # that is generated interactively, this function provides
 401     # a shorthand for "print str(some_exec_code_instance)", which
 402     # gives a reasonable nice look at the contents of the
 403     # exec_code object.
 404     def p(self) :
 405         print str(self)
 406
 407
 408
 409 ###############################################################################
 410 class CloudeebusService:
 411     '''
 412     support for sending DBus messages and registering for DBus signals
 413     '''
 414     def __init__(self, permissions):
 415         self.permissions = permissions;
 416         self.proxyObjects = {}
 417         self.proxyMethods = {}
 418         self.pendingCalls = []
 419         self.dynDBusClasses = {} # DBus class source code generated dynamically (a list because one by classname)
 420         self.services = {}  # DBus service created
 421         self.serviceAgents = {} # Instantiated DBus class previously generated dynamically, for now, one by classname
 422
 423
 424     def proxyObject(self, busName, serviceName, objectName):
 425         '''
 426         object hash id as busName#serviceName#objectName
 427         '''
 428         id = "#".join([busName, serviceName, objectName])
 429         if not self.proxyObjects.has_key(id):
 430             if not OPENDOOR:
 431                 # check permissions, array.index throws exception
 432                 self.permissions.index(serviceName)
 433             bus = cache.dbusConnexion(busName)
 434             self.proxyObjects[id] = bus.get_object(serviceName, objectName)
 435         return self.proxyObjects[id]
 436
 437
 438     def proxyMethod(self, busName, serviceName, objectName, interfaceName, methodName):
 439         '''
 440         method hash id as busName#serviceName#objectName#interfaceName#methodName
 441         '''
 442         id = "#".join([busName, serviceName, objectName, interfaceName, methodName])
 443         if not self.proxyMethods.has_key(id):
 444             obj = self.proxyObject(busName, serviceName, objectName)
 445             self.proxyMethods[id] = obj.get_dbus_method(methodName, interfaceName)
 446         return self.proxyMethods[id]
 447
 448
 449     @exportRpc
 450     def dbusRegister(self, list):
 451         '''
 452         arguments: bus, sender, object, interface, signal
 453         '''
 454         if len(list) < 5:
 455             raise Exception("Error: expected arguments: bus, sender, object, interface, signal)")
 456
 457         if not OPENDOOR:
 458             # check permissions, array.index throws exception
 459             self.permissions.index(list[1])
 460
 461         # check if a handler exists
 462         sigId = "#".join(list)
 463         if cache.signalHandlers.has_key(sigId):
 464             return sigId
 465
 466         # create a handler that will publish the signal
 467         dbusSignalHandler = DbusSignalHandler(*list)
 468         cache.signalHandlers[sigId] = dbusSignalHandler
 469
 470         return dbusSignalHandler.id
 471
 472
 473     @exportRpc
 474     def dbusSend(self, list):
 475         '''
 476         arguments: bus, destination, object, interface, message, [args]
 477         '''
 478         # clear pending calls
 479         for call in self.pendingCalls:
 480             if not call.pending:
 481                 self.pendingCalls.remove(call)
 482
 483         if len(list) < 5:
 484             raise Exception("Error: expected arguments: bus, destination, object, interface, message, [args])")
 485
 486         # parse JSON arg list
 487         args = []
 488         if len(list) == 6:
 489             args = json.loads(list[5])
 490
 491         # get dbus proxy method
 492         method = self.proxyMethod(*list[0:5])
 493
 494         # use a deferred call handler to manage dbus results
 495         dbusCallHandler = DbusCallHandler(method, args)
 496         self.pendingCalls.append(dbusCallHandler)
 497         return dbusCallHandler.callMethod()
 498
 499
 500     def srvCB(self, name, async_succes_cb, async_error_cb, *args):
 501         print "self.srvCB(name='%s', args=%s')\n\n" % (name, str(args))
 502         methodId = self.srvName + "#" + self.agentObjectPath + "#" + name
 503         print "factory.dispatch(methodId='%s', json.dumps(args)=%s')\n\n" % (methodId, json.dumps(args))
 504         factory.dispatch(methodId, json.dumps(args))
 505
 506     @exportRpc
 507     def serviceAdd(self, list):
 508         '''
 509         arguments: busName, srvName
 510         '''
 511         busName = list[0]
 512         self.bus =  cache.dbusConnexion( busName['name'] )
 513         self.srvName = list[1]
 514         if (self.services.has_key(self.srvName) == False):
 515             self.services[self.srvName] = dbus.service.BusName(name = self.srvName, bus = self.bus)
 516         return self.srvName
 517
 518     @exportRpc
 519     def serviceAddAgent(self, list):
 520         '''
 521         arguments: objectPath, xmlTemplate
 522         '''
 523         self.agentObjectPath = list[0]
 524         xmlTemplate = list[1]
 525         className = re.sub('/', '_', self.agentObjectPath[1:])
 526         if (self.dynDBusClasses.has_key(className) == False):
 527             self.dynDBusClasses[className] = dynDBusClass(className, globals(), locals())
 528             self.dynDBusClasses[className].createDBusServiceFromXML(xmlTemplate)
 529
 530             # For Debug only
 531             if (1):
 532                 if (1): ## Force deletion
 533                     if os.access('./MyDbusClass.py', os.R_OK) == True:
 534                         os.remove('./MyDbusClass.py')
 535
 536                     if os.access('./MyDbusClass.py', os.R_OK) == False:
 537                         f = open('./MyDbusClass.py', 'w')
 538                         f.write(self.dynDBusClasses[className].class_code.exec_string)
 539                         f.close()
 540 #                self.dynDBusClass[className].p()
 541                 self.dynDBusClasses[className].declare()
 542
 543             if (self.serviceAgents.has_key(className) == False):
 544                 exe_str = "self.serviceAgents[" + className +"] = " + className + "(self.bus, callback=self.srvCB, objName=self.agentObjectPath, busName=self.srvName)"
 545                 exec (exe_str, globals(), locals())
 546
 547     @exportRpc
 548     def getVersion(self):
 549         '''
 550         return current version string
 551         '''
 552         return VERSION
 553
 554
 555
 556 ###############################################################################
 557 class CloudeebusServerProtocol(WampCraServerProtocol):
 558     '''
 559     connexion and session authentication management
 560     '''
 561
 562     def onSessionOpen(self):
 563         # CRA authentication options
 564         self.clientAuthTimeout = 0
 565         self.clientAuthAllowAnonymous = OPENDOOR
 566         # CRA authentication init
 567         WampCraServerProtocol.onSessionOpen(self)
 568
 569
 570     def getAuthPermissions(self, key, extra):
 571         return json.loads(extra.get("permissions", "[]"))
 572
 573
 574     def getAuthSecret(self, key):
 575         secret = CREDENTIALS.get(key, None)
 576         if secret is None:
 577             return None
 578         # secret must be of str type to be hashed
 579         return secret.encode('utf-8')
 580
 581
 582     def onAuthenticated(self, key, permissions):
 583         if not OPENDOOR:
 584             # check authentication key
 585             if key is None:
 586                 raise Exception("Authentication failed")
 587             # check permissions, array.index throws exception
 588             for req in permissions:
 589                 WHITELIST.index(req)
 590         # create cloudeebus service instance
 591         self.cloudeebusService = CloudeebusService(permissions)
 592         # register it for RPC
 593         self.registerForRpc(self.cloudeebusService)
 594         # register for Publish / Subscribe
 595         self.registerForPubSub("", True)
 596
 597
 598     def connectionLost(self, reason):
 599         WampCraServerProtocol.connectionLost(self, reason)
 600         if factory.getConnectionCount() == 0:
 601             cache.reset()
 602
 603
 604
 605 ###############################################################################
 606
 607 if __name__ == '__main__':
 608
 609     cache = DbusCache()
 610
 611     parser = argparse.ArgumentParser(description='Javascript DBus bridge.')
 612     parser.add_argument('-v', '--version', action='store_true',
 613         help='print version and exit')
 614     parser.add_argument('-d', '--debug', action='store_true',
 615         help='log debug info on standard output')
 616     parser.add_argument('-o', '--opendoor', action='store_true',
 617         help='allow anonymous access to all services')
 618     parser.add_argument('-p', '--port', default='9000',
 619         help='port number')
 620     parser.add_argument('-c', '--credentials',
 621         help='path to credentials file')
 622     parser.add_argument('-w', '--whitelist',
 623         help='path to whitelist file')
 624
 625     args = parser.parse_args(sys.argv[1:])
 626
 627     if args.version:
 628         print("Cloudeebus version " + VERSION)
 629         exit(0)
 630
 631     if args.debug:
 632         log.startLogging(sys.stdout)
 633
 634     OPENDOOR = args.opendoor
 635
 636     if args.credentials:
 637         jfile = open(args.credentials)
 638         CREDENTIALS = json.load(jfile)
 639         jfile.close()
 640
 641     if args.whitelist:
 642         jfile = open(args.whitelist)
 643         WHITELIST = json.load(jfile)
 644         jfile.close()
 645
 646     uri = "ws://localhost:" + args.port
 647
 648     factory = WampServerFactory(uri, debugWamp = args.debug)
 649     factory.protocol = CloudeebusServerProtocol
 650     factory.setProtocolOptions(allowHixie76 = True)
 651
 652     listenWS(factory)
 653
 654     DBusGMainLoop(set_as_default=True)
 655
 656     reactor.run()