shutil.rmtree is vulnerable to a symlink attack

Lib/test/test_shutil.py

 # Copyright (C) 2003 Python Software Foundation
 
 import unittest
 import shutil
 import tempfile
 import sys
 import stat
 import os
 import os.path
 import functools
 from test import support
 from test.support import TESTFN
 from os.path import splitdrive
 from distutils.spawn import find_executable, spawn
 from shutil import (_make_tarball, _make_zipfile, make_archive,
                     register_archive_format, unregister_archive_format,
                     get_archive_formats, Error, unpack_archive,
                     register_unpack_format, RegistryError,
                     unregister_unpack_format, get_unpack_formats)
 import tarfile
 import warnings
+import time
+import errno
+try:
+    import threading
+except ImportError:
+    threading = None
 
 from test import support
 from test.support import TESTFN, check_warnings, captured_stdout, requires_zlib
 
 try:
     import bz2
     BZ2_SUPPORTED = True
 except ImportError:
     BZ2_SUPPORTED = False
 
@@ skipping 113 matching lines @@
         # 99.9% of the time it initially fails to remove
         # a file in the directory, so the first time through
         # func is os.remove.
         # However, some Linux machines running ZFS on
         # FUSE experienced a failure earlier in the process
         # at os.listdir.  The first failure may legally
         # be either.
         if self.errorState == 0:
             if func is os.remove:
                 self.assertEqual(arg, self.childpath)
+            elif func is os.unlinkat:
+                self.assertEqual(arg, (3, 'a'))
             else:
                 self.assertIs(func, os.listdir,
                               "func must be either os.remove or os.listdir")
                 self.assertEqual(arg, TESTFN)
             self.assertTrue(issubclass(exc[0], OSError))
             self.errorState = 1
         else:
             self.assertEqual(func, os.rmdir)
             self.assertEqual(arg, TESTFN)
             self.assertTrue(issubclass(exc[0], OSError))
@@ skipping 135 matching lines @@
         # bug 1669.
         os.mkdir(TESTFN)
         try:
             src = os.path.join(TESTFN, 'cheese')
             dst = os.path.join(TESTFN, 'shop')
             os.mkdir(src)
             os.symlink(src, dst)
             self.assertRaises(OSError, shutil.rmtree, dst)
         finally:
             shutil.rmtree(TESTFN, ignore_errors=True)
+
+    @unittest.skipIf(threading == None, 'requires threading')

[eric.araujo, 2011/10/07 19:29:47]

You can just say skipUnless(threading, 'msg')

[rosslagerwall, 2011/10/07 21:18:46]

Right.

+    def test_rmtree_4489(self):
+        # Issue #4489: test a symlink attack on rmtree
+
+        def hack(a, b):
+            time.sleep(0.1)
+            os.rename(b, b + 'x')
+            self.tempdirs.append(b + 'x')
+            os.symlink(a, b)
+
+        a = self.mkdtemp()
+        b = self.mkdtemp()
+
+        for i in range(25000):
+            open(os.path.join(a, str(i)), "w").close()
+            open(os.path.join(b, str(i)), "w").close()
+
+        l = sorted(os.listdir(a))
+            
+        hackt = threading.Thread(target=hack, args=(a, b))
+        hackt.start()
+        try:
+            shutil.rmtree(b)
+        except OSError as inst:
+            if inst.errno != errno.ENOTDIR:
+                raise

[eric.araujo, 2011/10/07 19:29:47]

Shouldn’t this use self.fail?

[rosslagerwall, 2011/10/07 21:18:46]

I would have thought it's better if the original exception is passed through and
displayed rather than some sort of failure message that just says "OSError
occurred".

+        finally:
+            support.unlink(b)
+            self.tempdirs.remove(b)
+        if hasattr(os, 'openat') and hasattr(os, 'fdlistdir') and hasattr(os,
+                'unlinkat') and hasattr(os, 'fstatat'):
+            self.assertEqual(l, sorted(os.listdir(a)))
+        
 
     if hasattr(os, "mkfifo"):
         # Issue #3002: copyfile and copytree block indefinitely on named pipes
         def test_copyfile_named_pipe(self):
             os.mkfifo(TESTFN)
             try:
                 self.assertRaises(shutil.SpecialFileError,
                                   shutil.copyfile, TESTFN, TESTFN2)
                 self.assertRaises(shutil.SpecialFileError,
                                   shutil.copyfile, __file__, TESTFN)
@@ skipping 684 matching lines @@
         finally:
             os.rmdir(dst_dir)
 
 
 
 def test_main():
     support.run_unittest(TestShutil, TestMove, TestCopyFile)
 
 if __name__ == '__main__':
     test_main()