浏览代码

[api][filebrowser] Refactor rename endpoint with improved validation and tests (#4198)

This commit refactors the file and directory rename API endpoint to enhance its robustness, validation, and test coverage.

Key Changes:
- Dedicated Serializer: The API now uses a RenameSerializer with a pydantic schema (RenameSchema) for comprehensive input validation, handling aspects like path traversal, restricted file extensions, and invalid characters.

- Core Logic Extraction: The main renaming logic is extracted into a dedicated rename_file_or_directory function in filebrowser/operations.py, promoting code reuse and easier testing.

- Improved Error Handling: The endpoint now returns structured and informative error messages for validation failures and other exceptions.

- Comprehensive Unit Tests: The following has been added:

 - Extensive unit tests for the rename API view, covering various success and failure scenarios using APIRequestFactory.
 - Dedicated tests for the rename_file_or_directory operation to ensure its correctness.
 - Thorough tests for the RenameSchema to validate all its validation rules.
 - Unit tests for the RenameSerializer to ensure it integrates correctly with the schema.

- URL Routing Update: The URL for storage/rename is updated to point to the new filebrowser.api.rename view.

These changes result in a more reliable and maintainable rename API with clear and predictable behavior.
Harsh Gupta 4 月之前
父节点
当前提交
1ab48c0cfc

+ 28 - 29
apps/filebrowser/src/filebrowser/api.py

@@ -28,8 +28,9 @@ from django.http import HttpResponse, HttpResponseNotModified, HttpResponseRedir
 from django.utils.http import http_date
 from django.views.static import was_modified_since
 from rest_framework import status
+from rest_framework.decorators import api_view, parser_classes
 from rest_framework.exceptions import NotFound
-from rest_framework.parsers import MultiPartParser
+from rest_framework.parsers import JSONParser, MultiPartParser
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
@@ -51,11 +52,12 @@ from filebrowser.conf import (
   ENABLE_EXTRACT_UPLOADED_ARCHIVE,
   FILE_DOWNLOAD_CACHE_CONTROL,
   REDIRECT_DOWNLOAD,
-  RESTRICT_FILE_EXTENSIONS,
   SHOW_DOWNLOAD_BUTTON,
 )
 from filebrowser.lib.rwx import compress_mode, filetype, rwx
-from filebrowser.serializers import UploadFileSerializer
+from filebrowser.operations import rename_file_or_directory
+from filebrowser.schemas import RenameSchema
+from filebrowser.serializers import RenameSerializer, UploadFileSerializer
 from filebrowser.utils import get_user_fs, parse_broker_url
 from filebrowser.views import (
   _can_inline_display,
@@ -673,39 +675,36 @@ def save_file(request):
   return HttpResponse(status=200)
 
 
-@api_error_handler
+@api_view(["POST"])
+@parser_classes([JSONParser])
 def rename(request):
-  source_path = request.POST.get("source_path", "")
-  destination_path = request.POST.get("destination_path", "")
-
-  # Check if source and destination paths are provided
-  if not source_path or not destination_path:
-    return HttpResponse("Missing required parameters: source_path and destination_path", status=400)
+  """
+  Rename a file or directory.
 
-  # Extract file extensions from paths
-  _, source_path_ext = os.path.splitext(source_path)
-  _, dest_path_ext = os.path.splitext(destination_path)
+  This endpoint renames a file or directory from a source path to a destination path.
+  The destination can be an absolute path or a relative path from the source's parent directory.
 
-  restricted_file_types = [ext.lower() for ext in RESTRICT_FILE_EXTENSIONS.get()] if RESTRICT_FILE_EXTENSIONS.get() else []
-  # Check if destination path has a restricted file type and it doesn't match the source file type
-  if dest_path_ext.lower() in restricted_file_types and (source_path_ext.lower() != dest_path_ext.lower()):
-    return HttpResponse(f'Cannot rename file to a restricted file type: "{dest_path_ext}"', status=403)
+  Args:
+    request (HttpRequest): The request object containing source and destination paths.
 
-  # Check if destination path contains a hash character
-  if "#" in destination_path:
-    return HttpResponse("Hashes are not allowed in file or directory names. Please choose a different name.", status=400)
+  Returns:
+    Response: A Response object with a success message or an error message with the appropriate status code.
+  """
+  serializer = RenameSerializer(data=request.data)
 
-  # If destination path doesn't have a directory specified, use the same directory as the source path
-  if "/" not in destination_path:
-    source_dir = os.path.dirname(source_path)
-    destination_path = request.fs.join(source_dir, destination_path)
+  if not serializer.is_valid():
+    return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
 
-  # Check if destination path already exists
-  if request.fs.exists(destination_path):
-    return HttpResponse(f"The destination path {destination_path} already exists.", status=409)
+  rename_params = RenameSchema(**serializer.validated_data)
+  try:
+    result = rename_file_or_directory(data=rename_params, username=request.user.username)
+    return Response(result, status=status.HTTP_200_OK)
 
-  request.fs.rename(source_path, destination_path)
-  return HttpResponse(status=200)
+  except ValueError as e:
+    return Response({"error": str(e)}, status=status.HTTP_400_BAD_REQUEST)
+  except Exception as e:
+    LOG.exception(f"Error renaming file: {e}")
+    return Response({"error": "An unexpected error occurred during rename operation"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
 
 def _is_destination_parent_of_source(request, source_path, destination_path):

+ 209 - 96
apps/filebrowser/src/filebrowser/api_test.py

@@ -21,12 +21,14 @@ from unittest.mock import MagicMock, Mock, patch
 
 import pytest
 from django.http import HttpResponseNotModified, HttpResponseRedirect, StreamingHttpResponse
+from rest_framework import status
 from rest_framework.exceptions import NotFound
+from rest_framework.test import APIRequestFactory
 
 from aws.s3.s3fs import S3ListAllBucketsException
 from desktop.lib.exceptions_renderable import PopupException
 from filebrowser.api import copy, download, get_all_filesystems, listdir_paged, mkdir, move, rename, touch, UploadFileAPI
-from filebrowser.conf import REDIRECT_DOWNLOAD, RESTRICT_FILE_EXTENSIONS, SHOW_DOWNLOAD_BUTTON
+from filebrowser.conf import REDIRECT_DOWNLOAD, SHOW_DOWNLOAD_BUTTON
 from hadoop.fs.exceptions import WebHdfsException
 
 
@@ -172,111 +174,222 @@ class TestMkdirAPI:
 
 
 class TestRenameAPI:
-  def test_rename_success(self):
-    request = Mock(
-      method="POST",
-      POST={"source_path": "s3a://test-bucket/test-user/source.txt", "destination_path": "new_name.txt"},
-      fs=Mock(
-        exists=Mock(return_value=False),
-        join=Mock(return_value="s3a://test-bucket/test-user/new_name.txt"),
-        rename=Mock(),
-      ),
-    )
-    reset = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
-    try:
-      response = rename(request)
+  @patch("filebrowser.api.RenameSerializer")
+  @patch("filebrowser.api.rename_file_or_directory")
+  def test_rename_success(self, mock_rename_operation, mock_serializer_class):
+    # Create mock schema and serializer
+    mock_schema = {"source_path": "/user/test/file.txt", "destination_path": "/user/test/file_new.txt"}
+    mock_serializer = Mock(is_valid=Mock(return_value=True), validated_data=mock_schema)
+    mock_serializer_class.return_value = mock_serializer
 
-      assert response.status_code == 200
-      request.fs.rename.assert_called_once_with("s3a://test-bucket/test-user/source.txt", "s3a://test-bucket/test-user/new_name.txt")
-    finally:
-      reset()
+    mock_rename_operation.return_value = {"message": "Renamed successfully"}
 
-  def test_rename_restricted_file_type(self):
-    request = Mock(
-      method="POST",
-      POST={"source_path": "s3a://test-bucket/test-user/source.txt", "destination_path": "new_name.exe"},
-      fs=Mock(
-        rename=Mock(),
-      ),
-    )
-    reset = RESTRICT_FILE_EXTENSIONS.set_for_testing(".exe,.txt")
-    try:
-      response = rename(request)
+    request = APIRequestFactory().post("/storage/rename", data=mock_schema, format="json")
+    request.user = Mock(username="test_user")
 
-      assert response.status_code == 403
-      assert response.content.decode("utf-8") == 'Cannot rename file to a restricted file type: ".exe"'
-    finally:
-      reset()
+    response = rename(request)
 
-  def test_rename_hash_in_path(self):
-    request = Mock(
-      method="POST",
-      POST={"source_path": "s3a://test-bucket/test-user/source.txt", "destination_path": "new#name.txt"},
-      fs=Mock(
-        rename=Mock(),
-      ),
-    )
-    reset = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
-    try:
-      response = rename(request)
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data == {"message": "Renamed successfully"}
+    mock_rename_operation.assert_called_once()
 
-      assert response.status_code == 400
-      assert response.content.decode("utf-8") == "Hashes are not allowed in file or directory names. Please choose a different name."
-    finally:
-      reset()
+  @patch("filebrowser.api.RenameSerializer")
+  def test_rename_invalid_data(self, mock_serializer_class):
+    # Mock invalid serializer
+    mock_serializer = Mock(is_valid=Mock(return_value=False), errors={"source_path": ["This field is required."]})
+    mock_serializer_class.return_value = mock_serializer
 
-  def test_rename_destination_exists(self):
-    request = Mock(
-      method="POST",
-      POST={"source_path": "s3a://test-bucket/test-user/source.txt", "destination_path": "new_name.txt"},
-      fs=Mock(
-        rename=Mock(),
-        exists=Mock(return_value=True),
-        join=Mock(return_value="s3a://test-bucket/test-user/new_name.txt"),
-      ),
-    )
-    reset = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
-    try:
-      response = rename(request)
+    request = APIRequestFactory().post("/storage/rename", data={}, format="json")
+    request.user = Mock(username="test_user")
 
-      assert response.status_code == 409
-      assert response.content.decode("utf-8") == "The destination path s3a://test-bucket/test-user/new_name.txt already exists."
-    finally:
-      reset()
+    response = rename(request)
 
-  def test_rename_no_source_path(self):
-    request = Mock(
-      method="POST",
-      POST={"destination_path": "new_name.txt"},
-      fs=Mock(
-        rename=Mock(),
-      ),
-    )
-    reset = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
-    try:
-      response = rename(request)
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.data == {"source_path": ["This field is required."]}
 
-      assert response.status_code == 400
-      assert response.content.decode("utf-8") == "Missing required parameters: source_path and destination_path"
-    finally:
-      reset()
+  @patch("filebrowser.api.RenameSerializer")
+  @patch("filebrowser.api.rename_file_or_directory")
+  def test_rename_value_error(self, mock_rename_operation, mock_serializer_class):
+    mock_schema = {"source_path": "/user/test/file.txt", "destination_path": "/user/test/existing.txt"}
+    mock_serializer = Mock(is_valid=Mock(return_value=True), validated_data=mock_schema)
+    mock_serializer_class.return_value = mock_serializer
 
-  def test_rename_no_destination_path(self):
-    request = Mock(
-      method="POST",
-      POST={"source_path": "s3a://test-bucket/test-user/source.txt"},
-      fs=Mock(
-        rename=Mock(),
-      ),
+    mock_rename_operation.side_effect = ValueError("Destination path already exists")
+
+    request = APIRequestFactory().post("/storage/rename", data=mock_schema, format="json")
+    request.user = Mock(username="test_user")
+
+    response = rename(request)
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.data == {"error": "Destination path already exists"}
+
+  @patch("filebrowser.api.RenameSerializer")
+  @patch("filebrowser.api.rename_file_or_directory")
+  def test_rename_general_exception(self, mock_rename_operation, mock_serializer_class):
+    mock_schema = {"source_path": "/user/test/file.txt", "destination_path": "/user/test/file_new.txt"}
+    mock_serializer = Mock(is_valid=Mock(return_value=True), validated_data=mock_schema)
+    mock_serializer_class.return_value = mock_serializer
+
+    mock_rename_operation.side_effect = Exception("Filesystem error")
+
+    request = APIRequestFactory().post("/storage/rename", data=mock_schema, format="json")
+    request.user = Mock(username="test_user")
+
+    response = rename(request)
+
+    assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+    assert response.data == {"error": "An unexpected error occurred during rename operation"}
+
+  @patch("filebrowser.api.RenameSerializer")
+  @patch("filebrowser.api.rename_file_or_directory")
+  def test_rename_relative_path(self, mock_rename_operation, mock_serializer_class):
+    mock_schema = {"source_path": "/user/test/document.txt", "destination_path": "document_v2.txt"}
+    mock_serializer = Mock(is_valid=Mock(return_value=True), validated_data=mock_schema)
+    mock_serializer_class.return_value = mock_serializer
+
+    mock_rename_operation.return_value = {"message": "Renamed '/user/test/document.txt' to '/user/test/document_v2.txt' successfully"}
+
+    request = APIRequestFactory().post("/storage/rename", data=mock_schema, format="json")
+    request.user = Mock(username="test_user")
+
+    response = rename(request)
+
+    assert response.status_code == status.HTTP_200_OK
+    assert "successfully" in response.data["message"]
+
+  @patch("filebrowser.api.RenameSerializer")
+  @patch("filebrowser.api.rename_file_or_directory")
+  def test_rename_s3_paths(self, mock_rename_operation, mock_serializer_class):
+    mock_schema = {"source_path": "s3a://bucket/folder/file.txt", "destination_path": "s3a://bucket/folder/file_renamed.txt"}
+    mock_serializer = Mock(is_valid=Mock(return_value=True), validated_data=mock_schema)
+    mock_serializer_class.return_value = mock_serializer
+
+    mock_rename_operation.return_value = {"message": "Renamed successfully"}
+
+    request = APIRequestFactory().post("/storage/rename", data=mock_schema, format="json")
+    request.user = Mock(username="test_user")
+
+    response = rename(request)
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data == {"message": "Renamed successfully"}
+
+  @patch("filebrowser.api.RenameSerializer")
+  @patch("filebrowser.api.rename_file_or_directory")
+  def test_rename_directory(self, mock_rename_operation, mock_serializer_class):
+    mock_schema = {"source_path": "/user/test/old_folder", "destination_path": "/user/test/new_folder"}
+    mock_serializer = Mock(is_valid=Mock(return_value=True), validated_data=mock_schema)
+    mock_serializer_class.return_value = mock_serializer
+
+    mock_rename_operation.return_value = {"message": "Renamed '/user/test/old_folder' to '/user/test/new_folder' successfully"}
+
+    request = APIRequestFactory().post("/storage/rename", data=mock_schema, format="json")
+    request.user = Mock(username="test_user")
+
+    response = rename(request)
+
+    assert response.status_code == status.HTTP_200_OK
+    assert "old_folder" in response.data["message"]
+    assert "new_folder" in response.data["message"]
+
+  @patch("filebrowser.api.RenameSerializer")
+  @patch("filebrowser.api.rename_file_or_directory")
+  @patch("filebrowser.api.LOG.exception")
+  def test_rename_logs_exception(self, mock_log_exception, mock_rename_operation, mock_serializer_class):
+    mock_schema = {"source_path": "/user/test/file.txt", "destination_path": "/user/test/file_new.txt"}
+    mock_serializer = Mock(is_valid=Mock(return_value=True), validated_data=mock_schema)
+    mock_serializer_class.return_value = mock_serializer
+
+    mock_rename_operation.side_effect = Exception("Test error")
+
+    request = APIRequestFactory().post("/storage/rename", data=mock_schema, format="json")
+    request.user = Mock(username="test_user")
+
+    response = rename(request)
+
+    assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+    mock_log_exception.assert_called_once()
+    assert "Error renaming file" in mock_log_exception.call_args[0][0]
+
+  @patch("filebrowser.api.RenameSerializer")
+  def test_rename_validation_errors(self, mock_serializer_class):
+    # Mock serializer with multiple validation errors
+    mock_serializer = Mock(
+      is_valid=Mock(return_value=False),
+      errors={"source_path": ["Path cannot be empty"], "destination_path": ["Path traversal patterns are not allowed"]},
     )
-    reset = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
-    try:
-      response = rename(request)
+    mock_serializer_class.return_value = mock_serializer
 
-      assert response.status_code == 400
-      assert response.content.decode("utf-8") == "Missing required parameters: source_path and destination_path"
-    finally:
-      reset()
+    request = APIRequestFactory().post("/storage/rename", data={"source_path": "", "destination_path": "../etc"}, format="json")
+    request.user = Mock(username="test_user")
+
+    response = rename(request)
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert "source_path" in response.data
+    assert "destination_path" in response.data
+
+  @patch("filebrowser.api.RenameSerializer")
+  @patch("filebrowser.api.rename_file_or_directory")
+  def test_rename_unicode_paths(self, mock_rename_operation, mock_serializer_class):
+    mock_schema = {"source_path": "/user/test/文档.txt", "destination_path": "/user/test/文档_新.txt"}
+    mock_serializer = Mock(is_valid=Mock(return_value=True), validated_data=mock_schema)
+    mock_serializer_class.return_value = mock_serializer
+
+    mock_rename_operation.return_value = {"message": "Renamed '/user/test/文档.txt' to '/user/test/文档_新.txt' successfully"}
+
+    request = APIRequestFactory().post("/storage/rename", data=mock_schema, format="json")
+    request.user = Mock(username="test_user")
+
+    response = rename(request)
+
+    assert response.status_code == status.HTTP_200_OK
+    assert "文档.txt" in response.data["message"]
+    assert "文档_新.txt" in response.data["message"]
+
+  @patch("filebrowser.api.RenameSerializer")
+  @patch("filebrowser.api.rename_file_or_directory")
+  def test_rename_special_characters(self, mock_rename_operation, mock_serializer_class):
+    mock_schema = {
+      "source_path": "/user/test/file with spaces & (special).txt",
+      "destination_path": "/user/test/file_with_spaces_and_special.txt",
+    }
+    mock_serializer = Mock(is_valid=Mock(return_value=True), validated_data=mock_schema)
+    mock_serializer_class.return_value = mock_serializer
+
+    mock_rename_operation.return_value = {"message": "Renamed successfully"}
+
+    request = APIRequestFactory().post("/storage/rename", data=mock_schema, format="json")
+    request.user = Mock(username="test_user")
+
+    response = rename(request)
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data == {"message": "Renamed successfully"}
+
+  @patch("filebrowser.api.RenameSerializer")
+  @patch("filebrowser.api.rename_file_or_directory")
+  @patch("filebrowser.api.RenameSchema")
+  def test_rename_schema_creation(self, mock_rename_schema_class, mock_rename_operation, mock_serializer_class):
+    mock_schema_data = {"source_path": "/user/test/file.txt", "destination_path": "/user/test/file_new.txt"}
+    mock_serializer = Mock(is_valid=Mock(return_value=True), validated_data=mock_schema_data)
+    mock_serializer_class.return_value = mock_serializer
+
+    mock_schema_instance = Mock()
+    mock_rename_schema_class.return_value = mock_schema_instance
+
+    mock_rename_operation.return_value = {"message": "Success"}
+
+    request = APIRequestFactory().post("/storage/rename", data=mock_schema_data, format="json")
+    request.user = Mock(username="test_user")
+
+    rename(request)
+
+    # Verify RenameSchema was created with the validated data
+    mock_rename_schema_class.assert_called_once_with(**mock_schema_data)
+    # Verify rename operation was called with the schema instance
+    mock_rename_operation.assert_called_once_with(data=mock_schema_instance, username="test_user")
 
 
 class TestMoveAPI:
@@ -1405,8 +1518,8 @@ class TestUploadFileAPI:
     response = self.view.post(self.request)
 
     assert response.status_code == 400
-    assert "error" in response.data
-    assert response.data["error"] == "File upload failed or was not handled correctly by the upload handler."
+    assert response.data is not None
+    assert response.data.get("error") == "File upload failed or was not handled correctly by the upload handler."
 
   def test_post_raises_popup_exception(self):
     self.request.FILES = MagicMock()

+ 75 - 0
apps/filebrowser/src/filebrowser/operations.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import os
+from typing import Any, Dict
+
+from filebrowser.schemas import RenameSchema
+from filebrowser.utils import get_user_fs
+
+LOG = logging.getLogger()
+
+
+def rename_file_or_directory(data: RenameSchema, username: str) -> Dict[str, Any]:
+  """
+  Renames a file or directory.
+
+  Args:
+    data (RenameSchema): The rename data.
+    username (str): The user performing the operation.
+
+  Returns:
+    dict: A dictionary with a success message.
+
+  Raises:
+    ValueError: If the username is empty, or if the source path does not exist, or if the destination path already exists.
+    Exception: If the rename operation fails.
+  """
+  if not username:
+    raise ValueError("Username is required and cannot be empty.")
+
+  source_path = data.source_path
+  destination_path = data.destination_path
+
+  fs = get_user_fs(username)
+
+  # Handle relative destination paths
+  if "/" not in destination_path:
+    source_dir = os.path.dirname(source_path)
+    destination_path = fs.join(source_dir, destination_path)
+
+  # Normalize destination path after potential modification
+  destination_path_normalized = fs.normpath(destination_path)
+
+  # Check if source exists
+  if not fs.exists(source_path):
+    raise ValueError(f"Source path does not exist: {source_path}")
+
+  # Check if destination already exists
+  if fs.exists(destination_path_normalized):
+    raise ValueError(f"Destination path already exists: {destination_path_normalized}")
+
+  LOG.info(f"User {username} is renaming {source_path} to {destination_path_normalized}")
+
+  try:
+    fs.rename(source_path, destination_path_normalized)
+  except Exception as e:
+    LOG.exception(f"Error during rename operation: {e}")
+    raise Exception(f"Failed to rename file: {str(e)}")
+
+  return {"message": f"Renamed '{source_path}' to '{destination_path_normalized}' successfully"}

+ 225 - 0
apps/filebrowser/src/filebrowser/operations_tests.py

@@ -0,0 +1,225 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest.mock import Mock, patch
+
+import pytest
+
+from filebrowser.operations import rename_file_or_directory
+from filebrowser.schemas import RenameSchema
+
+
+class TestRenameFileOrDirectory:
+  @patch("filebrowser.operations.get_user_fs")
+  def test_successful_rename_absolute_path(self, mock_get_user_fs):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]  # source exists, destination doesn't
+    mock_fs.normpath.return_value = "/user/test/file_new.txt"
+    mock_fs.rename.return_value = None
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/file.txt", destination_path="/user/test/file_new.txt")
+    result = rename_file_or_directory(data, "test_user")
+
+    assert result["message"] == "Renamed '/user/test/file.txt' to '/user/test/file_new.txt' successfully"
+    mock_fs.rename.assert_called_once_with("/user/test/file.txt", "/user/test/file_new.txt")
+
+  @patch("os.path.dirname")
+  @patch("filebrowser.operations.get_user_fs")
+  def test_successful_rename_relative_path(self, mock_get_user_fs, mock_dirname):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]  # source exists, destination doesn't
+    mock_fs.join.return_value = "/user/test/file_new.txt"
+    mock_fs.normpath.return_value = "/user/test/file_new.txt"
+    mock_fs.rename.return_value = None
+    mock_get_user_fs.return_value = mock_fs
+    mock_dirname.return_value = "/user/test"
+
+    data = RenameSchema(source_path="/user/test/file.txt", destination_path="file_new.txt")
+    result = rename_file_or_directory(data, "test_user")
+
+    assert result["message"] == "Renamed '/user/test/file.txt' to '/user/test/file_new.txt' successfully"
+    mock_fs.join.assert_called_once_with("/user/test", "file_new.txt")
+    mock_fs.rename.assert_called_once_with("/user/test/file.txt", "/user/test/file_new.txt")
+
+  def test_empty_username_raises_error(self):
+    data = RenameSchema(source_path="/user/test/file.txt", destination_path="/user/test/file_new.txt")
+
+    with pytest.raises(ValueError, match="Username is required and cannot be empty"):
+      rename_file_or_directory(data, "")
+
+    with pytest.raises(ValueError, match="Username is required and cannot be empty"):
+      rename_file_or_directory(data, None)
+
+  @patch("filebrowser.operations.get_user_fs")
+  def test_source_path_not_exists(self, mock_get_user_fs):
+    mock_fs = Mock()
+    mock_fs.exists.return_value = False  # source doesn't exist
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/nonexistent.txt", destination_path="/user/test/new.txt")
+    with pytest.raises(ValueError, match="Source path does not exist: /user/test/nonexistent.txt"):
+      rename_file_or_directory(data, "test_user")
+
+  @patch("filebrowser.operations.get_user_fs")
+  def test_destination_path_already_exists(self, mock_get_user_fs):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, True]  # both source and destination exist
+    mock_fs.normpath.return_value = "/user/test/existing_file.txt"
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/file.txt", destination_path="/user/test/existing_file.txt")
+    with pytest.raises(ValueError, match="Destination path already exists: /user/test/existing_file.txt"):
+      rename_file_or_directory(data, "test_user")
+
+  @patch("filebrowser.operations.get_user_fs")
+  def test_filesystem_rename_error(self, mock_get_user_fs):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]  # source exists, destination doesn't
+    mock_fs.normpath.return_value = "/user/test/file_new.txt"
+    mock_fs.rename.side_effect = Exception("Permission denied")
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/file.txt", destination_path="/user/test/file_new.txt")
+    with pytest.raises(Exception, match="Failed to rename file: Permission denied"):
+      rename_file_or_directory(data, "test_user")
+
+  @patch("filebrowser.operations.get_user_fs")
+  def test_rename_directory(self, mock_get_user_fs):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]  # source exists, destination doesn't
+    mock_fs.normpath.return_value = "/user/test/new_folder"
+    mock_fs.rename.return_value = None
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/old_folder", destination_path="/user/test/new_folder")
+    result = rename_file_or_directory(data, "test_user")
+
+    assert result["message"] == "Renamed '/user/test/old_folder' to '/user/test/new_folder' successfully"
+    mock_fs.rename.assert_called_once_with("/user/test/old_folder", "/user/test/new_folder")
+
+  @patch("filebrowser.operations.get_user_fs")
+  def test_move_to_different_directory(self, mock_get_user_fs):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]  # source exists, destination doesn't
+    mock_fs.normpath.return_value = "/user/test/archive/file.txt"
+    mock_fs.rename.return_value = None
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/documents/file.txt", destination_path="/user/test/archive/file.txt")
+    result = rename_file_or_directory(data, "test_user")
+
+    assert result["message"] == "Renamed '/user/test/documents/file.txt' to '/user/test/archive/file.txt' successfully"
+    mock_fs.rename.assert_called_once_with("/user/test/documents/file.txt", "/user/test/archive/file.txt")
+
+  @patch("filebrowser.operations.get_user_fs")
+  def test_rename_with_special_characters(self, mock_get_user_fs):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]
+    mock_fs.normpath.return_value = "/user/test/file (copy) 2024.txt"
+    mock_fs.rename.return_value = None
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/file & document.txt", destination_path="/user/test/file (copy) 2024.txt")
+    result = rename_file_or_directory(data, "test_user")
+
+    assert "successfully" in result["message"]
+
+  @patch("filebrowser.operations.get_user_fs")
+  def test_rename_s3_paths(self, mock_get_user_fs):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]
+    mock_fs.normpath.return_value = "s3a://bucket/folder/file_new.txt"
+    mock_fs.rename.return_value = None
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="s3a://bucket/folder/file.txt", destination_path="s3a://bucket/folder/file_new.txt")
+    result = rename_file_or_directory(data, "test_user")
+
+    assert result["message"] == "Renamed 's3a://bucket/folder/file.txt' to 's3a://bucket/folder/file_new.txt' successfully"
+
+  @patch("filebrowser.operations.get_user_fs")
+  def test_relative_path_with_subdirectory(self, mock_get_user_fs):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]
+    mock_fs.normpath.return_value = "/user/test/subfolder/file.txt"
+    mock_fs.rename.return_value = None
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/file.txt", destination_path="/user/test/subfolder/file.txt")
+    result = rename_file_or_directory(data, "test_user")
+
+    assert "successfully" in result["message"]
+    mock_fs.rename.assert_called_once_with("/user/test/file.txt", "/user/test/subfolder/file.txt")
+
+  @patch("filebrowser.operations.LOG.info")
+  @patch("filebrowser.operations.get_user_fs")
+  def test_logging_on_success(self, mock_get_user_fs, mock_log_info):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]
+    mock_fs.normpath.return_value = "/user/test/file_new.txt"
+    mock_fs.rename.return_value = None
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/file.txt", destination_path="/user/test/file_new.txt")
+    rename_file_or_directory(data, "test_user")
+
+    mock_log_info.assert_called_once_with("User test_user is renaming /user/test/file.txt to /user/test/file_new.txt")
+
+  @patch("filebrowser.operations.LOG.exception")
+  @patch("filebrowser.operations.get_user_fs")
+  def test_logging_on_error(self, mock_get_user_fs, mock_log_exception):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]
+    mock_fs.normpath.return_value = "/user/test/file_new.txt"
+    mock_fs.rename.side_effect = Exception("Access denied")
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/file.txt", destination_path="/user/test/file_new.txt")
+    with pytest.raises(Exception):
+      rename_file_or_directory(data, "test_user")
+
+    mock_log_exception.assert_called_once()
+    assert "Error during rename operation" in mock_log_exception.call_args[0][0]
+
+  @patch("filebrowser.operations.get_user_fs")
+  def test_unicode_paths(self, mock_get_user_fs):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]
+    mock_fs.normpath.return_value = "/user/test/文档_新.txt"
+    mock_fs.rename.return_value = None
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/文档.txt", destination_path="/user/test/文档_新.txt")
+    result = rename_file_or_directory(data, "test_user")
+
+    assert "successfully" in result["message"]
+    mock_fs.rename.assert_called_once_with("/user/test/文档.txt", "/user/test/文档_新.txt")
+
+  @patch("filebrowser.operations.get_user_fs")
+  def test_normpath_called_on_destination(self, mock_get_user_fs):
+    mock_fs = Mock()
+    mock_fs.exists.side_effect = [True, False]
+    mock_fs.normpath.return_value = "/user/test/file_new.txt"
+    mock_fs.rename.return_value = None
+    mock_get_user_fs.return_value = mock_fs
+
+    data = RenameSchema(source_path="/user/test/file.txt", destination_path="/user/test/./file_new.txt")
+    rename_file_or_directory(data, "test_user")
+
+    mock_fs.normpath.assert_called_with("/user/test/./file_new.txt")
+    mock_fs.rename.assert_called_once_with("/user/test/file.txt", "/user/test/file_new.txt")

+ 67 - 0
apps/filebrowser/src/filebrowser/schemas.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from filebrowser.utils import is_file_upload_allowed
+
+
+class RenameSchema(BaseModel):
+  """Schema for file/directory rename operation."""
+
+  model_config = ConfigDict(str_strip_whitespace=True)
+
+  source_path: str = Field(..., description="The source path to rename from")
+  destination_path: str = Field(..., description="The destination path to rename to")
+
+  @field_validator("source_path", "destination_path")
+  @classmethod
+  def validate_paths_not_empty(cls, v: str) -> str:
+    if not v or not v.strip():
+      raise ValueError("Path cannot be empty")
+    return v.strip()
+
+  @field_validator("source_path", "destination_path")
+  @classmethod
+  def validate_no_path_traversal(cls, v: str) -> str:
+    if ".." in v:
+      raise ValueError("Path traversal patterns are not allowed")
+    return v
+
+  @field_validator("destination_path")
+  @classmethod
+  def validate_no_hash_character(cls, v: str) -> str:
+    if "#" in v:
+      raise ValueError("Hashes are not allowed in file or directory names")
+    return v
+
+  @model_validator(mode="after")
+  def validate_file_extension_allowed(self) -> "RenameSchema":
+    # Extract file extensions
+    _, source_ext = os.path.splitext(self.source_path)
+    dest_filename = os.path.basename(self.destination_path)
+    _, dest_ext = os.path.splitext(dest_filename)
+
+    # Only check restrictions if changing extension
+    if source_ext.lower() != dest_ext.lower():
+      is_allowed, error_message = is_file_upload_allowed(dest_filename)
+      if not is_allowed:
+        raise ValueError(error_message)
+
+    return self

+ 207 - 0
apps/filebrowser/src/filebrowser/schemas_tests.py

@@ -0,0 +1,207 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import pytest
+from pydantic import ValidationError
+
+from filebrowser.conf import ALLOW_FILE_EXTENSIONS, RESTRICT_FILE_EXTENSIONS
+from filebrowser.schemas import RenameSchema
+
+
+class TestRenameSchema:
+  def test_valid_rename_same_directory(self):
+    data = {"source_path": "/user/test/documents/report.txt", "destination_path": "/user/test/documents/report_v2.txt"}
+    schema = RenameSchema(**data)
+    assert schema.source_path == "/user/test/documents/report.txt"
+    assert schema.destination_path == "/user/test/documents/report_v2.txt"
+
+  def test_valid_rename_relative_path(self):
+    data = {"source_path": "/user/test/documents/report.txt", "destination_path": "report_v2.txt"}
+    schema = RenameSchema(**data)
+    assert schema.source_path == "/user/test/documents/report.txt"
+    assert schema.destination_path == "report_v2.txt"
+
+  def test_valid_rename_different_directory(self):
+    data = {"source_path": "/user/test/documents/report.txt", "destination_path": "/user/test/archive/report.txt"}
+    schema = RenameSchema(**data)
+    assert schema.source_path == "/user/test/documents/report.txt"
+    assert schema.destination_path == "/user/test/archive/report.txt"
+
+  def test_whitespace_trimming(self):
+    data = {"source_path": "  /user/test/file.txt  ", "destination_path": "  /user/test/file_new.txt  "}
+    schema = RenameSchema(**data)
+    assert schema.source_path == "/user/test/file.txt"
+    assert schema.destination_path == "/user/test/file_new.txt"
+
+  def test_empty_source_path(self):
+    data = {"source_path": "", "destination_path": "/user/test/file.txt"}
+    with pytest.raises(ValidationError) as exc_info:
+      RenameSchema(**data)
+    assert "Path cannot be empty" in str(exc_info.value)
+
+  def test_empty_destination_path(self):
+    data = {"source_path": "/user/test/file.txt", "destination_path": ""}
+    with pytest.raises(ValidationError) as exc_info:
+      RenameSchema(**data)
+    assert "Path cannot be empty" in str(exc_info.value)
+
+  def test_whitespace_only_paths(self):
+    data = {"source_path": "   ", "destination_path": "/user/test/file.txt"}
+    with pytest.raises(ValidationError) as exc_info:
+      RenameSchema(**data)
+    assert "Path cannot be empty" in str(exc_info.value)
+
+  def test_path_traversal_in_source(self):
+    data = {"source_path": "/user/test/../admin/file.txt", "destination_path": "/user/test/file.txt"}
+    with pytest.raises(ValidationError) as exc_info:
+      RenameSchema(**data)
+    assert "Path traversal patterns are not allowed" in str(exc_info.value)
+
+  def test_path_traversal_in_destination(self):
+    data = {"source_path": "/user/test/file.txt", "destination_path": "/user/test/../admin/file.txt"}
+    with pytest.raises(ValidationError) as exc_info:
+      RenameSchema(**data)
+    assert "Path traversal patterns are not allowed" in str(exc_info.value)
+
+  def test_hash_character_in_destination(self):
+    data = {"source_path": "/user/test/file.txt", "destination_path": "/user/test/file#new.txt"}
+    with pytest.raises(ValidationError) as exc_info:
+      RenameSchema(**data)
+    assert "Hashes are not allowed in file or directory names" in str(exc_info.value)
+
+  def test_hash_character_allowed_in_source(self):
+    data = {"source_path": "/user/test/file#old.txt", "destination_path": "/user/test/file_new.txt"}
+    schema = RenameSchema(**data)
+    assert schema.source_path == "/user/test/file#old.txt"
+
+  def test_file_extension_change_without_restrictions(self):
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing(None)
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
+
+    try:
+      data = {"source_path": "/user/test/document.txt", "destination_path": "/user/test/document.pdf"}
+      schema = RenameSchema(**data)
+      assert schema.source_path == "/user/test/document.txt"
+      assert schema.destination_path == "/user/test/document.pdf"
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_file_extension_change_with_restrict_list(self):
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing(None)
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing([".exe", ".bat", ".cmd"])
+
+    try:
+      data = {"source_path": "/user/test/script.txt", "destination_path": "/user/test/script.exe"}
+      with pytest.raises(ValidationError) as exc_info:
+        RenameSchema(**data)
+      assert "is restricted" in str(exc_info.value)
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_file_extension_change_with_allow_list(self):
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt", ".csv", ".json"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
+
+    try:
+      # Allowed extension change
+      data = {"source_path": "/user/test/data.txt", "destination_path": "/user/test/data.csv"}
+      schema = RenameSchema(**data)
+      assert schema.destination_path == "/user/test/data.csv"
+
+      # Not allowed extension change
+      data = {"source_path": "/user/test/data.txt", "destination_path": "/user/test/data.xml"}
+      with pytest.raises(ValidationError) as exc_info:
+        RenameSchema(**data)
+      assert "is not permitted" in str(exc_info.value)
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_same_extension_case_insensitive(self):
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
+
+    try:
+      data = {"source_path": "/user/test/document.TXT", "destination_path": "/user/test/document_renamed.txt"}
+      schema = RenameSchema(**data)
+      assert schema.destination_path == "/user/test/document_renamed.txt"
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_rename_directory(self):
+    data = {"source_path": "/user/test/old_folder", "destination_path": "/user/test/new_folder"}
+    schema = RenameSchema(**data)
+    assert schema.source_path == "/user/test/old_folder"
+    assert schema.destination_path == "/user/test/new_folder"
+
+  def test_special_characters_in_paths(self):
+    data = {
+      "source_path": "/user/test/file with spaces & special-chars_123.txt",
+      "destination_path": "/user/test/file (renamed) @ 2024.txt",
+    }
+    schema = RenameSchema(**data)
+    assert schema.source_path == "/user/test/file with spaces & special-chars_123.txt"
+    assert schema.destination_path == "/user/test/file (renamed) @ 2024.txt"
+
+  def test_unicode_characters_in_paths(self):
+    data = {"source_path": "/user/test/文档.txt", "destination_path": "/user/test/документ.txt"}
+    schema = RenameSchema(**data)
+    assert schema.source_path == "/user/test/文档.txt"
+    assert schema.destination_path == "/user/test/документ.txt"
+
+  def test_very_long_paths(self):
+    long_path = "/user/test/" + "a" * 200 + "/file.txt"
+    data = {"source_path": long_path, "destination_path": long_path + "_new"}
+    schema = RenameSchema(**data)
+    assert schema.source_path == long_path
+    assert schema.destination_path == long_path + "_new"
+
+  def test_multiple_validation_errors(self):
+    data = {"source_path": "  ", "destination_path": "../etc/passwd"}
+    with pytest.raises(ValidationError) as exc_info:
+      RenameSchema(**data)
+    error_str = str(exc_info.value)
+    assert "Path cannot be empty" in error_str
+    assert "Path traversal patterns are not allowed" in error_str
+
+  def test_s3_paths(self):
+    data = {"source_path": "s3a://bucket/folder/file.txt", "destination_path": "s3a://bucket/folder/file_renamed.txt"}
+    schema = RenameSchema(**data)
+    assert schema.source_path == "s3a://bucket/folder/file.txt"
+    assert schema.destination_path == "s3a://bucket/folder/file_renamed.txt"
+
+  def test_both_allow_and_restrict_lists_configured(self):
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt", ".csv", ".exe"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing([".exe", ".bat"])
+
+    try:
+      # File in allow list but also in restrict list
+      data = {"source_path": "/user/test/script.txt", "destination_path": "/user/test/script.exe"}
+      with pytest.raises(ValidationError) as exc_info:
+        RenameSchema(**data)
+      assert "is restricted" in str(exc_info.value)
+
+      # File in allow list and not in restrict list
+      data = {"source_path": "/user/test/data.txt", "destination_path": "/user/test/data.csv"}
+      schema = RenameSchema(**data)
+      assert schema.destination_path == "/user/test/data.csv"
+    finally:
+      reset_allow()
+      reset_restrict()

+ 20 - 0
apps/filebrowser/src/filebrowser/serializers.py

@@ -14,8 +14,11 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+from pydantic import ValidationError
 from rest_framework import serializers
 
+from filebrowser.schemas import RenameSchema
+
 
 class UploadFileSerializer(serializers.Serializer):
   """
@@ -24,3 +27,20 @@ class UploadFileSerializer(serializers.Serializer):
 
   destination_path = serializers.CharField(required=True, allow_blank=False)
   overwrite = serializers.BooleanField(default=False)
+
+
+class RenameSerializer(serializers.Serializer):
+  """
+  Validates the parameters for the file/directory rename API.
+  """
+
+  source_path = serializers.CharField(required=True, allow_blank=False)
+  destination_path = serializers.CharField(required=True, allow_blank=False)
+
+  def validate(self, data):
+    try:
+      RenameSchema.model_validate(data)
+    except ValidationError as e:
+      raise serializers.ValidationError(e.errors())
+
+    return data

+ 194 - 2
apps/filebrowser/src/filebrowser/serializers_tests.py

@@ -15,7 +15,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from filebrowser.serializers import UploadFileSerializer
+from filebrowser.conf import ALLOW_FILE_EXTENSIONS, RESTRICT_FILE_EXTENSIONS
+from filebrowser.serializers import RenameSerializer, UploadFileSerializer
 
 
 class TestUploadFileSerializer:
@@ -68,7 +69,6 @@ class TestUploadFileSerializer:
     assert serializer.validated_data["overwrite"] is False
 
   def test_overwrite_numeric_values(self):
-    """Test serializer handles numeric boolean values correctly."""
     # Test with 1
     serializer = UploadFileSerializer(data={"destination_path": "s3a://test_bucket/test/", "overwrite": 1})
 
@@ -119,3 +119,195 @@ class TestUploadFileSerializer:
       serializer = UploadFileSerializer(data={"destination_path": path})
       assert serializer.is_valid(), f"Path '{path}' should be valid. Errors: {serializer.errors}"
       assert serializer.validated_data["destination_path"] == path
+
+
+class TestRenameSerializer:
+  def test_valid_data_absolute_paths(self):
+    serializer = RenameSerializer(
+      data={"source_path": "/user/test/documents/report.txt", "destination_path": "/user/test/documents/report_v2.txt"}
+    )
+
+    assert serializer.is_valid(), f"Serializer validation failed: {serializer.errors}"
+    assert serializer.validated_data["source_path"] == "/user/test/documents/report.txt"
+    assert serializer.validated_data["destination_path"] == "/user/test/documents/report_v2.txt"
+
+  def test_valid_data_relative_destination(self):
+    serializer = RenameSerializer(data={"source_path": "/user/test/documents/report.txt", "destination_path": "report_v2.txt"})
+
+    assert serializer.is_valid(), f"Serializer validation failed: {serializer.errors}"
+    assert serializer.validated_data["source_path"] == "/user/test/documents/report.txt"
+    assert serializer.validated_data["destination_path"] == "report_v2.txt"
+
+  def test_missing_source_path(self):
+    serializer = RenameSerializer(data={"destination_path": "/user/test/file_new.txt"})
+
+    assert not serializer.is_valid()
+    assert "source_path" in serializer.errors
+    assert any("required" in str(error).lower() for error in serializer.errors["source_path"])
+
+  def test_missing_destination_path(self):
+    serializer = RenameSerializer(data={"source_path": "/user/test/file.txt"})
+
+    assert not serializer.is_valid()
+    assert "destination_path" in serializer.errors
+    assert any("required" in str(error).lower() for error in serializer.errors["destination_path"])
+
+  def test_empty_source_path(self):
+    serializer = RenameSerializer(data={"source_path": "", "destination_path": "/user/test/file_new.txt"})
+
+    assert not serializer.is_valid()
+    assert "source_path" in serializer.errors
+    assert any("blank" in str(error).lower() for error in serializer.errors["source_path"])
+
+  def test_empty_destination_path(self):
+    serializer = RenameSerializer(data={"source_path": "/user/test/file.txt", "destination_path": ""})
+
+    assert not serializer.is_valid()
+    assert "destination_path" in serializer.errors
+    assert any("blank" in str(error).lower() for error in serializer.errors["destination_path"])
+
+  def test_null_values(self):
+    serializer = RenameSerializer(data={"source_path": None, "destination_path": None})
+
+    assert not serializer.is_valid()
+    assert "source_path" in serializer.errors
+    assert "destination_path" in serializer.errors
+
+  def test_schema_validation_integration(self):
+    # Test path traversal detection
+    serializer = RenameSerializer(data={"source_path": "/user/test/../admin/file.txt", "destination_path": "/user/test/file.txt"})
+
+    assert not serializer.is_valid()
+    errors_str = str(serializer.errors)
+    assert "Path traversal patterns are not allowed" in errors_str
+
+  def test_hash_character_validation(self):
+    serializer = RenameSerializer(data={"source_path": "/user/test/file.txt", "destination_path": "/user/test/file#new.txt"})
+
+    assert not serializer.is_valid()
+    errors_str = str(serializer.errors)
+    assert "Hashes are not allowed" in errors_str
+
+  def test_file_extension_restriction(self):
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing(None)
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing([".exe", ".bat"])
+
+    try:
+      serializer = RenameSerializer(data={"source_path": "/user/test/script.txt", "destination_path": "/user/test/script.exe"})
+
+      assert not serializer.is_valid()
+      errors_str = str(serializer.errors)
+      assert "is restricted" in errors_str
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_file_extension_allow_list(self):
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt", ".csv"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
+
+    try:
+      # Not allowed extension
+      serializer = RenameSerializer(data={"source_path": "/user/test/data.txt", "destination_path": "/user/test/data.xml"})
+
+      assert not serializer.is_valid()
+      errors_str = str(serializer.errors)
+      assert "is not permitted" in errors_str
+
+      # Allowed extension
+      serializer = RenameSerializer(data={"source_path": "/user/test/data.txt", "destination_path": "/user/test/data.csv"})
+
+      assert serializer.is_valid()
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_whitespace_trimming(self):
+    serializer = RenameSerializer(data={"source_path": "  /user/test/file.txt  ", "destination_path": "  /user/test/file_new.txt  "})
+
+    assert serializer.is_valid()
+    assert serializer.validated_data["source_path"] == "/user/test/file.txt"
+    assert serializer.validated_data["destination_path"] == "/user/test/file_new.txt"
+
+  def test_extra_fields_ignored(self):
+    serializer = RenameSerializer(
+      data={
+        "source_path": "/user/test/file.txt",
+        "destination_path": "/user/test/file_new.txt",
+        "extra_field": "should be ignored",
+        "another_field": 123,
+      }
+    )
+
+    assert serializer.is_valid()
+    assert serializer.validated_data["source_path"] == "/user/test/file.txt"
+    assert serializer.validated_data["destination_path"] == "/user/test/file_new.txt"
+    assert "extra_field" not in serializer.validated_data
+
+  def test_various_path_formats(self):
+    valid_paths = [
+      ("/user/test/file.txt", "/user/test/file_new.txt"),
+      ("s3a://bucket/folder/file.txt", "s3a://bucket/folder/file_new.txt"),
+      ("hdfs://namenode:9000/user/file.txt", "hdfs://namenode:9000/user/file_new.txt"),
+      ("abfs://container@account/path/file.txt", "abfs://container@account/path/file_new.txt"),
+      ("/path with spaces/file.txt", "/path with spaces/file_new.txt"),
+      ("/user/test/文档.txt", "/user/test/文档_新.txt"),  # Unicode paths
+    ]
+
+    for source, dest in valid_paths:
+      serializer = RenameSerializer(data={"source_path": source, "destination_path": dest})
+      assert serializer.is_valid(), f"Paths '{source}' -> '{dest}' should be valid. Errors: {serializer.errors}"
+
+  def test_complex_validation_errors(self):
+    serializer = RenameSerializer(
+      data={
+        "source_path": "  ",  # Empty after trimming
+        "destination_path": "../etc/passwd#hacked",  # Multiple issues
+      }
+    )
+
+    assert not serializer.is_valid()
+    errors_str = str(serializer.errors)
+    # The blank validation happens at serializer level
+    assert "blank" in errors_str.lower()
+
+  def test_same_extension_rename(self):
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing([".exe"])
+
+    try:
+      serializer = RenameSerializer(data={"source_path": "/user/test/document.txt", "destination_path": "/user/test/document_renamed.txt"})
+
+      assert serializer.is_valid()
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_directory_rename(self):
+    serializer = RenameSerializer(data={"source_path": "/user/test/old_folder", "destination_path": "/user/test/new_folder"})
+
+    assert serializer.is_valid()
+
+  def test_special_characters_in_names(self):
+    serializer = RenameSerializer(
+      data={"source_path": "/user/test/file & document (1).txt", "destination_path": "/user/test/file @ document [2024].txt"}
+    )
+
+    assert serializer.is_valid()
+
+  def test_very_long_paths(self):
+    long_name = "a" * 200
+    serializer = RenameSerializer(
+      data={"source_path": f"/user/test/{long_name}/file.txt", "destination_path": f"/user/test/{long_name}/file_new.txt"}
+    )
+
+    assert serializer.is_valid()
+
+  def test_validation_error_format(self):
+    serializer = RenameSerializer(data={"source_path": "/user/test/file.txt", "destination_path": "/user/test/file#new.txt"})
+
+    assert not serializer.is_valid()
+    # Check that errors are in the expected format
+    assert isinstance(serializer.errors, dict)
+    # The error should be in non_field_errors since it's from model validation
+    assert any("Hashes are not allowed" in str(error) for errors in serializer.errors.values() for error in errors)

+ 0 - 6
desktop/core/src/desktop/api_public.py

@@ -300,12 +300,6 @@ def storage_touch(request):
   return filebrowser_api.touch(django_request)
 
 
-@api_view(["POST"])
-def storage_rename(request):
-  django_request = get_django_request(request)
-  return filebrowser_api.rename(django_request)
-
-
 @api_view(["GET"])
 def storage_content_summary(request):
   django_request = get_django_request(request)

+ 1 - 1
desktop/core/src/desktop/api_public_urls_v1.py

@@ -112,7 +112,7 @@ urlpatterns += [
   re_path(r'^storage/create/file/?$', api_public.storage_touch, name='storage_touch'),
   re_path(r'^storage/create/directory/?$', api_public.storage_mkdir, name='storage_mkdir'),
   re_path(r'^storage/save/?$', api_public.storage_save_file, name="storage_save_file"),
-  re_path(r'^storage/rename/?$', api_public.storage_rename, name='storage_rename'),
+  re_path(r'^storage/rename/?$', filebrowser_api.rename, name='storage_rename'),
   re_path(r'^storage/move/?$', api_public.storage_move, name='storage_move'),
   re_path(r'^storage/copy/?$', api_public.storage_copy, name='storage_copy'),
   re_path(r'^storage/upload/file/?$', filebrowser_api.UploadFileAPI.as_view(), name='storage_upload_file'),