fix: address PR review for ZIP upload (#12253)

- Add BadZipFile handling in extract_flows_from_zip for defense-in-depth
- Wrap blocking ZIP I/O in asyncio.to_thread() to avoid event loop blocking
- Add post-read size check to guard against dishonest ZIP headers
- Add 7 adversarial tests: invalid JSON, max entries, oversized entries,
  mixed valid/invalid, filename fallback, corrupt ZIP, batch name dedup

Author: viktoravelino

src/backend/base/langflow/api/utils/zip_utils.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+import asyncio
 import io
 import zipfile
+from dataclasses import dataclass, field
 
 import orjson
 from lfx.log.logger import logger
@@ -13,36 +15,71 @@
 MAX_ENTRY_UNCOMPRESSED_BYTES = 50 * 1024 * 1024  # 50 MB per file
 
 
-async def extract_flows_from_zip(contents: bytes) -> list[dict]:
-    """Extract flow JSON data from a ZIP file.
+@dataclass
+class _ZipExtractionResult:
+    """Result of synchronous ZIP extraction, including warnings to log after."""
+
+    flows: list[dict] = field(default_factory=list)
+    warnings: list[str] = field(default_factory=list)
 
-    Reads all .json files from the ZIP archive and returns their parsed contents.
-    Enforces limits on entry count and individual file size to mitigate zip bomb attacks.
+
+def _extract_flows_sync(contents: bytes) -> _ZipExtractionResult:
+    """Synchronous helper that performs all blocking ZIP I/O.
 
     Raises:
-        ValueError: If the ZIP contains more than MAX_ZIP_ENTRIES JSON files.
+        ValueError: If the ZIP is corrupt or contains more than MAX_ZIP_ENTRIES JSON files.
     """
-    flows: list[dict] = []
+    result = _ZipExtractionResult()
+
+    try:
+        zf = zipfile.ZipFile(io.BytesIO(contents), "r")
+    except zipfile.BadZipFile as exc:
+        msg = f"Uploaded file is not a valid ZIP archive: {exc}"
+        raise ValueError(msg) from exc
 
-    with zipfile.ZipFile(io.BytesIO(contents), "r") as zip_file:
-        json_entries = [info for info in zip_file.infolist() if info.filename.lower().endswith(".json")]
+    with zf:
+        json_entries = [info for info in zf.infolist() if info.filename.lower().endswith(".json")]
 
         if len(json_entries) > MAX_ZIP_ENTRIES:
             msg = f"ZIP contains {len(json_entries)} JSON entries, exceeding the limit of {MAX_ZIP_ENTRIES}"
             raise ValueError(msg)
 
         for info in json_entries:
             if info.file_size > MAX_ENTRY_UNCOMPRESSED_BYTES:
-                await logger.awarning(
+                result.warnings.append(
                     f"Skipping ZIP entry '{info.filename}': uncompressed size "
                     f"{info.file_size} exceeds limit of {MAX_ENTRY_UNCOMPRESSED_BYTES} bytes"
                 )
                 continue
             try:
-                raw = zip_file.read(info.filename)
-                flows.append(orjson.loads(raw))
+                raw = zf.read(info.filename)
+                if len(raw) > MAX_ENTRY_UNCOMPRESSED_BYTES:
+                    result.warnings.append(
+                        f"Skipping ZIP entry '{info.filename}': actual size "
+                        f"{len(raw)} exceeds limit of {MAX_ENTRY_UNCOMPRESSED_BYTES} bytes"
+                    )
+                    continue
+                result.flows.append(orjson.loads(raw))
             except orjson.JSONDecodeError:
-                await logger.awarning(f"Skipping ZIP entry '{info.filename}': invalid JSON")
+                result.warnings.append(f"Skipping ZIP entry '{info.filename}': invalid JSON")
                 continue
 
-    return flows
+    return result
+
+
+async def extract_flows_from_zip(contents: bytes) -> list[dict]:
+    """Extract flow JSON data from a ZIP file.
+
+    Reads all .json files from the ZIP archive and returns their parsed contents.
+    Enforces limits on entry count and individual file size to mitigate zip bomb attacks.
+    Blocking I/O is offloaded to a thread to avoid blocking the event loop.
+
+    Raises:
+        ValueError: If the ZIP is corrupt or contains more than MAX_ZIP_ENTRIES JSON files.
+    """
+    result = await asyncio.to_thread(_extract_flows_sync, contents)
+
+    for warning in result.warnings:
+        await logger.awarning(warning)
+
+    return result.flows

src/backend/tests/unit/test_database.py

@@ -697,6 +697,199 @@ async def test_download_then_upload_roundtrip(client: AsyncClient, json_flow: st
     assert len(uploaded) == 2
 
 
+@pytest.mark.usefixtures("session")
+async def test_upload_zip_with_invalid_json(client: AsyncClient, json_flow: str, logged_in_headers):
+    """ZIP entries with invalid JSON are skipped; valid entries are still processed."""
+    flow = orjson.loads(json_flow)
+    data = flow["data"]
+    valid_flow = {"name": "valid_flow", "description": "good", "data": data}
+
+    zip_buffer = io.BytesIO()
+    with zipfile.ZipFile(zip_buffer, "w") as zf:
+        zf.writestr("valid.json", json.dumps(valid_flow))
+        zf.writestr("broken.json", "{not valid json!!!}")
+    zip_buffer.seek(0)
+
+    response = await client.post(
+        "api/v1/flows/upload/",
+        files={"file": ("mixed.zip", zip_buffer.getvalue(), "application/zip")},
+        headers=logged_in_headers,
+    )
+    assert response.status_code == 201
+    response_data = response.json()
+    assert len(response_data) == 1
+    assert response_data[0]["name"] == "valid_flow"
+
+
+@pytest.mark.usefixtures("session")
+async def test_upload_zip_exceeding_max_entries(client: AsyncClient, json_flow: str, logged_in_headers, monkeypatch):
+    """ZIP with more JSON entries than the limit raises 400."""
+    import langflow.api.utils.zip_utils as zip_utils_mod
+
+    monkeypatch.setattr(zip_utils_mod, "MAX_ZIP_ENTRIES", 3)
+
+    flow = orjson.loads(json_flow)
+    data = flow["data"]
+
+    zip_buffer = io.BytesIO()
+    with zipfile.ZipFile(zip_buffer, "w") as zf:
+        for i in range(5):
+            zf.writestr(f"flow_{i}.json", json.dumps({"name": f"flow_{i}", "data": data}))
+    zip_buffer.seek(0)
+
+    response = await client.post(
+        "api/v1/flows/upload/",
+        files={"file": ("too_many.zip", zip_buffer.getvalue(), "application/zip")},
+        headers=logged_in_headers,
+    )
+    assert response.status_code == 400
+    assert "exceeding the limit" in response.json()["detail"]
+
+
+@pytest.mark.usefixtures("session")
+async def test_upload_zip_with_oversized_entry(client: AsyncClient, json_flow: str, logged_in_headers, monkeypatch):
+    """Entries exceeding size limit are skipped; smaller valid entries are processed."""
+    import langflow.api.utils.zip_utils as zip_utils_mod
+
+    flow = orjson.loads(json_flow)
+    data = flow["data"]
+    small_flow = {"name": "small_flow", "data": {"nodes": [], "edges": []}}
+    big_flow = {"name": "big_flow", "data": data}
+
+    # Build the ZIP first, then set the limit between the two entry sizes
+    zip_buffer = io.BytesIO()
+    with zipfile.ZipFile(zip_buffer, "w") as zf:
+        zf.writestr("small.json", json.dumps(small_flow))
+        zf.writestr("big.json", json.dumps(big_flow))
+
+    # Re-read to find the actual sizes and pick a limit between them
+    with zipfile.ZipFile(io.BytesIO(zip_buffer.getvalue()), "r") as zf:
+        sizes = {info.filename: info.file_size for info in zf.infolist()}
+    limit = (sizes["small.json"] + sizes["big.json"]) // 2
+    monkeypatch.setattr(zip_utils_mod, "MAX_ENTRY_UNCOMPRESSED_BYTES", limit)
+
+    zip_buffer.seek(0)
+    response = await client.post(
+        "api/v1/flows/upload/",
+        files={"file": ("oversized.zip", zip_buffer.getvalue(), "application/zip")},
+        headers=logged_in_headers,
+    )
+    assert response.status_code == 201
+    response_data = response.json()
+    assert len(response_data) == 1
+    assert response_data[0]["name"] == "small_flow"
+
+
+@pytest.mark.usefixtures("session")
+async def test_upload_zip_with_mixed_valid_invalid(client: AsyncClient, json_flow: str, logged_in_headers, monkeypatch):
+    """Mix of valid flows, invalid JSON, and oversized entries → only valid flows returned."""
+    import langflow.api.utils.zip_utils as zip_utils_mod
+
+    flow = orjson.loads(json_flow)
+    data = flow["data"]
+    valid_flow = {"name": "keeper", "data": {"nodes": [], "edges": []}}
+    oversized_flow = {"name": "too_big", "data": data, "padding": "x" * 500}
+
+    zip_buffer = io.BytesIO()
+    with zipfile.ZipFile(zip_buffer, "w") as zf:
+        zf.writestr("valid.json", json.dumps(valid_flow))
+        zf.writestr("broken.json", "NOT JSON {{{")
+        zf.writestr("huge.json", json.dumps(oversized_flow))
+        zf.writestr("readme.txt", "ignored non-json")
+
+    # Set limit between valid entry size and oversized entry size
+    with zipfile.ZipFile(io.BytesIO(zip_buffer.getvalue()), "r") as zf:
+        sizes = {info.filename: info.file_size for info in zf.infolist()}
+    limit = (sizes["valid.json"] + sizes["huge.json"]) // 2
+    monkeypatch.setattr(zip_utils_mod, "MAX_ENTRY_UNCOMPRESSED_BYTES", limit)
+
+    zip_buffer.seek(0)
+    response = await client.post(
+        "api/v1/flows/upload/",
+        files={"file": ("mixed.zip", zip_buffer.getvalue(), "application/zip")},
+        headers=logged_in_headers,
+    )
+    assert response.status_code == 201
+    response_data = response.json()
+    assert len(response_data) == 1
+    assert response_data[0]["name"] == "keeper"
+
+
+@pytest.mark.usefixtures("session")
+async def test_upload_zip_to_projects_filename_none(client: AsyncClient, json_flow: str, logged_in_headers):
+    """When filename has no stem (e.g. '.zip'), the project name defaults to 'Imported Project'."""
+    flow = orjson.loads(json_flow)
+    data = flow["data"]
+    flow_data = {"name": "flow_none", "data": data}
+
+    zip_buffer = io.BytesIO()
+    with zipfile.ZipFile(zip_buffer, "w") as zf:
+        zf.writestr("flow.json", json.dumps(flow_data))
+    zip_buffer.seek(0)
+
+    # filename=".zip" → rsplit gives ("", "zip") → "" is falsy → "Imported Project"
+    response = await client.post(
+        "api/v1/projects/upload/",
+        files={"file": (".zip", zip_buffer.getvalue(), "application/zip")},
+        headers=logged_in_headers,
+    )
+    assert response.status_code == 201
+    response_data = response.json()
+    assert len(response_data) == 1
+
+    folder_id = response_data[0]["folder_id"]
+    project_response = await client.get(f"api/v1/projects/{folder_id}", headers=logged_in_headers)
+    assert project_response.status_code == 200
+    assert project_response.json()["name"].startswith("Imported Project")
+
+
+@pytest.mark.usefixtures("session")
+async def test_upload_bad_zip_file_returns_400(client: AsyncClient, logged_in_headers):
+    """Uploading a corrupt/invalid ZIP file returns 400 with a descriptive error."""
+    # Build a payload that passes zipfile.is_zipfile() but fails ZipFile() construction.
+    # We keep only the end-of-central-directory record (last 22 bytes of a real ZIP)
+    # prepended with garbage, so the EOCD signature is found but the central directory is invalid.
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, "w") as zf:
+        zf.writestr("dummy.json", '{"name":"x"}')
+    valid_zip = buf.getvalue()
+    # Minimal EOCD is 22 bytes; keep it and prepend garbage
+    corrupt_zip = b"garbage" * 10 + valid_zip[-22:]
+
+    response = await client.post(
+        "api/v1/flows/upload/",
+        files={"file": ("corrupt.zip", corrupt_zip, "application/zip")},
+        headers=logged_in_headers,
+    )
+    assert response.status_code == 400
+    assert "not a valid ZIP" in response.json()["detail"]
+
+
+@pytest.mark.usefixtures("session")
+async def test_upload_zip_to_projects_batch_name_dedup(client: AsyncClient, json_flow: str, logged_in_headers):
+    """Multiple flows with the same name get unique names within the batch."""
+    flow = orjson.loads(json_flow)
+    data = flow["data"]
+
+    zip_buffer = io.BytesIO()
+    with zipfile.ZipFile(zip_buffer, "w") as zf:
+        for i in range(3):
+            zf.writestr(f"flow_{i}.json", json.dumps({"name": "duplicate_name", "data": data}))
+    zip_buffer.seek(0)
+
+    response = await client.post(
+        "api/v1/projects/upload/",
+        files={"file": ("dedup_test.zip", zip_buffer.getvalue(), "application/zip")},
+        headers=logged_in_headers,
+    )
+    assert response.status_code == 201
+    response_data = response.json()
+    assert len(response_data) == 3
+    names = [r["name"] for r in response_data]
+    # All names must be unique
+    assert len(set(names)) == 3
+
+
 @pytest.mark.usefixtures("active_user")
 async def test_create_flow_with_invalid_data(client: AsyncClient, logged_in_headers):
     flow = {"name": "a" * 256, "data": "Invalid flow data"}