Changeset 6553

Timestamp:

10/02/10 17:04:46 (11 years ago)

Author:

pjkersha

Message:

Unit testing AuthzDecisionStatement? added to a Response.

Location:

TI12-security/trunk/ndg_security_saml/saml

Files:

• saml2/core.py (modified) (11 diffs)
• test/test_saml.py (modified) (10 diffs)
• xml/etree.py (modified) (6 diffs)

TI12-security/trunk/ndg_security_saml/saml/saml2/core.py

@@ -322 +322 @@
         '''
         raise NotImplementedError()
-            
+
+
+class DecisionType(object):
+    """Define decision types for the authorisation decisions"""
+    
+    # "Permit" decision type
+    PERMIT = "Permit"
+    
+    # "Deny" decision type
+    DENY = "Deny"
+    
+    # "Indeterminate" decision type
+    INDETERMINATE = "Indeterminate"
+        
+    TYPES = (PERMIT, DENY, INDETERMINATE)
+    
+    __slots__ = ('__value',)
+    
+    def __init__(self, decisionType):
+        self.__value = None
+        self.value = decisionType
+
+    def _setValue(self, value):
+        if not isinstance(value, basestring):
+            raise TypeError('Expecting string type for "value" attribute; got '
+                            'instead' % type(value))
+            
+        if value not in DecisionType.TYPES:
+            raise AttributeError('Permissable decision types are %r; got %r '
+                                 'instead' % (DecisionType.TYPES,
+                                              value))
+        self.__value = value
+        
+    def _getValue(self):
+        return self.__value
+    
+    value = property(fget=_getValue, fset=_setValue, doc="Decision value")
+    
+    def __str__(self):
+        return self.__value
+
 
 class AuthzDecisionStatement(Statement):
@@ -359 +399 @@
 
         # Resource attribute value. 
-        self.__resource = None
+        self.__resource = None  
+        
+        self.__decision = DecisionType(DecisionType.INDETERMINATE)    
+        self.__actions = TypedList(Action)
+        self.__evidence = None
         
         # Tuning for normalization of resource URIs in property set method
@@ -454 +498 @@
         @return: the decision of the authorization request
         '''
-        raise NotImplementedError()
+        return self.__decision
 
     def _setDecision(self, value):
@@ -462 +506 @@
         @param value: the decision of the authorization request
         '''
-        raise NotImplementedError()
+        if not isinstance(value, basestring):
+            raise TypeError('Expecting %r type for "decision" attribute; '
+                            'got instead' % (DecisionType, type(value)))
+        self.__decision = value
     
     @property
@@ -1043 +1090 @@
         # TODO: Implement AuthnStatement and AuthzDecisionStatement classes
         self.__authnStatements = []
-        self.__authzDecisionStatements = []
+        self.__authzDecisionStatements = TypedList(AuthzDecisionStatement)
         self.__attributeStatements = TypedList(AttributeStatement)
         
@@ -1166 +1213 @@
                       doc="Attribute Assertion advice")
     
-    def _get_statements(self):
-        """Get statements string."""
+    @property
+    def statements(self):
+        """Attribute Assertion statements"""
         return self.__statements
-
-    statements = property(fget=_get_statements,
-                          doc="Attribute Assertion statements")
-    
-    def _get_authnStatements(self):
-        """Get authnStatements string."""
+    
+    @property
+    def authnStatements(self):
+        """Attribute Assertion authentication"""
         return self.__authnStatements
-
-    authnStatements = property(fget=_get_authnStatements,
-                               doc="Attribute Assertion authentication "
-                                   "statements")
-    
-    def _get_authzDecisionStatements(self):
-        """Get authorisation decision statements."""
+    
+    @property
+    def authzDecisionStatements(self):
+        """Attribute Assertion authorisation decision statements"""
         return self.__authzDecisionStatements
-
-    authzDecisionStatements = property(fget=_get_authzDecisionStatements,
-                                       doc="Attribute Assertion authorisation "
-                                           "decision statements")
-    
-    def _get_attributeStatements(self):
-        """Get attributeStatements string."""
+    
+    @property
+    def attributeStatements(self):
+        """Attribute Assertion attribute statements"""
         return self.__attributeStatements
-
-    attributeStatements = property(fget=_get_attributeStatements,
-                                   doc="Attribute Assertion attribute "
-                                       "statements")
     
 
@@ -1652 +1688 @@
     UNIX_NS_URI = "urn:oasis:names:tc:SAML:1.0:action:unix"
 
+    ACTION_NS_IDENTIFIERS = (
+        RWEDC_NS_URI,
+        RWEDC_NEGATION_NS_URI,    
+        GHPP_NS_URI,
+        UNIX_NS_URI       
+    )
+    
     # Read action. 
     READ_ACTION = "Read"
@@ -1694 +1737 @@
     HTTP_POST_ACTION = "POST"
     
+    ACTION_TYPES = {
+        RWEDC_NS_URI: (READ_ACTION, WRITE_ACTION, EXECUTE_ACTION, DELETE_ACTION,
+                       CONTROL_ACTION),
+        RWEDC_NEGATION_NS_URI: (READ_ACTION, WRITE_ACTION, EXECUTE_ACTION, 
+                                DELETE_ACTION, CONTROL_ACTION, NEG_READ_ACTION, 
+                                NEG_WRITE_ACTION, NEG_EXECUTE_ACTION, 
+                                NEG_CONTROL_ACTION),    
+        GHPP_NS_URI: (HTTP_GET_ACTION, HTTP_HEAD_ACTION, HTTP_PUT_ACTION,
+                      HTTP_POST_ACTION),
+                      
+        # This namespace uses octal bitmask for file permissions
+        UNIX_NS_URI: ()   
+    }
+    
     def __init__(self, **kw):
         '''Create an authorization action type
@@ -1699 +1756 @@
         super(Action, self).__init__(**kw)
 
-        # URI of the Namespace of this action
-        self.__namespace = None
+        # URI of the Namespace of this action.  Default to read/write/negation 
+        # type - 2.7.4.2 SAML 2 Core Spec. 15 March 2005
+        self.__namespace = Action.RWEDC_NEGATION_NS_URI
 
         #Value value
         self.__action = None       
     
+        self.__actionTypes = Action.ACTION_TYPES
+
+    def _getActionTypes(self):
+        return self.__actionTypes
+
+    def _setActionTypes(self, value):
+        if not isinstance(value, dict):
+            raise TypeError('Expecting list or tuple type for "actionTypes" '
+                            'attribute; got %r' % type(value))
+            
+        for k, v in value.items():
+            if not isinstance(v, (tuple, type(None))):
+                raise TypeError('Expecting None or tuple type for '
+                                '"actionTypes" dictionary values; got %r for '
+                                '%r key' % (type(value), k))
+        self.__actionTypes = value
+
+    actionTypes = property(_getActionTypes, 
+                           _setActionTypes, 
+                           doc="Restrict vocabulary of action types")
+        
     def _getNamespace(self):
         '''
-        gets the namespace scope of the specifiedvalue.
-        
-        @return: the namespace scope of the specifiedvalue
+        gets the namespace scope of the specified value.
+        
+        @return: the namespace scope of the specified value
         '''
         return self.__namespace
@@ -1715 +1794 @@
     def _setNamespace(self, value):
         '''
-        Sets the namespace scope of the specifiedvalue.
-        
-        @param value: the namespace scope of the specifiedvalue
+        Sets the namespace scope of the specified value.
+        
+        @param value: the namespace scope of the specified value
         '''
         if not isinstance(value, basestring):
             raise TypeError('Expecting string type for "namespace" '
                             'attribute; got %r' % type(value))
+            
+        if value not in self.__actionTypes.keys():
+            raise AttributeError('"namespace" action type %r not recognised. '
+                                 'It must be one of these action types: %r' % 
+                                 self.__actionNsIdentifiers)
+            
         self.__namespace = value
 
@@ -1738 +1823 @@
         Sets the URI of the action to be performed.
         
-        @param value: the URI of thevalue to be performed
-        '''
-        if not isinstance(value, basestring):
-            raise TypeError('Expecting string type for "action" '
+        @param value: the URI of the value to be performed
+        '''
+        # int and oct allow for UNIX file permissions action type
+        if not isinstance(value, (basestring, int)):
+            raise TypeError('Expecting string or int type for "action" '
                             'attribute; got %r' % type(value))
+            
+        # Default to read/write/negation type - 2.7.4.2 SAML 2 Core Spec.
+        # 15 March 2005
+        allowedActions = self.__actionTypes.get(self.__namespace,
+                                                Action.RWEDC_NEGATION_NS_URI)
+        
+        # Only apply restriction for action type that has a restricted 
+        # vocabulary - UNIX type is missed out of this because its an octal
+        # mask
+        if len(allowedActions) > 0 and value not in allowedActions:
+            raise AttributeError('%r action not recognised; known actions for '
+                                 'the %r namespace identifier are: %r.  ' 
+                                 'If this is not as expected make sure to set '
+                                 'the "namespace" attribute to an alternative '
+                                 'value first or override completely by '
+                                 'explicitly setting the "allowTypes" '
+                                 'attribute' % 
+                                 (value, self.__namespace, allowedActions))
         self.__value = value
 

TI12-security/trunk/ndg_security_saml/saml/test/test_saml.py

@@ -22 +22 @@
 
 from saml.saml2.core import (SAMLVersion, Attribute, AttributeStatement, 
-                             Assertion, AttributeQuery, Response, Issuer, 
-                             Subject, NameID, StatusCode, 
+                             AuthzDecisionStatement, Assertion, AttributeQuery, 
+                             Response, Issuer, Subject, NameID, StatusCode, 
                              StatusMessage, Status, Conditions, 
                              XSStringAttributeValue, Action,
@@ -196 +196 @@
         for action, namespace in actions:
             authzDecisionQuery.actions.append(Action())
+            authzDecisionQuery.actions[-1].namespace = namespace
             authzDecisionQuery.actions[-1].value = action
-            authzDecisionQuery.actions[-1].namespace = namespace
             
         return authzDecisionQuery
@@ -210 +210 @@
     RESOURCE_URI = SAMLUtil.RESOURCE_URI
     
-    def _createAssertionHelper(self):
+    def _createAttributeAssertionHelper(self):
         samlUtil = SAMLUtil()
         
@@ -237 +237 @@
     def test01CreateAssertion(self):
          
-        assertion = self._createAssertionHelper()
+        assertion = self._createAttributeAssertionHelper()
 
         
@@ -254 +254 @@
 
     def test02ParseAssertion(self):
-        assertion = self._createAssertionHelper()
+        assertion = self._createAttributeAssertionHelper()
         
         # Create ElementTree Assertion Element
@@ -327 +327 @@
         print("_"*80)
 
-    def test05CreateResponse(self):
+    def test05CreateAttributeQueryResponse(self):
         response = Response()
         response.issueInstant = datetime.utcnow()
@@ -347 +347 @@
         response.status.statusMessage.value = "Response created successfully"
            
-        assertion = self._createAssertionHelper()
+        assertion = self._createAttributeAssertionHelper()
         
         # Add a conditions statement for a validity of 8 hours
@@ -407 +407 @@
         
         authzDecisionQuery.actions.append(Action())
+        authzDecisionQuery.actions[0].namespace = Action.GHPP_NS_URI
         authzDecisionQuery.actions[0].value = Action.HTTP_GET_ACTION
-        authzDecisionQuery.actions[0].namespace = Action.GHPP_NS_URI
         
         self.assert_(
@@ -414 +414 @@
         self.assert_(
             authzDecisionQuery.actions[0].namespace == Action.GHPP_NS_URI)
+        
+        # Try out the restricted vocabulary
+        try:
+            authzDecisionQuery.actions[0].value = "delete everything"
+            self.fail("Expecting AttributeError raised for incorrect action "
+                      "setting.")
+        except AttributeError, e:
+            print("Caught incorrect action type setting: %s" % e)
+        
+        authzDecisionQuery.actions[0].actionTypes = {'urn:malicious': 
+                                                     ("delete everything",)}
+        
+        # Try again now that the actipn types have been adjusted
+        authzDecisionQuery.actions[0].namespace = 'urn:malicious'
+        authzDecisionQuery.actions[0].value = "delete everything"
         
     def test07SerializeAuthzDecisionQuery(self):
@@ -471 +486 @@
         self.assert_(authzDecisionQuery2.evidence is None)
 
+
+    def test05CreateAuthzDecisionQueryResponse(self):
+        response = Response()
+        response.issueInstant = datetime.utcnow()
+        
+        # Make up a request ID that this response is responding to
+        response.inResponseTo = str(uuid4())
+        response.id = str(uuid4())
+        response.version = SAMLVersion(SAMLVersion.VERSION_20)
+            
+        response.issuer = Issuer()
+        response.issuer.format = Issuer.X509_SUBJECT
+        response.issuer.value = \
+                        SAMLTestCase.ISSUER_DN
+        
+        response.status = Status()
+        response.status.statusCode = StatusCode()
+        response.status.statusCode.value = StatusCode.SUCCESS_URI
+        response.status.statusMessage = StatusMessage()        
+        response.status.statusMessage.value = "Response created successfully"
+           
+        assertion = Assertion()
+        authzDecisionStatement = AuthzDecisionStatement()
+        authzDecisionStatement.resource = SAMLTestCase.RESOURCE_URI
+        authzDecisionStatement.actions.append(Action())
+        authzDecisionStatement.actions[-1].namespace = Action.GHPP_NS_URI
+        authzDecisionStatement.actions[-1].value = Action.HTTP_GET_ACTION
+        assertion.authzDecisionStatements.append(authzDecisionStatement)
+        
+#        assertion.subject = Subject()  
+#        assertion.subject.nameID = NameID()
+#        assertion.subject.nameID.format = SAMLTestCase.NAMEID_FORMAT
+#        assertion.subject.nameID.value = SAMLTestCase.NAMEID_VALUE    
+#            
+#        assertion.issuer = Issuer()
+#        assertion.issuer.format = Issuer.X509_SUBJECT
+#        assertion.issuer.value = SAMLTestCase.ISSUER_DN
+
+        response.assertions.append(assertion)
+        
+        # Create ElementTree Assertion Element
+        responseElem = ResponseElementTree.toXML(response)
+        
+        self.assert_(iselement(responseElem))
+        
+        # Serialise to output        
+        xmlOutput = prettyPrint(responseElem)       
+        self.assert_(len(xmlOutput))
+        print("\n"+"_"*80)
+        print(xmlOutput)
+        print("_"*80)
+
        
 if __name__ == "__main__":

TI12-security/trunk/ndg_security_saml/saml/xml/etree.py

@@ -43 +43 @@
                              AuthzDecisionQuery, Subject, NameID, Issuer, 
                              Response, Status, StatusCode, StatusMessage, 
-                             StatusDetail, Advice, Action,
+                             StatusDetail, Advice, Action, Evidence,
                              XSStringAttributeValue) 
                              
@@ -371 +371 @@
         
         for authzDecisionStatement in assertion.authzDecisionStatements:
-            raise NotImplementedError("Assertion Authorisation Decision "
-                                      "Statement creation is not implemented")
+            authzDecisionStatementElem = AuthzDecisionStatementElementTree.toXML(
+                                        authzDecisionStatement,
+                                        **authzDecisionValueElementTreeFactoryKw)
+            elem.append(authzDecisionStatementElem)
             
         for attributeStatement in assertion.attributeStatements:
@@ -417 +419 @@
             attributeValues.append(attributeValue)
         
-        assertion = Assertion()
+        assertion = cls()
         assertion.version = SAMLVersion(attributeValues[0])
         if assertion.version != SAMLVersion.VERSION_20:
@@ -451 +453 @@
                                           "parsing is not implemented")
         
-            elif localName==AuthzDecisionStatement.DEFAULT_ELEMENT_LOCAL_NAME:
-                raise NotImplementedError("Assertion Authorisation Decision "
-                                          "Statement parsing is not "
-                                          "implemented")
+            elif localName == AuthzDecisionStatement.DEFAULT_ELEMENT_LOCAL_NAME:
+                authzDecisionStatement = \
+                    AuthzDecisionStatementElementTree.fromXML(childElem)
+                assertion.authzDecisionStatements.append(authzDecisionStatement)
             
             elif localName == AttributeStatement.DEFAULT_ELEMENT_LOCAL_NAME:
@@ -463 +465 @@
             else:
                 raise XMLTypeParseError('Assertion child element name "%s" '
-                                          'not recognised' % localName)
+                                        'not recognised' % localName)
         
         return assertion
@@ -533 +535 @@
         
         return attributeStatement
+
+  
+class AuthzDecisionStatementElementTree(AuthzDecisionStatement):
+    """ElementTree XML representation of AuthzDecisionStatement"""
+    
+    @classmethod
+    def toXML(cls, authzDecisionStatement):
+        """Make a tree of a XML elements based on the authzDecision statement
+        
+        @type assertion: saml.saml2.core.AuthzDecisionStatement
+        @param assertion: AuthzDecision Statement to be represented as an 
+        ElementTree Element
+        factory
+        @rtype: ElementTree.Element
+        @return: ElementTree Element
+        """
+        if not isinstance(authzDecisionStatement, AuthzDecisionStatement):
+            raise TypeError("Expecting %r type got: %r" % 
+                            (AuthzDecisionStatement, authzDecisionStatement))
+          
+        if not authzDecisionStatement.resource:
+            raise AttributeError("Resource for AuthzDecisionStatement is not "
+                                 "set")
+              
+        attrib = {
+            cls.DECISION_ATTRIB_NAME: authzDecisionStatement.decision,
+            cls.RESOURCE_ATTRIB_NAME: authzDecisionStatement.resource
+        }
+            
+        tag = str(QName.fromGeneric(cls.DEFAULT_ELEMENT_NAME))  
+        elem = ElementTree.Element(tag, **attrib)
+        ElementTree._namespace_map[cls.DEFAULT_ELEMENT_NAME.namespaceURI
+                                   ] = cls.DEFAULT_ELEMENT_NAME.prefix 
+
+        for action in authzDecisionStatement.actions:
+            # Factory enables support for multiple authzDecision types
+            actionElem = ActionElementTree.toXML(action)
+            elem.append(actionElem)
+        
+        if (authzDecisionStatement.evidence and 
+            len(authzDecisionStatement.evidence.values) > 0):
+            raise NotImplementedError("authzDecisionStatementElementTree does "
+                                      "not currently support the Evidence type")
+            
+        return elem
+    
+    @classmethod
+    def fromXML(cls, elem, **authzDecisionValueElementTreeFactoryKw):
+        """Parse an ElementTree SAML AuthzDecisionStatement element into an
+        AuthzDecisionStatement object
+        
+        @type elem: ElementTree.Element
+        @param elem: ElementTree element containing the AuthzDecisionStatement
+        @type authzDecisionValueElementTreeFactoryKw: dict
+        @param authzDecisionValueElementTreeFactoryKw: keywords for AuthzDecisionValue
+        factory
+        @rtype: saml.saml2.core.AuthzDecisionStatement
+        @return: AuthzDecision Statement"""
+        
+        if not ElementTree.iselement(elem):
+            raise TypeError("Expecting %r input type for parsing; got %r" %
+                            (ElementTree.Element, elem))
+
+        localName = QName.getLocalPart(elem.tag)
+        if localName != cls.DEFAULT_ELEMENT_LOCAL_NAME:
+            raise XMLTypeParseError("No \"%s\" element found" %
+                                    cls.DEFAULT_ELEMENT_LOCAL_NAME)
+        
+        
+        authzDecisionStatement = cls()
+
+        for childElem in elem:
+            localName = QName.getLocalPart(childElem.tag)
+            
+            if localName == Action.DEFAULT_ELEMENT_LOCAL_NAME:
+                action = ActionElementTree.fromXML(childElem)
+                authzDecisionStatement.actions.append(action)
+                
+            elif localName == Evidence.DEFAULT_ELEMENT_LOCAL_NAME:
+                raise NotImplementedError("XML parse of %s element is not "
+                                          "implemented" %
+                                          Evidence.DEFAULT_ELEMENT_LOCAL_NAME)
+            else:
+                raise XMLTypeParseError("AuthzDecisionStatement child element "
+                                        "name %r not recognised" % localName)
+        
+        return authzDecisionStatement