fix(uploads): reject NUL characters in filenames (#5915)

backend/packages/harness/deerflow/uploads/manager.py

@@ -65,6 +65,8 @@ def normalize_filename(filename: str) -> str:
     safe = Path(filename).name
     if not safe or safe in {".", ".."}:
         raise ValueError(f"Filename is unsafe: {filename!r}")
+    if "\x00" in safe:
+        raise ValueError(f"Filename contains NUL: {filename!r}")
     # Reject backslashes — on Linux Path.name keeps them as literal chars,
     # but they indicate a Windows-style path that should be stripped or rejected.
     if "\\" in safe:

backend/tests/test_uploads_manager.py

@@ -74,6 +74,10 @@ class TestNormalizeFilename:
         with pytest.raises(ValueError, match="unsafe"):
             normalize_filename(".")
 
+    def test_rejects_embedded_nul(self):
+        with pytest.raises(ValueError, match="NUL"):
+            normalize_filename("report\x00.pdf")
+
 
 # ---------------------------------------------------------------------------
 # claim_unique_filename

backend/tests/test_uploads_router.py

@@ -1049,6 +1049,33 @@ def test_upload_files_rejects_dotdot_and_dot_filenames(tmp_path):
     assert [f.name for f in thread_uploads_dir.iterdir()] == ["passwd"]
 
 
+def test_upload_files_skips_nul_filename_without_rolling_back_valid_file(tmp_path):
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir()
+    provider = _mounted_provider()
+
+    with (
+        patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "get_sandbox_provider", return_value=provider),
+    ):
+        result = asyncio.run(
+            call_unwrapped(
+                uploads.upload_files,
+                "thread-local",
+                request=MagicMock(),
+                files=[
+                    UploadFile(filename="valid.txt", file=BytesIO(b"kept")),
+                    UploadFile(filename="report\x00.pdf", file=BytesIO(b"invalid")),
+                ],
+                config=SimpleNamespace(),
+            )
+        )
+
+    assert [file.filename for file in result.files] == ["valid.txt"]
+    assert (thread_uploads_dir / "valid.txt").read_bytes() == b"kept"
+
+
 def test_upload_files_rejects_preexisting_symlink_destination(tmp_path):
     thread_uploads_dir = tmp_path / "uploads"
     thread_uploads_dir.mkdir(parents=True)