Changeset 7356 for TI12-security/trunk/NDGSecurity/python/ndg_security_test/ndg/security/test

Timestamp:

23/08/10 16:31:46 (11 years ago)

Author:

pjkersha

Message:

Incomplete - task 2: XACML-Security Integration

• implemented caching of authorisation decision statements in the PEP to cut down on calls to authorisation service.

Location:

TI12-security/trunk/NDGSecurity/python/ndg_security_test/ndg/security/test/unit/credentialwallet

Files:

• . (modified) (1 prop)
• README (deleted)
• credWallet.cfg (deleted)
• credWalletTest.cfg (deleted)
• test_credentialwallet.py (modified) (11 diffs)

TI12-security/trunk/NDGSecurity/python/ndg_security_test/ndg/security/test/unit/credentialwallet

@@ -1 +1 @@
 NDGCredentialWalletPickle.dat
 SAMLCredentialWalletPickle.dat
+*.dat

@@ -1 +1 @@
 NDGCredentialWalletPickle.dat
 SAMLCredentialWalletPickle.dat
+*.dat

TI12-security/trunk/NDGSecurity/python/ndg_security_test/ndg/security/test/unit/credentialwallet/test_credentialwallet.py

@@ -30 +30 @@
 from ndg.security.test.unit import BaseTestCase
 from ndg.security.common.utils.etree import prettyPrint
-from ndg.security.common.credentialwallet import SAMLCredentialWallet
-
-
-class SAMLCredentialWalletTestCase(BaseTestCase):
+from ndg.security.common.credentialwallet import (SAMLAttributeWallet,
+                                                  SAMLAuthzDecisionWallet)
+
+
+class CredentialWalletBaseTestCase(BaseTestCase):
     THIS_DIR = os.path.dirname(__file__)
     CONFIG_FILENAME = 'test_samlcredentialwallet.cfg'
     CONFIG_FILEPATH = os.path.join(THIS_DIR, CONFIG_FILENAME)
-    PICKLE_FILENAME = 'SAMLCredentialWalletPickle.dat'
-    PICKLE_FILEPATH = os.path.join(THIS_DIR, PICKLE_FILENAME)
+    
+    
+class SAMLAttributeWalletTestCase(CredentialWalletBaseTestCase):
+    PICKLE_FILENAME = 'SAMLAttributeWalletPickle.dat'
+    PICKLE_FILEPATH = os.path.join(CredentialWalletBaseTestCase.THIS_DIR, 
+                                   PICKLE_FILENAME)
     
     ASSERTION_STR = (
@@ -63 +68 @@
     
     def __init__(self, *arg, **kw):
-        super(SAMLCredentialWalletTestCase, self).__init__(*arg, **kw)
+        super(SAMLAttributeWalletTestCase, self).__init__(*arg, **kw)
         
     def setUp(self):
@@ -75 +80 @@
         timeExpires = timeNow + timedelta(seconds=validityDuration)
         assertionStr = Template(
-            SAMLCredentialWalletTestCase.ASSERTION_STR).substitute(
+            self.__class__.ASSERTION_STR).substitute(
                 dict(
                  issuerName=issuerName,
@@ -90 +95 @@
 
     def _addCredential(self):
-        wallet = SAMLCredentialWallet()   
-        wallet.addCredential(
-            self.assertion, 
-            attributeAuthorityURI=\
-                SAMLCredentialWalletTestCase.SITEA_ATTRIBUTEAUTHORITY_SAML_URI)
+        wallet = SAMLAttributeWallet()   
+        wallet.addCredential(self.assertion, issuerEndpoint=\
+        self.__class__.SITEA_ATTRIBUTEAUTHORITY_SAML_URI)
         return wallet
     
@@ -101 +104 @@
         
         self.assert_(len(wallet.credentials) == 1)
-        self.assert_(
-            SAMLCredentialWalletTestCase.SITEA_ATTRIBUTEAUTHORITY_SAML_URI in \
-            wallet.credentialsKeyedByURI)
-        self.assert_(SAMLCredentialWalletTestCase.SITEA_SAML_ISSUER_NAME in \
+        self.assert_(self.__class__.SITEA_ATTRIBUTEAUTHORITY_SAML_URI in \
+                     wallet.credentialsKeyedByURI)
+        self.assert_(self.__class__.SITEA_SAML_ISSUER_NAME in \
                      wallet.credentials)
         
         assertion = wallet.credentials[
-            SAMLCredentialWalletTestCase.SITEA_SAML_ISSUER_NAME
+            self.__class__.SITEA_SAML_ISSUER_NAME
         ].credential
         
@@ -115 +117 @@
     
     def test02VerifyCredential(self):
-        wallet = SAMLCredentialWallet()
+        wallet = SAMLAttributeWallet()
         self.assert_(wallet.isValidCredential(self.assertion))
         
@@ -132 +134 @@
         # is carried to prune expired credentials
         shortExpiryAssertion = self._createAssertion(validityDuration=1)
-        wallet = SAMLCredentialWallet()
+        wallet = SAMLAttributeWallet()
         wallet.addCredential(shortExpiryAssertion)
         
@@ -144 +146 @@
         # a clock skew of 
         shortExpiryAssertion = self._createAssertion(validityDuration=1)
-        wallet = SAMLCredentialWallet()
+        wallet = SAMLAttributeWallet()
         
         # Set a tolerance of five seconds
@@ -167 +169 @@
         self.assert_(newAssertion.conditions.notOnOrAfter == \
                      wallet.credentials[
-                        SAMLCredentialWalletTestCase.SITEA_SAML_ISSUER_NAME
+                        self.__class__.SITEA_SAML_ISSUER_NAME
                     ].credential.conditions.notOnOrAfter)
         
@@ -177 +179 @@
     def test07Pickle(self):
         wallet = self._addCredential()
-        outFile = open(SAMLCredentialWalletTestCase.PICKLE_FILEPATH, 'w')
+        outFile = open(self.__class__.PICKLE_FILEPATH, 'w')
         pickle.dump(wallet, outFile)
         outFile.close()
         
-        inFile = open(SAMLCredentialWalletTestCase.PICKLE_FILEPATH)
+        inFile = open(self.__class__.PICKLE_FILEPATH)
         unpickledWallet = pickle.load(inFile)
         self.assert_(unpickledWallet.credentialsKeyedByURI.get(
-            SAMLCredentialWalletTestCase.SITEA_ATTRIBUTEAUTHORITY_SAML_URI))
+            self.__class__.SITEA_ATTRIBUTEAUTHORITY_SAML_URI))
         
         self.assert_(unpickledWallet.credentials.items()[0][1].issuerName == \
@@ -190 +192 @@
 
     def test08CreateFromConfig(self):
-        wallet = SAMLCredentialWallet.fromConfig(
-                                SAMLCredentialWalletTestCase.CONFIG_FILEPATH)
+        wallet = SAMLAttributeWallet.fromConfig(
+                                self.__class__.CONFIG_FILEPATH)
         self.assert_(wallet.clockSkewTolerance == timedelta(seconds=0.01))
         self.assert_(wallet.userId == 'https://openid.localhost/philip.kershaw')
         
+
+class SAMLAuthzDecisionWalletTestCase(CredentialWalletBaseTestCase):
+    """Test wallet for caching Authorisation Decision statements"""
+    PICKLE_FILENAME = 'SAMLAuthzDecisionWalletPickle.dat'
+    PICKLE_FILEPATH = os.path.join(CredentialWalletBaseTestCase.THIS_DIR, 
+                                   PICKLE_FILENAME)
+    
+    RESOURCE_ID = 'http://localhost/My%20Secured%20URI'
+    ASSERTION_STR = """
+    <saml:Assertion xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion" Version="2.0" IssueInstant="$timeNow" ID="c32235a9-85df-4325-99a2-bad73668c01d">
+        <saml:Issuer Format="urn:oasis:names:tc:SAML:1.1:nameid-format:x509SubjectName">/O=NDG/OU=BADC/CN=attributeauthority.badc.rl.ac.uk</saml:Issuer>
+        <saml:Subject>
+            <saml:NameID Format="urn:esg:openid">https://openid.localhost/philip.kershaw</saml:NameID>
+        </saml:Subject>
+        <saml:Conditions NotOnOrAfter="$timeExpires" NotBefore="$timeNow"></saml:Conditions>
+        <saml:AuthzDecisionStatement Decision="Permit" Resource="$resourceId">
+            <saml:Action Namespace="urn:oasis:names:tc:SAML:1.0:action:ghpp">GET</saml:Action>
+        </saml:AuthzDecisionStatement>
+    </saml:Assertion>
+    """
+    
+    def setUp(self):
+        self.assertion = self._createAssertion()
+        
+    def _createAssertion(self, timeNow=None, validityDuration=60*60*8,
+                         issuerName=BaseTestCase.SITEA_SAML_ISSUER_NAME):
+        if timeNow is None:
+            timeNow = datetime.utcnow()
+            
+        timeExpires = timeNow + timedelta(seconds=validityDuration)
+        assertionStr = Template(
+            self.__class__.ASSERTION_STR).substitute(
+                dict(
+                 issuerName=issuerName,
+                 timeNow=SAMLDateTime.toString(timeNow), 
+                 timeExpires=SAMLDateTime.toString(timeExpires),
+                 resourceId=self.__class__.RESOURCE_ID,
+                )
+            )
+
+        assertionStream = StringIO()
+        assertionStream.write(assertionStr)
+        assertionStream.seek(0)
+        assertionElem = ElementTree.parse(assertionStream).getroot()
+        return AssertionElementTree.fromXML(assertionElem)
+                    
+    def _addCredential(self):
+        wallet = SAMLAuthzDecisionWallet()   
+        wallet.addCredential(self.assertion)
+        return wallet
+    
+    def test01AddCredential(self):
+        wallet = self._addCredential()
+        
+        self.assert_(len(wallet.credentials) == 1)
+        self.assert_(self.__class__.RESOURCE_ID in wallet.credentials)
+
+        assertion = wallet.credentials[self.__class__.RESOURCE_ID].credential
+        
+        print("SAML Assertion:\n%s" % 
+              prettyPrint(AssertionElementTree.toXML(assertion)))
+    
+    def test02VerifyCredential(self):
+        wallet = SAMLAttributeWallet()
+        self.assert_(wallet.isValidCredential(self.assertion))
+        
+        expiredAssertion = self._createAssertion(
+                                timeNow=datetime.utcnow() - timedelta(hours=24))
+                                
+        self.assert_(not wallet.isValidCredential(expiredAssertion))
+        
+        futureAssertion = self._createAssertion(
+                                timeNow=datetime.utcnow() + timedelta(hours=24))
+
+        self.assert_(not wallet.isValidCredential(futureAssertion))
+        
+    def test03AuditCredential(self):
+        # Add a short lived credential and ensure it's removed when an audit
+        # is carried to prune expired credentials
+        shortExpiryAssertion = self._createAssertion(validityDuration=1)
+        wallet = SAMLAttributeWallet()
+        wallet.addCredential(shortExpiryAssertion)
+        
+        self.assert_(len(wallet.credentials) == 1)
+        sleep(2)
+        wallet.audit()
+        self.assert_(len(wallet.credentials) == 0)
+
+    def test04ClockSkewTolerance(self):
+        # Add a short lived credential but with the wallet set to allow for
+        # a clock skew of 
+        shortExpiryAssertion = self._createAssertion(validityDuration=1)
+        wallet = SAMLAuthzDecisionWallet()
+        
+        # Set a tolerance of five seconds
+        wallet.clockSkewTolerance = 5.*60*60
+        wallet.addCredential(shortExpiryAssertion)
+        
+        self.assert_(len(wallet.credentials) == 1)
+        sleep(2)
+        wallet.audit()
+        self.assert_(len(wallet.credentials) == 1)
+        
+    def test05ReplaceCredential(self):
+        # Replace an existing credential from a given institution with a more
+        # up to date one
+        wallet = self._addCredential()
+        self.assert_(len(wallet.credentials) == 1)
+        
+        newAssertion = self._createAssertion()  
+
+        wallet.addCredential(newAssertion)
+        self.assert_(len(wallet.credentials) == 1)
+        self.assert_(newAssertion.conditions.notOnOrAfter == \
+                     wallet.credentials[
+                        self.__class__.RESOURCE_ID
+                    ].credential.conditions.notOnOrAfter)
+
+    def test06Pickle(self):
+        wallet = self._addCredential()
+        outFile = open(self.__class__.PICKLE_FILEPATH, 'w')
+        pickle.dump(wallet, outFile)
+        outFile.close()
+        
+        inFile = open(self.__class__.PICKLE_FILEPATH)
+        unpickledWallet = pickle.load(inFile)
+        self.assert_(unpickledWallet.credentials.get(
+                                                    self.__class__.RESOURCE_ID))
+        
+    def test07CreateFromConfig(self):
+        wallet = SAMLAttributeWallet.fromConfig(
+                                self.__class__.CONFIG_FILEPATH)
+        self.assert_(wallet.clockSkewTolerance == timedelta(seconds=0.01))
+        self.assert_(wallet.userId == 'https://openid.localhost/philip.kershaw')
+
+
 if __name__ == "__main__":
     unittest.main()        

@@ -1 +1 @@
 NDGCredentialWalletPickle.dat
 SAMLCredentialWalletPickle.dat
+*.dat