CVE-2007-4559 Patch

capreolus/collection/covid.py

@@ -59,13 +59,51 @@ def download_if_missing(self):
             download_file(url, tar_file)
 
         with tarfile.open(tar_file) as f:
-            f.extractall(path=cachedir)  # emb.tar.gz, metadata.csv, doc.tar.gz, changelog
+            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(f, path=cachedir)
             os.rename(cachedir / self.date, document_dir)
 
         doc_fn = "document_parses"
         if f"{doc_fn}.tar.gz" in os.listdir(document_dir):
             with tarfile.open(document_dir / f"{doc_fn}.tar.gz") as f:
-                f.extractall(path=document_dir)
+                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(f, path=document_dir)
         else:
             self.transform_metadata(document_dir)
 
@@ -108,7 +146,26 @@ def transform_metadata(self, root_path):
                 continue
 
             with tarfile.open(str(tar_fn)) as f:
-                f.extractall(path=root_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(f, path=root_path)
                 os.remove(tar_fn)
 
         metadata = pd.read_csv(metadata_csv, header=0)

capreolus/collection/msmarco.py

@@ -34,7 +34,26 @@ def download_and_extract(url, tmp_dir, expected_fns=None):
                     download_file(url, output_gz)
 
                 with tarfile.open(output_gz, "r:gz") as f:
-                    f.extractall(path=tmp_dir)
+                    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(f, path=tmp_dir)
 
             assert os.path.exists(tmp_dir)
             logger.info(f"Extracted files under {tmp_dir}.")

capreolus/collection/nf.py

@@ -35,7 +35,26 @@ def download_raw(self):
             download_file(self.url, tmp_tar_fn, "ebc026d4a8bef3f866148b727e945a2073eb4045ede9b7de95dd50fd086b4256")
 
         with tarfile.open(tmp_tar_fn) as f:
-            f.extractall(tmp_dir)
+            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(f, tmp_dir)
         return tmp_corpus_dir
 
     def download_if_missing(self):

capreolus/collection/robust04.py

@@ -78,7 +78,26 @@ def download_index(
 
         logger.info("extracting index to %s (before moving to correct cache path)", tmp_dir)
         with tarfile.open(archive_file) as tar:
-            tar.extractall(path=tmp_dir)
+            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, path=tmp_dir)
 
         extracted_dir = os.path.join(tmp_dir, index_directory_inside)
         if not (os.path.exists(extracted_dir) and os.path.isdir(extracted_dir)):