src/python/dm/common/plugins/fileTransferPlugin.py

+#!/usr/bin/env python
+
+import os
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.utility.dmSubprocess import DmSubprocess
+from dm.common.exceptions.invalidArgument import InvalidArgument
+from dm.common.exceptions.invalidRequest import InvalidRequest
+from fileProcessorInterface import FileProcessorInterface
+
+class FileTransferPlugin(FileProcessorInterface):
+
+    def __init__(self, command, src=None, dest=None):
+        self.src = src
+        self.dest = dest
+        self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+        if command is None or not len(command):
+            raise InvalidArgument('File transfer command must be non-empty string.')
+        self.command = command
+        self.subprocess = None
+
+    def processFile(self, filePath, daqPath, experiment):
+        storageHost = experiment.get('storageHost')
+        storageDirectory = experiment.get('storageDirectory')
+        dest = '%s:%s' % (storageHost, storageDirectory)
+        # Use relative path with respect to daq directory as a source
+        os.chdir(daqPath)
+        src = os.path.relpath(filePath, daqPath)
+        self.start(src, dest)
+
+    def getFullCommand(self, src, dest):
+        return '%s %s %s' % (self.command, src, dest)
+
+    def setSrc(self, src):
+        self.src = src
+
+    def setDest(self, dest):
+        self.dest = dest
+
+    def start(self, src=None, dest=None):
+        # Use preconfigured source if provided source is None
+        fileSrc = src
+        if src is None:
+            fileSrc = self.src
+        # Use provided destination only if preconfigured destination is None
+        # Plugins may have desired destination preconfigured for all files
+        fileDest = self.dest
+        if self.dest is None:
+            fileDest = dest
+
+        if not fileSrc or not fileDest:
+            raise InvalidRequest('Both source and destination must be non-empty strings.')
+        self.subprocess = DmSubprocess.getSubprocess(self.getFullCommand(fileSrc, fileDest))
+        return self.subprocess.run()
+
+    def wait(self):
+        if self.subprocess:
+            return self.subprocess.wait()
+        return None
+
+    def poll(self):
+        if self.subprocess:
+            return self.subprocess.poll()
+        return None
+
+    def getStdOut(self):
+        if self.subprocess:
+            return self.subprocess.getStdOut()
+        return None
+
+
+    def getStdErr(self):
+        if self.subprocess:
+            return self.subprocess.getStdErr()
+        return None
+
+    def getExitStatus(self):
+        if self.subprocess:
+            return self.subprocess.getExitStatus()
+        return None
+
+    def reset(self):
+        self.subprocess = None
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    ft = FileTransfer('rsync -arv', '/tmp/xyz', '/tmp/xyz2')
+    ft.start()
+    print 'StdOut: ', ft.getStdOut()
+    print 'StdErr: ', ft.getStdErr()
+    print 'Exit Status: ', ft.getExitStatus()

src/python/dm/common/plugins/rsyncFileTransferPlugin.py

+#!/usr/bin/env python
+
+from fileTransferPlugin import FileTransferPlugin
+class RsyncFileTransferPlugin(FileTransferPlugin):
+
+    COMMAND = 'rsync -arvlPR'
+
+    def __init__(self, src=None, dest=None):
+        FileTransferPlugin.__init__(self, self.COMMAND, src, dest)
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    ft = RsyncFileTransferPlugin('/tmp/xyz', '/tmp/xyz2')
+    ft.start()
+    print 'StdOut: ', ft.getStdOut()
+    print 'StdErr: ', ft.getStdErr()
+    print 'Exit Status: ', ft.getExitStatus()

src/python/dm/common/processing/fileProcessingManager.py

+#!/usr/bin/env python
+
+import threading
+import time
+
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.utility.configurationManager import ConfigurationManager
+from dm.common.utility.objectUtility import ObjectUtility
+from dm.common.utility.timeBasedProcessingQueue import TimeBasedProcessingQueue
+from dm.common.utility.singleton import Singleton
+from fileProcessingThread import FileProcessingThread
+
+class FileProcessingManager(threading.Thread,Singleton):
+
+    CONFIG_SECTION_NAME = 'FileProcessingManager'
+    NUMBER_OF_PROCESSING_THREADS_KEY = 'numberofprocessingthreads'
+    DEFAULT_NUMBER_OF_RETRIES_KEY = 'defaultnumberofretries'
+    DEFAULT_RETRY_WAIT_PERIOD_IN_SECONDS_KEY = 'defaultretrywaitperiodinseconds'
+    FILE_PROCESSOR_KEY = 'fileprocessor'
+
+    # Singleton.
+    __instanceLock = threading.RLock()
+    __instance = None
+
+    def __init__(self):
+        FileProcessingManager.__instanceLock.acquire()
+        try:
+            if FileProcessingManager.__instance:
+                return
+            FileProcessingManager.__instance = self
+            self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+
+            self.logger.debug('Initializing')
+            self.lock = threading.RLock()
+            self.fileProcessingThreadDict = {}
+            self.eventFlag = threading.Event()
+            self.fileProcessorDict = {}
+            self.fileProcessorKeyList = []
+            self.fileProcessingQueue = TimeBasedProcessingQueue()
+            self.processedFileDict = {}
+            self.unprocessedFileDict = {}
+            self.__configure()
+            self.logger.debug('Initialization complete')
+        finally:
+            FileProcessingManager.__instanceLock.release()
+
+    def __configure(self):
+        cm = ConfigurationManager.getInstance()
+        configItems = cm.getConfigItems(FileProcessingManager.CONFIG_SECTION_NAME)
+        self.logger.debug('Got config items: %s' % configItems)
+        self.nProcessingThreads = int(cm.getConfigOption(FileProcessingManager.CONFIG_SECTION_NAME, FileProcessingManager.NUMBER_OF_PROCESSING_THREADS_KEY))
+        self.defaultNumberOfRetries = int(cm.getConfigOption(FileProcessingManager.CONFIG_SECTION_NAME, FileProcessingManager.DEFAULT_NUMBER_OF_RETRIES_KEY))
+        self.defaultRetryWaitPeriodInSeconds = int(cm.getConfigOption(FileProcessingManager.CONFIG_SECTION_NAME, FileProcessingManager.DEFAULT_RETRY_WAIT_PERIOD_IN_SECONDS_KEY))
+
+        # Create processors  
+        for (key,value) in configItems:
+            if key.startswith(FileProcessingManager.FILE_PROCESSOR_KEY):
+                (moduleName,className,constructor) = cm.getModuleClassConstructorTuple(value)
+                self.logger.debug('Creating file processor instance of class %s' % className)
+                fileProcessor = ObjectUtility.createObjectInstance(moduleName, className, constructor)
+                self.logger.debug('Configuring file processor %s' % fileProcessor)
+                fileProcessor.setNumberOfRetries(self.defaultNumberOfRetries)
+                fileProcessor.setRetryWaitPeriodInSeconds(self.defaultRetryWaitPeriodInSeconds)
+                fileProcessor.configure()
+                self.fileProcessorDict[key] = fileProcessor
+        self.fileProcessorKeyList = self.fileProcessorDict.keys()
+        self.fileProcessorKeyList.sort()
+        self.logger.debug('File processor keys: %s' % self.fileProcessorKeyList)
+
+    def processFile(self, fileInfo):
+        self.fileProcessingQueue.push(fileInfo)
+        self.eventFlag.set()
+
+    def start(self):
+        self.lock.acquire()
+        try:
+            self.logger.debug('Starting file processing threads')
+            for i in range(0, self.nProcessingThreads):
+                tName = 'FileProcessingThread-%s' % i
+                t = FileProcessingThread(tName, self, self.fileProcessorDict, self.fileProcessorKeyList, self.fileProcessingQueue, self.processedFileDict, self.unprocessedFileDict)
+                t.start()
+                self.fileProcessingThreadDict[tName] = t
+        finally:
+            self.lock.release()
+
+    def stop(self):
+        self.logger.debug('Stopping file processing threads')
+        for (tName, t) in self.fileProcessingThreadDict.items():
+            t.stop()
+        self.lock.acquire()
+        try:
+            self.eventFlag.set()
+        finally:
+             self.lock.release()
+        for (tName, t) in self.fileProcessingThreadDict.items():
+            t.join()
+
+    def setEvent(self):
+        self.lock.acquire()
+        try:
+             self.eventFlag.set()
+        finally:
+            self.lock.release()
+
+    def clearEvent(self):
+        self.lock.acquire()
+        try:
+            self.eventFlag.clear()
+        finally:
+            self.lock.release()
+
+    def waitOnEvent(self, timeoutInSeconds=None):
+        self.eventFlag.wait(timeoutInSeconds)
+
+####################################################################
+# Testing
+
+if __name__ == '__main__':
+    fp = FileProcessingManager.getInstance()
+    print fp
+    #fp.start()
+    #time.sleep(30)
+    #fp.stop()
+
+

src/python/dm/common/processing/fileProcessingThread.py

+#!/usr/bin/env python
+
+import threading
+
+from dm.common.utility.loggingManager import LoggingManager
+
+class FileProcessingThread(threading.Thread):
+
+    THREAD_EVENT_TIMEOUT_IN_SECONDS = 10.0
+
+    def __init__ (self, name, fileProcessingManager, fileProcessorDict, fileProcessorKeyList, fileProcessingQueue, processedFileDict, unprocessedFileDict):
+    
+        threading.Thread.__init__(self)
+        self.setName(name)
+        self.exitFlag = False
+        self.fileProcessingManager = fileProcessingManager
+        self.fileProcessorDict = fileProcessorDict
+        self.fileProcessorKeyList = fileProcessorKeyList
+        self.fileProcessingQueue = fileProcessingQueue
+        self.processedFileDict = processedFileDict 
+        self.unprocessedFileDict = unprocessedFileDict 
+        self.logger = LoggingManager.getInstance().getLogger(name)
+
+    def run(self):
+        self.logger.debug('Starting thread: %s' % self.getName())
+        while True:
+            self.fileProcessingManager.clearEvent()
+            if self.exitFlag:
+                self.logger.debug('Exit flag is set')
+                break
+
+            while True:
+                fileInfo = self.fileProcessingQueue.pop()
+                if fileInfo is None:
+                    break
+                filePath = fileInfo.get('filePath')
+                
+                try:
+                    for processorKey in self.fileProcessorKeyList: 
+                        processor = self.fileProcessorDict.get(processorKey)
+                        processorName = processor.__class__.__name__
+                        fileProcessedByDict = fileInfo.get('processedByDict', {})
+                        fileInfo['processedByDict'] = fileProcessedByDict
+
+                        if fileProcessedByDict.has_key(processorName):
+                            self.logger.debug('%s has already been processed by %s ' % (fileInfo, processorName))
+                            continue
+
+                        self.logger.debug('%s is about to process file %s ' % (processorName, fileInfo))
+                        try:
+                            processor.processFile(fileInfo)
+                            fileProcessedByDict[processorName] = True
+                            self.logger.debug('%s processed file at path %s ' % (processorName, filePath))
+                        except Exception, ex:
+                            self.logger.exception(ex)
+                            self.logger.debug('%s processing failed for file at path %s ' % (processorName, filePath))
+                            fileProcessingDict = fileInfo.get('processingDict', {})
+                            fileInfo['processingDict'] = fileProcessingDict
+                            processorDict = fileProcessingDict.get(processorName, {}) 
+                            fileProcessingDict[processorName] = processorDict
+
+                            processorDict['lastError'] = str(ex)
+                            nRetriesLeft = processorDict.get('numberOfRetriesLeft', processor.getNumberOfRetries()) 
+                            self.logger.debug('Number of %s retries left for file %s: %s' % (processorName, filePath, nRetriesLeft))
+                            processorDict['numberOfRetriesLeft'] = nRetriesLeft - 1
+                            if nRetriesLeft <= 0:
+                                self.logger.debug('No more %s retries left for file %s' % (processorName, fileInfo))
+                                self.unprocessedFileDict[filePath] = fileInfo
+                            else:
+                                retryWaitPeriod = processor.getRetryWaitPeriodInSeconds()
+                                self.logger.debug('%s will retry processing file %s in %s seconds' % (processorName, filePath, retryWaitPeriod))
+                                self.fileProcessingQueue.push(fileInfo, retryWaitPeriod)
+
+                except Exception, ex:
+                    self.logger.exception(ex)
+
+            self.fileProcessingManager.waitOnEvent(self.THREAD_EVENT_TIMEOUT_IN_SECONDS)
+        self.logger.debug('%s is done' % self.getName())
+
+    def stop(self):
+        self.exitFlag = True
+
+####################################################################
+# Testing
+
+if __name__ == '__main__':
+    pass
+
+

src/python/dm/common/processing/plugins/fileProcessor.py

+#!/usr/bin/env python
+
+import abc
+class FileProcessor:
+
+    DEFAULT_NUMBER_OF_RETRIES = 0
+    DEFAULT_RETRY_WAIT_PERIOD_IN_SECONDS = 60
+
+    def __init__(self):
+        self.configDict = {}
+
+    @abc.abstractmethod
+    def processFile(self, fileInfo):
+        return NotImplemented
+        
+    def configure(self):
+        # Use this method for processor configuration
+        pass
+
+    def setConfigKeyValue(self, key, value):
+        self.configDict[key] = value
+
+    def getConfigKeyValue(self, key):
+        return self.configDict.get(key)
+
+    def setNumberOfRetries(self, nRetries):
+        self.configDict['numberOfRetries'] = nRetries
+
+    def getNumberOfRetries(self):
+        return self.configDict.get('numberOfRetries', self.DEFAULT_NUMBER_OF_RETRIES)
+
+    def setRetryWaitPeriodInSeconds(self, waitPeriod):
+        self.configDict['retryWaitPeriodInSeconds'] = waitPeriod
+
+    def getRetryWaitPeriodInSeconds(self):
+        return self.configDict.get('retryWaitPeriodInSeconds', self.DEFAULT_RETRY_WAIT_PERIOD_IN_SECONDS)
+

src/python/dm/common/processing/plugins/fileTransferPlugin.py

+#!/usr/bin/env python
+
+import os
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.utility.dmSubprocess import DmSubprocess
+from dm.common.exceptions.invalidArgument import InvalidArgument
+from dm.common.exceptions.invalidRequest import InvalidRequest
+from fileProcessor import FileProcessor
+
+class FileTransferPlugin(FileProcessor):
+
+    def __init__(self, command, src=None, dest=None):
+        FileProcessor.__init__(self)
+        self.src = src
+        self.dest = dest
+        self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+        if command is None or not len(command):
+            raise InvalidArgument('File transfer command must be non-empty string.')
+        self.command = command
+        self.subprocess = None
+
+    def processFile(self, fileInfo):
+        filePath = fileInfo.get('filePath')
+        dataDirectory = fileInfo.get('dataDirectory')
+        experiment = fileInfo.get('experiment')
+
+        storageHost = experiment.get('storageHost')
+        storageDirectory = experiment.get('storageDirectory')
+        dest = '%s:%s' % (storageHost, storageDirectory)
+        # Use relative path with respect to data directory as a source
+        os.chdir(dataDirectory)
+        src = os.path.relpath(filePath, dataDirectory)
+        self.start(src, dest)
+
+    def getFullCommand(self, src, dest):
+        return '%s %s %s' % (self.command, src, dest)
+
+    def setSrc(self, src):
+        self.src = src
+
+    def setDest(self, dest):
+        self.dest = dest
+
+    def start(self, src=None, dest=None):
+        # Use preconfigured source if provided source is None
+        fileSrc = src
+        if src is None:
+            fileSrc = self.src
+        # Use provided destination only if preconfigured destination is None
+        # Plugins may have desired destination preconfigured for all files
+        fileDest = self.dest
+        if self.dest is None:
+            fileDest = dest
+
+        if not fileSrc or not fileDest:
+            raise InvalidRequest('Both source and destination must be non-empty strings.')
+        self.subprocess = DmSubprocess.getSubprocess(self.getFullCommand(fileSrc, fileDest))
+        return self.subprocess.run()
+
+    def wait(self):
+        if self.subprocess:
+            return self.subprocess.wait()
+        return None
+
+    def poll(self):
+        if self.subprocess:
+            return self.subprocess.poll()
+        return None
+
+    def getStdOut(self):
+        if self.subprocess:
+            return self.subprocess.getStdOut()
+        return None
+
+
+    def getStdErr(self):
+        if self.subprocess:
+            return self.subprocess.getStdErr()
+        return None
+
+    def getExitStatus(self):
+        if self.subprocess:
+            return self.subprocess.getExitStatus()
+        return None
+
+    def reset(self):
+        self.subprocess = None
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    ft = FileTransfer('rsync -arv', '/tmp/xyz', '/tmp/xyz2')
+    ft.start()
+    print 'StdOut: ', ft.getStdOut()
+    print 'StdErr: ', ft.getStdErr()
+    print 'Exit Status: ', ft.getExitStatus()

src/python/dm/common/processing/plugins/rsyncFileTransferPlugin.py

+#!/usr/bin/env python
+
+from fileTransferPlugin import FileTransferPlugin
+class RsyncFileTransferPlugin(FileTransferPlugin):
+
+    COMMAND = 'rsync -arvlPR'
+
+    def __init__(self, src=None, dest=None):
+        FileTransferPlugin.__init__(self, self.COMMAND, src, dest)
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    ft = RsyncFileTransferPlugin('/tmp/xyz', '/tmp/xyz2')
+    ft.start()
+    print 'StdOut: ', ft.getStdOut()
+    print 'StdErr: ', ft.getStdErr()
+    print 'Exit Status: ', ft.getExitStatus()

src/python/dm/common/service/auth/authorizationPrincipalAuthenticator.py

+#!/usr/bin/env python
+
+from dm.common.utility.loggingManager import LoggingManager
+
+class AuthorizationPrincipalAuthenticator:
+
+    def __init__(self, name=None):
+        self.name = name
+        self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+
+    def getName(self):
+        return self.name
+
+    def authenticatePrincipal(self, principal, password):
+        return None
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    pass
+

src/python/dm/common/service/auth/authorizationPrincipalManager.py

+#!/usr/bin/env python
+
+import os
+
+from dm.common.exceptions.authorizationError import AuthorizationError
+from dm.common.objects.dmObjectManager import DmObjectManager
+from dm.common.utility.configurationManager import ConfigurationManager
+from dm.common.utility.objectCache import ObjectCache
+from dm.common.utility.cryptUtility import CryptUtility
+from dm.common.utility.objectUtility import ObjectUtility
+from dm.common.utility.singleton import Singleton
+
+class AuthorizationPrincipalManager(DmObjectManager, Singleton):
+
+    DEFAULT_CACHE_SIZE = 10000 # number of items
+    DEFAULT_CACHE_OBJECT_LIFETIME = 3600 # seconds
+
+    CONFIG_SECTION_NAME = 'AuthorizationPrincipalManager'
+    ADMIN_ROLE_NAME_KEY = 'adminrolename'
+    PRINCIPAL_RETRIEVER_KEY = 'principalretriever'
+    PRINCIPAL_AUTHENTICATOR_KEY = 'principalauthenticator'
+
+    # Singleton instance.
+    __instance = None
+
+    def __init__(self):
+        if AuthorizationPrincipalManager.__instance:
+            return
+        AuthorizationPrincipalManager.__instance = self
+        DmObjectManager.__init__(self)
+        self.configurationManager = ConfigurationManager.getInstance()
+        self.principalRetriever = None
+        self.principalAuthenticatorList = []
+        self.objectCache = ObjectCache(AuthorizationPrincipalManager.DEFAULT_CACHE_SIZE, AuthorizationPrincipalManager.DEFAULT_CACHE_OBJECT_LIFETIME)
+        self.configure()
+
+    @classmethod
+    def cryptPassword(cls, cleartext):
+        return CryptUtility.cryptPassword(cleartext)
+
+    @classmethod
+    def cryptPasswordWithPbkdf2(cls, cleartext):
+        return CryptUtility.cryptPasswordWithPbkdf2(cleartext)
+
+    def configure(self):
+        configItems = self.configurationManager.getConfigItems(AuthorizationPrincipalManager.CONFIG_SECTION_NAME)
+        self.logger.debug('Got config items: %s' % configItems)
+        adminRoleName = self.configurationManager.getConfigOption(AuthorizationPrincipalManager.CONFIG_SECTION_NAME, AuthorizationPrincipalManager.ADMIN_ROLE_NAME_KEY)
+
+        # Create principal retriever
+        principalRetriever = self.configurationManager.getConfigOption(AuthorizationPrincipalManager.CONFIG_SECTION_NAME, AuthorizationPrincipalManager.PRINCIPAL_RETRIEVER_KEY)
+        (moduleName,className,constructor) = self.configurationManager.getModuleClassConstructorTuple(principalRetriever, AuthorizationPrincipalManager)    
+        self.logger.debug('Creating principal retriever class: %s' % className)
+        self.principalRetriever = ObjectUtility.createObjectInstance(moduleName, className, constructor)
+        if adminRoleName is not None:
+            self.principalRetriever.setAdminRoleName(adminRoleName)
+        self.logger.debug('Authorization principal retriever: %s' % (self.principalRetriever))
+
+        # Create principal authenticators
+        for (key,value) in configItems:
+            if key.startswith(AuthorizationPrincipalManager.PRINCIPAL_AUTHENTICATOR_KEY):
+                (moduleName,className,constructor) = self.configurationManager.getModuleClassConstructorTuple(value, AuthorizationPrincipalManager)    
+                self.logger.debug('Creating principal authenticator class: %s' % className)
+                principalAuthenticator = ObjectUtility.createObjectInstance(moduleName, className, constructor)
+                self.addAuthorizationPrincipalAuthenticator(principalAuthenticator)
+                self.logger.debug('Authorization principal authenticator: %s' % (principalAuthenticator))
+
+    def addAuthorizationPrincipalAuthenticator(self, principalAuthenticator):
+        self.principalAuthenticatorList.append(principalAuthenticator)
+
+    def getAuthenticatedAuthorizationPrincipal(self, username, password):
+        """ Get principal based on a username and password """
+        # First try cache.
+        #self.logger.debug('Trying username %s from the cache' % username)
+        principal = None
+        principal = self.objectCache.get(username)
+        if principal is None:
+            # Try principal retriever
+            principal = self.principalRetriever.getAuthorizationPrincipal(username)
+
+        if principal is None:
+            self.logger.debug('No principal for username: %s' % username)
+            return 
+        
+        # Try all authorization principal authenticators.
+        for principalAuthenticator in self.principalAuthenticatorList:
+            self.logger.debug('Attempting to authenticate %s by %s' % (username, principalAuthenticator.getName()))
+            authenticatedPrincipal = principalAuthenticator.authenticatePrincipal(principal, password)
+            if authenticatedPrincipal is not None:
+                self.logger.debug('Adding authorization principal %s to the cache, authenticated by %s' % (principal.getName(),principalAuthenticator.getName()))
+                self.objectCache.put(username, authenticatedPrincipal)
+                return authenticatedPrincipal
+        return None
+
+    def removeAuthorizationPrincipal(self, username):
+        """ Clear principal from the cache. """
+        self.objectCache.remove(username)
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    am = AuthorizationPrincipalManager.getInstance()
+    authPrincipal = am.getAuthorizationPrincipal('sveseli', 'sv')
+    print 'Auth principal: ', authPrincipal
+
+

src/python/dm/common/service/auth/authorizationPrincipalRetriever.py

+#!/usr/bin/env python
+
+from dm.common.constants import dmRole
+from dm.common.utility.loggingManager import LoggingManager
+
+class AuthorizationPrincipalRetriever:
+
+    def __init__(self, name=None):
+        self.adminRoleName = dmRole.DM_ADMIN_ROLE
+        self.name = name
+        self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+
+    def getName(self):
+        return self.name
+
+    def setAdminRoleName(self, adminRoleName):
+        self.adminRoleName = adminRoleName
+
+    def getAuthorizationPrincipal(self, username):
+        return None
+
+    def checkAutorizationPrincipalRole(self, principal):
+        if principal is None or self.adminRoleName is None:
+            return
+        userInfo = principal.getUserInfo()
+        if userInfo is None:
+            return
+        for userSystemRoleName in userInfo.get('userSystemRoleNameList', []):
+            if userSystemRoleName == self.adminRoleName:
+                principal.setRole(dmRole.DM_ADMIN_ROLE)
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    pass
+

src/python/dm/common/service/auth/cryptedPasswordPrincipalAuthenticator.py

+#!/usr/bin/env python
+
+from dm.common.utility.cryptUtility import CryptUtility
+from authorizationPrincipalAuthenticator import AuthorizationPrincipalAuthenticator 
+
+class CryptedPasswordPrincipalAuthenticator(AuthorizationPrincipalAuthenticator):
+
+    def __init__(self):
+        AuthorizationPrincipalAuthenticator.__init__(self, self.__class__.__name__)
+
+    def authenticatePrincipal(self, principal, password):
+        if principal is not None:
+            principalToken = principal.getToken()
+            if principalToken is not None and len(principalToken):
+                if CryptUtility.verifyPasswordWithPbkdf2(password, principalToken):
+                    self.logger.debug('Authentication successful for %s' % principal.getName())
+                    return principal
+                else:
+                    self.logger.debug('Authentication failed for %s' % principal.getName())
+            else:
+                self.logger.debug('Token is empty for %s, authentication not performed' % principal.getName())
+        return None
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    pass
+

src/python/dm/common/service/auth/dbPrincipalRetriever.py

+#!/usr/bin/env python
+
+from dm.common.constants import dmRole
+from dm.common.objects.authorizationPrincipal import AuthorizationPrincipal
+from dm.common.db.api.userDbApi import UserDbApi
+from authorizationPrincipalRetriever import AuthorizationPrincipalRetriever 
+
+class DbPrincipalRetriever(AuthorizationPrincipalRetriever):
+
+    def __init__(self):
+        AuthorizationPrincipalRetriever.__init__(self, self.__class__.__name__)
+        self.dbApi = UserDbApi()
+
+    def getAuthorizationPrincipal(self, username):
+        principal = None
+        try:
+            user = self.dbApi.getUserWithPasswordByUsername(username)
+            principal = AuthorizationPrincipal(name=username, token=user.get('password'))
+            principal.setRole(dmRole.DM_USER_ROLE)
+            principal.setUserInfo(user)
+            self.checkAutorizationPrincipalRole(principal)
+        except Exception, ex:
+            self.logger.debug(ex)
+        return principal
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    pass
+

src/python/dm/common/service/auth/ldapPasswordPrincipalAuthenticator.py

+#!/usr/bin/env python
+
+from dm.common.utility.ldapUtility import LdapUtility
+from authorizationPrincipalAuthenticator import AuthorizationPrincipalAuthenticator 
+
+class LdapPasswordPrincipalAuthenticator(AuthorizationPrincipalAuthenticator):
+
+    def __init__(self, serverUrl, dnFormat):
+        AuthorizationPrincipalAuthenticator.__init__(self, self.__class__.__name__)
+        self.ldapUtility = LdapUtility(serverUrl, dnFormat)
+
+    def authenticatePrincipal(self, principal, password):
+        if principal is not None:
+            try:
+                self.logger.debug('Checking credentials for %s' % principal.getName())
+                self.ldapUtility.checkCredentials(principal.getName(), password)
+                return principal
+            except Exception, ex:
+                self.logger.debug(ex)
+        return None
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    pass
+

src/python/dm/common/service/auth/noOpPrincipalRetriever.py

+#!/usr/bin/env python
+
+from dm.common.constants import dmRole
+from dm.common.objects.authorizationPrincipal import AuthorizationPrincipal
+from dm.common.utility.cryptUtility import CryptUtility
+from authorizationPrincipalRetriever import AuthorizationPrincipalRetriever 
+
+class NoOpPrincipalRetriever(AuthorizationPrincipalRetriever):
+
+    def __init__(self):
+        AuthorizationPrincipalRetriever.__init__(self, self.__class__.__name__)
+
+    def getAuthorizationPrincipal(self, username):
+        # Set password to be the same as username
+        noOpPassword = CryptUtility.cryptPasswordWithPbkdf2(username)
+        principal = AuthorizationPrincipal(name=username, token=noOpPassword)
+        principal.setRole(dmRole.DM_USER_ROLE)
+        if self.adminRoleName is not None:
+            principal.setRole(dmRole.DM_ADMIN_ROLE)
+        return principal
+
+#######################################################################
+# Testing.
+if __name__ == '__main__':
+    pass
+

src/python/dm/common/service/dmController.py

+#!/usr/bin/env python
+
+#
+# Base DM controller class.
+#
+
+#######################################################################
+
+import cherrypy
+import httplib
+import json
+
+from sys import exc_info
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.constants import dmStatus
+from dm.common.constants import dmHttpHeaders
+from dm.common.exceptions.dmException import DmException
+from dm.common.exceptions import dmExceptionMap
+from dm.common.exceptions.internalError import InternalError
+
+#######################################################################
+
+class DmController(object):
+    """ Base controller class. """
+    def __init__(self):
+        self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+
+    @classmethod
+    def getLogger(cls):
+        logger = LoggingManager.getInstance().getLogger(cls.__name__)
+        return logger
+
+    @classmethod
+    def addDmResponseHeaders(cls, status=dmStatus.DM_OK, msg='Success', exceptionType=None):
+        cherrypy.response.headers[dmHttpHeaders.DM_STATUS_CODE_HTTP_HEADER] = status
+        cherrypy.response.headers[dmHttpHeaders.DM_STATUS_MESSAGE_HTTP_HEADER] = msg
+        if exceptionType is not None:
+            cherrypy.response.headers[dmHttpHeaders.DM_EXCEPTION_TYPE_HTTP_HEADER] = exceptionType
+
+    @classmethod
+    def addDmSessionRoleHeaders(cls, role):
+        cherrypy.response.headers[dmHttpHeaders.DM_SESSION_ROLE_HTTP_HEADER] = role
+
+    @classmethod
+    def addDmExceptionHeaders(cls, ex):
+        cls.handleException(ex)
+
+    @classmethod
+    def handleCpException(cls):
+        cherrypy.response.status = httplib.OK
+        ex = exc_info()[1]
+        if ex == None:
+            ex = InternalError('Internal Webservice Error')
+        cls.handleException(ex)
+
+    @classmethod
+    def handleException(cls, ex):
+        exClass = ex.__class__.__name__.split('.')[-1]
+        status = None
+        msg = '%s' % ex
+        msg = msg.split('\n')[0]
+        for code in dmExceptionMap.DM_EXCEPTION_MAP.keys():
+            exStr = dmExceptionMap.DM_EXCEPTION_MAP.get(code).split('.')[-1]
+            if exStr == exClass:
+                status = code
+        if not status:
+            status = dmStatus.DM_ERROR
+        cls.addDmResponseHeaders(status, msg, exClass)
+
+    @classmethod
+    def formatJsonResponse(cls, response):
+        cherrypy.response.headers['Content-Type'] = 'application/json'
+        return '%s' % (response)
+
+    @classmethod
+    def toJson(cls, o):
+        return json.dumps(o)
+
+    @classmethod
+    def fromJson(cls, s):
+        return json.loads(s)
+
+    @classmethod
+    def listToJson(cls, dmObjectList):
+        jsonList = []
+        for dmObject in dmObjectList:
+            jsonList.append(dmObject.getDictRep(keyList='__all__'))
+        return json.dumps(jsonList)
+
+    @classmethod
+    def getSessionUser(cls):
+        return cherrypy.serving.session.get('user')
+
+    @classmethod
+    def getSessionUsername(cls):
+        return cherrypy.serving.session.get('_cp_username')
+
+    # Exception decorator for all exposed method calls
+    @classmethod
+    def execute(cls, func):
+        def decorate(*args, **kwargs):
+            try:
+                response = func(*args, **kwargs)
+            except DmException, ex:
+                cls.getLogger().error('%s' % ex)
+                cls.handleException(ex)
+                response = ex.getFullJsonRep()
+            except Exception, ex:
+                cls.getLogger().error('%s' % ex)
+                cls.handleException(ex)
+                response = InternalError(ex).getFullJsonRep()
+            return cls.formatJsonResponse(response)
+        return decorate
+

src/python/dm/common/service/dmRestWebServiceBase.py

+#!/usr/bin/env python
+
+#
+# Base web service class.
+#
+
+####################################################################
+import sys
+import os
+import cherrypy
+from cherrypy.process import plugins
+from cherrypy import server
+
+from dm.common.constants import dmStatus
+from dm.common.utility.configurationManager import ConfigurationManager
+from dm.common.utility.loggingManager import LoggingManager
+from dm.common.utility.dmModuleManager import DmModuleManager
+from dm.common.exceptions.configurationError import ConfigurationError
+
+####################################################################
+
+class DmRestWebServiceBase:
+
+    DEFAULT_N_SERVER_REQUEST_THREADS = 10
+    DEFAULT_SERVER_SOCKET_TIMEOUT = 30
+    CONFIG_SECTION_NAME = 'WebService'
+    CONFIG_OPTION_NAME_LIST = [ 'serviceHost', 'servicePort', 
+        'sslCertFile', 'sslKeyFile', 'sslCaCertFile' ]
+
+    class SignalHandler:
+        def __init__(self, signal, oldSignalHandler):
+            self.signal = signal
+            self.oldSignalHandler = oldSignalHandler
+            self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+
+        def signalHandler(self):
+            self.logger.debug('%s signal handler called' % self.signal)
+            DmModuleManager.getInstance().stopModules()
+            self.oldSignalHandler()
+
+    def __init__(self, routeMapper):
+        self.configurationManager = ConfigurationManager.getInstance()
+        self.routeMapper = routeMapper
+        self.options = None
+        self.args = None
+        self.logger = None
+
+    def prepareOptions(self):
+        from optparse import OptionParser
+        p = OptionParser()
+        p.add_option('-d', '--daemon', action="store_true",
+            dest='daemonFlag', default=False, 
+            help="Run as a daemon.")
+        p.add_option('-p', '--pid-file',
+            dest='pidFile', default=None,
+            help="Store process id in the given file.")
+        p.add_option('', '--config-file',
+            dest='configFile', default=None,
+            help="Service configuration file.")
+        p.add_option('-P', '--port',
+            dest='servicePort', default=None,
+            help="Service port.")
+        p.add_option('-H', '--host',
+            dest='serviceHost', default=None,
+            help="Service host.")
+        p.add_option('-C', '--ssl-ca-cert', 
+            dest='sslCaCertFile', default=None,
+            help='SSL CA certificate path (used for client SSL certificate verification). Requires --ssl-key and --ssl-cert.')
+        p.add_option('-c', '--ssl-cert', 
+            dest='sslCertFile', default=None,
+            help='SSL certificate path. SSL operation requires both --ssl-key and --ssl-cert. Client SSL certificate verification also requires --ssl-ca-cert.')
+        p.add_option('-k', '--ssl-key', 
+            dest='sslKeyFile', default=None, 
+            help='SSL key path. SSL operation requires both --ssl-key and --ssl-cert. Client SSL certificate verification also requires --ssl-ca-cert.')
+        p.add_option('', '--n-server-threads', 
+            dest='nServerThreads', default=DmRestWebServiceBase.DEFAULT_N_SERVER_REQUEST_THREADS,
+            help='Number of service request handler threads (defaut: %s).' % DmRestWebServiceBase.DEFAULT_N_SERVER_REQUEST_THREADS)
+        return p
+
+    def initDmModules(self):
+        return None
+
+    def getDefaultServerHost(self):
+        return None
+
+    def getDefaultServerPort(self):
+        return None
+
+    # Instantiate modified signal handler that stops dm modules first, 
+    # and then does the default action.
+    def modifySignalHandlers(self, engine):
+        pluginsSignalHandler = plugins.SignalHandler(engine)
+        handlers = pluginsSignalHandler.handlers
+
+        # Add handler for interrupt
+        handlers['SIGINT'] = engine.exit
+
+        # Modify all signal handlers
+        for signal in handlers.keys():
+            self.logger.debug('Modifying signal: %s' % signal)
+            oldSignalHandler = handlers[signal]
+            self.logger.debug('Old signal handler: %s' % oldSignalHandler)
+            signalHandler = DmRestWebServiceBase.SignalHandler(signal, oldSignalHandler)
+            self.logger.debug('Setting signal handler to: %s' % signalHandler.signalHandler)
+            handlers[signal] = signalHandler.signalHandler
+            pluginsSignalHandler.subscribe()
+
+    def initServerLog(self):
+        cherrypyLogLevel = self.configurationManager.getCherrypyLogLevel()
+        cherrypy.log.error_log.setLevel(cherrypyLogLevel)
+        cherrypy.log.error_file = self.configurationManager.getCherrypyLogFile()
+        cherrypy.log.error_log.propagate = False
+        cherrypy.log.access_log.setLevel(cherrypyLogLevel)
+        cherrypy.log.access_file = self.configurationManager.getCherrypyAccessFile()
+        cherrypy.log.access_log.propagate = False
+
+    def updateServerConfig(self):
+        serviceHost = self.configurationManager.getServiceHost()
+        servicePort = int(self.configurationManager.getServicePort())
+        nServerThreads = int(self.options.nServerThreads)
+        configDict = {
+            'server.socket_host' : serviceHost,
+            'server.socket_port' : servicePort,
+            'server.thread_pool' : nServerThreads,
+            'log.screen' : (self.options.daemonFlag != True),
+        }
+        cherrypy.config.update(configDict)
+
+    def readConfigFile(self, configFile):
+        configFile = self.options.configFile
+        if not configFile:
+            configFile = self.configurationManager.getConfigFile()
+        else:
+            self.configurationManager.setConfigFile(configFile)
+
+        if not os.path.exists(configFile):
+            raise ConfigurationError('Configuration file %s does not exist.' % configFile)
+        # Read file and set config options
+        self.configurationManager.setOptionsFromConfigFile(DmRestWebServiceBase.CONFIG_SECTION_NAME, DmRestWebServiceBase.CONFIG_OPTION_NAME_LIST, configFile)
+
+    def readCommandLineOptions(self):
+        # This method should be called after reading config file
+        # in case some options are overridden
+        if self.options.sslCaCertFile != None:
+            self.configurationManager.setSslCaCertFile(self.options.sslCaCertFile)
+        if self.options.sslCertFile != None:
+            self.configurationManager.setSslCertFile(self.options.sslCertFile)
+
+        if self.options.sslKeyFile != None:
+            self.configurationManager.setSslKeyFile(self.options.sslKeyFile)
+
+        if self.options.serviceHost != None:
+            self.configurationManager.setServiceHost(self.options.serviceHost)
+
+        if self.options.servicePort != None:
+            self.configurationManager.setServicePort(self.options.servicePort)
+
+    def prepareServer(self):
+        try:
+            optionParser = self.prepareOptions()
+            (self.options, self.args) = optionParser.parse_args()
+      
+            # Read config file and override with command line options
+            self.readConfigFile(self.options.configFile)
+            self.readCommandLineOptions()
+             
+            # Turn off console log for daemon mode.
+            self.logger = LoggingManager.getInstance().getLogger(self.__class__.__name__)
+            if self.options.daemonFlag:
+                LoggingManager.getInstance().setConsoleLogLevel('CRITICAL')
+
+            dispatch = self.routeMapper.setupRoutes()
+            self.logger.debug('Using route dispatch: %s' % dispatch)
+
+            config = {
+                '/' : {
+                    'request.dispatch' : dispatch,
+                },
+            }
+                   
+            # No root controller as we provided our own.
+            cherrypy.tree.mount(root=None, config=config)
+            self.initServerLog()
+            self.updateServerConfig()
+
+            self.logger.info('Using host %s' % self.configurationManager.getServiceHost())
+            self.logger.info('Using port %s' % self.configurationManager.getServicePort())
+            self.logger.debug('Using %s request handler threads' % self.options.nServerThreads)
+        except Exception, ex:
+            if self.logger is not None:
+                self.logger.exception(ex)
+            else:
+                import traceback
+                print '\n%s' % sys.exc_info()[1]
+                traceback.print_exc(file=sys.stderr)
+            sys.exit(dmStatus.DM_ERROR)
+
+    # Run server.
+    def __runServer(self):
+        self.logger.info('Starting service')
+        engine = cherrypy.engine
+
+        # Set up Deamonization
+        if self.options.daemonFlag:
+            plugins.Daemonizer(engine).subscribe()
+        self.logger.debug('Daemon mode: %s' % self.options.daemonFlag)
+
+        if self.options.pidFile != None:
+            plugins.PIDFile(engine, self.options.pidFile).subscribe()
+        self.logger.debug('Using PID file: %s' % self.options.pidFile)
+
+        sslCertFile = self.configurationManager.getSslCertFile()
+        sslKeyFile = self.configurationManager.getSslKeyFile()
+        sslCaCertFile = self.configurationManager.getSslCaCertFile()
+        if sslCertFile != None and sslKeyFile != None:
+            server.ssl_ca_certificate = None
+            if sslCaCertFile != None:
+                server.ssl_ca_certificate = self.options.sslCaCertFile
+                self.logger.info('Using SSL CA cert file: %s' % sslCaCertFile)
+            server.ssl_certificate = sslCertFile
+            self.logger.info('Using SSL cert file: %s' % sslCertFile)
+
+            server.ssl_private_key = sslKeyFile
+            self.logger.info('Using SSL key file: %s' % sslKeyFile)
+
+        server.ssl_module = 'builtin'
+        #server.ssl_module = 'pyopenssl'
+
+        # Increase timeout to prevent early SSL connection terminations
+        server.socket_timeout = DmRestWebServiceBase.DEFAULT_SERVER_SOCKET_TIMEOUT
+        
+        # Setup the signal handler to stop the application while running.
+        if hasattr(engine, 'signal_handler'):
+            self.logger.debug('Subscribing signal handler')
+            engine.signal_handler.subscribe()
+            self.modifySignalHandlers(engine)
+
+        # Turn off autoreloader.
+        self.logger.debug('Turning off autoreloader')
+        engine.autoreload.unsubscribe()
+
+        # Start the engine.
+        try:
+            self.logger.debug('Starting engine')
+            engine.start()
+
+            # Prepare dm services.
+            self.logger.debug('Starting modules')
+            self.initDmModules()
+            DmModuleManager.getInstance().startModules()
+
+        except Exception, ex:
+            self.logger.exception('Service exiting: %s' % ex)
+            DmModuleManager.getInstance().stopModules()
+            return dmStatus.DM_ERROR
+        self.logger.info('Service ready')
+        engine.block()
+        DmModuleManager.getInstance().stopModules()
+        self.logger.info('Service done')
+        return dmStatus.DM_OK
+
+    # Run server instance.
+    def run(self):
+        self.prepareServer()
+        sys.exit(self.__runServer())
+
+####################################################################
+# Testing
+
+if __name__ == '__main__':
+    pass