source: TI05-delivery/ows_framework/trunk/ows_server/ows_server/lib/security_util.py @ 2949

# Copyright (C) 2007 STFC & NERC (Science and Technology Facilities Council).
# This software may be distributed under the terms of the
# Q Public License, version 1.0 or later.
# http://ndg.nerc.ac.uk/public_docs/QPublic_license.txt
"""
Utilities for transfor of credentials over login service

@author: Philip Kershaw
"""
__revision__ = '$Id:$'

import logging
log = logging.getLogger(__name__)

from pylons import session, request


class SecuritySession(object):
    """Utility for Pylons security session keys enables correct
    keys to be set
    
    @type key: string
    @cvar key: name of security key in session object
    
    @type subKeys: tuple
    @cvar subKeys: list of valid security keys to set"""
    
    key = 'ndgSec'
    subKeys = ('h', 'sid', 'u', 'roles')

    def __init__(self, **subKeys):
        """Update the security key of session object with the
        input sub keys
        
        type **subKeys: dict
        param **subKeys: set any of the security keywords as contained in
        SecuritySession.subKeys"""
        
        # Set the security session keys from request.params if no keywords 
        # were input
        if subKeys == {}:
            subKeys = LoginServiceQuery.decodeRequestParams()
            
        # Ensure security key is present
        if SecuritySession.key not in session:
            session[SecuritySession.key] = {}
         
        # Ensure valid keys are being set   
        for k in subKeys:
            if k not in SecuritySession.subKeys:
                raise KeyError, 'Invalid key Security session dict: "%s"' % k
        
        # Update security values
        session[SecuritySession.key].update(subKeys)            
        session.save()
        log.debug("Set security session: %s" % session[SecuritySession.key])

    @classmethod
    def delete(self):
        """Delete security key from session object"""
        if SecuritySession.key in session:
            del session[SecuritySession.key]
            session.save()
        log.debug("Deleted security key to session object: %s" % session)
            
setSecuritySession = SecuritySession
           
 
class LoginServiceQuery(object):
    """Create query string containing security credentials.  This is used by
    the Identity Provider pass the credentials over a HTTP GET back to the 
    Service Provider"""
    
    keys = SecuritySession.subKeys
    rolesSep = ","
    argSep = "&"
        
    def __str__(self):
        """Provide convenient short-cut for call to make query string

        @rtype: string
        @return: URL query string with security args"""
        return self.makeQueryStr()
   
    @classmethod
    def makeQueryStr(cls):
        """Create the query string containing the required security 
        credentials to return to the service provider
        
        @rtype: string
        @return: URL query string with security args"""
        
        # Make a copy of the security session dict reseting the
        # roles to a single string ready for passing over URL
        secDict = session[SecuritySession.key].copy()
        secDict['roles'] = cls.rolesSep.join(secDict['roles'])
        
        # Return the full query as a string
        return cls.argSep.join(["%s=%s" % (k, secDict[k]) for k in cls.keys])

    @classmethod
    def stripFromURI(cls):
        """Make a new query string using Pylons request.params but stripping
        args relating to security
        
        @rtype: string
        @return: URL query string with security args removed"""
        return cls.argSep.join(['%s=%s' % (i, request.params[i]) \
                                for i in request.params if i not in cls.keys])

    @classmethod
    def decodeRequestParams(cls):
        """Get security parameters from request.params received from Login 
        Service (IdP).  Decode parameters where necessary: roles are sent as a
        comma delimited list - convert into a list type
        
        @rtype: dict
        @return: dictionary of security parameters 
        """
        
        try:
            # request.params is actually a MultiDict type but for the purposes
            # of this code it can be treated as a regular dict type
            keys = dict([(k, request.params[k]) for k in cls.keys])
        except KeyError, e:
            OwsError, \
                '%s argument is missing from URL returned by Login Service' %\
                str(e)
                
        # Modify roles from a comma delimited string into a list
        if 'roles' in keys:
            keys['roles'] = keys['roles'].split(cls.rolesSep)

        return keys
    
import sys

class SecurityConfigError(Exception):
    """Handle errors from parsing security config items"""
       
class SecurityConfig(object):
    """Get Security related parameters from the Pylons NDG config file"""

    def __init__(self, cfg=None):
        '''Get PKI settings for Attribute Authority and Session Manager from
        the configuration file
        
        @type param: pylons config file object
        @param cfg: reference to NDG configuration file.  If omitted defaults
        to request.environ['ndgConfig']'''
        
        if cfg is None:
            cfg = request.environ['ndgConfig']

        tracefileExpr = cfg.get('NDG_SECURITY', 'tracefile')
        if tracefileExpr:
            self.tracefile = eval(tracefileExpr)

        self.smURI = cfg.get('NDG_SECURITY', 'sessionMgrURI')        
        self.aaURI = cfg.get('NDG_SECURITY', 'attAuthorityURI')
        
        # ... for SSL connections to security web services
        try:
            self.sslCACertFilePathList = \
            cfg.get('NDG_SECURITY', 'sslCACertFilePathList').split()
                
        except AttributeError:
            raise SecurityConfigError, \
                        'No "sslCACertFilePathList" security setting'

        self.sslPeerCertCN = cfg.get('NDG_SECURITY', 'sslPeerCertCN')

        # ...and for WS-Security digital signature
        self.wssCertFilePath = cfg.get('NDG_SECURITY', 'wssCertFilePath')
        self.wssPriKeyFilePath = cfg.get('NDG_SECURITY', 'wssKeyFilePath')
        self.wssPriKeyPwd = cfg.get('NDG_SECURITY', 'wssKeyPwd')

        try:
            self.wssCACertFilePathList = \
                cfg.get('NDG_SECURITY', 'wssCACertFilePathList').split()
                
        except AttributeError:
            raise SecurityConfigError, \
                                'No "wssCACertFilePathList" security setting'

        # Gatekeeper params
        
        # Attribute Certificate Issuer
        self.acIssuer = cfg.get('NDG_SECURITY', 'acIssuer')
        
        # verification of X.509 cert back to CA
        self.acCACertFilePathList = cfg.get('NDG_SECURITY', 
                                            'acCACertFilePathList')

             
    def __repr__(self):
        return '\n'.join(["%s=%s" % (k,v) for k,v in self.__dict__.items() \
                if k[:2] != "__"])