source: TI12-security/trunk/python/ndg.security.test/ndg/security/test/sessionMgrClient/SessionMgrClientTest.py @ 3044

#!/usr/bin/env python
"""Test harness for NDG Session Manager client - makes requests for 
authentication and authorisation.  An Attribute Authority and Simple CA
services must be running for the reqAuthorisation and addUser tests

NERC Data Grid Project
"""
__author__ = "P J Kershaw"
__date__ = "23/02/06"
__copyright__ = "(C) 2007 STFC & NERC"
__license__ = \
"""This software may be distributed under the terms of the Q Public 
License, version 1.0 or later."""
__contact__ = "P.J.Kershaw@rl.ac.uk"
__revision__ = '$Id$'

import unittest
import os, sys, getpass, re
from ConfigParser import SafeConfigParser

from ndg.security.common.SessionMgr import SessionMgrClient, \
    AttributeRequestDenied
    
from ndg.security.common.SessionCookie import SessionCookie
from ndg.security.common.X509 import X509CertParse, X509CertRead


class SessionMgrClientTestCase(unittest.TestCase):
    pemPat = "-----BEGIN CERTIFICATE-----[^\-]*-----END CERTIFICATE-----"
        
    test2Passphrase = None
    test3Passphrase = None

    def _getCertChainFromProxyCertFile(self, proxyCertFilePath):
        '''Read proxy cert and user cert from a single PEM file and put in
        a list ready for input into SignatureHandler'''               
        proxyCertFileTxt = open(proxyCertFilePath).read()
        
        pemPatRE = re.compile(self.__class__.pemPat, re.S)
        x509CertList = pemPatRE.findall(proxyCertFileTxt)
        
        signingCertChain = [X509CertParse(x509Cert) for x509Cert in \
                            x509CertList]
    
        # Expecting proxy cert first - move this to the end.  This will
        # be the cert used to verify the message signature
        signingCertChain.reverse()
        
        return signingCertChain


    def setUp(self):
        
        configParser = SafeConfigParser()
        configParser.read("./sessionMgrClientTest.cfg")
        
        self.cfg = {}
        for section in configParser.sections():
            self.cfg[section] = dict(configParser.items(section))

        tracefile = sys.stderr

        try:
            if self.cfg['setUp'].get('clntprikeypwd') is None:
                clntPriKeyPwd = getpass.getpass(\
                            prompt="\nsetUp - client private key password: ")
            else:
                clntPriKeyPwd = self.cfg['setUp'].get('clntprikeypwd')
        except KeyboardInterrupt:
            sys.exit(0)

        # List of CA certificates for use in validation of certs used in
        # signature for server reponse
        try:
            caCertFilePathList=self.cfg['setUp']['cacertfilepathlist'].split()
        except:
            caCertFilePathList = []
          
        try:
            sslCACertList = [X509CertRead(file) for file in \
                         self.cfg['setUp']['sslcacertfilepathlist'].split()]
        except KeyError:
            sslCACertList = []
          
          
        reqBinSecTokValType = self.cfg['setUp'].get('reqbinsectokvaltype')

        # Check certificate types proxy or standard
        proxyCertFilePath = self.cfg['setUp'].get('proxycertfilepath')
        if proxyCertFilePath:
            signingCertChain = \
                        self._getCertChainFromProxyCertFile(proxyCertFilePath)
        else:
            signingCertChain = None
                
        setSignatureHandler = eval(self.cfg['setUp']['setsignaturehandler'])
            
        # Initialise the Session Manager client connection
        # Omit traceFile keyword to leave out SOAP debug info
        self.clnt = SessionMgrClient(uri=self.cfg['setUp']['smuri'],
                sslCACertList=sslCACertList,
                sslPeerCertCN=self.cfg['setUp'].get('sslpeercertcn'),
                setSignatureHandler=setSignatureHandler,
                reqBinSecTokValType=reqBinSecTokValType,
                signingCertFilePath=self.cfg['setUp'].get('clntcertfilepath'),
                signingCertChain=signingCertChain,
                signingPriKeyFilePath=self.cfg['setUp']['clntprikeyfilepath'],
                signingPriKeyPwd=clntPriKeyPwd,
                caCertFilePathList=caCertFilePathList,
                tracefile=tracefile) 
        
        self.sessID = None
        self.proxyCert = None
        self.proxyPriKey = None
        self.userCert = None

# TODO: is addUser part of session manager?
#    def test1AddUser(self):
#        """Add a new user ID to the MyProxy repository"""
#        
#        passphrase = self.cfg['test1AddUser'].get('passphrase') or \
#            getpass.getpass(prompt="\ntest1AddUser pass-phrase for new user: ")
#            
#        # Note the pass-phrase is read from the file tmp.  To pass
#        # explicitly as a string use the 'passphrase' keyword instead
#        self.clnt.addUser(self.cfg['test1AddUser']['username'], 
#                          passphrase=passphrase)
#        print "Added user '%s'" % self.cfg['test1AddUser']['username']
        

    def test1Connect(self):
        """test1Connect: Connect as if acting as a browser client - 
        a session ID is returned"""
        
        if self.__class__.test2Passphrase is None:
            self.__class__.test2Passphrase = \
                                    self.cfg['test1Connect'].get('passphrase')
        
        if not self.__class__.test2Passphrase:
            self.__class__.test2Passphrase = getpass.getpass(\
                               prompt="\ntest2Connect pass-phrase for user: ")

        self.proxyCert, self.proxyPriKey, self.userCert, self.sessID = \
            self.clnt.connect(self.cfg['test1Connect']['username'], 
                              passphrase=self.__class__.test2Passphrase)

        print "User '%s' connected to Session Manager:\n%s" % \
            (self.cfg['test1Connect']['username'], self.sessID)
            
            
    def test2GetSessionStatus(self):
        """test2GetSessionStatus: check a session is alive"""
        print "\n\t" + self.test2GetSessionStatus.__doc__
        
        self.test1Connect()
        assert self.clnt.getSessionStatus(sessID=self.sessID), \
                "Session is dead"
                
        print "User connected to Session Manager with sessID=%s" % self.sessID

        assert not self.clnt.getSessionStatus(sessID='abc'), \
            "sessID=abc shouldn't exist!"
            
        print "CORRECT: sessID=abc doesn't exist"


    def test3ConnectNoCreateServerSess(self):
        """test3ConnectNoCreateServerSess: Connect as a non browser client - 
        sessID should be None"""

        if self.__class__.test3Passphrase is None:
            self.__class__.test3Passphrase = \
                self.cfg['test3ConnectNoCreateServerSess'].get('passphrase')
                
        if not self.__class__.test3Passphrase:
            self.__class__.test3Passphrase = getpass.getpass(\
            prompt="\ntest3ConnectNoCreateServerSess pass-phrase for user: ")

        self.proxyCert, self.proxyPriKey, self.userCert, sessID = \
            self.clnt.connect(\
                      self.cfg['test3ConnectNoCreateServerSess']['username'], 
                      passphrase=self.__class__.test3Passphrase,
                      createServerSess=False)
        
        # Expect null session ID
        assert(not sessID)
          
        print "User '%s' connected to Session Manager:\n%s" % \
                    (self.cfg['test3ConnectNoCreateServerSess']['username'], 
                     self.proxyCert)
            

    def test4DisconnectUsingSessID(self):
        """test4DisconnectUsingSessID: disconnect as if acting as a browser client 
        """
        
        print "\n\t" + self.test4DisconnectUsingSessID.__doc__
        self.test1Connect()
        
        self.clnt.disconnect(sessID=self.sessID)
        
        print "User disconnected from Session Manager:\n%s" % self.sessID
            

    def test5DisconnectUsingProxyCert(self):
        """test5DisconnectUsingProxyCert: Disconnect as a command line client 
        """
        
        print "\n\t" + self.test5DisconnectUsingProxyCert.__doc__
        self.test1Connect()
        
        # Use proxy cert / private key just obtained from connect call for
        # signature generation         
        self.clnt.signatureHandler.reqBinSecTokValType = 'X509PKIPathv1'
        self.clnt.signatureHandler.signingPriKey = self.proxyPriKey        
        self.clnt.signatureHandler.signingCertChain = (self.userCert,
                                                       self.proxyCert)
        
        # Proxy cert in signature determines ID of session to
        # delete
        self.clnt.disconnect()
        print "User disconnected from Session Manager:\n%s" % self.proxyCert


    def test6GetAttCertUsingSessID(self):
        """test6GetAttCertUsingSessID: make an attribute request using
        a session ID as authentication credential"""

        print "\n\t" + self.test6GetAttCertUsingSessID.__doc__        
        self.test1Connect()
        
        attCert = self.clnt.getAttCert(\
            sessID=self.sessID, 
            attAuthorityURI=self.cfg['test6GetAttCertUsingSessID']['aauri'])
        
        print "Attribute Certificate:\n%s" % attCert 
        attCert.filePath = \
            self.cfg['test6GetAttCertUsingSessID']['acoutfilepath'] 
        attCert.write()


    def test6aGetAttCertRefusedUsingSessID(self):
        """test6aGetAttCertRefusedUsingSessID: make an attribute request using
        a sessID as authentication credential requesting an AC from an
        Attribute Authority where the user is NOT registered"""

        print "\n\t" + self.test6aGetAttCertRefusedUsingSessID.__doc__        
        self.test1Connect()
        
        aaURI = self.cfg['test6aGetAttCertRefusedUsingSessID']['aauri']
        
        try:
            attCert = self.clnt.getAttCert(sessID=self.sessID, 
                                           attAuthorityURI=aaURI,
                                           mapFromTrustedHosts=False)
        except AttributeRequestDenied, e:
            print "SUCCESS - obtained expected result: %s" % e
            return
        
        self.fail("Request allowed from AA where user is NOT registered!")


    def test6bGetMappedAttCertUsingSessID(self):
        """test6bGetMappedAttCertUsingSessID: make an attribute request using
        a session ID as authentication credential"""

        print "\n\t" + self.test6bGetMappedAttCertUsingSessID.__doc__        
        self.test1Connect()
        
        aaURI = self.cfg['test6bGetMappedAttCertUsingSessID']['aauri']
        
        attCert=self.clnt.getAttCert(sessID=self.sessID,attAuthorityURI=aaURI)
        
        print "Attribute Certificate:\n%s" % attCert  


    def test6cGetAttCertWithExtAttCertListUsingSessID(self):
        """test6cGetAttCertUsingSessID: make an attribute request using
        a session ID as authentication credential"""
        
        print "\n\t" + \
            self.test6cGetAttCertWithExtAttCertListUsingSessID.__doc__        
        self.test1Connect()
        
        aaURI = \
            self.cfg['test6cGetAttCertWithExtAttCertListUsingSessID']['aauri']
        
        # Use output from test6GetAttCertUsingSessID!
        extACFilePath = \
    self.cfg['test6cGetAttCertWithExtAttCertListUsingSessID']['extacfilepath']   
        extAttCert = open(extACFilePath).read()
        
        attCert = self.clnt.getAttCert(sessID=self.sessID, 
                                       attAuthorityURI=aaURI,
                                       extAttCertList=[extAttCert])
          
        print "Attribute Certificate:\n%s" % attCert  


    def test7GetAttCertUsingProxyCert(self):
        """test7GetAttCertUsingProxyCert: make an attribute request using
        a proxy cert as authentication credential"""
        print "\n\t" + self.test7GetAttCertUsingProxyCert.__doc__
        self.test1Connect()

        self.clnt.signatureHandler.reqBinSecTokValType = 'X509PKIPathv1'
        self.clnt.signatureHandler.signingPriKey = self.proxyPriKey        
        self.clnt.signatureHandler.signingCertChain = (self.userCert,
                                                       self.proxyCert)
        
        # Request an attribute certificate from an Attribute Authority 
        # using the proxyCert returned from connect()
        
        aaURI = self.cfg['test7GetAttCertUsingProxyCert']['aauri']
        attCert = self.clnt.getAttCert(attAuthorityURI=aaURI)
          
        print "Attribute Certificate:\n%s" % attCert  


    def test8GetX509Cert(self):
        "test8GetX509Cert: return the Session Manager's X.509 Cert."
        cert = self.clnt.getX509Cert()
                                             
        print "Session Manager X.509 Certificate:\n" + cert
            
            
#_____________________________________________________________________________       
class SessionMgrClientTestSuite(unittest.TestSuite):
    
    def __init__(self):
        map = map(SessionMgrClientTestCase,
                  (
                    "test1Connect",
                    "test2GetSessionStatus",
                    "test3ConnectNoCreateServerSess",
                    "test4DisconnectUsingSessID",
                    "test5DisconnectUsingProxyCert",
                    "test6GetAttCertUsingSessID",
                    "test6bGetMappedAttCertUsingSessID",
                    "test6cGetAttCertWithExtAttCertListUsingSessID",
                    "test7GetAttCertUsingProxyCert",
                    "test8GetX509Cert",
                  ))
        unittest.TestSuite.__init__(self, map)
            
                                                    
if __name__ == "__main__":
    unittest.main()