CVE-2012-3360: Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal') (CWE-22)
value: Directory traversal vulnerability in virt/disk/api.py in OpenStack Compute (Nova) Folsom (2012.2) and Essex (2012.1), when used over libvirt-based hypervisors, allows remote authenticated users to write arbitrary files to the disk image via a .. (dot dot) in the path attribute of a file element.
CWE description: The product uses external input to construct a pathname that is intended to identify a file or directory that is located underneath a restricted parent directory, but the product does not properly neutralize special elements within the pathname that can cause the pathname to resolve to a location that is outside of the restricted directory.
Published: 2012-07-22T16:55Z
Repository: https://github.com/openstack/nova
Language: Python
Files changed: nova/tests/test_virt.py, nova/virt/disk/api.py
Commit date: 2012-07-03 16:24:55 +0200
Commit message: Prevent file injection writing to host filesystem.

Fix bug 1015531, CVE-2012-3360, CVE-2012-3361

This patch prevents the file injection code from writing into the host
filesystem if a user specifies a path for the injected file that
contains '..'.  The check is to make sure that the final normalized path
that is about to be written to is within the mounted guest filesystem.

Signed-off-by: Russell Bryant <rbryant@redhat.com>
Signed-off-by: Pádraig Brady <pbrady@redhat.com>
Signed-off-by: Mark McLoughlin <markmc@redhat.com>

Change-Id: I658cd12fd319cee91eb9544cdf53c862c5d2c560

Diff: diff --git a/nova/tests/test_virt.py b/nova/tests/test_virt.py
index 388f075afb..c4aa828ad0 100644
--- a/nova/tests/test_virt.py
+++ b/nova/tests/test_virt.py
@@ -13,12 +13,14 @@
 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from nova import exception
 from nova import flags
 from nova import test
+from nova.virt.disk import api as disk_api
 from nova.virt import driver
 
 FLAGS = flags.FLAGS
 
 
@@ -79,5 +81,23 @@ class TestVirtDriver(test.TestCase):
         self.assertFalse(driver.swap_is_usable({'device_name': None}))
         self.assertFalse(driver.swap_is_usable({'device_name': '/dev/sdb',
                                                 'swap_size': 0}))
         self.assertTrue(driver.swap_is_usable({'device_name': '/dev/sdb',
                                                 'swap_size': 1}))
+
+
+class TestVirtDisk(test.TestCase):
+    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')
diff --git a/nova/virt/disk/api.py b/nova/virt/disk/api.py
index 42d9d91626..3e66cd2a7b 100644
--- a/nova/virt/disk/api.py
+++ b/nova/virt/disk/api.py
@@ -309,59 +309,81 @@ def inject_data_into_fs(fs, key, net, metadata, admin_password, execute):
         _inject_metadata_into_fs(metadata, fs, execute=execute)
     if admin_password:
         _inject_admin_password_into_fs(admin_password, fs, execute=execute)
 
 
-def _inject_file_into_fs(fs, path, contents):
-    absolute_path = os.path.join(fs, path.lstrip('/'))
+def _join_and_check_path_within_fs(fs, *args):
+    '''os.path.join() with safety check for injected file paths.
+
+    Join the supplied path components and make sure that the
+    resulting path we are injecting into is within the
+    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))
+    if not absolute_path.startswith(os.path.realpath(fs) + '/'):
+        raise exception.Invalid(_('injected file path not valid'))
+    return absolute_path
+
+
+def _inject_file_into_fs(fs, path, contents, append=False):
+    absolute_path = _join_and_check_path_within_fs(fs, path.lstrip('/'))
+
     parent_dir = os.path.dirname(absolute_path)
     utils.execute('mkdir', '-p', parent_dir, run_as_root=True)
-    utils.execute('tee', absolute_path, process_input=contents,
-          run_as_root=True)
+
+    args = []
+    if append:
+        args.append('-a')
+    args.append(absolute_path)
+
+    kwargs = dict(process_input=contents, run_as_root=True)
+
+    utils.execute('tee', *args, **kwargs)
 
 
 def _inject_metadata_into_fs(metadata, fs, execute=None):
-    metadata_path = os.path.join(fs, "meta.js")
     metadata = dict([(m.key, m.value) for m in metadata])
-
-    utils.execute('tee', metadata_path,
-                  process_input=jsonutils.dumps(metadata), run_as_root=True)
+    _inject_file_into_fs(fs, 'meta.js', jsonutils.dumps(metadata))
 
 
 def _inject_key_into_fs(key, fs, execute=None):
     """Add the given public ssh key to root's authorized_keys.
 
     key is an ssh key string.
     fs is the path to the base of the filesystem into which to inject the key.
     """
-    sshdir = os.path.join(fs, 'root', '.ssh')
+    sshdir = _join_and_check_path_within_fs(fs, 'root', '.ssh')
     utils.execute('mkdir', '-p', sshdir, run_as_root=True)
     utils.execute('chown', 'root', sshdir, run_as_root=True)
     utils.execute('chmod', '700', sshdir, run_as_root=True)
-    keyfile = os.path.join(sshdir, 'authorized_keys')
-    key_data = [
+
+    keyfile = os.path.join('root', '.ssh', 'authorized_keys')
+
+    key_data = ''.join([
         '\n',
         '# The following ssh key was injected by Nova',
         '\n',
         key.strip(),
         '\n',
-    ]
-    utils.execute('tee', '-a', keyfile,
-                  process_input=''.join(key_data), run_as_root=True)
+    ])
+
+    _inject_file_into_fs(fs, keyfile, key_data, append=True)
 
 
 def _inject_net_into_fs(net, fs, execute=None):
     """Inject /etc/network/interfaces into the filesystem rooted at fs.
 
     net is the contents of /etc/network/interfaces.
     """
-    netdir = os.path.join(os.path.join(fs, 'etc'), 'network')
+    netdir = _join_and_check_path_within_fs(fs, 'etc', 'network')
     utils.execute('mkdir', '-p', netdir, run_as_root=True)
     utils.execute('chown', 'root:root', netdir, run_as_root=True)
     utils.execute('chmod', 755, netdir, run_as_root=True)
-    netfile = os.path.join(netdir, 'interfaces')
-    utils.execute('tee', netfile, process_input=net, run_as_root=True)
+
+    netfile = os.path.join('etc', 'network', 'interfaces')
+    _inject_file_into_fs(fs, netfile, net)
 
 
 def _inject_admin_password_into_fs(admin_passwd, fs, execute=None):
     """Set the root password to admin_passwd
 
@@ -382,20 +404,19 @@ def _inject_admin_password_into_fs(admin_passwd, fs, execute=None):
     fd, tmp_passwd = tempfile.mkstemp()
     os.close(fd)
     fd, tmp_shadow = tempfile.mkstemp()
     os.close(fd)
 
-    utils.execute('cp', os.path.join(fs, 'etc', 'passwd'), tmp_passwd,
-                  run_as_root=True)
-    utils.execute('cp', os.path.join(fs, 'etc', 'shadow'), tmp_shadow,
-                  run_as_root=True)
+    passwd_path = _join_and_check_path_within_fs(fs, 'etc', 'passwd')
+    shadow_path = _join_and_check_path_within_fs(fs, 'etc', 'shadow')
+
+    utils.execute('cp', passwd_path, tmp_passwd, run_as_root=True)
+    utils.execute('cp', shadow_path, tmp_shadow, run_as_root=True)
     _set_passwd(admin_user, admin_passwd, tmp_passwd, tmp_shadow)
-    utils.execute('cp', tmp_passwd, os.path.join(fs, 'etc', 'passwd'),
-                  run_as_root=True)
+    utils.execute('cp', tmp_passwd, passwd_path, run_as_root=True)
     os.unlink(tmp_passwd)
-    utils.execute('cp', tmp_shadow, os.path.join(fs, 'etc', 'shadow'),
-                  run_as_root=True)
+    utils.execute('cp', tmp_shadow, shadow_path, run_as_root=True)
     os.unlink(tmp_shadow)
 
 
 def _set_passwd(username, admin_passwd, passwd_file, shadow_file):
     """set the password for username to admin_passwd

Vulnerable code: def _inject_file_into_fs(fs, path, contents):
    absolute_path = os.path.join(fs, path.lstrip('/'))
    utils.execute('tee', absolute_path, process_input=contents,
          run_as_root=True)
    metadata_path = os.path.join(fs, "meta.js")

    utils.execute('tee', metadata_path,
                  process_input=jsonutils.dumps(metadata), run_as_root=True)
    sshdir = os.path.join(fs, 'root', '.ssh')
    keyfile = os.path.join(sshdir, 'authorized_keys')
    key_data = [
    ]
    utils.execute('tee', '-a', keyfile,
                  process_input=''.join(key_data), run_as_root=True)
    netdir = os.path.join(os.path.join(fs, 'etc'), 'network')
    netfile = os.path.join(netdir, 'interfaces')
    utils.execute('tee', netfile, process_input=net, run_as_root=True)
    utils.execute('cp', os.path.join(fs, 'etc', 'passwd'), tmp_passwd,
                  run_as_root=True)
    utils.execute('cp', os.path.join(fs, 'etc', 'shadow'), tmp_shadow,
                  run_as_root=True)
    utils.execute('cp', tmp_passwd, os.path.join(fs, 'etc', 'passwd'),
                  run_as_root=True)
    utils.execute('cp', tmp_shadow, os.path.join(fs, 'etc', 'shadow'),
                  run_as_root=True)
Fixed code: from nova import exception
from nova.virt.disk import api as disk_api


class TestVirtDisk(test.TestCase):
    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 _join_and_check_path_within_fs(fs, *args):
    '''os.path.join() with safety check for injected file paths.

    Join the supplied path components and make sure that the
    resulting path we are injecting into is within the
    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))
    if not absolute_path.startswith(os.path.realpath(fs) + '/'):
        raise exception.Invalid(_('injected file path not valid'))
    return absolute_path


def _inject_file_into_fs(fs, path, contents, append=False):
    absolute_path = _join_and_check_path_within_fs(fs, path.lstrip('/'))


    args = []
    if append:
        args.append('-a')
    args.append(absolute_path)

    kwargs = dict(process_input=contents, run_as_root=True)

    utils.execute('tee', *args, **kwargs)
    _inject_file_into_fs(fs, 'meta.js', jsonutils.dumps(metadata))
    sshdir = _join_and_check_path_within_fs(fs, 'root', '.ssh')

    keyfile = os.path.join('root', '.ssh', 'authorized_keys')

    key_data = ''.join([
    ])

    _inject_file_into_fs(fs, keyfile, key_data, append=True)
    netdir = _join_and_check_path_within_fs(fs, 'etc', 'network')

    netfile = os.path.join('etc', 'network', 'interfaces')
    _inject_file_into_fs(fs, netfile, net)
    passwd_path = _join_and_check_path_within_fs(fs, 'etc', 'passwd')
    shadow_path = _join_and_check_path_within_fs(fs, 'etc', 'shadow')

    utils.execute('cp', passwd_path, tmp_passwd, run_as_root=True)
    utils.execute('cp', shadow_path, tmp_shadow, run_as_root=True)
    utils.execute('cp', tmp_passwd, passwd_path, run_as_root=True)
    utils.execute('cp', tmp_shadow, shadow_path, run_as_root=True)