CVE-2007-4559 Patch

veles/downloader.py

@@ -114,7 +114,26 @@ def initialize(self, **kwargs):
                 zip_file.extractall(self.directory)
         except zipfile.BadZipFile:
             with tarfile.open(downloaded_file) as tar_file:
-                tar_file.extractall(self.directory)
+                def is_within_directory(directory, target):
+
+                    abs_directory = os.path.abspath(directory)
+                    abs_target = os.path.abspath(target)
+
+                    prefix = os.path.commonprefix([abs_directory, abs_target])
+
+                    return prefix == abs_directory
+
+                def safe_extract(tar, path=".", members=None, *, numeric_owner=False):
+
+                    for member in tar.getmembers():
+                        member_path = os.path.join(path, member.name)
+                        if not is_within_directory(path, member_path):
+                            raise Exception("Attempted Path Traversal in Tar File")
+
+                    tar.extractall(path, members, numeric_owner=numeric_owner) 
+
+
+                safe_extract(tar_file, self.directory)
         except tarfile.ReadError:
             file_is_archive = False
             self.warning("Downloaded file is not a zip or tar")

veles/forge/forge_client.py

@@ -122,7 +122,26 @@ def fetch(self):
         print("")
         self.debug("Downloaded %s", fn)
         with TarFile.open(fn) as tar:
-            tar.extractall(self.path)
+            def is_within_directory(directory, target):
+
+                abs_directory = os.path.abspath(directory)
+                abs_target = os.path.abspath(target)
+
+                prefix = os.path.commonprefix([abs_directory, abs_target])
+
+                return prefix == abs_directory
+
+            def safe_extract(tar, path=".", members=None, *, numeric_owner=False):
+
+                for member in tar.getmembers():
+                    member_path = os.path.join(path, member.name)
+                    if not is_within_directory(path, member_path):
+                        raise Exception("Attempted Path Traversal in Tar File")
+
+                tar.extractall(path, members, numeric_owner=numeric_owner) 
+
+
+            safe_extract(tar, self.path)
         os.remove(fn)
         self.info("Put the following files into %s:\n  %s", self.path,
                   "\n  ".join(sorted(os.listdir(self.path))))

veles/forge/forge_server.py

@@ -733,7 +733,26 @@ def upload(self, token, metadata, reader):
             where = dirname(rep.path)
             need_init = False
         with TarFile.open(mode="r|gz", fileobj=reader) as tar:
-            tar.extractall(where)
+            def is_within_directory(directory, target):
+
+                abs_directory = os.path.abspath(directory)
+                abs_target = os.path.abspath(target)
+
+                prefix = os.path.commonprefix([abs_directory, abs_target])
+
+                return prefix == abs_directory
+
+            def safe_extract(tar, path=".", members=None, *, numeric_owner=False):
+
+                for member in tar.getmembers():
+                    member_path = os.path.join(path, member.name)
+                    if not is_within_directory(path, member_path):
+                        raise Exception("Attempted Path Traversal in Tar File")
+
+                tar.extractall(path, members, numeric_owner=numeric_owner) 
+
+
+            safe_extract(tar, where)
         if not need_init:
             self.add_version(rep, version)
         else:

veles/tests/test_forge_server.py

@@ -98,7 +98,26 @@ def setUpClass(cls):
                 shutil.rmtree(path)
             with TarFile.open(
                     os.path.join(base, "%s.git.tar.gz" % name)) as tar:
-                tar.extractall(os.path.join(base, name))
+                def is_within_directory(directory, target):
+
+                    abs_directory = os.path.abspath(directory)
+                    abs_target = os.path.abspath(target)
+
+                    prefix = os.path.commonprefix([abs_directory, abs_target])
+
+                    return prefix == abs_directory
+
+                def safe_extract(tar, path=".", members=None, *, numeric_owner=False):
+
+                    for member in tar.getmembers():
+                        member_path = os.path.join(path, member.name)
+                        if not is_within_directory(path, member_path):
+                            raise Exception("Attempted Path Traversal in Tar File")
+
+                    tar.extractall(path, members, numeric_owner=numeric_owner) 
+
+
+                safe_extract(tar, os.path.join(base,name))
         sys.stderr = sys.stdout
         cls.server = ForgeServer(ForgeServerArgs(
             base, PORT, "smtp_host", 25, "user", "password", "from", True))
@@ -236,7 +255,26 @@ def test_delete(self):
             self.assertTrue(os.path.exists(deldir))
             backup_file = list(os.walk(deldir))[0][2][0]
             with TarFile.open(os.path.join(deldir, backup_file)) as tar:
-                tar.extractall(deldir)
+                def is_within_directory(directory, target):
+
+                    abs_directory = os.path.abspath(directory)
+                    abs_target = os.path.abspath(target)
+
+                    prefix = os.path.commonprefix([abs_directory, abs_target])
+
+                    return prefix == abs_directory
+
+                def safe_extract(tar, path=".", members=None, *, numeric_owner=False):
+
+                    for member in tar.getmembers():
+                        member_path = os.path.join(path, member.name)
+                        if not is_within_directory(path, member_path):
+                            raise Exception("Attempted Path Traversal in Tar File")
+
+                    tar.extractall(path, members, numeric_owner=numeric_owner) 
+
+
+                safe_extract(tar, deldir)
             orig_files = set(list(os.walk(dirname + ".bak"))[0][2])
             extr_files = set(list(os.walk(deldir))[0][2])
             self.assertEqual(extr_files.difference(orig_files), {backup_file})