Source Code for Package ndg.xacml.test.context

  1  #!/usr/bin/env python 
  2  """NDG XACML Context unit test package  
  3   
  4  NERC DataGrid 
  5  """ 
  6  __author__ = "P J Kershaw" 
  7  __date__ = "28/10/10" 
  8  __copyright__ = "(C) 2010 Science and Technology Facilities Council" 
  9  __contact__ = "Philip.Kershaw@stfc.ac.uk" 
 10  __license__ = "BSD - see LICENSE file in top-level directory" 
 11  __contact__ = "Philip.Kershaw@stfc.ac.uk" 
 12  __revision__ = "$Id$" 
 13  import unittest 
 14  from os import path 
 15   
 16  from ndg.xacml.core import Identifiers 
 17  from ndg.xacml.core.attribute import Attribute 
 18  from ndg.xacml.core.attributevalue import (AttributeValue,  
 19                                             AttributeValueClassFactory) 
 20   
 21  from ndg.xacml.core.context.environment import Environment 
 22  from ndg.xacml.core.context.request import Request 
 23  from ndg.xacml.core.context.subject import Subject 
 24  from ndg.xacml.core.context.resource import Resource 
 25  from ndg.xacml.core.context.action import Action 
 26  from ndg.xacml.core.context.pdp import PDP 
 27  from ndg.xacml.core.context.handler import CtxHandlerBase 
 28  from ndg.xacml.parsers.etree.factory import ReaderFactory 
 29   
 30  from ndg.xacml.test import XACML_NDGTEST1_FILEPATH 
 31    
 32  ROLE_ATTRIBUTE_ID = "urn:ndg:security:authz:1.0:attr" 
 33  SUBJECT_ID = 'https://my.name.somewhere.ac.uk' 
 34   
 35  attributeValueFactory = AttributeValueClassFactory() 
 36  AnyUriAttributeValue = attributeValueFactory(AttributeValue.ANY_TYPE_URI) 
 37  StringAttributeValue = attributeValueFactory(AttributeValue.STRING_TYPE_URI) 
 38   
 39   
 40 -class TestContextHandler(CtxHandlerBase): 
 41      """Test implementation of Context Handler which includes an implemented PIP 
 42      interface""" 
 43       
 44 -    def __init__(self): 
 45          """Add an attribute to hold a reference to a policy information point""" 
 46           
 47          super(TestContextHandler, self).__init__() 
 48           
 49 -    def handlePEPRequest(self, myRequest): 
 50          """Handle request from Policy Enforcement Point 
 51           
 52          @param pepRequest: request from PEP, derived class determines its type 
 53          e.g. SAML AuthzDecisionQuery 
 54          @type myRequest: type 
 55          @return: PEP response - derived class determines type 
 56          @rtype: None 
 57          """ 
 58           
 59          # Convert myRequest to XACML context request - var assignment here is  
 60          # representative of this process rather than actually doing anything. 
 61          xacmlRequest = myRequest 
 62           
 63          if self.pdp is None: 
 64              raise TypeError('No "pdp" attribute set') 
 65           
 66          # Add a reference to this context so that the PDP can invoke queries 
 67          # back to the PIP 
 68          xacmlRequest.ctxHandler = self  
 69                  
 70          xacmlResponse = self.pdp.evaluate(xacmlRequest) 
 71           
 72          # Convert XACML context response to domain specific request 
 73          myResponse = xacmlResponse 
 74           
 75          return myResponse 
 76       
 77 -    def pipQuery(self, request, designator): 
 78          '''PIP adds admin attribute value for given attribute ID and for any  
 79          subject''' 
 80          if designator.attributeId == ROLE_ATTRIBUTE_ID: 
 81              attrVal = StringAttributeValue(value='admin') 
 82              return [attrVal] 
 83          else: 
 84              return None  
 85           
 86           
 87 -class XacmlContextBaseTestCase(unittest.TestCase): 
 88      """Base class containing common methods for test initialisation""" 
 89       
 90      @staticmethod 
 91 -    def _createRequestCtx(resourceId,  
 92                            includeSubject=True, 
 93                            subjectId=SUBJECT_ID, 
 94                            subjectRoles=None, 
 95                            roleAttributeId=ROLE_ATTRIBUTE_ID, 
 96                            action='read', 
 97                            resourceContent=None): 
 98          """Create an example XACML Request Context for tests""" 
 99          if subjectRoles is None: 
100              subjectRoles = ('staff',) 
101               
102          request = Request() 
103           
104          if includeSubject: 
105              subject = Subject() 
106              openidSubjectAttribute = Attribute() 
107               
108              openidSubjectAttribute.attributeId = "urn:esg:openid" 
109              openidSubjectAttribute.dataType = AnyUriAttributeValue.IDENTIFIER 
110               
111              openidSubjectAttribute.attributeValues.append( 
112                                                          AnyUriAttributeValue()) 
113              openidSubjectAttribute.attributeValues[-1].value = subjectId 
114                                           
115               
116              subject.attributes.append(openidSubjectAttribute) 
117       
118              for role in subjectRoles: 
119                  roleAttribute = Attribute() 
120                   
121                  roleAttribute.attributeId = roleAttributeId 
122                  roleAttribute.dataType = StringAttributeValue.IDENTIFIER 
123                   
124                  roleAttribute.attributeValues.append(StringAttributeValue()) 
125                  roleAttribute.attributeValues[-1].value = role  
126               
127                  subject.attributes.append(roleAttribute) 
128                                         
129              request.subjects.append(subject) 
130           
131          resource = Resource() 
132          resourceAttribute = Attribute() 
133          resource.attributes.append(resourceAttribute) 
134           
135          resourceAttribute.attributeId = Identifiers.Resource.RESOURCE_ID 
136                               
137          resourceAttribute.dataType = AnyUriAttributeValue.IDENTIFIER 
138          resourceAttribute.attributeValues.append(AnyUriAttributeValue()) 
139          resourceAttribute.attributeValues[-1].value = resourceId 
140   
141          resource.resourceContent = resourceContent 
142   
143          request.resources.append(resource) 
144           
145          request.action = Action() 
146          actionAttribute = Attribute() 
147          request.action.attributes.append(actionAttribute) 
148           
149          actionAttribute.attributeId = Identifiers.Action.ACTION_ID 
150          actionAttribute.dataType = StringAttributeValue.IDENTIFIER 
151          actionAttribute.attributeValues.append(StringAttributeValue()) 
152          actionAttribute.attributeValues[-1].value = action 
153   
154          request.environment = Environment() 
155           
156          return request 
157           
158      @staticmethod 
159 -    def _createPDPfromNdgTest1Policy(): 
160          """Create PDP from NDG test policy file""" 
161          pdp = PDP.fromPolicySource(XACML_NDGTEST1_FILEPATH, ReaderFactory) 
162          return pdp 
163