action_utils: Introduce utility to move uploaded files

Tests:

- Unit tests pass.

Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: Veiko Aasa <veiko17@disroot.org>

plinth/action_utils.py

@@ -688,3 +688,63 @@ def podman_uninstall(container_name: str, volume_name: str, image_name: str,
     service_file.unlink(missing_ok=True)
     shutil.rmtree(volume_path, ignore_errors=True)
     service_daemon_reload()
+
+
+def move_uploaded_file(source: str | pathlib.Path,
+                       destination_dir: str | pathlib.Path,
+                       destination_file_name: str,
+                       allow_overwrite: bool = False, user: str = 'root',
+                       group: str = 'root', permissions: int = 0o644):
+    """Move an uploaded file from Django upload directory to a destination.
+
+    Sets the expected ownership and permissions on the destination file. If
+    possible, performs a simple rename operation. Otherwise, copies the file to
+    the destination.
+
+    The source must be regular file under the currently configured Django file
+    upload directory. It must be a absolute path that can be verified to be
+    under Django settings FILE_UPLOAD_TEMP_DIR.
+
+    The destination_dir must be a directory. destination_file_name must be a
+    simple file name without any other path components. When concatenated
+    together, they specify the full destination path to move the file to.
+
+    If allow_overwrite is set to False and destination file exists, an
+    exception is raised.
+    """
+    from plinth import settings
+
+    if isinstance(source, str):
+        source = pathlib.Path(source)
+
+    if isinstance(destination_dir, str):
+        destination_dir = pathlib.Path(destination_dir)
+
+    source = source.resolve(strict=True)
+    destination_dir = destination_dir.resolve()
+
+    # Verify source file
+    if not source.is_file():
+        raise ValueError('Source is not a file')
+
+    tmp_dir = pathlib.Path(getattr(settings, 'FILE_UPLOAD_TEMP_DIR', '/tmp'))
+    if not source.is_relative_to(tmp_dir):
+        raise ValueError('Uploaded file is not in expected temp directory')
+
+    # Verify destination directory
+    if not destination_dir.is_dir():
+        raise ValueError('Destination is not a directory')
+
+    # Verify destination file name
+    if len(pathlib.Path(destination_file_name).parts) != 1:
+        raise ValueError('Invalid destination file name')
+
+    destination = destination_dir / destination_file_name
+
+    if destination.exists() and not allow_overwrite:
+        raise FileExistsError('Destination already exists')
+
+    # Move or copy
+    shutil.move(source, destination)
+    shutil.chown(destination, user, group)
+    destination.chmod(permissions)

plinth/tests/test_action_utils.py

@@ -6,16 +6,16 @@ Test module for key/value store.
 import json
 import pathlib
 import subprocess
-from unittest.mock import patch
+from unittest.mock import call, patch
 
 import pytest
 
 from plinth.action_utils import (get_addresses, get_hostname,
-                                 is_systemd_running, service_action,
-                                 service_disable, service_enable,
-                                 service_is_enabled, service_is_running,
-                                 service_reload, service_restart,
-                                 service_start, service_stop,
+                                 is_systemd_running, move_uploaded_file,
+                                 service_action, service_disable,
+                                 service_enable, service_is_enabled,
+                                 service_is_running, service_reload,
+                                 service_restart, service_start, service_stop,
                                  service_try_reload_or_restart,
                                  service_try_restart, service_unmask)
 
@@ -140,3 +140,92 @@ def test_get_addresses():
     assert len(ips) > 3  # min: ip, 2x'localhost', hostname
     for address in ips:
         assert address['kind'] in ('4', '6')
+
+
+@pytest.fixture(name='upload_dir')
+def fixture_update_dir(tmp_path):
+    """Patch Django file upload directory."""
+    tmp_path /= 'source'
+    tmp_path.mkdir()
+
+    import plinth.settings
+    old_value = plinth.settings.FILE_UPLOAD_TEMP_DIR
+    plinth.settings.FILE_UPLOAD_TEMP_DIR = tmp_path
+    yield tmp_path
+    plinth.settings.FILE_UPLOAD_TEMP_DIR = old_value
+
+
+def test_move_uploaded_file(tmp_path, upload_dir):
+    """Test moving Django uploaded file to destination directory."""
+    tmp_path /= 'destination'
+    tmp_path.mkdir()
+
+    # Source file does not exist
+    source = tmp_path / 'does-non-exist'
+    destination = tmp_path / 'destination'
+    destination_file_name = 'destination-file-name'
+    with pytest.raises(FileNotFoundError):
+        move_uploaded_file(source, destination, destination_file_name)
+
+    # Source is not a file
+    source = tmp_path / 'source-dir'
+    source.mkdir()
+    with pytest.raises(ValueError, match='Source is not a file'):
+        move_uploaded_file(source, destination, destination_file_name)
+
+    # Source is not in expected temporary upload directory
+    source = tmp_path / 'source-file'
+    source.touch()
+    with pytest.raises(
+            ValueError,
+            match='Uploaded file is not in expected temp directory'):
+        move_uploaded_file(source, destination, destination_file_name)
+
+    # Destination does not exist
+    source = upload_dir / 'source-file'
+    source.touch()
+    with pytest.raises(ValueError, match='Destination is not a directory'):
+        move_uploaded_file(source, destination, destination_file_name)
+
+    # Destination is not a file
+    destination.touch()
+    with pytest.raises(ValueError, match='Destination is not a directory'):
+        move_uploaded_file(source, destination, destination_file_name)
+
+    # Destination file name is a multi-component path
+    destination.unlink()
+    destination.mkdir()
+    destination_file_name = '../destination-file-name'
+    with pytest.raises(ValueError, match='Invalid destination file name'):
+        move_uploaded_file(source, destination, destination_file_name)
+
+    # Destination file exists and override is not allowed
+    destination_file_name = 'destination-file-exists'
+    (destination / destination_file_name).touch()
+    with pytest.raises(FileExistsError, match='Destination already exists'):
+        move_uploaded_file(source, destination, destination_file_name)
+
+    with pytest.raises(FileExistsError, match='Destination already exists'):
+        move_uploaded_file(source, destination, destination_file_name,
+                           allow_overwrite=False)
+
+    # Successful move
+    with patch('shutil.chown') as chown:
+        destination_file = destination / destination_file_name
+        destination_file.unlink()
+        source.write_text('x-contents-1')
+        move_uploaded_file(source, destination, destination_file_name)
+        chown.mock_calls = [call(destination_file, 'root', 'root')]
+        assert destination_file.stat().st_mode & 0o777 == 0o644
+        assert destination_file.read_text() == 'x-contents-1'
+        assert not source.exists()
+
+        chown.reset_mock()
+        source.write_text('x-contents-2')
+        move_uploaded_file(source, destination, destination_file_name,
+                           allow_overwrite=True, user='x-user',
+                           group='x-group', permissions=0o600)
+        chown.mock_calls = [call(destination_file, 'x-user', 'x-group')]
+        assert destination_file.stat().st_mode & 0o777 == 0o600
+        assert destination_file.read_text() == 'x-contents-2'
+        assert not source.exists()