- Merge change for CMF Collector #259

doc/CHANGES.txt

@@ -144,6 +144,9 @@ Zope Changes
 
     Bugs fixed
 
+     - OFS.CopySupport: Enforced "Delete objects" permission during
+       move (CMF Collector #259).
+
      - Removed DWIM'y attempt to filter acquired-but-not-aceessible
        results from 'guarded_getattr'.
 

lib/python/OFS/CopySupport.py

@@ -21,6 +21,7 @@ from zlib import compress, decompress
 
 from App.Dialogs import MessageDialog
 from AccessControl import getSecurityManager
+from AccessControl.Permissions import delete_objects as DeleteObjects
 from Acquisition import aq_base, aq_inner, aq_parent
 from zExceptions import Unauthorized, BadRequest
 from webdav.Lockable import ResourceLockedError
@@ -152,7 +153,7 @@ class CopyContainer(ExtensionClass.Base):
             m = Moniker.loadMoniker(mdata)
             try: ob = m.bind(app)
             except: raise CopyError, eNotFound
-            self._verifyObjectPaste(ob)
+            self._verifyObjectPaste(ob, validate_src=op+1)
             oblist.append(ob)
 
         if op==0:
@@ -379,13 +380,23 @@ class CopyContainer(ExtensionClass.Base):
                   action = 'manage_main')
 
             if validate_src:
+
+                sm = getSecurityManager()
+
                 # Ensure the user is allowed to access the object on the
                 # clipboard.
-                try:    parent = aq_parent(aq_inner(object))
-                except: parent = None
-                if not getSecurityManager().validate(None,parent,None,object):
+                try:
+                    parent = aq_parent(aq_inner(object))
+                except:
+                    parent = None
+
+                if not sm.validate(None,parent,None,object):
                     raise Unauthorized, absattr(object.id)
 
+                if validate_src == 2: # moving
+                    if not sm.checkPermission(DeleteObjects, parent):
+                        raise Unauthorized, 'Delete not allowed.'
+
         else: # /if method_name
             raise CopyError, MessageDialog(
                   title   = 'Not Supported',

lib/python/OFS/tests/testCopySupport.py

-import os, sys, unittest
+import unittest
+import cStringIO
+from mimetools import Message
+from multifile import MultiFile
 
-import string, cStringIO, re
-import ZODB, Acquisition
+from AccessControl import SecurityManager
+from AccessControl.SecurityManagement import newSecurityManager
+from AccessControl.SecurityManagement import noSecurityManager
+from Acquisition import Implicit
+from Acquisition import aq_base
 from OFS.Application import Application
 from OFS.Folder import manage_addFolder
 from OFS.Image import manage_addFile
 from Testing.makerequest import makerequest
 from webdav.common import rfc1123_date
-from AccessControl import SecurityManager
-from AccessControl.SecurityManagement import newSecurityManager
-from AccessControl.SecurityManagement import noSecurityManager
 
-from mimetools import Message
-from multifile import MultiFile
+
+ADD_IMAGES_AND_FILES = 'Add images and files'
+FILE_META_TYPES = ( { 'name'        : 'File'
+                    , 'action'      : 'manage_addFile'
+                    , 'permission'  : ADD_IMAGES_AND_FILES
+                    }
+                  ,
+                  )
 
 class UnitTestSecurityPolicy:
     """
@@ -35,7 +44,7 @@ class UnitTestSecurityPolicy:
     def checkPermission( self, permission, object, context) :
         return 1
 
-class UnitTestUser( Acquisition.Implicit ):
+class UnitTestUser( Implicit ):
     """
         Stubbed out manager for unit testing purposes.
     """
@@ -54,9 +63,9 @@ def makeConnection():
     s = DemoStorage(quota=(1<<20))
     return ZODB.DB( s ).open()
 
-class TestCopySupport( unittest.TestCase ):
+class CopySupportTestBase(unittest.TestCase):
 
-    def setUp( self ):
+    def _initFolders(self):
 
         self.connection = makeConnection()
         try:
@@ -71,14 +80,6 @@ class TestCopySupport( unittest.TestCase ):
             folder1 = getattr( self.app, 'folder1' )
             folder2 = getattr( self.app, 'folder2' )
 
-            folder1.all_meta_types = folder2.all_meta_types = \
-                                    ( { 'name'        : 'File'
-                                      , 'action'      : 'manage_addFile'
-                                      , 'permission'  : 'Add images and files'
-                                      }
-                                    ,
-                                    )
-
             manage_addFile( folder1, 'file'
                           , file='', content_type='text/plain')
 
@@ -90,27 +91,44 @@ class TestCopySupport( unittest.TestCase ):
             self.connection.close()
             raise
         get_transaction().begin()
-        self.folder1 = getattr( self.app, 'folder1' )
-        self.folder2 = getattr( self.app, 'folder2' )
+
+        return self.app._getOb( 'folder1' ), self.app._getOb( 'folder2' )
+
+    def _cleanApp( self ):
+
+        get_transaction().abort()
+        self.app._p_jar.sync()
+        self.connection.close()
+        del self.app
+        del self.responseOut
+        del self.root
+        del self.connection
+
+class TestCopySupport( CopySupportTestBase ):
+
+    def setUp( self ):
+
+        folder1, folder2 = self._initFolders()
+
+        folder1.all_meta_types = folder2.all_meta_types = FILE_META_TYPES
+
+        self.folder1 = folder1
+        self.folder2 = folder2
 
         self.policy = UnitTestSecurityPolicy()
         self.oldPolicy = SecurityManager.setSecurityPolicy( self.policy )
         newSecurityManager( None, UnitTestUser().__of__( self.root ) )
 
     def tearDown( self ):
+
         noSecurityManager()
         SecurityManager.setSecurityPolicy( self.oldPolicy )
         del self.oldPolicy
         del self.policy
         del self.folder2
         del self.folder1
-        get_transaction().abort()
-        self.app._p_jar.sync()
-        self.connection.close()
-        del self.app
-        del self.responseOut
-        del self.root
-        del self.connection
+
+        self._cleanApp()
 
     def testRename( self ):
         self.failUnless( 'file' in self.folder1.objectIds() )
@@ -219,10 +237,242 @@ class TestCopySupport( unittest.TestCase ):
                                     {'id':'file1', 'new_id':'copy_of_file1'},
                                     {'id':'file2', 'new_id':'copy_of_file2'}])
 
+class _SensitiveSecurityPolicy:
+
+    def __init__( self, validate_lambda, checkPermission_lambda ):
+        self._lambdas = ( validate_lambda, checkPermission_lambda )
+
+    def validate( self, *args, **kw ):
+        return self._lambdas[ 0 ]( *args, **kw )
+
+    def checkPermission( self, *args, **kw ) :
+        return self._lambdas[ 1 ]( *args, **kw )
+
+class _AllowedUser( UnitTestUser ):
+
+    def __init__( self, allowed_lambda ):
+        self._lambdas = ( allowed_lambda, )
+
+    def allowed( self, object, object_roles=None ):
+        return self._lambdas[ 0 ]( object, object_roles )
+
+class TestCopySupportSecurity( CopySupportTestBase ):
+
+    _old_policy = None
+
+    def setUp( self ):
+        self._scrubSecurity()
+
+    def tearDown( self ):
+
+        self._scrubSecurity()
+        self._cleanApp()
+
+    def _scrubSecurity( self ):
+
+        noSecurityManager()
+
+        if self._old_policy is not None:
+            SecurityManager.setSecurityPolicy( self._old_policy )
+
+    def _assertCopyErrorUnauth( self, callable, *args, **kw ):
+
+        import re
+        from zExceptions import Unauthorized
+        from OFS.CopySupport import CopyError
+
+        ce_regex = kw.get( 'ce_regex' )
+        if ce_regex is not None:
+            del kw[ 'ce_regex' ]
+
+        try:
+            callable( *args, **kw )
+
+        except CopyError, e:
+
+            if ce_regex is not None:
+                
+                pattern = re.compile( ce_regex, re.DOTALL )
+                if pattern.search( e ) is None:
+                    self.fail( "Paste failed; didn't match pattern:\n%s" % e )
+
+            else:
+                self.fail( "Paste failed; no pattern:\n%s" % e )
+
+        except Unauthorized, e:
+            pass
+
+        else:
+            self.fail( "Paste allowed unexpectedly." )
+
+    def _initPolicyAndUser( self    
+                          , a_lambda=None
+                          , v_lambda=None
+                          , c_lambda=None
+                          ):
+        def _promiscuous( *args, **kw ):
+            return 1
+
+        if a_lambda is None:
+            a_lambda = _promiscuous
+
+        if v_lambda is None:
+            v_lambda = _promiscuous
+
+        if c_lambda is None:
+            c_lambda = _promiscuous
+
+        scp = _SensitiveSecurityPolicy( v_lambda, c_lambda )
+        self._old_policy = SecurityManager.setSecurityPolicy( scp )
+
+        newSecurityManager( None
+                          , _AllowedUser( a_lambda ).__of__( self.root ) )
+
+    def test_copy_baseline( self ):
+
+        folder1, folder2 = self._initFolders()
+        folder2.all_meta_types = FILE_META_TYPES
+
+        self._initPolicyAndUser()
+
+        self.failUnless( 'file' in folder1.objectIds() )
+        self.failIf( 'file' in folder2.objectIds() )
+
+        cookie = folder1.manage_copyObjects( ids=( 'file', ) )
+        folder2.manage_pasteObjects( cookie )
+
+        self.failUnless( 'file' in folder1.objectIds() )
+        self.failUnless( 'file' in folder2.objectIds() )
+
+    def test_copy_cant_read_source( self ):
+
+        folder1, folder2 = self._initFolders()
+        folder2.all_meta_types = FILE_META_TYPES
+
+        a_file = folder1._getOb( 'file' )
+
+        def _validate( a, c, n, v, *args, **kw ):
+            return aq_base( v ) is not aq_base( a_file )
+
+        self._initPolicyAndUser( v_lambda=_validate )
+
+        cookie = folder1.manage_copyObjects( ids=( 'file', ) )
+        self._assertCopyErrorUnauth( folder2.manage_pasteObjects
+                                   , cookie
+                                   , ce_regex='Insufficient privileges'
+                                   )
+
+    def test_copy_cant_create_target_metatype_not_supported( self ):
+        
+        from OFS.CopySupport import CopyError
+
+        folder1, folder2 = self._initFolders()
+        folder2.all_meta_types = ()
+
+        self._initPolicyAndUser()
+
+        cookie = folder1.manage_copyObjects( ids=( 'file', ) )
+        self._assertCopyErrorUnauth( folder2.manage_pasteObjects
+                                   , cookie
+                                   , ce_regex='Not Supported'
+                                   )
+
+    def test_move_baseline( self ):
+
+        folder1, folder2 = self._initFolders()
+        folder2.all_meta_types = FILE_META_TYPES
+
+        self.failUnless( 'file' in folder1.objectIds() )
+        self.failIf( 'file' in folder2.objectIds() )
+
+        self._initPolicyAndUser()
+
+        cookie = folder1.manage_cutObjects( ids=( 'file', ) )
+        folder2.manage_pasteObjects( cookie )
+
+        self.failIf( 'file' in folder1.objectIds() )
+        self.failUnless( 'file' in folder2.objectIds() )
+
+    def test_move_cant_read_source( self ):
+        
+        from OFS.CopySupport import CopyError
+
+        folder1, folder2 = self._initFolders()
+        folder2.all_meta_types = FILE_META_TYPES
+
+        a_file = folder1._getOb( 'file' )
+
+        def _validate( a, c, n, v, *args, **kw ):
+            return aq_base( v ) is not aq_base( a_file )
+
+        self._initPolicyAndUser( v_lambda=_validate )
+
+        cookie = folder1.manage_cutObjects( ids=( 'file', ) )
+        self._assertCopyErrorUnauth( folder2.manage_pasteObjects
+                                   , cookie
+                                   , ce_regex='Insufficient privileges'
+                                   )
+
+    def test_move_cant_create_target_metatype_not_supported( self ):
+        
+        from OFS.CopySupport import CopyError
+
+        folder1, folder2 = self._initFolders()
+        folder2.all_meta_types = ()
+
+        self._initPolicyAndUser()
+
+        cookie = folder1.manage_cutObjects( ids=( 'file', ) )
+        self._assertCopyErrorUnauth( folder2.manage_pasteObjects
+                                   , cookie
+                                   , ce_regex='Not Supported'
+                                   )
+
+    def test_move_cant_create_target_metatype_not_allowed( self ):
+        
+        from OFS.CopySupport import CopyError
+
+        folder1, folder2 = self._initFolders()
+        folder2.all_meta_types = FILE_META_TYPES
+
+        def _no_manage_addFile( a, c, n, v, *args, **kw ):
+            return n != 'manage_addFile'
+
+        self._initPolicyAndUser( v_lambda=_no_manage_addFile )
+
+        cookie = folder1.manage_cutObjects( ids=( 'file', ) )
+        self._assertCopyErrorUnauth( folder2.manage_pasteObjects
+                                   , cookie
+                                   , ce_regex='Insufficient Privileges'
+                                             + '.*%s' % ADD_IMAGES_AND_FILES
+                                   )
+
+    def test_move_cant_delete_source( self ):
+        
+        from OFS.CopySupport import CopyError
+        from AccessControl.Permissions import delete_objects as DeleteObjects
+
+        folder1, folder2 = self._initFolders()
+        folder1.manage_permission( DeleteObjects, roles=(), acquire=0 )
+        folder2.all_meta_types = FILE_META_TYPES
+
+        def _no_delete_objects(permission, object, context):
+            return permission != DeleteObjects
+
+        self._initPolicyAndUser( c_lambda=_no_delete_objects )
+
+        cookie = folder1.manage_cutObjects( ids=( 'file', ) )
+        self._assertCopyErrorUnauth( folder2.manage_pasteObjects
+                                   , cookie
+                                   , ce_regex='Insufficient Privileges'
+                                             + '.*%s' % DeleteObjects
+                                   )
+
 
 def test_suite():
     suite = unittest.TestSuite()
     suite.addTest( unittest.makeSuite( TestCopySupport ) )
+    suite.addTest( unittest.makeSuite( TestCopySupportSecurity ) )
     return suite
 
 def main():