CVE-2013-4251

@@ -34,6 +34,7 @@
 
 import os
 import sys
+import stat
 import pickle
 import socket
 import tempfile
@@ -133,7 +134,7 @@ def is_writable(dir):
 
     # Do NOT use a hardcoded name here due to the danger from race conditions
     # on NFS when multiple processes are accessing the same base directory in
-    # parallel.  We use both hostname and pocess id for the prefix in an
+    # parallel.  We use both hostname and process id for the prefix in an
     # attempt to ensure that there can really be no name collisions (tempfile
     # appends 6 random chars to this prefix).
     prefix = 'dummy_%s_%s_' % (socket.gethostname(),os.getpid())
@@ -150,6 +151,88 @@ def whoami():
     """return a string identifying the user."""
     return os.environ.get("USER") or os.environ.get("USERNAME") or "unknown"
 
+
+def _create_dirs(path):
+    """ create provided path, ignore errors """
+    try:
+        os.makedirs(path, mode=0o700)
+    except OSError:
+        pass
+
+
+def default_dir_posix(tmp_dir=None):
+    """
+    Create or find default catalog store for posix systems
+
+    purpose of 'tmp_dir' is to enable way how to test this function easily
+    """
+    path_candidates = []
+    python_name = "python%d%d_compiled" % tuple(sys.version_info[:2])
+
+    if tmp_dir:
+        home_dir = tmp_dir
+    else:
+        home_dir = os.path.expanduser('~')
+    tmp_dir = tmp_dir or tempfile.gettempdir()
+
+    home_temp_dir_name = '.' + python_name
+    home_temp_dir = os.path.join(home_dir, home_temp_dir_name)
+    path_candidates.append(home_temp_dir)
+
+    temp_dir_name = repr(os.getuid()) + '_' + python_name
+    temp_dir_path = os.path.join(tmp_dir, temp_dir_name)
+    path_candidates.append(temp_dir_path)
+
+    for path in path_candidates:
+        _create_dirs(path)
+        if check_dir(path):
+            return path
+
+    # since we got here, both dirs are not useful
+    tmp_dir_path = find_valid_temp_dir(temp_dir_name, tmp_dir)
+    if not tmp_dir_path:
+        tmp_dir_path = create_temp_dir(temp_dir_name, tmp_dir=tmp_dir)
+    return tmp_dir_path
+
+
+def default_dir_win(tmp_dir=None):
+    """
+    Create or find default catalog store for Windows systems
+
+    purpose of 'tmp_dir' is to enable way how to test this function easily
+    """
+    def create_win_temp_dir(prefix, inner_dir=None, tmp_dir=None):
+        """
+        create temp dir starting with 'prefix' in 'tmp_dir' or
+        'tempfile.gettempdir'; if 'inner_dir' is specified, it should be
+        created inside
+        """
+        tmp_dir_path = find_valid_temp_dir(prefix, tmp_dir)
+        if tmp_dir_path:
+            if inner_dir:
+                tmp_dir_path = os.path.join(tmp_dir_path, inner_dir)
+                if not os.path.isdir(tmp_dir_path):
+                    os.mkdir(tmp_dir_path, 0o700)
+        else:
+            tmp_dir_path = create_temp_dir(prefix, inner_dir, tmp_dir)
+        return tmp_dir_path
+
+    python_name = "python%d%d_compiled" % tuple(sys.version_info[:2])
+    tmp_dir = tmp_dir or tempfile.gettempdir()
+
+    temp_dir_name = "%s" % whoami()
+    temp_root_dir = os.path.join(tmp_dir, temp_dir_name)
+    temp_dir_path = os.path.join(temp_root_dir, python_name)
+    _create_dirs(temp_dir_path)
+    if check_dir(temp_dir_path) and check_dir(temp_root_dir):
+        return temp_dir_path
+    else:
+        if check_dir(temp_root_dir):
+            return create_win_temp_dir(python_name, tmp_dir=temp_root_dir)
+        else:
+            return create_win_temp_dir(temp_dir_name, python_name, tmp_dir)
+
+
 def default_dir():
     """ Return a default location to store compiled files and catalogs.
 
@@ -164,61 +247,151 @@ def default_dir():
         in the user's home, /tmp/<uid>_pythonXX_compiled is used.  If it
         doesn't exist, it is created.  The directory is marked rwx------
         to try and keep people from being able to sneak a bad module
-        in on you.
-
+        in on you. If the directory already exists in /tmp/ and is not
+        secure, new one is created.
     """
-
     # Use a cached value for fast return if possible
-    if hasattr(default_dir,"cached_path") and \
-       os.path.exists(default_dir.cached_path) and \
-       os.access(default_dir.cached_path, os.W_OK):
+    if hasattr(default_dir, "cached_path") and \
+       check_dir(default_dir.cached_path):
         return default_dir.cached_path
 
-    python_name = "python%d%d_compiled" % tuple(sys.version_info[:2])
-    path_candidates = []
-    if sys.platform != 'win32':
-        try:
-            path_candidates.append(os.path.join(os.environ['HOME'],
-                                                '.' + python_name))
-        except KeyError:
-            pass
-
-        temp_dir = repr(os.getuid()) + '_' + python_name
-        path_candidates.append(os.path.join(tempfile.gettempdir(), temp_dir))
+    if sys.platform == 'win32':
+        path = default_dir_win()
     else:
-        path_candidates.append(os.path.join(tempfile.gettempdir(),
-                                           "%s" % whoami(), python_name))
-
-    writable = False
-    for path in path_candidates:
-        if not os.path.exists(path):
-            try:
-                os.makedirs(path, mode=0o700)
-            except OSError:
-                continue
-        if is_writable(path):
-            writable = True
-            break
-
-    if not writable:
-        print('warning: default directory is not write accessible.')
-        print('default:', path)
+        path = default_dir_posix()
 
     # Cache the default dir path so that this function returns quickly after
     # being called once (nothing in it should change after the first call)
     default_dir.cached_path = path
 
     return path
 
-def intermediate_dir():
-    """ Location in temp dir for storing .cpp and .o  files during
-        builds.
+
+def check_dir(im_dir):
     """
-    python_name = "python%d%d_intermediate" % tuple(sys.version_info[:2])
-    path = os.path.join(tempfile.gettempdir(),"%s"%whoami(),python_name)
-    if not os.path.exists(path):
-        os.makedirs(path, mode=0o700)
-    return path
+    Check if dir is safe; if it is, return True.
+    These checks make sense only on posix:
+     * directory has correct owner
+     * directory has correct permissions (0700)
+     * directory is not a symlink
+    """
+    def check_is_dir():
+        return os.path.isdir(im_dir)
+
+    def check_permissions():
+        """ If on posix, permissions should be 0700. """
+        writable = is_writable(im_dir)
+        if sys.platform != 'win32':
+            try:
+                im_dir_stat = os.stat(im_dir)
+            except OSError:
+                return False
+            writable &= stat.S_IMODE(im_dir_stat.st_mode) == 0o0700
+        return writable
+
+    def check_ownership():
+        """ Intermediate dir owner should be same as owner of process. """
+        if sys.platform != 'win32':
+            try:
+                im_dir_stat = os.stat(im_dir)
+            except OSError:
+                return False
+            proc_uid = os.getuid()
+            return proc_uid == im_dir_stat.st_uid
+        return True
+
+    def check_is_symlink():
+        """ Check if intermediate dir is symlink. """
+        try:
+            return not os.path.islink(im_dir)
+        except OSError:
+            return False
+
+    checks = [check_is_dir, check_permissions,
+              check_ownership, check_is_symlink]
+
+    for check in checks:
+        if not check():
+            return False
+
+    return True
+
+
+def create_temp_dir(prefix, inner_dir=None, tmp_dir=None):
+    """
+    Create intermediate dirs <tmp>/<prefix+random suffix>/<inner_dir>/
+
+    argument 'tmp_dir' is used in unit tests
+    """
+    if not tmp_dir:
+        tmp_dir_path = tempfile.mkdtemp(prefix=prefix)
+    else:
+        tmp_dir_path = tempfile.mkdtemp(prefix=prefix, dir=tmp_dir)
+    if inner_dir:
+        tmp_dir_path = os.path.join(tmp_dir_path, inner_dir)
+        os.mkdir(tmp_dir_path, 0o700)
+    return tmp_dir_path
+
+
+def intermediate_dir_prefix():
+    """ Prefix of root intermediate dir (<tmp>/<root_im_dir>). """
+    return "%s-%s-" % ("scipy", whoami())
+
+
+def find_temp_dir(prefix, tmp_dir=None):
+    """ Find temp dirs in 'tmp_dir' starting with 'prefix'"""
+    matches = []
+    tmp_dir = tmp_dir or tempfile.gettempdir()
+    for tmp_file in os.listdir(tmp_dir):
+        if tmp_file.startswith(prefix):
+            matches.append(os.path.join(tmp_dir, tmp_file))
+    return matches
+
+
+def find_valid_temp_dir(prefix, tmp_dir=None):
+    """
+    Try to look for existing temp dirs.
+    If there is one suitable found, return it, otherwise return None.
+    """
+    matches = find_temp_dir(prefix, tmp_dir)
+    for match in matches:
+        if check_dir(match):
+            # as soon as we find correct dir, we can stop searching
+            return match
+
+
+def py_intermediate_dir():
+    """
+    Name of intermediate dir for current python interpreter:
+    <temp dir>/<name>/pythonXY_intermediate/
+    """
+    name = "python%d%d_intermediate" % tuple(sys.version_info[:2])
+    return name
+
+
+def create_intermediate_dir(tmp_dir=None):
+    py_im_dir = py_intermediate_dir()
+    return create_temp_dir(intermediate_dir_prefix(), py_im_dir, tmp_dir)
+
+
+def intermediate_dir(tmp_dir=None):
+    """
+    Temporary directory for storing .cpp and .o files during builds.
+
+    First, try to find the dir and if it exists, verify it is safe.
+    Otherwise, create it.
+    """
+    im_dir = find_valid_temp_dir(intermediate_dir_prefix(), tmp_dir)
+    py_im_dir = py_intermediate_dir()
+    if im_dir is None:
+        py_im_dir = py_intermediate_dir()
+        im_dir = create_intermediate_dir(tmp_dir)
+    else:
+        im_dir = os.path.join(im_dir, py_im_dir)
+        if not os.path.isdir(im_dir):
+            os.mkdir(im_dir, 0o700)
+    return im_dir
+
 
 def default_temp_dir():
     path = os.path.join(default_dir(),'temp')

@@ -2,16 +2,234 @@
 
 import sys
 import os
+import stat
+import tempfile
+
+from distutils.dir_util import remove_tree
 
 from numpy.testing import TestCase, assert_
+from numpy.testing.noseclasses import KnownFailureTest
 
 from scipy.weave import catalog
 from weave_test_utils import clear_temp_catalog, restore_temp_catalog, \
         empty_temp_dir, cleanup_temp_dir
 
 
+class TestIntermediateDir(TestCase):
+    """
+    Tests for intermediate dir (store of .cpp and .o during builds).
+    These tests test whether intermediate dir is safe. If it's not,
+    new one should be created.
+    """
+    def dirs_are_valid(self, wrong_dir, tmpdir):
+        """ test if new dir is created and is consistent """
+        new_im_dir = catalog.intermediate_dir(tmpdir)
+        assert_(not os.path.samefile(new_im_dir, wrong_dir))
+        new_im_dir2 = catalog.intermediate_dir(tmpdir)
+        assert_(os.path.samefile(new_im_dir, new_im_dir2))
+
+    def test_ownership(self):
+        """ test if intermediate dir is owned by correct user """
+        if sys.platform != 'win32':
+            im_dir = catalog.intermediate_dir()
+            im_dir_stat = os.stat(im_dir)
+            proc_uid = os.getuid()
+            assert_(proc_uid == im_dir_stat.st_uid)
+            r_im_dir_stat = os.stat(os.path.dirname(im_dir))
+            assert_(proc_uid == r_im_dir_stat.st_uid)
+
+    def test_incorrect_ownership(self):
+        """
+        test if new intermediate dir is created when there is only one
+        im dir owned by improper user
+        """
+        if sys.platform != 'win32':
+            import pwd
+            tmpdir = tempfile.mkdtemp()
+            try:
+                im_dir = catalog.create_intermediate_dir(tmpdir)
+                root_im_dir = os.path.dirname(im_dir)
+                nobody = pwd.getpwnam('nobody')[2]
+                nobody_g = pwd.getpwnam('nobody')[3]
+                try:
+                    os.chown(root_im_dir, nobody, nobody_g)
+                except OSError:
+                    raise KnownFailureTest("Can't change owner.")
+                else:
+                    self.dirs_are_valid(im_dir, tmpdir)
+            finally:
+                remove_tree(tmpdir)
+
+    def test_permissions(self):
+        """ im dir should have permissions 0700 """
+        if sys.platform != 'win32':
+            im_dir = catalog.intermediate_dir()
+            im_dir_stat = os.stat(im_dir)
+            assert_(stat.S_IMODE(im_dir_stat.st_mode) == 0o0700)
+            r_im_dir_stat = os.stat(os.path.dirname(im_dir))
+            assert_(stat.S_IMODE(r_im_dir_stat.st_mode) == 0o0700)
+
+    def test_incorrect_permissions(self):
+        """
+        if permissions on existing im dir are not correct,
+        new one should be created
+        """
+        if sys.platform != 'win32':
+            tmpdir = tempfile.mkdtemp()
+            try:
+                im_dir = catalog.create_intermediate_dir(tmpdir)
+                root_im_dir = os.path.dirname(im_dir)
+                try:
+                    os.chmod(root_im_dir, 0o777)
+                except OSError:
+                    raise KnownFailureTest("Can't set file permissions.")
+                else:
+                    self.dirs_are_valid(im_dir, tmpdir)
+            finally:
+                remove_tree(tmpdir)
+
+    def test_symlink(self):
+        """ im dir shouldn't be a symlink """
+        if sys.platform != 'win32':
+            r_im_dir = os.path.dirname(catalog.intermediate_dir())
+            assert_(os.path.islink(r_im_dir) is False)
+
+    def test_symlink_raise(self):
+        """ if existing im dir is a symlink, new one should be created """
+        if sys.platform != 'win32':
+            tmpdir = tempfile.mkdtemp()
+            try:
+                im_dir = catalog.create_intermediate_dir(tmpdir)
+                root_im_dir = os.path.dirname(im_dir)
+
+                tempdir = tempfile.mkdtemp(prefix='scipy-test', dir=tmpdir)
+                try:
+                    os.rename(root_im_dir, tempdir)
+                except OSError:
+                    raise KnownFailureTest("Can't move intermediate dir.")
+
+                try:
+                    os.symlink(tempdir, root_im_dir)
+                except OSError:
+                    raise KnownFailureTest(
+                        "Can't create symlink to intermediate dir.")
+                else:
+                    self.dirs_are_valid(im_dir, tmpdir)
+            finally:
+                remove_tree(tmpdir)
+
+
 class TestDefaultDir(TestCase):
+    """
+    Tests for 'catalog.default_dir()'.
+    These should verified posix and win default_dir function.
+    """
+    def test_win(self):
+        """
+        test if default_dir for Windows platform is accessible
+
+        since default_dir_win() does not have any Windows specific code,
+        let's test it everywhere
+        """
+        d = catalog.default_dir_win()
+        assert_(catalog.is_writable(d))
+
+    def test_win_inaccessible_root(self):
+        """
+        there should be a new root dir created if existing one is not accessible
+        """
+        tmpdir = tempfile.mkdtemp()
+        try:
+            d_dir = catalog.default_dir_win(tmpdir)
+            root_ddir = os.path.dirname(d_dir)
+
+            try:
+                os.chmod(root_ddir, stat.S_IREAD | stat.S_IEXEC)
+            except OSError:
+                raise KnownFailureTest("Can't change permissions of root default_dir.")
+
+            new_ddir = catalog.default_dir_win(tmpdir)
+            assert_(not os.path.samefile(new_ddir, d_dir))
+            new_ddir2 = catalog.default_dir_win(tmpdir)
+            assert_(os.path.samefile(new_ddir, new_ddir2))
+        finally:
+            os.chmod(root_ddir, 0o700)
+            remove_tree(tmpdir)
+
+    def test_win_inaccessible_ddir(self):
+        """
+        create new defualt_dir if current one is not accessible
+        """
+        tmpdir = tempfile.mkdtemp()
+        try:
+            d_dir = catalog.default_dir_win(tmpdir)
+
+            try:
+                os.chmod(d_dir, stat.S_IREAD | stat.S_IEXEC)
+            except OSError:
+                raise KnownFailureTest("Can't change permissions of default_dir.")
+
+            new_ddir = catalog.default_dir_win(tmpdir)
+            assert_(not os.path.samefile(new_ddir, d_dir))
+            new_ddir2 = catalog.default_dir_win(tmpdir)
+            assert_(os.path.samefile(new_ddir, new_ddir2))
+        finally:
+            os.chmod(d_dir, 0o700)
+            remove_tree(tmpdir)
+
+    def test_posix(self):
+        """ test if posix default_dir is writable """
+        d = catalog.default_dir_posix()
+        assert_(catalog.is_writable(d))
+
+    def test_posix_home_inaccessible(self):
+        """ what happens when home catalog dir is innaccessible """
+        tmpdir = tempfile.mkdtemp()
+        try:
+            d_dir = catalog.default_dir_posix(tmpdir)
+
+            try:
+                os.chmod(d_dir, 0o000)
+            except OSError:
+                raise KnownFailureTest("Can't change permissions of default_dir.")
+
+            new_ddir = catalog.default_dir_posix(tmpdir)
+            assert_(not os.path.samefile(new_ddir, d_dir))
+            new_ddir2 = catalog.default_dir_posix(tmpdir)
+            assert_(os.path.samefile(new_ddir, new_ddir2))
+        finally:
+            os.chmod(d_dir, 0o700)
+            remove_tree(tmpdir)
+
+    def test_posix_dirs_inaccessible(self):
+        """ test if new dir is created if both implicit dirs are not valid"""
+        tmpdir = tempfile.mkdtemp()
+        try:
+            d_dir = catalog.default_dir_posix(tmpdir)
+
+            try:
+                os.chmod(d_dir, 0o000)
+            except OSError:
+                raise KnownFailureTest("Can't change permissions of default_dir.")
+
+            d_dir2 = catalog.default_dir_posix(tmpdir)
+
+            try:
+                os.chmod(d_dir2, 0o000)
+            except OSError:
+                raise KnownFailureTest("Can't change permissions of default_dir.")
+
+            new_ddir = catalog.default_dir_posix(tmpdir)
+            assert_(not (os.path.samefile(new_ddir, d_dir) or os.path.samefile(new_ddir, d_dir2)))
+            new_ddir2 = catalog.default_dir_posix(tmpdir)
+            assert_(os.path.samefile(new_ddir, new_ddir2))
+        finally:
+            os.chmod(d_dir, 0o700)
+            os.chmod(d_dir2, 0o700)
+            remove_tree(tmpdir)
+
     def test_is_writable(self):
+        """ default_dir has to be writable """
         path = catalog.default_dir()
         name = os.path.join(path,'dummy_catalog')
         test_file = open(name,'w')
@@ -93,22 +311,18 @@ def get_test_dir(self,erase = 0):
                 os.remove(cat_file)
         return pardir
 
-    def remove_dir(self,d):
-        import distutils.dir_util
-        distutils.dir_util.remove_tree(d)
-
     def test_nonexistent_catalog_is_none(self):
         pardir = self.get_test_dir(erase=1)
         cat = catalog.get_catalog(pardir,'r')
-        self.remove_dir(pardir)
+        remove_tree(pardir)
         assert_(cat is None)
 
     def test_create_catalog(self):
         pardir = self.get_test_dir(erase=1)
         cat = catalog.get_catalog(pardir,'c')
         assert_(cat is not None)
         cat.close()
-        self.remove_dir(pardir)
+        remove_tree(pardir)
 
 
 class TestCatalog(TestCase):