Merge pull request from GHSA-g95c-2jgm-hqc6

Co-authored-by: Dave Quinlan <[email protected]>

Author: galvana

src/fides/api/service/connectors/saas/connector_registry_service.py

@@ -40,6 +40,7 @@
     replace_dataset_placeholders,
     replace_version,
 )
+from fides.api.util.unsafe_file_util import verify_zip
 from fides.config import CONFIG
 
 
@@ -181,6 +182,9 @@ def save_template(cls, db: Session, zip_file: ZipFile) -> None:
         custom connector template, registers the template, and saves it to the database.
         """
 
+        # verify the zip file before we use it
+        verify_zip(zip_file)
+
         config_contents = None
         dataset_contents = None
         icon_contents = None

src/fides/api/util/unsafe_file_util.py

@@ -0,0 +1,34 @@
+from typing import Optional
+from zipfile import ZipFile
+
+MAX_FILE_SIZE = 16 * 1024 * 1024  # 16 MB
+CHUNK_SIZE = 1024
+
+
+def verify_zip(zip_file: ZipFile, max_file_size: Optional[int] = None) -> None:
+    """
+    Function to safely verify the contents of zipped files. It prevents potential
+    'zip bomb' attacks by checking the file size of the files in the zip without fully
+    extracting them. If the size of any file in the zip exceeds the specified
+    max_file_size, it raises a ValueError. If the max_file_size is not provided,
+    it uses a default value of 16 MB.
+
+    :param zip_file: A ZipFile object to be verified.
+    :param max_file_size: An optional integer specifying the maximum bytes allowed per file. If not provided, a default value is used.
+    :raises ValueError: If a file in the zip file exceeds the maximum allowed size
+    """
+
+    if max_file_size is None:
+        max_file_size = MAX_FILE_SIZE
+
+    for file_info in zip_file.infolist():
+        file_size = 0
+
+        with zip_file.open(file_info) as file:
+            # wraps the file read in an iterator that stops once no bytes
+            # are returned or the max file size is reached
+            for chunk in iter(lambda: file.read(CHUNK_SIZE), b""):
+                file_size += len(chunk)
+
+                if file_size > max_file_size:
+                    raise ValueError("File size exceeds maximum allowed size")

tests/ops/util/test_unsafe_file_util.py

@@ -0,0 +1,26 @@
+from io import BytesIO
+from zipfile import ZipFile
+
+import pytest
+
+from fides.api.util.unsafe_file_util import verify_zip
+from tests.ops.test_helpers.saas_test_utils import create_zip_file
+
+
+class TestVerifyZip:
+    @pytest.fixture
+    def zip_file(self) -> BytesIO:
+        return create_zip_file(
+            {
+                "config.yml": "This file isn't that big, but it will be considered suspicious if the max file size is set too low",
+            }
+        )
+
+    def test_verify_zip(self, zip_file):
+        verify_zip(ZipFile(zip_file))
+
+    def test_verify_zip_with_small_file_size_limit(self, zip_file):
+        """We set the max file size to 1 byte, so the zip file should be rejected."""
+        with pytest.raises(ValueError) as exc:
+            verify_zip(ZipFile(zip_file), 1)
+        assert "File size exceeds maximum allowed size" in str(exc.value)