From e95bb8b4da6488dc4cdc3ccbda7149ac794db848 Mon Sep 17 00:00:00 2001
From: lelia <2418071+lelia@users.noreply.github.com>
Date: Sat, 14 Nov 2026 00:35:25 -0500
Subject: [PATCH] Reject duplicate multipart upload keys before sending files

---
 CHANGELOG.md                             |   9 +
 socketdev/fullscans/__init__.py          |  23 ++-
 socketdev/uploadmanifests/__init__.py    |  79 +--------
 socketdev/utils/__init__.py              | 183 +++++++++++--------
 socketdev/version.py                     |   2 +-
 tests/unit/test_duplicate_upload_keys.py | 215 +++++++++++++++++++++++
 6 files changed, 360 insertions(+), 151 deletions(-)
 create mode 100644 tests/unit/test_duplicate_upload_keys.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 11bc25b..f076059 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,14 @@
 # Changelog
 
+## 3.6.3
+
+### Fixed: reject duplicate multipart upload keys
+
+- Manifest uploads and full scans now raise `ValueError` before sending a
+  request when multiple files map to the same multipart field name. The error
+  includes the colliding key and both source paths; files are never silently
+  dropped. Key calculation is shared by lazy and eager manifest uploads.
+
 ## 3.6.2
 
 ### Fixed: reject retired API credentials before use
diff --git a/socketdev/fullscans/__init__.py b/socketdev/fullscans/__init__.py
index 89526ae..18a7d60 100644
--- a/socketdev/fullscans/__init__.py
+++ b/socketdev/fullscans/__init__.py
@@ -871,6 +871,9 @@ def post(
         Returns:
             dict or CreateFullScanResponse: API response containing scan results
             
+        Raises:
+            ValueError: If two files have the same multipart upload key.
+
         Note:
             When use_lazy_loading=True, files are opened only when needed during upload,
             preventing file descriptor exhaustion. The max_open_files parameter controls how many
@@ -902,7 +905,25 @@ def post(
         if use_lazy_loading:
             prepared_files = Utils.load_files_for_sending_lazy(files, workspace, max_open_files, base_path, base_paths)
         else:
-            prepared_files = files
+            # This mode accepts requests-style multipart entries already prepared by callers.
+            is_mapping = hasattr(files, "items")
+            entries = list(files.items()) if is_mapping else list(files or [])
+            key_sources = []
+            for key, value in entries:
+                if isinstance(value, (tuple, list)):
+                    filename, content = value[:2]
+                    source = getattr(content, "name", filename)
+                else:
+                    content = value
+                    source = getattr(content, "name", key)
+                # Requests omits parts with no content and decodes byte field names as UTF-8.
+                if content is None:
+                    continue
+                normalized_key = key.decode("utf-8") if isinstance(key, bytes) else key
+                key_sources.append((normalized_key, source))
+            Utils.validate_upload_keys(key_sources)
+            # Validation must not consume a one-shot iterable before Requests encodes it.
+            prepared_files = files if files is None or is_mapping or isinstance(files, (list, tuple)) else entries
 
         response = self.api.do_request(path=path, method="POST", files=prepared_files)
 
diff --git a/socketdev/uploadmanifests/__init__.py b/socketdev/uploadmanifests/__init__.py
index dc6b634..2de2ae5 100644
--- a/socketdev/uploadmanifests/__init__.py
+++ b/socketdev/uploadmanifests/__init__.py
@@ -11,77 +11,8 @@ def __init__(self, api):
         self.api = api
 
     def _calculate_key_name(self, file_path: str, workspace: Optional[str] = None, base_path: Optional[str] = None, base_paths: Optional[List[str]] = None) -> str:
-        """
-        Calculate the key name for a file using the same logic as load_files_for_sending_lazy.
-        This ensures consistency between lazy and non-lazy loading modes.
-        """
-        # Normalize file path
-        if "\\" in file_path:
-            file_path = file_path.replace("\\", "/")
-        
-        # Normalize paths
-        if workspace and "\\" in workspace:
-            workspace = workspace.replace("\\", "/")
-        if base_path and "\\" in base_path:
-            base_path = base_path.replace("\\", "/")
-        if base_paths:
-            base_paths = [bp.replace("\\", "/") if "\\" in bp else bp for bp in base_paths]
-
-        # Calculate the key name for the form data
-        key = file_path
-        path_stripped = False
-
-        # If base_paths is provided, try to strip one of the paths from the file path
-        if base_paths:
-            for bp in base_paths:
-                normalized_base_path = bp.rstrip("/") + "/" if not bp.endswith("/") else bp
-                if key.startswith(normalized_base_path):
-                    key = key[len(normalized_base_path):]
-                    path_stripped = True
-                    break
-                elif key.startswith(bp.rstrip("/")):
-                    stripped_base = bp.rstrip("/")
-                    if key.startswith(stripped_base + "/") or key == stripped_base:
-                        key = key[len(stripped_base):]
-                        key = key.lstrip("/")
-                        path_stripped = True
-                        break
-        elif base_path:
-            normalized_base_path = base_path.rstrip("/") + "/" if not base_path.endswith("/") else base_path
-            if key.startswith(normalized_base_path):
-                key = key[len(normalized_base_path):]
-                path_stripped = True
-            elif key.startswith(base_path.rstrip("/")):
-                stripped_base = base_path.rstrip("/")
-                if key.startswith(stripped_base + "/") or key == stripped_base:
-                    key = key[len(stripped_base):]
-                    key = key.lstrip("/")
-                    path_stripped = True
-
-        # If workspace is provided and no base paths matched, fall back to workspace logic
-        if not path_stripped and workspace and file_path.startswith(workspace):
-            key = file_path[len(workspace):]
-            # Remove all leading slashes (for absolute paths)
-            while key.startswith("/"):
-                key = key[1:]
-            path_stripped = True
-
-        # Clean up relative path prefixes, but preserve filename dots
-        while key.startswith("./"):
-            key = key[2:]
-        while key.startswith("../"):
-            key = key[3:]
-        # Remove any remaining leading slashes (for absolute paths)
-        while key.startswith("/"):
-            key = key[1:]
-
-        # Remove Windows drive letter if present (C:/...)
-        if len(key) > 2 and key[1] == ':' and (key[2] == '/' or key[2] == '\\'):
-            key = key[2:]
-            while key.startswith("/"):
-                key = key[1:]
-
-        return key
+        """Calculate a multipart key using the shared upload path logic."""
+        return Utils.calculate_upload_key(file_path, workspace, base_path, base_paths)
 
     def upload_manifest_files(self, org_slug: str, file_paths: List[str], workspace: Optional[str] = None, base_path: Optional[str] = None, base_paths: Optional[List[str]] = None, use_lazy_loading: bool = True) -> str:
         """
@@ -99,6 +30,7 @@ def upload_manifest_files(self, org_slug: str, file_paths: List[str], workspace:
             str: The tarHash from the upload response
             
         Raises:
+            ValueError: If two source paths produce the same upload key.
             Exception: If upload fails
         """
         # Filter to only existing files
@@ -118,9 +50,8 @@ def upload_manifest_files(self, org_slug: str, file_paths: List[str], workspace:
         else:
             # Fallback to basic file loading if needed
             loaded_files = []
-            for file_path in valid_files:
-                # Use the same key generation logic as lazy loading for consistency
-                key = self._calculate_key_name(file_path, workspace, base_path, base_paths)
+            key_sources = Utils.prepare_upload_keys(valid_files, workspace, base_path, base_paths)
+            for key, file_path in key_sources:
                 with open(file_path, 'rb') as f:
                     loaded_files.append((key, (key, f.read())))
         
diff --git a/socketdev/utils/__init__.py b/socketdev/utils/__init__.py
index 74be447..c2ee014 100644
--- a/socketdev/utils/__init__.py
+++ b/socketdev/utils/__init__.py
@@ -1,4 +1,5 @@
 from typing import Literal, List, Tuple, Optional, Union
+from urllib3.fields import RequestField
 import logging
 import os
 import weakref
@@ -238,6 +239,105 @@ def validate_integration_type(integration_type: str) -> IntegrationType:
             raise ValueError(f"Invalid integration type: {integration_type}")
         return integration_type  # type: ignore
     
+    @staticmethod
+    def calculate_upload_key(file_path: str, workspace: Optional[str] = None, base_path: Optional[str] = None, base_paths: Optional[List[str]] = None) -> str:
+        """Calculate the repo-relative multipart field name for a manifest."""
+        workspace, base_path, base_paths = Utils._normalize_upload_options(workspace, base_path, base_paths)
+        return Utils._calculate_normalized_upload_key(file_path.replace("\\", "/"), workspace, base_path, base_paths)
+
+    @staticmethod
+    def _normalize_upload_options(workspace, base_path, base_paths):
+        return (
+            workspace.replace("\\", "/") if workspace else workspace,
+            base_path.replace("\\", "/") if base_path else base_path,
+            [path.replace("\\", "/") for path in base_paths] if base_paths else base_paths,
+        )
+
+    @staticmethod
+    def _calculate_normalized_upload_key(file_path, workspace, base_path, base_paths):
+        # Calculate the key name for the form data
+        key = file_path
+        path_stripped = False
+
+        # If base_paths is provided, try to strip one of the paths from the file path
+        if base_paths:
+            for bp in base_paths:
+                normalized_base_path = bp.rstrip("/") + "/" if not bp.endswith("/") else bp
+                if key.startswith(normalized_base_path):
+                    key = key[len(normalized_base_path):]
+                    path_stripped = True
+                    break
+                elif key.startswith(bp.rstrip("/")):
+                    stripped_base = bp.rstrip("/")
+                    if key.startswith(stripped_base + "/") or key == stripped_base:
+                        key = key[len(stripped_base):]
+                        key = key.lstrip("/")
+                        path_stripped = True
+                        break
+        elif base_path:
+            normalized_base_path = base_path.rstrip("/") + "/" if not base_path.endswith("/") else base_path
+            if key.startswith(normalized_base_path):
+                key = key[len(normalized_base_path):]
+                path_stripped = True
+            elif key.startswith(base_path.rstrip("/")):
+                stripped_base = base_path.rstrip("/")
+                if key.startswith(stripped_base + "/") or key == stripped_base:
+                    key = key[len(stripped_base):]
+                    key = key.lstrip("/")
+                    path_stripped = True
+
+        # If workspace is provided and no base paths matched, fall back to workspace logic
+        if not path_stripped and workspace and file_path.startswith(workspace):
+            key = file_path[len(workspace):]
+            # Remove all leading slashes (for absolute paths)
+            while key.startswith("/"):
+                key = key[1:]
+            path_stripped = True
+
+        # Clean up relative path prefixes, but preserve filename dots
+        while key.startswith("./"):
+            key = key[2:]
+        while key.startswith("../"):
+            key = key[3:]
+        # Remove any remaining leading slashes (for absolute paths)
+        while key.startswith("/"):
+            key = key[1:]
+
+        # Remove Windows drive letter if present (C:/...)
+        if len(key) > 2 and key[1] == ':' and (key[2] == '/' or key[2] == '\\'):
+            key = key[2:]
+            while key.startswith("/"):
+                key = key[1:]
+
+        return key
+
+    @staticmethod
+    def validate_upload_keys(key_sources: List[Tuple[str, str]]) -> None:
+        """Reject colliding multipart keys, preserving source paths for diagnostics."""
+        seen = {}
+        formatter = RequestField(name="", data=None).header_formatter
+        for key, source in key_sources:
+            if key is None:
+                continue
+            # Use the installed encoder's escaping rules; the result is name="<key>".
+            wire_key = formatter("name", key)[6:-1]
+            if wire_key in seen:
+                raise ValueError(
+                    f"Duplicate upload key '{wire_key}' from '{seen[wire_key]}' and '{source}'"
+                )
+            seen[wire_key] = source
+
+    @staticmethod
+    def prepare_upload_keys(files: List[str], workspace: Optional[str] = None, base_path: Optional[str] = None, base_paths: Optional[List[str]] = None) -> List[Tuple[str, str]]:
+        """Compute and validate all keys before reading any file contents."""
+        workspace, base_path, base_paths = Utils._normalize_upload_options(workspace, base_path, base_paths)
+        key_sources = [
+            (Utils._calculate_normalized_upload_key(path.replace("\\", "/"), workspace, base_path, base_paths), path)
+            for path in files
+        ]
+        Utils.validate_upload_keys(key_sources)
+        return key_sources
+
     @staticmethod
     def load_files_for_sending_lazy(files: List[str], workspace: Optional[str] = None, max_open_files: int = 100, base_path: Optional[str] = None, base_paths: Optional[List[str]] = None) -> List[Tuple[str, Tuple[str, LazyFileLoader]]]:
         """
@@ -258,86 +358,19 @@ def load_files_for_sending_lazy(files: List[str], workspace: Optional[str] = Non
         Returns:
             List of tuples formatted for requests multipart upload:
             [(field_name, (filename, lazy_file_object)), ...]
+
+        Raises:
+            ValueError: If two source paths produce the same upload key.
         """
         # Configure the file descriptor manager with the specified limit
         _fd_manager.set_max_open_files(max_open_files)
         
         send_files = []
-        if workspace and "\\" in workspace:
-            workspace = workspace.replace("\\", "/")
-        if base_path and "\\" in base_path:
-            base_path = base_path.replace("\\", "/")
-        if base_paths:
-            base_paths = [bp.replace("\\", "/") if "\\" in bp else bp for bp in base_paths]
-        
-        for file_path in files:
-            # Normalize file path
-            if "\\" in file_path:
-                file_path = file_path.replace("\\", "/")
-
-            # Skip directories
-            if os.path.isdir(file_path):
-                continue
-
-            # Handle file path splitting safely
-            if "/" in file_path:
-                _, name = file_path.rsplit("/", 1)
-            else:
-                name = file_path
-
-            # Calculate the key name for the form data
-            key = file_path
-            path_stripped = False
-
-            # If base_paths is provided, try to strip one of the paths from the file path
-            if base_paths:
-                for bp in base_paths:
-                    normalized_base_path = bp.rstrip("/") + "/" if not bp.endswith("/") else bp
-                    if key.startswith(normalized_base_path):
-                        key = key[len(normalized_base_path):]
-                        path_stripped = True
-                        break
-                    elif key.startswith(bp.rstrip("/")):
-                        stripped_base = bp.rstrip("/")
-                        if key.startswith(stripped_base + "/") or key == stripped_base:
-                            key = key[len(stripped_base):]
-                            key = key.lstrip("/")
-                            path_stripped = True
-                            break
-            elif base_path:
-                normalized_base_path = base_path.rstrip("/") + "/" if not base_path.endswith("/") else base_path
-                if key.startswith(normalized_base_path):
-                    key = key[len(normalized_base_path):]
-                    path_stripped = True
-                elif key.startswith(base_path.rstrip("/")):
-                    stripped_base = base_path.rstrip("/")
-                    if key.startswith(stripped_base + "/") or key == stripped_base:
-                        key = key[len(stripped_base):]
-                        key = key.lstrip("/")
-                        path_stripped = True
-
-            # If workspace is provided and no base paths matched, fall back to workspace logic
-            if not path_stripped and workspace and file_path.startswith(workspace):
-                key = file_path[len(workspace):]
-                # Remove all leading slashes (for absolute paths)
-                while key.startswith("/"):
-                    key = key[1:]
-                path_stripped = True
-
-            # Clean up relative path prefixes, but preserve filename dots
-            while key.startswith("./"):
-                key = key[2:]
-            while key.startswith("../"):
-                key = key[3:]
-            # Remove any remaining leading slashes (for absolute paths)
-            while key.startswith("/"):
-                key = key[1:]
-
-            # Remove Windows drive letter if present (C:/...)
-            if len(key) > 2 and key[1] == ':' and (key[2] == '/' or key[2] == '\\'):
-                key = key[2:]
-                while key.startswith("/"):
-                    key = key[1:]
+        # Directories do not produce multipart parts, so exclude them before validating keys.
+        upload_files = [path for path in files if not os.path.isdir(path.replace("\\", "/"))]
+        key_sources = Utils.prepare_upload_keys(upload_files, workspace, base_path, base_paths)
+        for key, source_path in key_sources:
+            file_path = source_path.replace("\\", "/")
 
             # Create lazy file loader instead of opening file immediately
             lazy_file = LazyFileLoader(file_path, key)
diff --git a/socketdev/version.py b/socketdev/version.py
index 4f82193..6a45bf5 100644
--- a/socketdev/version.py
+++ b/socketdev/version.py
@@ -1 +1 @@
-__version__ = "3.6.2"
+__version__ = "3.6.3"
diff --git a/tests/unit/test_duplicate_upload_keys.py b/tests/unit/test_duplicate_upload_keys.py
new file mode 100644
index 0000000..8d7d98a
--- /dev/null
+++ b/tests/unit/test_duplicate_upload_keys.py
@@ -0,0 +1,215 @@
+"""Multipart key collisions must fail before files are read or requests are sent."""
+
+from collections import UserDict
+from io import BytesIO
+from unittest.mock import Mock, patch
+
+import pytest
+from requests import Request
+from urllib3.fields import RequestField
+
+from socketdev.fullscans import FullScans, FullScanParams
+from socketdev.uploadmanifests import UploadManifests
+from socketdev.utils import Utils
+
+
+COLLISIONS = [
+    (["./services/a/package.json"] * 2, {"base_paths": ["."]}, "services/a/package.json"),
+    (["/x/a/package.json", "/x/b/package.json"], {"base_paths": ["/x/a", "/x/b"]}, "package.json"),
+    ([r"C:\repo\package.json", "C:/repo/package.json"], {"base_path": r"C:\repo"}, "package.json"),
+    (["./package.json", "package.json"], {}, "package.json"),
+    (['/repo/a"b/package.json', "/repo/a%22b/package.json"], {"base_path": "/repo"}, "a%22b/package.json"),
+    (["/repo/a\nb/package.json", "/repo/a%0Ab/package.json"], {"base_path": "/repo"}, "a%0Ab/package.json"),
+    (["/repo/a\rb/package.json", "/repo/a%0Db/package.json"], {"base_path": "/repo"}, "a%0Db/package.json"),
+    (["/repo/package.json", "/repo//package.json"], {"workspace": "/repo"}, "package.json"),
+]
+
+
+def assert_collision(error, key, paths):
+    assert str(error.value) == (
+        f"Duplicate upload key '{key}' from '{paths[0]}' and '{paths[1]}'"
+    )
+
+
+@pytest.mark.parametrize("paths,options,key", COLLISIONS)
+def test_lazy_utility_rejects_collisions_before_creating_loaders(paths, options, key):
+    with patch("socketdev.utils.LazyFileLoader") as loader:
+        with pytest.raises(ValueError) as error:
+            Utils.load_files_for_sending_lazy(paths, **options)
+    assert_collision(error, key, paths)
+    loader.assert_not_called()
+
+
+@pytest.mark.parametrize("paths,options,key", COLLISIONS)
+@pytest.mark.parametrize("lazy", [True, False])
+def test_manifest_upload_rejects_collisions_before_reading_or_sending(paths, options, key, lazy):
+    api = Mock()
+    with patch("os.path.exists", return_value=True), patch("os.path.isfile", return_value=True):
+        with patch("builtins.open") as opened:
+            with pytest.raises(ValueError) as error:
+                UploadManifests(api).upload_manifest_files("org", paths, use_lazy_loading=lazy, **options)
+    assert_collision(error, key, paths)
+    opened.assert_not_called()
+    api.do_request.assert_not_called()
+
+
+@pytest.mark.parametrize("paths,options,key", COLLISIONS)
+def test_fullscan_rejects_collisions_before_sending(paths, options, key):
+    api = Mock()
+    with pytest.raises(ValueError) as error:
+        FullScans(api).post(paths, FullScanParams(repo="repo", org_slug="org"), use_lazy_loading=True, **options)
+    assert_collision(error, key, paths)
+    api.do_request.assert_not_called()
+
+
+@pytest.mark.parametrize("lazy", [True, False])
+def test_clean_manifest_upload_preserves_keys_order_and_contents(tmp_path, lazy):
+    paths = [tmp_path / "frontend" / "package.json", tmp_path / "backend" / "package.json"]
+    for i, path in enumerate(paths):
+        path.parent.mkdir()
+        path.write_bytes(f"manifest {i}".encode())
+    api = Mock()
+    api.do_request.return_value.status_code = 200
+    api.do_request.return_value.json.return_value = {"tarHash": "hash"}
+    assert UploadManifests(api).upload_manifest_files(
+        "org", [str(path) for path in paths], base_path=str(tmp_path), use_lazy_loading=lazy
+    ) == "hash"
+    files = api.do_request.call_args.kwargs["files"]
+    assert [key for key, _ in files] == ["frontend/package.json", "backend/package.json"]
+    for i, (key, (filename, content)) in enumerate(files):
+        assert filename == key
+        if lazy:
+            try:
+                assert content.read() == f"manifest {i}".encode()
+            finally:
+                content.close()
+        else:
+            assert content == f"manifest {i}".encode()
+
+
+def test_clean_fullscan_preserves_lazy_multipart_keys(tmp_path):
+    paths = [str(tmp_path / "a" / "package.json"), str(tmp_path / "b" / "package.json")]
+    api = Mock()
+    api.do_request.return_value.status_code = 201
+    api.do_request.return_value.json.return_value = {"id": "scan"}
+    assert FullScans(api).post(
+        paths, FullScanParams(repo="repo", org_slug="org"), use_lazy_loading=True, base_path=str(tmp_path)
+    ) == {"id": "scan"}
+    files = api.do_request.call_args.kwargs["files"]
+    assert [(key, value[0], value[1].file_path) for key, value in files] == [
+        ("a/package.json", "a/package.json", paths[0]),
+        ("b/package.json", "b/package.json", paths[1]),
+    ]
+
+
+def test_lazy_utility_skips_directories(tmp_path):
+    directory = str(tmp_path)
+    assert Utils.load_files_for_sending_lazy([directory, directory]) == []
+
+
+def test_prepared_fullscan_rejects_duplicate_keys():
+    api = Mock()
+    files = [("package.json", (path, BytesIO(b"{}"))) for path in ("/a/package.json", "/b/package.json")]
+    with pytest.raises(ValueError) as error:
+        FullScans(api).post(files, FullScanParams(repo="repo", org_slug="org"))
+    assert_collision(error, "package.json", ["/a/package.json", "/b/package.json"])
+    assert all(value[1].tell() == 0 for _, value in files)
+    api.do_request.assert_not_called()
+
+
+@pytest.mark.parametrize("mapping", [True, False])
+def test_prepared_fullscan_preserves_valid_files(mapping):
+    api = Mock()
+    api.do_request.return_value.status_code = 201
+    files = [("a/package.json", ("a/package.json", b"{}")), ("b/package.json", BytesIO(b"{}"))]
+    if mapping:
+        files = dict(files)
+    FullScans(api).post(files, FullScanParams(repo="repo", org_slug="org"))
+    assert api.do_request.call_args.kwargs["files"] is files
+
+
+@pytest.mark.parametrize("shape", ["iterator", "mapping", "none", "tuple-none", "metadata", "list-value"])
+def test_prepared_fullscan_preserves_encoded_parts(shape):
+    part = ("package.json", ("package.json", b"manifest content"))
+    entries = [part]
+    if shape == "none":
+        entries.insert(0, ("package.json", None))
+    elif shape == "tuple-none":
+        entries.insert(0, ("package.json", ("missing.json", None)))
+    elif shape == "metadata":
+        entries = [("package.json", ("package.json", b"manifest content", "application/json", {"X-Manifest": "yes"}))]
+    elif shape == "list-value":
+        entries = [("package.json", ["package.json", b"manifest content"])]
+    files = iter(entries) if shape == "iterator" else UserDict(entries) if shape == "mapping" else entries
+    api = Mock()
+    api.do_request.return_value.status_code = 201
+    FullScans(api).post(files, FullScanParams(repo="repo", org_slug="org"))
+    forwarded = api.do_request.call_args.kwargs["files"]
+    # A fixed boundary makes the encoded body comparable byte for byte.
+    with patch("urllib3.filepost.choose_boundary", return_value="test-boundary"):
+        expected = Request("POST", "https://example.test", files=entries).prepare()
+        actual = Request("POST", "https://example.test", files=forwarded).prepare()
+    assert actual.body == expected.body
+    assert actual.body.count(b'name="package.json";') == 1
+
+
+def test_prepared_fullscan_rejects_equivalent_byte_and_string_keys():
+    api = Mock()
+    files = [("package.json", ("a.json", b"first")), (b"package.json", ("b.json", b"second"))]
+    with pytest.raises(ValueError) as error:
+        FullScans(api).post(files, FullScanParams(repo="repo", org_slug="org"))
+    assert_collision(error, "package.json", ["a.json", "b.json"])
+    api.do_request.assert_not_called()
+
+
+@pytest.mark.parametrize("character,escape", [('"', "%22"), ("\n", "%0A"), ("\r", "%0D")])
+def test_prepared_fullscan_rejects_keys_with_identical_encoded_names(character, escape):
+    keys = [f"a{character}b/package.json", f"a{escape}b/package.json"]
+    files = [(keys[0], ("a.json", b"first")), (keys[1], ("b.json", b"second"))]
+    encoded = Request("POST", "https://example.test", files=files).prepare()
+    assert encoded.body.count(f'name="{keys[1]}";'.encode()) == 2
+    api = Mock()
+    with pytest.raises(ValueError) as error:
+        FullScans(api).post(files, FullScanParams(repo="repo", org_slug="org"))
+    assert_collision(error, keys[1], ["a.json", "b.json"])
+    api.do_request.assert_not_called()
+
+
+@pytest.mark.parametrize("prepared", [True, False])
+def test_collision_validation_uses_installed_encoder(prepared):
+    def legacy_formatter(name, value):
+        replacements = {code: f"%{code:02X}" for code in range(32) if code != 27}
+        replacements.update({34: "%22", 92: "\\\\"})
+        return f'{name}="{value.translate(replacements)}"'
+
+    keys = ["a\tb/package.json", "a%09b/package.json"]
+    files = [(keys[0], ("a.json", b"first")), (keys[1], ("b.json", b"second"))]
+    api = Mock()
+    original_init = RequestField.__init__
+
+    def legacy_init(self, *args, **kwargs):
+        original_init(self, *args, **kwargs)
+        self.header_formatter = legacy_formatter
+
+    with patch.object(RequestField, "__init__", legacy_init):
+        encoded = Request("POST", "https://example.test", files=files).prepare()
+        assert encoded.body.count(b'name="a%09b/package.json";') == 2
+        with pytest.raises(ValueError) as error:
+            if prepared:
+                FullScans(api).post(files, FullScanParams(repo="repo", org_slug="org"))
+            else:
+                Utils.load_files_for_sending_lazy(keys)
+    assert_collision(error, keys[1], ["a.json", "b.json"] if prepared else keys)
+    api.do_request.assert_not_called()
+
+
+def test_prepared_fullscan_preserves_unnamed_part():
+    api = Mock()
+    api.do_request.return_value.status_code = 201
+    files = [(None, ("package.json", b"manifest content"))]
+    FullScans(api).post(files, FullScanParams(repo="repo", org_slug="org"))
+    forwarded = api.do_request.call_args.kwargs["files"]
+    with patch("urllib3.filepost.choose_boundary", return_value="test-boundary"):
+        assert Request("POST", "https://example.test", files=files).prepare().body == Request(
+            "POST", "https://example.test", files=forwarded
+        ).prepare().body