Browse Source

[api] Prevent renaming files to restricted file types (#3886)

- We need to prevent users to rename files to restricted file types. This is to counter the scenario where user uploads a file with different type then renames it to a different type for malicious intent. 
- Although we are not restricting when the source file type is already restricted and then user is renaming to same file type because we don't know how the file is present in the first place in the filesystem as Hue will not allow normal upload of restricted file type.
Harsh Gupta 1 year ago
parent
commit
251b017ee4
2 changed files with 134 additions and 7 deletions
  1. 18 5
      apps/filebrowser/src/filebrowser/api.py
  2. 116 2
      apps/filebrowser/src/filebrowser/api_test.py

+ 18 - 5
apps/filebrowser/src/filebrowser/api.py

@@ -496,18 +496,31 @@ 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)
+
+  # Extract file extensions from paths
+  _, source_path_ext = os.path.splitext(source_path)
+  _, dest_path_ext = os.path.splitext(destination_path)
+
+  restricted_file_types = [ext.lower() for ext in RESTRICT_FILE_EXTENSIONS.get()]
+  # 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)
+
+   # Check if destination path contains a hash character
   if "#" in destination_path:
-    return HttpResponse(
-      f"Error creating {os.path.basename(source_path)} to {destination_path}: Hashes are not allowed in file or directory names", status=400
-    )
+    return HttpResponse("Hashes are not allowed in file or directory names. Please choose a different name.", status=400)
 
-  # If dest_path doesn't have a directory specified, use same directory.
+  # 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)
 
+  # Check if destination path already exists
   if request.fs.exists(destination_path):
-    return HttpResponse(f"The destination path {destination_path} already exists.", status=500)  # TODO: Status code?
+    return HttpResponse(f"The destination path {destination_path} already exists.", status=409)
 
   request.fs.rename(source_path, destination_path)
   return HttpResponse(status=200)

+ 116 - 2
apps/filebrowser/src/filebrowser/api_test.py

@@ -20,14 +20,14 @@ from unittest.mock import Mock, patch
 
 from django.core.files.uploadedfile import SimpleUploadedFile
 
-from filebrowser.api import upload_file
+from filebrowser.api import rename, upload_file
 from filebrowser.conf import (
   MAX_FILE_SIZE_UPLOAD_LIMIT,
   RESTRICT_FILE_EXTENSIONS,
 )
 
 
-class TestNormalFileUpload:
+class TestSimpleFileUploadAPI:
   def test_file_upload_success(self):
     with patch('filebrowser.api.string_io') as string_io:
       with patch('filebrowser.api.stat_absolute_path') as stat_absolute_path:
@@ -232,3 +232,117 @@ class TestNormalFileUpload:
       finally:
         for reset in resets:
           reset()
+
+
+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'},
+      body=Mock(),
+      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('')
+    try:
+      response = rename(request)
+
+      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()
+
+  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'},
+      body=Mock(),
+      fs=Mock(
+        rename=Mock(),
+      ),
+    )
+    reset = RESTRICT_FILE_EXTENSIONS.set_for_testing('.exe,.txt')
+    try:
+      response = rename(request)
+
+      assert response.status_code == 403
+      assert response.content.decode('utf-8') == 'Cannot rename file to a restricted file type: ".exe"'
+    finally:
+      reset()
+
+  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'},
+      body=Mock(),
+      fs=Mock(
+        rename=Mock(),
+      ),
+    )
+    reset = RESTRICT_FILE_EXTENSIONS.set_for_testing('')
+    try:
+      response = rename(request)
+
+      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()
+
+  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'},
+      body=Mock(),
+      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('')
+    try:
+      response = rename(request)
+
+      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()
+
+  def test_rename_no_source_path(self):
+    request = Mock(
+      method='POST',
+      POST={'destination_path': 'new_name.txt'},
+      body=Mock(),
+      fs=Mock(
+        rename=Mock(),
+      ),
+    )
+    reset = RESTRICT_FILE_EXTENSIONS.set_for_testing('')
+    try:
+      response = rename(request)
+
+      assert response.status_code == 400
+      assert response.content.decode('utf-8') == 'Missing required parameters: source_path and destination_path'
+    finally:
+      reset()
+
+  def test_rename_no_destination_path(self):
+    request = Mock(
+      method='POST',
+      POST={'source_path': 's3a://test-bucket/test-user/source.txt'},
+      body=Mock(),
+      fs=Mock(
+        rename=Mock(),
+      ),
+    )
+    reset = RESTRICT_FILE_EXTENSIONS.set_for_testing('')
+    try:
+      response = rename(request)
+
+      assert response.status_code == 400
+      assert response.content.decode('utf-8') == 'Missing required parameters: source_path and destination_path'
+    finally:
+      reset()