Add tarfile member sanitization to extractall()

Fix test code to use CVE-2007-4559 safe tarfile.extractall, and write
tgz file to disk first to avoid seeking request.get response.

tests/integration/conftest.py

@@ -91,6 +91,7 @@ def fixture_kafka_description(request: SubRequest) -> KafkaDescription:
 
     return KafkaDescription(
         version=kafka_version,
+        kafka_tgz=RUNTIME_DIR / kafka_tgz,
         install_dir=kafka_dir,
         download_url=kafka_url,
         protocol_version="2.7",

tests/integration/utils/config.py

@@ -24,6 +24,7 @@ def from_dict(data: dict) -> "ZKConfig":
 @dataclass(frozen=True)
 class KafkaDescription:
     version: str
+    kafka_tgz: str
     install_dir: Path
     download_url: str
     protocol_version: str

tests/integration/utils/kafka_server.py

@@ -73,10 +73,37 @@ def maybe_download_kafka(kafka_description: KafkaDescription) -> None:
     """If necessary download kafka to run the tests."""
     if not os.path.exists(kafka_description.install_dir):
         log.info("Downloading Kafka '%s'", kafka_description.download_url)
-
         download = requests.get(kafka_description.download_url, stream=True)
-        with tarfile.open(mode="r:gz", fileobj=download.raw) as file:
-            file.extractall(str(kafka_description.install_dir.parent))
+        with open(kafka_description.kafka_tgz, "wb") as fd:
+            for chunk in download.iter_content(chunk_size=None):
+                fd.write(chunk)
+
+    if os.path.exists(kafka_description.install_dir):
+        return
+
+    log.info("Extracting Kafka '%s'", kafka_description.kafka_tgz)
+    with open(kafka_description.kafka_tgz, "rb") as kafkatgz:
+        with tarfile.open(mode="r:gz", fileobj=kafkatgz) as file:
+
+            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(file, str(kafka_description.install_dir.parent))
 
 
 def kafka_java_args(