Forward port fix for collector #1774 from 2.8 branch.

lib/python/AccessControl/ImplC.py

@@ -18,7 +18,8 @@ try:
     from cAccessControl import rolesForPermissionOn, \
          PermissionRole, imPermissionRole, _what_not_even_god_should_do, \
          RestrictedDTMLMixin, aq_validate, guarded_getattr, \
-         ZopeSecurityPolicy, setDefaultBehaviors
+         setDefaultBehaviors
+    from cAccessControl import ZopeSecurityPolicy as cZopeSecurityPolicy
     from cAccessControl import SecurityManager as cSecurityManager
 except ImportError:
     import sys
@@ -26,12 +27,17 @@ except ImportError:
     del sys.modules[__name__]
 
 
-from ImplPython import RestrictedDTML, SecurityManager
+from ImplPython import RestrictedDTML, SecurityManager, ZopeSecurityPolicy
 
 
 class RestrictedDTML(RestrictedDTMLMixin, RestrictedDTML):
     """A mix-in for derivatives of DT_String.String that adds Zope security."""
 
+class ZopeSecurityPolicy(cZopeSecurityPolicy, ZopeSecurityPolicy):
+    """A security manager provides methods for checking access and managing
+    executable context and policies
+    """
+
 class SecurityManager(cSecurityManager, SecurityManager):
     """A security manager provides methods for checking access and managing
     executable context and policies

lib/python/AccessControl/ImplPython.py

@@ -34,6 +34,7 @@ except ImportError:
 
 from AccessControl import SecurityManagement
 from AccessControl import Unauthorized
+from AccessControl.interfaces import ISecurityPolicy
 from AccessControl.interfaces import ISecurityManager
 from AccessControl.SimpleObjectPolicies import Containers, _noroles
 from AccessControl.ZopeGuards import guarded_getitem
@@ -199,6 +200,8 @@ from AccessControl.ZopeSecurityPolicy import getRoles  # XXX
 
 class ZopeSecurityPolicy:
 
+    implements(ISecurityPolicy)
+
     def __init__(self, ownerous=1, authenticated=1, verbose=0):
         """Create a Zope security policy.
 
@@ -459,12 +462,41 @@ class ZopeSecurityPolicy:
         raise Unauthorized(name, value)
 
     def checkPermission(self, permission, object, context):
-        # XXX proxy roles and executable owner are not checked
         roles = rolesForPermissionOn(permission, object)
         if isinstance(roles, basestring):
             roles = [roles]
-        return context.user.allowed(object, roles)
 
+        # check executable owner and proxy roles
+        stack = context.stack
+        if stack:
+            eo = stack[-1]
+            # If the executable had an owner, can it execute?
+            if self._ownerous:
+                owner = eo.getOwner()
+                if (owner is not None) and not owner.allowed(object, roles):
+                    # We don't want someone to acquire if they can't 
+                    # get an unacquired!
+                    return 0
+            proxy_roles = getattr(eo, '_proxy_roles', None)
+            if proxy_roles:
+                # Verify that the owner actually can state the proxy role
+                # in the context of the accessed item; users in subfolders
+                # should not be able to use proxy roles to access items 
+                # above their subfolder!
+                owner = eo.getWrappedOwner()
+                if owner is not None:
+                    if object is not aq_base(object):
+                        if not owner._check_context(object):
+                            # object is higher up than the owner, 
+                            # deny access
+                            return 0
+
+                for r in proxy_roles:
+                    if r in roles:
+                        return 1
+                return 0
+
+        return context.user.allowed(object, roles)
 
 # AccessControl.SecurityManager
 # -----------------------------

lib/python/AccessControl/cAccessControl.c

@@ -337,8 +337,6 @@ typedef struct {
 */
 
 static PyObject *ZopeSecurityPolicy_validate(PyObject *self, PyObject *args);
-static PyObject *ZopeSecurityPolicy_checkPermission(PyObject *self,
-	PyObject *args);
 static void ZopeSecurityPolicy_dealloc(ZopeSecurityPolicy *self);
 
 
@@ -418,11 +416,6 @@ static PyMethodDef ZopeSecurityPolicy_methods[] = {
 		METH_VARARGS,
 		""
 	},
-	{"checkPermission",
-		(PyCFunction)ZopeSecurityPolicy_checkPermission,
-		METH_VARARGS,
-		""
-	},
 	{ NULL, NULL }
 };
 
@@ -1286,69 +1279,6 @@ static PyObject *ZopeSecurityPolicy_validate(PyObject *self, PyObject *args) {
 }
 
 
-/*
-** ZopeSecurityPolicy_checkPermission
-**
-*/
-
-static PyObject *ZopeSecurityPolicy_checkPermission(PyObject *self,
-	PyObject *args) {
-
-	PyObject *permission = NULL;
-	PyObject *object = NULL;
-	PyObject *context = NULL;
-	PyObject *roles;
-	PyObject *result = NULL;
-	PyObject *user;
-
-	/*| def checkPermission(self, permission, object, context)
-	*/
-
-	if (unpacktuple3(args, "checkPermission", 3, 
-                         &permission, &object, &context) < 0)
-		return NULL;
-
-	/*| roles = rolesForPermissionOn(permission, object)
-	*/
-
-	roles = c_rolesForPermissionOn(permission, object, NULL, NULL);
-	if (roles == NULL)
-          return NULL;
-
-	/*| if type(roles) in (StringType, UnicodeType):
-	**|	roles = [roles]
-	*/
-
-	if ( PyString_Check(roles) || PyUnicode_Check(roles) ) {
-          PyObject *r;
-
-          r = PyList_New(1);
-          if (r == NULL) {
-            Py_DECREF(roles);
-            return NULL;
-          }
-          /* Note: ref to roles is passed to the list object. */
-          PyList_SET_ITEM(r, 0, roles);
-          roles = r;
-	}
-
-	/*| return context.user.allowed(object, roles)
-	*/
-
-	user = PyObject_GetAttr(context, user_str);
-	if (user != NULL) {
-          ASSIGN(user, PyObject_GetAttr(user, allowed_str));
-          if (user != NULL) {
-            result = callfunction2(user, object, roles);
-            Py_DECREF(user);
-          }
-	}
-
-	Py_DECREF(roles);
-
-	return result;
-}
-
 /*
 ** ZopeSecurityPolicy_dealloc
 **

lib/python/AccessControl/interfaces.py

@@ -282,8 +282,37 @@ class IStandardUserFolder(Interface):
         """Get a sequence of names of the users which reside in the user folder.
         """
 
+class ISecurityPolicy(Interface):
+    """Plug-in policy for checking access to objects within untrusted code.
+    """
+    def validate(accessed, container, name, value, context, roles=_noroles):
+        """Check that the current user (from context) has access.
+
+        o Raise Unauthorized if access is not allowed;  otherwise, return
+          a true value.
+
+        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.
+
+        context -- the security context (normally supplied by the security
+                   manager).
+
+        roles -- The roles of the object if already known.
+        """
+
+    def checkPermission(permission, object, context):
+        """Check whether the current user has a permission w.r.t. an object.
+        """
+
 class ISecurityManager(Interface):
-    """Checks access and manages executable context and policies.
+    """Check access and manages executable context and policies.
     """
     _policy = Attribute(u'Current Security Policy')
 

lib/python/AccessControl/tests/testZopeSecurityPolicy.py

@@ -23,7 +23,6 @@ try:
     from zExceptions import Unauthorized
 except ImportError:
     Unauthorized = 'Unauthorized'
-from AccessControl.ZopeSecurityPolicy import ZopeSecurityPolicy
 from AccessControl.User import UserFolder
 from AccessControl.SecurityManagement import SecurityContext
 from Acquisition import Implicit, Explicit, aq_base
@@ -126,6 +125,8 @@ class RestrictedSimpleItem (SimpleItemish):
 
     __allow_access_to_unprotected_subobjects__ = 0
 
+    _Foo_Permission = user_roles + eo_roles
+    _Kill_Permission = sysadmin_roles
     _View_Permission = eo_roles
 
 
@@ -152,9 +153,7 @@ class SimpleClass:
     attr = 1
 
 
-class ZopeSecurityPolicyTests (unittest.TestCase):
-
-    policy = ZopeSecurityPolicy()
+class ZopeSecurityPolicyTestBase(unittest.TestCase):
 
     def setUp(self):
         a = App()
@@ -174,6 +173,10 @@ class ZopeSecurityPolicyTests (unittest.TestCase):
         self.user = user
         context = SecurityContext(user)
         self.context = context
+        self.policy = self._makeOne()
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
 
     def assertPolicyAllows(self, ob, attrname):
         res = self.policy.validate(ob, ob, attrname, getattr(ob, attrname),
@@ -276,6 +279,52 @@ class ZopeSecurityPolicyTests (unittest.TestCase):
         v = self.policy.checkPermission('View', r_item, o_context)
         self.assert_(v, '_View_Permission should grant access to theowner')
 
+    def test_checkPermission_respects_proxy_roles(self):
+        r_item = self.a.r_item
+        context = self.context
+        self.failIf(self.policy.checkPermission('View', r_item, context))
+        o_context = SecurityContext(self.uf.getUserById('joe'))
+        # Push an executable with proxy roles on the stack
+        eo = OwnedSetuidMethod().__of__(r_item)
+        eo._proxy_roles = eo_roles
+        context.stack.append(eo)
+        self.failUnless(self.policy.checkPermission('View', r_item, context))
+
+    def test_checkPermission_proxy_roles_limit_access(self):
+        r_item = self.a.r_item
+        context = self.context
+        self.failUnless(self.policy.checkPermission('Foo', r_item, context))
+        o_context = SecurityContext(self.uf.getUserById('joe'))
+        # Push an executable with proxy roles on the stack
+        eo = OwnedSetuidMethod().__of__(r_item)
+        eo._proxy_roles = sysadmin_roles
+        context.stack.append(eo)
+        self.failIf(self.policy.checkPermission('Foo', r_item, context))
+
+    def test_checkPermission_proxy_role_scope(self):
+        self.a.subobject = ImplictAcqObject()
+        subobject = self.a.subobject
+        subobject.acl_users = UserFolder()
+        subobject.acl_users._addUser('theowner', 'password', 'password', 
+                                      eo_roles + sysadmin_roles, ())
+        subobject.r_item = RestrictedSimpleItem()
+        r_subitem = subobject.r_item
+        r_subitem.owned_setuid_m = OwnedSetuidMethod()
+        r_subitem.getPhysicalRoot = lambda root=self.a: root
+
+        r_item = self.a.r_item
+        r_item.getPhysicalRoot = lambda root=self.a: root
+        context = self.context
+        context.stack.append(r_subitem.owned_setuid_m.__of__(r_subitem))
+
+        # Out of owner context
+        self.failIf(self.policy.checkPermission('View', r_item, context))
+        self.failIf(self.policy.checkPermission('Kill', r_item, context))
+
+        # Inside owner context
+        self.failIf(self.policy.checkPermission('View', r_subitem, context))
+        self.failUnless(self.policy.checkPermission('Kill', r_subitem, context))
+
     def testUnicodeRolesForPermission(self):
         r_item = self.a.r_item
         context = self.context
@@ -351,6 +400,27 @@ class ZopeSecurityPolicyTests (unittest.TestCase):
                 self.fail('Policy accepted bad __roles__')
 
 
+class ISecurityPolicyConformance:
+
+    def test_conforms_to_ISecurityPolicy(self):
+        from AccessControl.interfaces import ISecurityPolicy
+        from zope.interface.verify import verifyClass
+        verifyClass(ISecurityPolicy, self._getTargetClass())
+
+class Python_ZSPTests(ZopeSecurityPolicyTestBase,
+                      ISecurityPolicyConformance,
+                     ):
+    def _getTargetClass(self):
+        from AccessControl.ImplPython import ZopeSecurityPolicy
+        return ZopeSecurityPolicy
+
+class C_ZSPTests(ZopeSecurityPolicyTestBase,
+                 ISecurityPolicyConformance,
+                ):
+    def _getTargetClass(self):
+        from AccessControl.ImplC import ZopeSecurityPolicy
+        return ZopeSecurityPolicy
+
 def test_getRoles():
     """
 
@@ -445,6 +515,7 @@ def test_getRoles():
 
 def test_zsp_gets_right_roles_for_methods():
     """
+    >>> from AccessControl.ZopeSecurityPolicy import ZopeSecurityPolicy
     >>> zsp = ZopeSecurityPolicy()
     >>> from ExtensionClass import Base
     >>> class C(Base):
@@ -499,7 +570,8 @@ from zope.testing.doctest import DocTestSuite
 
 def test_suite():
     suite = unittest.TestSuite()
-    suite.addTest(unittest.makeSuite(ZopeSecurityPolicyTests, 'test'))
+    suite.addTest(unittest.makeSuite(Python_ZSPTests, 'test'))
+    suite.addTest(unittest.makeSuite(C_ZSPTests, 'test'))
     suite.addTest(DocTestSuite())
     return suite