diff --git a/plinth/action_utils.py b/plinth/action_utils.py index 667714923..f8104d73b 100644 --- a/plinth/action_utils.py +++ b/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) diff --git a/plinth/tests/test_action_utils.py b/plinth/tests/test_action_utils.py index 3887c29ef..b1a01416e 100644 --- a/plinth/tests/test_action_utils.py +++ b/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()