CVE-2012-3447

@@ -189,6 +189,10 @@
     # nova/virt/libvirt/utils.py: 'qemu-img'
     filters.CommandFilter("/usr/bin/qemu-img", "root"),
 
+    # nova/virt/disk/api.py: 'readlink', '-e'
+    filters.CommandFilter("/usr/bin/readlink", "root"),
+    filters.CommandFilter("/bin/readlink", "root"),
+
     # nova/virt/disk/api.py: 'touch', target
     filters.CommandFilter("/usr/bin/touch", "root"),
 

@@ -18,6 +18,7 @@
 from nova import exception
 from nova import flags
 from nova import test
+from nova import utils
 from nova.virt.disk import api as disk_api
 from nova.virt import driver
 
@@ -86,6 +87,17 @@ def test_swap_is_usable(self):
 
 
 class TestVirtDisk(test.TestCase):
+    def setUp(self):
+        super(TestVirtDisk, self).setUp()
+
+        real_execute = utils.execute
+
+        def nonroot_execute(*cmd_parts, **kwargs):
+            kwargs.pop('run_as_root', None)
+            return real_execute(*cmd_parts, **kwargs)
+
+        self.stubs.Set(utils, 'execute', nonroot_execute)
+
     def test_check_safe_path(self):
         ret = disk_api._join_and_check_path_within_fs('/foo', 'etc',
                                                       'something.conf')

@@ -597,9 +597,13 @@ def _tee_handler(cmd, **kwargs):
             self._tee_executed = True
             return '', ''
 
+        def _readlink_handler(cmd_parts, **kwargs):
+            return os.path.realpath(cmd_parts[2]), ''
+
         fake_utils.fake_execute_set_repliers([
             # Capture the tee .../etc/network/interfaces command
             (r'tee.*interfaces', _tee_handler),
+            (r'readlink -nm.*', _readlink_handler),
         ])
         self._test_spawn(glance_stubs.FakeGlance.IMAGE_MACHINE,
                          glance_stubs.FakeGlance.IMAGE_KERNEL,

@@ -314,7 +314,9 @@ def _join_and_check_path_within_fs(fs, *args):
     mounted guest fs.  Trying to be clever and specifying a
     path with '..' in it will hit this safeguard.
     '''
-    absolute_path = os.path.realpath(os.path.join(fs, *args))
+    absolute_path, _err = utils.execute('readlink', '-nm',
+                                        os.path.join(fs, *args),
+                                        run_as_root=True)
     if not absolute_path.startswith(os.path.realpath(fs) + '/'):
         raise exception.Invalid(_('injected file path not valid'))
     return absolute_path

@@ -99,22 +99,6 @@ def fake_execute(*cmd, **kwargs):
 
         self.stubs.Set(utils, 'execute', fake_execute)
 
-    def test_check_safe_path(self):
-        ret = disk_api._join_and_check_path_within_fs('/foo', 'etc',
-                                                      'something.conf')
-        self.assertEquals(ret, '/foo/etc/something.conf')
-
-    def test_check_unsafe_path(self):
-        self.assertRaises(exception.Invalid,
-                          disk_api._join_and_check_path_within_fs,
-                          '/foo', 'etc/../../../something.conf')
-
-    def test_inject_files_with_bad_path(self):
-        self.assertRaises(exception.Invalid,
-                          disk_api._inject_file_into_fs,
-                          '/tmp', '/etc/../../../../etc/passwd',
-                          'hax')
-
     def test_lxc_destroy_container(self):
 
         def proc_mounts(self, mount_point):
@@ -165,3 +149,32 @@ def proc_mounts(self, mount_point):
         self.executes.pop()
 
         self.assertEqual(self.executes, expected_commands)
+
+
+class TestVirtDiskPaths(test.TestCase):
+    def setUp(self):
+        super(TestVirtDiskPaths, self).setUp()
+
+        real_execute = utils.execute
+
+        def nonroot_execute(*cmd_parts, **kwargs):
+            kwargs.pop('run_as_root', None)
+            return real_execute(*cmd_parts, **kwargs)
+
+        self.stubs.Set(utils, 'execute', nonroot_execute)
+
+    def test_check_safe_path(self):
+        ret = disk_api._join_and_check_path_within_fs('/foo', 'etc',
+                                                      'something.conf')
+        self.assertEquals(ret, '/foo/etc/something.conf')
+
+    def test_check_unsafe_path(self):
+        self.assertRaises(exception.Invalid,
+                          disk_api._join_and_check_path_within_fs,
+                          '/foo', 'etc/../../../something.conf')
+
+    def test_inject_files_with_bad_path(self):
+        self.assertRaises(exception.Invalid,
+                          disk_api._inject_file_into_fs,
+                          '/tmp', '/etc/../../../../etc/passwd',
+                          'hax')

@@ -677,9 +677,13 @@ def _tee_handler(cmd, **kwargs):
             self._tee_executed = True
             return '', ''
 
+        def _readlink_handler(cmd_parts, **kwargs):
+            return os.path.realpath(cmd_parts[2]), ''
+
         fake_utils.fake_execute_set_repliers([
             # Capture the tee .../etc/network/interfaces command
             (r'tee.*interfaces', _tee_handler),
+            (r'readlink -nm.*', _readlink_handler),
         ])
         self._test_spawn(IMAGE_MACHINE,
                          IMAGE_KERNEL,

@@ -363,7 +363,9 @@ def _join_and_check_path_within_fs(fs, *args):
     mounted guest fs.  Trying to be clever and specifying a
     path with '..' in it will hit this safeguard.
     '''
-    absolute_path = os.path.realpath(os.path.join(fs, *args))
+    absolute_path, _err = utils.execute('readlink', '-nm',
+                                        os.path.join(fs, *args),
+                                        run_as_root=True)
     if not absolute_path.startswith(os.path.realpath(fs) + '/'):
         raise exception.Invalid(_('injected file path not valid'))
     return absolute_path