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 # Frederic Paut <frederic.paut@intel.com>
  22 #
  23
  24
  25 import argparse, dbus, json, sys
  26
  27 from twisted.internet import glib2reactor
  28 # Configure the twisted mainloop to be run inside the glib mainloop.
  29 # This must be done before importing the other twisted modules
  30 glib2reactor.install()
  31 from twisted.internet import reactor, defer
  32
  33 from autobahn.websocket import listenWS
  34 from autobahn.wamp import exportRpc, WampServerFactory, WampCraServerProtocol
  35
  36 from dbus.mainloop.glib import DBusGMainLoop
  37
  38 import gobject
  39 gobject.threads_init()
  40
  41 from dbus import glib
  42 glib.init_threads()
  43
  44 # enable debug log
  45 from twisted.python import log
  46
  47
  48
  49 ###############################################################################
  50
  51 VERSION = "0.5.1"
  52 OPENDOOR = False
  53 CREDENTIALS = {}
  54 WHITELIST = []
  55 NETMASK =  []
  56
  57 ###############################################################################
  58 def ipV4ToHex(mask):
  59     ## Convert an ip or an IP mask (such as ip/24 or ip/255.255.255.0) in hex value (32bits)
  60     maskHex = 0
  61     byte = 0
  62     if mask.rfind(".") == -1:
  63         if (int(mask) < 32):
  64             maskHex = (2**(int(mask))-1)
  65             maskHex = maskHex << (32-int(mask))
  66         else:
  67             raise Exception("Illegal mask (larger than 32 bits) " + mask)
  68     else:
  69         maskField = mask.split(".")
  70         # Check if mask has four fields (byte)
  71         if len(maskField) != 4:
  72             raise Exception("Illegal ip address / mask (should be 4 bytes) " + mask)
  73         for maskQuartet in maskField:
  74             byte = int(maskQuartet)
  75             # Check if each field is really a byte
  76             if byte > 255:
  77                 raise Exception("Illegal ip address / mask (digit larger than a byte) " + mask)
  78             maskHex += byte
  79             maskHex = maskHex << 8
  80         maskHex = maskHex >> 8
  81     return maskHex
  82
  83 ###############################################################################
  84 class DbusCache:
  85     '''
  86     Global cache of DBus connexions and signal handlers
  87     '''
  88     def __init__(self):
  89         self.dbusConnexions = {}
  90         self.signalHandlers = {}
  91
  92
  93     def reset(self):
  94         '''
  95         Disconnect signal handlers before resetting cache.
  96         '''
  97         self.dbusConnexions = {}
  98         # disconnect signal handlers
  99         for key in self.signalHandlers:
 100             self.signalHandlers[key].disconnect()
 101         self.signalHandlers = {}
 102
 103
 104     def dbusConnexion(self, busName):
 105         if not self.dbusConnexions.has_key(busName):
 106             if busName == "session":
 107                 self.dbusConnexions[busName] = dbus.SessionBus()
 108             elif busName == "system":
 109                 self.dbusConnexions[busName] = dbus.SystemBus()
 110             else:
 111                 raise Exception("Error: invalid bus: %s" % busName)
 112         return self.dbusConnexions[busName]
 113
 114
 115
 116 ###############################################################################
 117 class DbusSignalHandler:
 118     '''
 119     signal hash id as busName#senderName#objectName#interfaceName#signalName
 120     '''
 121     def __init__(self, busName, senderName, objectName, interfaceName, signalName):
 122         self.id = "#".join([busName, senderName, objectName, interfaceName, signalName])
 123         # connect handler to signal
 124         self.bus = cache.dbusConnexion(busName)
 125         self.bus.add_signal_receiver(self.handleSignal, signalName, interfaceName, senderName, objectName)
 126
 127
 128     def disconnect(self):
 129         names = self.id.split("#")
 130         self.bus.remove_signal_receiver(self.handleSignal, names[4], names[3], names[1], names[2])
 131
 132
 133     def handleSignal(self, *args):
 134         '''
 135         publish dbus args under topic hash id
 136         '''
 137         factory.dispatch(self.id, json.dumps(args))
 138
 139
 140
 141 ###############################################################################
 142 class DbusCallHandler:
 143     '''
 144     deferred reply to return dbus results
 145     '''
 146     def __init__(self, method, args):
 147         self.pending = False
 148         self.request = defer.Deferred()
 149         self.method = method
 150         self.args = args
 151
 152
 153     def callMethod(self):
 154         '''
 155         dbus method async call
 156         '''
 157         self.pending = True
 158         self.method(*self.args, reply_handler=self.dbusSuccess, error_handler=self.dbusError)
 159         return self.request
 160
 161
 162     def dbusSuccess(self, *result):
 163         '''
 164         return JSON string result array
 165         '''
 166         self.request.callback(json.dumps(result))
 167         self.pending = False
 168
 169
 170     def dbusError(self, error):
 171         '''
 172         return dbus error message
 173         '''
 174         self.request.errback(Exception(error.get_dbus_message()))
 175         self.pending = False
 176
 177
 178
 179 ###############################################################################
 180 class CloudeebusService:
 181     '''
 182     support for sending DBus messages and registering for DBus signals
 183     '''
 184     def __init__(self, permissions):
 185         self.permissions = {};
 186         self.permissions['permissions'] = permissions['permissions']
 187         self.permissions['authextra'] = permissions['authextra']
 188         self.proxyObjects = {}
 189         self.proxyMethods = {}
 190         self.pendingCalls = []
 191
 192
 193     def proxyObject(self, busName, serviceName, objectName):
 194         '''
 195         object hash id as busName#serviceName#objectName
 196         '''
 197         id = "#".join([busName, serviceName, objectName])
 198         if not self.proxyObjects.has_key(id):
 199             if not OPENDOOR:
 200                 # check permissions, array.index throws exception
 201                 self.permissions['permissions'].index(serviceName)
 202             bus = cache.dbusConnexion(busName)
 203             self.proxyObjects[id] = bus.get_object(serviceName, objectName)
 204         return self.proxyObjects[id]
 205
 206
 207     def proxyMethod(self, busName, serviceName, objectName, interfaceName, methodName):
 208         '''
 209         method hash id as busName#serviceName#objectName#interfaceName#methodName
 210         '''
 211         id = "#".join([busName, serviceName, objectName, interfaceName, methodName])
 212         if not self.proxyMethods.has_key(id):
 213             obj = self.proxyObject(busName, serviceName, objectName)
 214             self.proxyMethods[id] = obj.get_dbus_method(methodName, interfaceName)
 215         return self.proxyMethods[id]
 216
 217
 218     @exportRpc
 219     def dbusRegister(self, list):
 220         '''
 221         arguments: bus, sender, object, interface, signal
 222         '''
 223         if len(list) < 5:
 224             raise Exception("Error: expected arguments: bus, sender, object, interface, signal)")
 225
 226         if not OPENDOOR:
 227             # check permissions, array.index throws exception
 228             self.permissions['permissions'].index(list[1])
 229
 230         # check if a handler exists
 231         sigId = "#".join(list)
 232         if cache.signalHandlers.has_key(sigId):
 233             return sigId
 234
 235         # create a handler that will publish the signal
 236         dbusSignalHandler = DbusSignalHandler(*list)
 237         cache.signalHandlers[sigId] = dbusSignalHandler
 238
 239         return dbusSignalHandler.id
 240
 241
 242     @exportRpc
 243     def dbusSend(self, list):
 244         '''
 245         arguments: bus, destination, object, interface, message, [args]
 246         '''
 247         # clear pending calls
 248         for call in self.pendingCalls:
 249             if not call.pending:
 250                 self.pendingCalls.remove(call)
 251
 252         if len(list) < 5:
 253             raise Exception("Error: expected arguments: bus, destination, object, interface, message, [args])")
 254
 255         # parse JSON arg list
 256         args = []
 257         if len(list) == 6:
 258             args = json.loads(list[5])
 259
 260         # get dbus proxy method
 261         method = self.proxyMethod(*list[0:5])
 262
 263         # use a deferred call handler to manage dbus results
 264         dbusCallHandler = DbusCallHandler(method, args)
 265         self.pendingCalls.append(dbusCallHandler)
 266         return dbusCallHandler.callMethod()
 267
 268
 269     @exportRpc
 270     def getVersion(self):
 271         '''
 272         return current version string
 273         '''
 274         return VERSION
 275
 276
 277
 278 ###############################################################################
 279 class CloudeebusServerProtocol(WampCraServerProtocol):
 280     '''
 281     connexion and session authentication management
 282     '''
 283
 284     def onSessionOpen(self):
 285         # CRA authentication options
 286         self.clientAuthTimeout = 0
 287         self.clientAuthAllowAnonymous = OPENDOOR
 288         # CRA authentication init
 289         WampCraServerProtocol.onSessionOpen(self)
 290
 291
 292     def getAuthPermissions(self, key, extra):
 293          return {'permissions': extra.get("permissions", None),
 294                  'authextra': extra.get("authextra", None)}
 295
 296     def getAuthSecret(self, key):
 297         secret = CREDENTIALS.get(key, None)
 298         if secret is None:
 299             return None
 300         # secret must be of str type to be hashed
 301         return str(secret)
 302
 303
 304     def onAuthenticated(self, key, permissions):
 305         if not OPENDOOR:
 306             # check net filter
 307             if NETMASK != []:
 308                 ipAllowed = False
 309                 for netfilter in NETMASK:
 310                     ipHex=ipV4ToHex(self.peer.host)
 311                     ipAllowed = (ipHex & netfilter['mask']) == netfilter['ipAllowed'] & netfilter['mask']
 312                     if ipAllowed:
 313                         break
 314                 if not ipAllowed:
 315                     raise Exception("host " + self.peer.host + " is not allowed!")
 316             # check authentication key
 317             if key is None:
 318                 raise Exception("Authentication failed")
 319             # check permissions, array.index throws exception
 320             for req in permissions['permissions']:
 321                     WHITELIST.index(req);
 322         # create cloudeebus service instance
 323         self.cloudeebusService = CloudeebusService(permissions)
 324         # register it for RPC
 325         self.registerForRpc(self.cloudeebusService)
 326         # register for Publish / Subscribe
 327         self.registerForPubSub("", True)
 328
 329
 330     def connectionLost(self, reason):
 331         WampCraServerProtocol.connectionLost(self, reason)
 332         if factory.getConnectionCount() == 0:
 333             cache.reset()
 334
 335
 336
 337 ###############################################################################
 338
 339 if __name__ == '__main__':
 340
 341     cache = DbusCache()
 342
 343     parser = argparse.ArgumentParser(description='Javascript DBus bridge.')
 344     parser.add_argument('-v', '--version', action='store_true',
 345         help='print version and exit')
 346     parser.add_argument('-d', '--debug', action='store_true',
 347         help='log debug info on standard output')
 348     parser.add_argument('-o', '--opendoor', action='store_true',
 349         help='allow anonymous access to all services')
 350     parser.add_argument('-p', '--port', default='9000',
 351         help='port number')
 352     parser.add_argument('-c', '--credentials',
 353         help='path to credentials file')
 354     parser.add_argument('-w', '--whitelist',
 355         help='path to whitelist file')
 356     parser.add_argument('-n', '--netmask',
 357         help='netmask,IP filter (comma separated.) eg. : -n 127.0.0.1,192.168.2.0/24,10.12.16.0/255.255.255.0')
 358
 359     args = parser.parse_args(sys.argv[1:])
 360
 361     if args.version:
 362         print("Cloudeebus version " + VERSION)
 363         exit(0)
 364
 365     if args.debug:
 366         log.startLogging(sys.stdout)
 367
 368     OPENDOOR = args.opendoor
 369
 370     if args.credentials:
 371         jfile = open(args.credentials)
 372         CREDENTIALS = json.load(jfile)
 373         jfile.close()
 374
 375     if args.whitelist:
 376         jfile = open(args.whitelist)
 377         WHITELIST = json.load(jfile)
 378         jfile.close()
 379
 380     if args.netmask:
 381         iplist = args.netmask.split(",")
 382         for ip in iplist:
 383             if ip.rfind("/") != -1:
 384                 ip=ip.split("/")
 385                 ipAllowed = ip[0]
 386                 mask = ip[1]
 387             else:
 388                 ipAllowed = ip
 389                 mask = "255.255.255.255"
 390             NETMASK.append( {'ipAllowed': ipV4ToHex(ipAllowed), 'mask' : ipV4ToHex(mask)} )
 391
 392     uri = "ws://localhost:" + args.port
 393
 394     factory = WampServerFactory(uri, debugWamp = args.debug)
 395     factory.protocol = CloudeebusServerProtocol
 396     factory.setProtocolOptions(allowHixie76 = True)
 397
 398     listenWS(factory)
 399
 400     DBusGMainLoop(set_as_default=True)
 401
 402     reactor.run()