Merge pull request #14408 from DefectDojo/zip-handling-consolidation

Refactor zip handling with safe_open_zip and safe_read_all_zip

Author: rossops

dojo/tools/blackduck/importer.py

@@ -7,6 +7,8 @@
 from collections.abc import Iterable
 from pathlib import Path
 
+from dojo.tools.utils import safe_open_zip
+
 from .model import BlackduckFinding
 
 
@@ -48,7 +50,7 @@ def _process_zipfile(self, report):
         files = {}
         security_issues = {}
 
-        with zipfile.ZipFile(report) as zipf:
+        with safe_open_zip(report) as zipf:
             for full_file_name in zipf.namelist():
                 file_name = full_file_name.split("/")[-1]
                 # Backwards compatibility, newer versions of Blackduck have a source file rather

dojo/tools/blackduck_component_risk/importer.py

@@ -4,6 +4,8 @@
 import zipfile
 from pathlib import Path
 
+from dojo.tools.utils import safe_open_zip
+
 logger = logging.getLogger(__name__)
 
 
@@ -42,7 +44,7 @@ def _process_zipfile(self, report: Path) -> (dict, dict, dict):
         components = {}
         source = {}
         try:
-            with zipfile.ZipFile(report) as zipf:
+            with safe_open_zip(report) as zipf:
                 c_file = False
                 s_file = False
                 for full_file_name in zipf.namelist():

dojo/tools/fortify/fpr_parser.py

@@ -1,12 +1,12 @@
 import logging
 import re
-import zipfile
 from xml.etree.ElementTree import Element
 
 from defusedxml import ElementTree
 
 from dojo.models import Finding, Test
 from dojo.tools.fortify.fortify_data import DescriptionData, RuleData, SnippetData, VulnerabilityData
+from dojo.tools.utils import safe_read_all_zip
 
 logger = logging.getLogger(__name__)
 
@@ -26,13 +26,9 @@ def __init__(self):
         pass
 
     def parse_fpr(self, filename, test):
-        if str(filename.__class__) == "<class '_io.TextIOWrapper'>":
-            input_zip = zipfile.ZipFile(filename.name, "r")
-        else:
-            input_zip = zipfile.ZipFile(filename, "r")
         # Read each file from the zip artifact into a dict with the format of
         # filename: file_content
-        zip_data = {name: input_zip.read(name) for name in input_zip.namelist()}
+        zip_data = safe_read_all_zip(filename)
         root, self.namespaces = self.identify_root(zip_data, "audit.fvdl", "No audit.fvdl file found in the zip")
         audit_log, self.namespaces_audit_log = self.identify_root(zip_data, "audit.xml")
         return self.convert_vulnerabilities_to_findings(root, audit_log, test)

dojo/tools/ms_defender/parser.py

@@ -1,10 +1,10 @@
 import json
 import logging
-import zipfile
 
 from django.conf import settings
 
 from dojo.models import Endpoint, Finding
+from dojo.tools.utils import safe_read_all_zip
 from dojo.url.models import URL
 
 logger = logging.getLogger(__name__)
@@ -37,12 +37,7 @@ def get_findings(self, file, test):
                 logger.warning("Error parsing JSON file %s: %s", file.name, e)
                 return []
         elif str(file.name).endswith(".zip"):
-            if str(file.__class__) == "<class '_io.TextIOWrapper'>":
-                input_zip = zipfile.ZipFile(file.name, "r")
-            else:
-                input_zip = zipfile.ZipFile(file, "r")
-
-            zipdata = {name: input_zip.read(name) for name in input_zip.namelist()}
+            zipdata = safe_read_all_zip(file)
             vulnerabilityfiles = []
             machinefiles = []
             for content in list(zipdata):

dojo/tools/sonarqube/parser.py

@@ -1,13 +1,13 @@
 import json
 import logging
-import zipfile
 
 from lxml import etree
 
 from dojo.tools.sonarqube.sonarqube_restapi_json import SonarQubeRESTAPIJSON
 from dojo.tools.sonarqube.sonarqube_restapi_zip import SonarQubeRESTAPIZIP
 from dojo.tools.sonarqube.soprasteria_html import SonarQubeSoprasteriaHTML
 from dojo.tools.sonarqube.soprasteria_json import SonarQubeSoprasteriaJSON
+from dojo.tools.utils import safe_read_all_zip
 
 logger = logging.getLogger(__name__)
 
@@ -38,11 +38,7 @@ def get_findings(self, file, test):
                 return SonarQubeRESTAPIJSON().get_json_items(json_content, test, self.mode)
             return []
         if file.name.endswith(".zip"):
-            if str(file.__class__) == "<class '_io.TextIOWrapper'>":
-                input_zip = zipfile.ZipFile(file.name, "r")
-            else:
-                input_zip = zipfile.ZipFile(file, "r")
-            zipdata = {name: input_zip.read(name) for name in input_zip.namelist()}
+            zipdata = safe_read_all_zip(file)
             return SonarQubeRESTAPIZIP().get_items(zipdata, test, self.mode)
         parser = etree.HTMLParser()
         tree = etree.parse(file, parser)

dojo/tools/utils.py

@@ -1,8 +1,83 @@
+import io
 import json
 import logging
+import zipfile
 
 logger = logging.getLogger(__name__)
 
+# Zip bomb protection limits
+MAX_ZIP_MEMBERS = 1000
+MAX_ZIP_MEMBER_SIZE = 512 * 1024 * 1024  # 512 MB per member (uncompressed)
+MAX_ZIP_TOTAL_SIZE = 1 * 1024 * 1024 * 1024  # 1 GB total (uncompressed)
+MAX_ZIP_RATIO = 100  # max compression ratio (uncompressed / compressed)
+
+
+def safe_open_zip(file):
+    """
+    Open a zip file with protection against zip bomb attacks.
+
+    Validates member count, per-member uncompressed size, total uncompressed
+    size, and compression ratios using the central-directory metadata before
+    any data is extracted.
+
+    Accepts a file-like object or an io.TextIOWrapper (in which case
+    file.name is used as the path).
+
+    Returns an open ZipFile. Use as a context manager or call .close()
+    explicitly when done.
+
+    Raises ValueError if any limit is exceeded.
+    """
+    zf = zipfile.ZipFile(file.name, "r") if isinstance(file, io.TextIOWrapper) else zipfile.ZipFile(file, "r")
+
+    infos = zf.infolist()
+
+    if len(infos) > MAX_ZIP_MEMBERS:
+        zf.close()
+        msg = f"Zip file contains {len(infos)} members, exceeding the limit of {MAX_ZIP_MEMBERS}."
+        raise ValueError(msg)
+
+    total_size = 0
+    for info in infos:
+        if info.file_size > MAX_ZIP_MEMBER_SIZE:
+            zf.close()
+            msg = (
+                f"Zip member '{info.filename}' has uncompressed size {info.file_size} bytes, "
+                f"exceeding the per-member limit of {MAX_ZIP_MEMBER_SIZE} bytes."
+            )
+            raise ValueError(msg)
+        if info.compress_size > 0 and (info.file_size / info.compress_size) > MAX_ZIP_RATIO:
+            zf.close()
+            ratio = info.file_size / info.compress_size
+            msg = (
+                f"Zip member '{info.filename}' has a compression ratio of "
+                f"{ratio:.1f}:1, exceeding the limit of {MAX_ZIP_RATIO}:1."
+            )
+            raise ValueError(msg)
+        total_size += info.file_size
+        if total_size > MAX_ZIP_TOTAL_SIZE:
+            zf.close()
+            msg = f"Zip file total uncompressed size exceeds the limit of {MAX_ZIP_TOTAL_SIZE} bytes."
+            raise ValueError(msg)
+
+    return zf
+
+
+def safe_read_all_zip(file):
+    """
+    Open a zip file safely and read all members into a dict {name: bytes}.
+
+    Applies the same zip bomb protections as safe_open_zip before reading
+    any data.
+
+    Raises ValueError if any limit is exceeded.
+    """
+    zf = safe_open_zip(file)
+    try:
+        return {name: zf.read(name) for name in zf.namelist()}
+    finally:
+        zf.close()
+
 
 def get_npm_cwe(item_node):
     """