Browse Source

[filebrowser] Add file extension restrictions to old APIs for rename and touch operations

Harsh Gupta 1 tháng trước cách đây
mục cha
commit
2480e0ee4e

+ 38 - 0
apps/filebrowser/src/filebrowser/views.py

@@ -93,6 +93,7 @@ from filebrowser.forms import (
 )
 )
 from filebrowser.lib import xxd
 from filebrowser.lib import xxd
 from filebrowser.lib.rwx import filetype, rwx
 from filebrowser.lib.rwx import filetype, rwx
+from filebrowser.utils import is_file_upload_allowed
 from hadoop.core_site import get_trash_interval
 from hadoop.core_site import get_trash_interval
 from hadoop.fs.exceptions import WebHdfsException
 from hadoop.fs.exceptions import WebHdfsException
 from hadoop.fs.fsutils import do_overwrite_save
 from hadoop.fs.fsutils import do_overwrite_save
@@ -1225,6 +1226,11 @@ def generic_op(form_class, request, op, parameter_names, piggyback=None, templat
       args = arg_extractor(request, form, parameter_names)
       args = arg_extractor(request, form, parameter_names)
       try:
       try:
         op(*args)
         op(*args)
+      except PopupException as e:
+        if is_ajax(request):
+          return JsonResponse({'detail': str(e)}, status=500)
+        else:
+          raise
       except (IOError, WebHdfsException) as e:
       except (IOError, WebHdfsException) as e:
         msg = _("Cannot perform operation.")
         msg = _("Cannot perform operation.")
         raise PopupException(msg, detail=e)
         raise PopupException(msg, detail=e)
@@ -1269,6 +1275,24 @@ def generic_op(form_class, request, op, parameter_names, piggyback=None, templat
   return render(template, request, ret)
   return render(template, request, ret)
 
 
 
 
+def _validate_file_extension_allowed(filename):
+  """
+  Validate that a file's extension is allowed based on configured restrictions.
+
+  Args:
+    filename: The filename to validate
+
+  Raises:
+    PopupException: If the file extension is not allowed
+
+  Returns:
+    None if validation passes
+  """
+  is_allowed, error_message = is_file_upload_allowed(filename)
+  if not is_allowed:
+    raise PopupException(error_message)
+
+
 def rename(request):
 def rename(request):
   def smart_rename(src_path, dest_path):
   def smart_rename(src_path, dest_path):
     """If dest_path doesn't have a directory specified, use same dir."""
     """If dest_path doesn't have a directory specified, use same dir."""
@@ -1277,6 +1301,16 @@ def rename(request):
     if "/" not in dest_path:
     if "/" not in dest_path:
       src_dir = os.path.dirname(src_path)
       src_dir = os.path.dirname(src_path)
       dest_path = request.fs.join(src_dir, dest_path)
       dest_path = request.fs.join(src_dir, dest_path)
+
+    # Extract file extensions from source and destination paths
+    _, source_ext = os.path.splitext(src_path)
+    dest_filename = os.path.basename(dest_path)
+    _, dest_ext = os.path.splitext(dest_filename)
+
+    # Check if extension is changing and validate if allowed
+    if source_ext.lower() != dest_ext.lower():
+      _validate_file_extension_allowed(dest_filename)
+
     if request.fs.exists(dest_path):
     if request.fs.exists(dest_path):
       raise PopupException(_('The destination path "%s" already exists.') % dest_path)
       raise PopupException(_('The destination path "%s" already exists.') % dest_path)
     request.fs.rename(src_path, dest_path)
     request.fs.rename(src_path, dest_path)
@@ -1310,6 +1344,10 @@ def touch(request):
     # No absolute path specification allowed.
     # No absolute path specification allowed.
     if posixpath.sep in name:
     if posixpath.sep in name:
       raise PopupException(_("Could not name file \"%s\": Slashes are not allowed in filenames." % name))
       raise PopupException(_("Could not name file \"%s\": Slashes are not allowed in filenames." % name))
+
+    # Validate file extension
+    _validate_file_extension_allowed(name)
+
     request.fs.create(
     request.fs.create(
         request.fs.join(
         request.fs.join(
             (path.encode('utf-8') if not isinstance(path, str) else path),
             (path.encode('utf-8') if not isinstance(path, str) else path),

+ 232 - 2
apps/filebrowser/src/filebrowser/views_test.py

@@ -43,11 +43,18 @@ from aws.s3.s3test_utils import get_test_bucket
 from azure.conf import ABFS_CLUSTERS, is_abfs_enabled, is_adls_enabled
 from azure.conf import ABFS_CLUSTERS, is_abfs_enabled, is_adls_enabled
 from desktop.conf import is_ofs_enabled, is_oozie_enabled, OZONE, RAZ, USE_STORAGE_CONNECTORS
 from desktop.conf import is_ofs_enabled, is_oozie_enabled, OZONE, RAZ, USE_STORAGE_CONNECTORS
 from desktop.lib.django_test_util import make_logged_in_client
 from desktop.lib.django_test_util import make_logged_in_client
+from desktop.lib.exceptions_renderable import PopupException
 from desktop.lib.test_utils import add_permission, add_to_group, grant_access, remove_from_group
 from desktop.lib.test_utils import add_permission, add_to_group, grant_access, remove_from_group
 from desktop.lib.view_util import location_to_url
 from desktop.lib.view_util import location_to_url
-from filebrowser.conf import ENABLE_EXTRACT_UPLOADED_ARCHIVE, MAX_SNAPPY_DECOMPRESSION_SIZE, REMOTE_STORAGE_HOME
+from filebrowser.conf import (
+  ALLOW_FILE_EXTENSIONS,
+  ENABLE_EXTRACT_UPLOADED_ARCHIVE,
+  MAX_SNAPPY_DECOMPRESSION_SIZE,
+  REMOTE_STORAGE_HOME,
+  RESTRICT_FILE_EXTENSIONS,
+)
 from filebrowser.lib.rwx import expand_mode
 from filebrowser.lib.rwx import expand_mode
-from filebrowser.views import _normalize_path, _read_parquet, snappy_installed
+from filebrowser.views import _normalize_path, _read_parquet, _validate_file_extension_allowed, snappy_installed
 from hadoop import pseudo_hdfs4
 from hadoop import pseudo_hdfs4
 from hadoop.conf import UPLOAD_CHUNK_SIZE
 from hadoop.conf import UPLOAD_CHUNK_SIZE
 from hadoop.fs.webhdfs import WebHdfs
 from hadoop.fs.webhdfs import WebHdfs
@@ -259,6 +266,229 @@ class TestFileBrowser:
         ), response.content
         ), response.content
 
 
 
 
+class TestFileExtensionRestrictions:
+  """Test file extension restrictions for filebrowser rename and touch APIs validation logic."""
+
+  # Tests for rename operation validation logic
+  def test_rename_without_extension_change(self):
+    """Test rename validation when extension doesn't change - should pass regardless of restrictions."""
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
+
+    try:
+      src_path = '/user/test/document.txt'
+      dest_path = '/user/test/document_renamed.txt'
+
+      # Extract extensions (same logic as smart_rename)
+      _, source_ext = os.path.splitext(src_path)
+      dest_filename = os.path.basename(dest_path)
+      _, dest_ext = os.path.splitext(dest_filename)
+
+      # Validation: only check if extension changes
+      if source_ext.lower() != dest_ext.lower():
+        _validate_file_extension_allowed(dest_filename)
+
+      # Should pass - extensions are the same
+      assert source_ext.lower() == dest_ext.lower()
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_rename_with_extension_change_to_restricted(self):
+    """Test rename validation with change to restricted extension - should fail."""
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing(None)
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing([".exe", ".bat", ".cmd"])
+
+    try:
+      src_path = '/user/test/script.txt'
+      dest_path = '/user/test/script.exe'
+
+      # Extract extensions (same logic as smart_rename)
+      _, source_ext = os.path.splitext(src_path)
+      dest_filename = os.path.basename(dest_path)
+      _, dest_ext = os.path.splitext(dest_filename)
+
+      # Validation: only check if extension changes
+      if source_ext.lower() != dest_ext.lower():
+        with pytest.raises(PopupException) as exc_info:
+          _validate_file_extension_allowed(dest_filename)
+        assert "is restricted" in str(exc_info.value)
+      else:
+        pytest.fail("Extensions should be different in this test")
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_rename_with_extension_change_to_allowed(self):
+    """Test rename validation with change to allowed extension - should pass."""
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt", ".csv", ".json"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
+
+    try:
+      src_path = '/user/test/data.txt'
+      dest_path = '/user/test/data.csv'
+
+      # Extract extensions (same logic as smart_rename)
+      _, source_ext = os.path.splitext(src_path)
+      dest_filename = os.path.basename(dest_path)
+      _, dest_ext = os.path.splitext(dest_filename)
+
+      # Validation: only check if extension changes
+      if source_ext.lower() != dest_ext.lower():
+        _validate_file_extension_allowed(dest_filename)  # Should not raise
+
+      # Should pass - .csv is in allow list
+      assert source_ext.lower() != dest_ext.lower()
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_rename_with_extension_change_to_not_allowed(self):
+    """Test rename validation with change to non-allowed extension - should fail."""
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt", ".csv"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
+
+    try:
+      src_path = '/user/test/data.txt'
+      dest_path = '/user/test/data.xml'
+
+      # Extract extensions (same logic as smart_rename)
+      _, source_ext = os.path.splitext(src_path)
+      dest_filename = os.path.basename(dest_path)
+      _, dest_ext = os.path.splitext(dest_filename)
+
+      # Validation: only check if extension changes
+      if source_ext.lower() != dest_ext.lower():
+        with pytest.raises(PopupException) as exc_info:
+          _validate_file_extension_allowed(dest_filename)
+        assert "is not permitted" in str(exc_info.value)
+      else:
+        pytest.fail("Extensions should be different in this test")
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_rename_case_insensitive_extension_check(self):
+    """Test that extension comparison is case-insensitive."""
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
+
+    try:
+      src_path = '/user/test/document.TXT'
+      dest_path = '/user/test/document_renamed.txt'
+
+      # Extract extensions (same logic as smart_rename)
+      _, source_ext = os.path.splitext(src_path)
+      dest_filename = os.path.basename(dest_path)
+      _, dest_ext = os.path.splitext(dest_filename)
+
+      # Validation: only check if extension changes
+      if source_ext.lower() != dest_ext.lower():
+        _validate_file_extension_allowed(dest_filename)
+
+      # Should pass - extensions are same (case-insensitive: .TXT == .txt)
+      assert source_ext.lower() == dest_ext.lower()
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_rename_directory_not_affected(self):
+    """Test that renaming directories is not affected by file extension restrictions."""
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
+
+    try:
+      src_path = '/user/test/old_folder'
+      dest_path = '/user/test/new_folder'
+
+      # Extract extensions (same logic as smart_rename)
+      _, source_ext = os.path.splitext(src_path)
+      dest_filename = os.path.basename(dest_path)
+      _, dest_ext = os.path.splitext(dest_filename)
+
+      # Validation: only check if extension changes
+      if source_ext.lower() != dest_ext.lower():
+        _validate_file_extension_allowed(dest_filename)
+
+      # Should pass - directories have no extensions, so extensions are same (both empty)
+      assert source_ext == dest_ext == ''
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  # Tests for touch (create file) operation
+  def test_touch_with_allowed_extension(self):
+    """Test creating file with allowed extension - should pass validation."""
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt", ".csv", ".json"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
+
+    try:
+      name = 'newfile.txt'
+      # Should not raise exception
+      _validate_file_extension_allowed(name)
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_touch_with_restricted_extension(self):
+    """Test creating file with restricted extension - should fail validation."""
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing(None)
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing([".exe", ".bat", ".cmd"])
+
+    try:
+      name = 'malicious.exe'
+      # Should raise PopupException
+      with pytest.raises(PopupException) as exc_info:
+        _validate_file_extension_allowed(name)
+      assert "is restricted" in str(exc_info.value)
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_touch_with_not_allowed_extension(self):
+    """Test creating file with non-allowed extension when allow list configured - should fail validation."""
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt", ".csv"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing(None)
+
+    try:
+      name = 'data.xml'
+      # Should raise PopupException
+      with pytest.raises(PopupException) as exc_info:
+        _validate_file_extension_allowed(name)
+      assert "is not permitted" in str(exc_info.value)
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_touch_without_extension(self):
+    """Test creating file without extension - should pass when no restrictions."""
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing(None)
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing([".exe"])
+
+    try:
+      name = 'README'
+      # Should not raise exception
+      _validate_file_extension_allowed(name)
+    finally:
+      reset_allow()
+      reset_restrict()
+
+  def test_touch_both_allow_and_restrict_lists(self):
+    """Test that restrict list takes precedence when both lists are configured."""
+    reset_allow = ALLOW_FILE_EXTENSIONS.set_for_testing([".txt", ".exe"])
+    reset_restrict = RESTRICT_FILE_EXTENSIONS.set_for_testing([".exe", ".bat"])
+
+    try:
+      name = 'program.exe'
+      # Should raise PopupException (restrict takes precedence)
+      with pytest.raises(PopupException) as exc_info:
+        _validate_file_extension_allowed(name)
+      assert "is restricted" in str(exc_info.value)
+    finally:
+      reset_allow()
+      reset_restrict()
+
+
 @pytest.mark.requires_hadoop
 @pytest.mark.requires_hadoop
 @pytest.mark.integration
 @pytest.mark.integration
 class TestFileBrowserWithHadoop(object):
 class TestFileBrowserWithHadoop(object):