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