Add interface and tests for AccessControl.SecurityManager.

o The new tests are amphibious:  they exercise both the Python and the C
  implementations, ensuring that they remain in sync.

lib/python/AccessControl/ImplPython.py

@@ -22,6 +22,7 @@ from Acquisition import aq_inner
 from Acquisition import aq_acquire
 from ExtensionClass import Base
 from zLOG import LOG, BLATHER, PROBLEM
+from zope.interface import implements
 
 # This is used when a permission maps explicitly to no permission.  We
 # try and get this from cAccessControl first to make sure that if both
@@ -33,6 +34,7 @@ except ImportError:
 
 from AccessControl import SecurityManagement
 from AccessControl import Unauthorized
+from AccessControl.interfaces import ISecurityManager
 from AccessControl.SimpleObjectPolicies import Containers, _noroles
 from AccessControl.ZopeGuards import guarded_getitem
 
@@ -491,7 +493,7 @@ class SecurityManager:
     """A security manager provides methods for checking access and managing
     executable context and policies
     """
-
+    implements(ISecurityManager)
     __allow_access_to_unprotected_subobjects__ = {
         'validate': 1, 'checkPermission': 1,
         'getUser': 1, 'calledByExecutable': 1

lib/python/AccessControl/interfaces.py

+##############################################################################
+#
+# Copyright (c) 2005 Zope Corporation and Contributors. All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE
+#
+##############################################################################
+"""Access control interfaces
+"""
+
+from AccessControl.SimpleObjectPolicies import _noroles
+from zope.interface import Interface
+from zope.interface import Attribute
+
+class ISecurityManager(Interface):
+    """Checks access and manages executable context and policies.
+    """
+    _policy = Attribute(u'Current Security Policy')
+
+    def validate(accessed=None,
+                 container=None,
+                 name=None,
+                 value=None,
+                 roles=_noroles,
+                ):
+        """Validate access.
+
+        Arguments:
+
+        accessed -- the object that was being accessed
+
+        container -- the object the value was found in
+
+        name -- The name used to access the value
+
+        value -- The value retrieved though the access.
+
+        roles -- The roles of the object if already known.
+
+        The arguments may be provided as keyword arguments. Some of these
+        arguments may be ommitted, however, the policy may reject access
+        in some cases when arguments are ommitted.  It is best to provide
+        all the values possible.
+        """
+
+    def DTMLValidate(accessed=None,
+                     container=None,
+                     name=None,
+                     value=None,
+                     md=None,
+                    ):
+        """Validate access.
+        * THIS EXISTS FOR DTML COMPATIBILITY *
+
+        Arguments:
+
+        accessed -- the object that was being accessed
+
+        container -- the object the value was found in
+
+        name -- The name used to access the value
+
+        value -- The value retrieved though the access.
+
+        md -- multidict for DTML (ignored)
+
+        The arguments may be provided as keyword arguments. Some of these
+        arguments may be ommitted, however, the policy may reject access
+        in some cases when arguments are ommitted.  It is best to provide
+        all the values possible.
+
+        """
+
+    def checkPermission(permission, object):
+        """Check whether the security context allows the given permission on
+        the given object.
+
+        Arguments:
+
+        permission -- A permission name
+
+        object -- The object being accessed according to the permission
+        """
+
+    def addContext(anExecutableObject):
+        """Add an ExecutableObject to the current security context.
+        
+        o If it declares a custom security policy,  make that policy
+          "current";  otherwise, make the "default" security policy
+          current.
+        """
+
+    def removeContext(anExecutableObject):
+        """Remove an ExecutableObject from the current security context.
+        
+        o Remove all objects from the top of the stack "down" to the
+          supplied object.
+
+        o If the top object on the stack declares a custom security policy,
+          make that policy "current".
+
+        o If the stack is empty, or if the top declares no custom security
+          policy, restore the 'default" security policy as current.
+        """
+
+    def getUser():
+        """Get the currently authenticated user
+        """
+
+    def calledByExecutable():
+        """Return a boolean value indicating whether this context was called
+           in the context of an by an executable (i.e., one added via
+           'addContext').
+        """

lib/python/AccessControl/tests/testSecurityManager.py

+##############################################################################
+#
+# Copyright (c) 2005 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""Tests for the SecurityManager implementations
+
+$Id$
+"""
+
+import unittest
+
+_THREAD_ID = 123
+
+class DummyContext:
+
+    def __init__(self):
+        self.user = object()
+        self.stack = []
+
+class DummyPolicy:
+
+    CHECK_PERMISSION_ARGS = None
+    CHECK_PERMISSION_RESULT = object()
+
+    VALIDATE_ARGS = None
+
+    def checkPermission(self, *args):
+        self.CHECK_PERMISSION_ARGS = args
+        return self.CHECK_PERMISSION_RESULT
+
+    def validate(self, *args):
+        self.VALIDATE_ARGS = args
+        return True
+
+class ExecutableObject:
+    def __init__(self, new_policy):
+        self._new_policy = new_policy
+
+    def _customSecurityPolicy(self):
+        return self._new_policy
+
+class ISecurityManagerConformance:
+
+    def test_conforms_to_ISecurityManager(self):
+        from AccessControl.interfaces import ISecurityManager
+        from zope.interface.verify import verifyClass
+        verifyClass(ISecurityManager, self._getTargetClass())
+
+class SecurityManagerTestBase(unittest.TestCase):
+
+    def _makeOne(self, thread_id, context):
+        return self._getTargetClass()(thread_id, context)
+
+    def test_getUser(self):
+        context = DummyContext()
+        mgr = self._makeOne(_THREAD_ID, context)
+        self.failUnless(mgr.getUser() is context.user)
+
+    def test_calledByExecutable_no_stack(self):
+        context = DummyContext()
+        mgr = self._makeOne(_THREAD_ID, context)
+        self.failIf(mgr.calledByExecutable())
+
+    def test_calledByExecutable_with_stack(self):
+        context = DummyContext()
+        mgr = self._makeOne(_THREAD_ID, context)
+        executableObject = object()
+        mgr.addContext(executableObject)
+        self.failUnless(mgr.calledByExecutable())
+
+    def test_addContext_no_custom_policy(self):
+        context = DummyContext()
+        mgr = self._makeOne(_THREAD_ID, context)
+        original_policy = mgr._policy
+        executableObject = object()
+        mgr.addContext(executableObject)
+        self.failUnless(mgr._policy is original_policy)
+
+    def test_addContext_with_custom_policy(self):
+        context = DummyContext()
+        mgr = self._makeOne(_THREAD_ID, context)
+        new_policy = DummyPolicy()
+        executableObject = ExecutableObject(new_policy)
+        mgr.addContext(executableObject)
+        self.failUnless(mgr._policy is new_policy)
+
+    def test_addContext_with_custom_policy_then_none(self):
+        context = DummyContext()
+        mgr = self._makeOne(_THREAD_ID, context)
+        original_policy = mgr._policy
+        new_policy = DummyPolicy()
+        executableObject = ExecutableObject(new_policy)
+        mgr.addContext(executableObject)
+        mgr.addContext(object())
+        self.failUnless(mgr._policy is original_policy)
+
+    def test_removeContext_pops_items_above_EO(self):
+        context = DummyContext()
+        ALPHA, BETA, GAMMA, DELTA = object(), object(), object(), object()
+        context.stack.append(ALPHA)
+        context.stack.append(BETA)
+        context.stack.append(GAMMA)
+        context.stack.append(DELTA)
+        mgr = self._makeOne(_THREAD_ID, context)
+
+        mgr.removeContext(GAMMA)
+
+        self.assertEqual(len(context.stack), 2)
+        self.failUnless(context.stack[0] is ALPHA)
+        self.failUnless(context.stack[1] is BETA)
+
+    def test_removeContext_last_EO_restores_default_policy(self):
+        context = DummyContext()
+        mgr = self._makeOne(_THREAD_ID, context)
+        original_policy = mgr._policy
+        new_policy = mgr._policy = DummyPolicy()
+        top = object()
+        context.stack.append(top)
+        mgr.removeContext(top)
+        self.failUnless(mgr._policy is original_policy)
+
+    def test_removeContext_with_top_having_custom_policy(self):
+        context = DummyContext()
+        mgr = self._makeOne(_THREAD_ID, context)
+        new_policy = DummyPolicy()
+        context.stack.append(ExecutableObject(new_policy))
+        top = object()
+        context.stack.append(top)
+        mgr.removeContext(top)
+        self.failUnless(mgr._policy is new_policy)
+
+    def test_removeContext_with_top_having_no_custom_policy(self):
+        context = DummyContext()
+        mgr = self._makeOne(_THREAD_ID, context)
+        original_policy = mgr._policy
+        new_policy = DummyPolicy()
+        executableObject = ExecutableObject(new_policy)
+        context.stack.append(executableObject)
+        top = object()
+        context.stack.append(top)
+        mgr.removeContext(executableObject)
+        self.failUnless(mgr._policy is original_policy)
+
+    def test_checkPermission_delegates_to_policy(self):
+        context = DummyContext()
+        PERMISSION = 'PERMISSION'
+        TARGET = object()
+        mgr = self._makeOne(_THREAD_ID, context)
+        new_policy = mgr._policy = DummyPolicy()
+        result = mgr.checkPermission(PERMISSION, TARGET)
+        self.failUnless(result is DummyPolicy.CHECK_PERMISSION_RESULT)
+        self.failUnless(new_policy.CHECK_PERMISSION_ARGS[0] is PERMISSION)
+        self.failUnless(new_policy.CHECK_PERMISSION_ARGS[1] is TARGET)
+        self.failUnless(new_policy.CHECK_PERMISSION_ARGS[2] is context)
+
+    def test_validate_without_roles_delegates_to_policy(self):
+        from AccessControl.SimpleObjectPolicies import _noroles
+
+        context = DummyContext()
+        ACCESSED = object()
+        CONTAINER = object()
+        NAME = 'NAME'
+        VALUE = object()
+        mgr = self._makeOne(_THREAD_ID, context)
+        new_policy = mgr._policy = DummyPolicy()
+
+        result = mgr.validate(ACCESSED,
+                              CONTAINER,
+                              NAME,
+                              VALUE,
+                             )
+
+        self.failUnless(result)
+        self.assertEqual(len(new_policy.VALIDATE_ARGS), 5)
+        self.failUnless(new_policy.VALIDATE_ARGS[0] is ACCESSED)
+        self.failUnless(new_policy.VALIDATE_ARGS[1] is CONTAINER)
+        self.assertEqual(new_policy.VALIDATE_ARGS[2], NAME)
+        self.failUnless(new_policy.VALIDATE_ARGS[3] is VALUE)
+        self.failUnless(new_policy.VALIDATE_ARGS[4] is context)
+
+    def test_validate_with_roles_delegates_to_policy(self):
+        from AccessControl.SimpleObjectPolicies import _noroles
+
+        context = DummyContext()
+        ACCESSED = object()
+        CONTAINER = object()
+        NAME = 'NAME'
+        VALUE = object()
+        ROLES = ('Hamlet', 'Othello')
+        mgr = self._makeOne(_THREAD_ID, context)
+        new_policy = mgr._policy = DummyPolicy()
+
+        result = mgr.validate(ACCESSED,
+                              CONTAINER,
+                              NAME,
+                              VALUE,
+                              ROLES,
+                             )
+
+        self.failUnless(result)
+        self.assertEqual(len(new_policy.VALIDATE_ARGS), 6)
+        self.failUnless(new_policy.VALIDATE_ARGS[0] is ACCESSED)
+        self.failUnless(new_policy.VALIDATE_ARGS[1] is CONTAINER)
+        self.assertEqual(new_policy.VALIDATE_ARGS[2], NAME)
+        self.failUnless(new_policy.VALIDATE_ARGS[3] is VALUE)
+        self.failUnless(new_policy.VALIDATE_ARGS[4] is context)
+        self.assertEqual(new_policy.VALIDATE_ARGS[5], ROLES)
+
+    def test_DTMLValidate_delegates_to_policy_validate(self):
+        from AccessControl.SimpleObjectPolicies import _noroles
+
+        context = DummyContext()
+        ACCESSED = object()
+        CONTAINER = object()
+        NAME = 'NAME'
+        VALUE = object()
+        MD = {}
+        mgr = self._makeOne(_THREAD_ID, context)
+        new_policy = mgr._policy = DummyPolicy()
+
+        result = mgr.DTMLValidate(ACCESSED,
+                                  CONTAINER,
+                                  NAME,
+                                  VALUE,
+                                  MD,
+                                 )
+
+        self.failUnless(result)
+        self.assertEqual(len(new_policy.VALIDATE_ARGS), 5)
+        self.failUnless(new_policy.VALIDATE_ARGS[0] is ACCESSED)
+        self.failUnless(new_policy.VALIDATE_ARGS[1] is CONTAINER)
+        self.assertEqual(new_policy.VALIDATE_ARGS[2], NAME)
+        self.failUnless(new_policy.VALIDATE_ARGS[3] is VALUE)
+        self.failUnless(new_policy.VALIDATE_ARGS[4] is context)
+
+class PythonSecurityManagerTests(SecurityManagerTestBase,
+                                 ISecurityManagerConformance,
+                                ):
+
+    def _getTargetClass(self):
+        from AccessControl.ImplPython import SecurityManager
+        return SecurityManager
+
+
+# N.B.:  The C version mixes in the Python version, which is why we
+#        can test for conformance to ISecurityManager.
+class C_SecurityManagerTests(SecurityManagerTestBase,
+                             ISecurityManagerConformance,
+                            ):
+
+    def _getTargetClass(self):
+        from AccessControl.ImplC import SecurityManager
+        return SecurityManager
+
+
+def test_suite():
+    suite = unittest.TestSuite()
+    suite.addTest( unittest.makeSuite( PythonSecurityManagerTests ) )
+    suite.addTest( unittest.makeSuite( C_SecurityManagerTests ) )
+    return suite
+
+if __name__ == '__main__':
+    unittest.main(defaultTest='test_suite')
+