0%
diff --git a/storages/templates/storage.html b/storages/templates/storage.html
index e4e726d5..e82fe939 100644
--- a/storages/templates/storage.html
+++ b/storages/templates/storage.html
@@ -223,8 +223,11 @@
}
});
+ var pendingUpload = null;
+ var uploadInProgress = false;
$('#isoUploadForm').on('submit', function(e) {
e.preventDefault();
+ if (uploadInProgress) return;
var fileInput = $('#id_file')[0];
if (fileInput.files.length === 0) {
@@ -234,17 +237,42 @@
var file = fileInput.files[0];
var fileName = file.name;
- var chunkSize = 10 * 1024 * 1024; // 10MB
+ var errorMessage = $('#upload-error-message');
+ if (!/\.iso$/i.test(fileName) || fileName.length <= 4) {
+ errorMessage.text("{% trans 'Only .iso files can be uploaded.'|escapejs %}").show();
+ return;
+ }
+ if (file.size === 0) {
+ alert("{% trans 'The selected file is empty.'|escapejs %}");
+ return;
+ }
+ var chunkSize = 32 * 1024 * 1024; // Fewer SSH connections per upload
var totalChunks = Math.ceil(file.size / chunkSize);
- var chunkIndex = 0;
+ var fileSignature = [fileName, file.size, file.lastModified].join(':');
+ if (!pendingUpload || pendingUpload.signature !== fileSignature) {
+ pendingUpload = {
+ signature: fileSignature,
+ chunkIndex: 0,
+ uploadId: (window.crypto && window.crypto.randomUUID) ? window.crypto.randomUUID() : (Date.now() + '-' + Math.random().toString(36).substring(2, 10))
+ };
+ }
+ var chunkIndex = pendingUpload.chunkIndex;
+ var uploadId = pendingUpload.uploadId;
+ var retryCount = 0;
+ uploadInProgress = true;
var progressBar = $('#upload-progress-bar');
var progressContainer = $('#upload-progress-container');
- var errorMessage = $('#upload-error-message');
+ var uploadStatus = $('#upload-status');
+ var uploadStatusText = $('#upload-status-text');
+ var uploadButton = $('#isoUploadForm button[type="submit"]');
progressContainer.show();
progressBar.width('0%').attr('aria-valuenow', 0).text('0%');
errorMessage.hide();
+ uploadStatusText.text("{% trans 'Starting upload...'|escapejs %}");
+ uploadStatus.removeClass('d-none').addClass('d-flex');
+ uploadButton.prop('disabled', true);
function uploadChunk() {
if (chunkIndex >= totalChunks) {
@@ -260,6 +288,8 @@
formData.append('file_name', fileName);
formData.append('chunk_index', chunkIndex);
formData.append('total_chunks', totalChunks);
+ formData.append('file_size', file.size);
+ formData.append('upload_id', uploadId);
formData.append('iso_upload', 'true');
formData.append('csrfmiddlewaretoken', '{{ csrf_token }}');
@@ -270,17 +300,39 @@
processData: false,
contentType: false,
success: function(data) {
+ retryCount = 0;
chunkIndex++;
+ pendingUpload.chunkIndex = chunkIndex;
var percentComplete = Math.round((chunkIndex / totalChunks) * 100);
progressBar.width(percentComplete + '%').attr('aria-valuenow', percentComplete).text(percentComplete + '%');
+ uploadStatusText.text("{% trans 'Uploading...'|escapejs %} " + percentComplete + '%');
if (data.reload) {
+ pendingUpload = null;
+ uploadInProgress = false;
location.reload();
} else if (chunkIndex < totalChunks) {
uploadChunk();
+ } else {
+ pendingUpload = null;
+ uploadInProgress = false;
+ uploadStatus.removeClass('d-flex').addClass('d-none');
+ uploadButton.prop('disabled', false);
}
},
error: function(jqXHR, textStatus, errorThrown) {
+ if ((jqXHR.status === 0 || jqXHR.status === 429 || jqXHR.status >= 500) && retryCount < 3) {
+ retryCount++;
+ uploadStatusText.text("{% trans 'Retrying upload...'|escapejs %}");
+ setTimeout(uploadChunk, 1000 * Math.pow(2, retryCount - 1));
+ return;
+ }
+ uploadInProgress = false;
+ uploadStatus.removeClass('d-flex').addClass('d-none');
+ uploadButton.prop('disabled', false);
+ if (jqXHR.status >= 400 && jqXHR.status < 500 && jqXHR.status !== 429) {
+ pendingUpload = null;
+ }
var errorText = jqXHR.responseJSON && jqXHR.responseJSON.error ? jqXHR.responseJSON.error : "{% trans 'An error occurred during upload.'|escapejs %}";
errorMessage.text(errorText).show();
progressContainer.hide();
diff --git a/storages/tests.py b/storages/tests.py
index fd632b98..414fad1f 100644
--- a/storages/tests.py
+++ b/storages/tests.py
@@ -1,6 +1,7 @@
import os
import shutil
import tempfile
+import time
from unittest.mock import MagicMock, patch
from django.core.files.uploadedfile import SimpleUploadedFile
@@ -9,12 +10,102 @@ from django.urls import reverse
from django.contrib.auth import get_user_model
from computes.models import Compute
-from storages.views import handle_uploaded_file
-from vrtManager.connection import CONN_SOCKET, CONN_SSH
+from storages.upload import (
+ handle_uploaded_file,
+ _acquire_upload_lock,
+ _atomic_finalize_ssh,
+ _cleanup_stale_local_uploads,
+ _get_upload_lock_directory,
+ _is_session_completed,
+ _sftp_replace_file,
+)
+from vrtManager.connection import CONN_SOCKET, CONN_SSH, CONN_TCP, CONN_TLS
User = get_user_model()
+class LibvirtUploadTests(TestCase):
+ def setUp(self):
+ self.temp_dir = tempfile.mkdtemp()
+ self.lock_patch = patch("storages.upload._get_upload_lock_directory", return_value=self.temp_dir)
+ self.lock_patch.start()
+ self.conn = MagicMock()
+ self.conn.conn = CONN_TCP
+ self.volumes = {}
+ self.conn.get_volumes.side_effect = lambda: list(self.volumes)
+ self.conn.get_volume.side_effect = lambda name: self.volumes[name]
+
+ def create_volume(xml, flags):
+ from xml.etree import ElementTree
+ root = ElementTree.fromstring(xml)
+ name = root.findtext("name")
+ self.assertNotIn(name, self.volumes)
+ self.assertEqual(root.find("capacity").get("unit"), "bytes")
+ self.assertEqual(root.find("target/format").get("type"), "raw")
+ self.volumes[name] = MagicMock()
+
+ self.conn.pool.createXML.side_effect = create_volume
+ self.conn.wvm.newStream.return_value.send.side_effect = lambda data: len(data)
+
+ def tearDown(self):
+ self.lock_patch.stop()
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+ def upload(self, data, index=0, total=1, upload_id="tcp-session", file_size=None):
+ return handle_uploaded_file(
+ self.conn, "/var/lib/libvirt/images", "ubuntu.iso",
+ SimpleUploadedFile("ubuntu.iso", data), index == total - 1,
+ chunk_index=index, total_chunks=total, upload_id=upload_id,
+ user_id=1, compute_id=2, pool="default", file_size=file_size,
+ )
+
+ def test_tcp_upload_retries_and_stream_offsets(self):
+ self.upload(b"abc", index=0, total=2, file_size=5)
+ self.upload(b"abc", index=0, total=2, file_size=5)
+ self.upload(b"de", index=1, total=2, file_size=5)
+ self.upload(b"de", index=1, total=2, file_size=5)
+ self.conn.pool.createXML.assert_called_once()
+ self.assertEqual(self.volumes["ubuntu.iso"].upload.call_count, 2)
+ self.assertEqual(self.volumes["ubuntu.iso"].upload.call_args.args[1:], (3, 2, 0))
+ self.assertEqual(self.conn.wvm.newStream.return_value.finish.call_count, 2)
+
+ def test_tls_uses_libvirt_stream(self):
+ self.conn.conn = CONN_TLS
+ self.upload(b"iso", file_size=3)
+ self.volumes["ubuntu.iso"].upload.assert_called_once()
+
+ def test_existing_volume_is_preserved(self):
+ self.volumes["ubuntu.iso"] = MagicMock()
+ with self.assertRaises(FileExistsError):
+ self.upload(b"iso", file_size=3)
+ self.conn.pool.createXML.assert_not_called()
+
+ def test_size_and_sequence_are_validated(self):
+ with self.assertRaises(ValueError):
+ self.upload(b"iso", file_size=None)
+ with self.assertRaises(ValueError):
+ self.upload(b"iso", file_size=4)
+ self.conn.pool.createXML.assert_not_called()
+ self.upload(b"ab", index=0, total=2, file_size=4)
+ with self.assertRaises(ValueError):
+ self.upload(b"cd", index=1, total=2, file_size=5)
+ with self.assertRaises(ValueError):
+ self.upload(b"cd", index=1, total=3, file_size=4)
+
+ def test_stream_failure_aborts_and_can_retry(self):
+ stream = self.conn.wvm.newStream.return_value
+ stream.send.side_effect = [1, OSError("connection lost")]
+ with self.assertRaises(OSError):
+ self.upload(b"iso", file_size=3)
+ stream.abort.assert_called_once()
+ self.assertEqual(stream.finish.call_count, 0)
+ stream.send.side_effect = lambda data: len(data)
+ self.upload(b"iso", file_size=3)
+ self.conn.pool.createXML.assert_called_once()
+
+
+
+
class StorageUploadSecurityTests(TestCase):
def setUp(self):
self.temp_dir = tempfile.mkdtemp()
@@ -30,49 +121,507 @@ class StorageUploadSecurityTests(TestCase):
def tearDown(self):
shutil.rmtree(self.temp_dir, ignore_errors=True)
+ def test_handle_uploaded_file_missing_or_invalid_upload_id(self):
+ dummy_chunk = SimpleUploadedFile("test.iso", b"data")
+ # Missing upload_id
+ with self.assertRaises(ValueError) as ctx:
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, "test.iso", dummy_chunk, is_last_chunk=True)
+ self.assertIn("Missing required upload_id", str(ctx.exception))
+
+ # Invalid upload_id format
+ for bad_id in ["", " ", "sess/../id", "id with spaces", "a" * 65]:
+ with self.assertRaises(ValueError):
+ handle_uploaded_file(
+ self.mock_socket_conn,
+ self.temp_dir,
+ "test.iso",
+ dummy_chunk,
+ is_last_chunk=True,
+ upload_id=bad_id,
+ )
+
def test_handle_uploaded_file_invalid_names(self):
dummy_chunk = SimpleUploadedFile("test.iso", b"data")
invalid_names = ["", " ", ".", "..", "test\x00evil.iso", "../evil.iso", "../../etc/shadow"]
for bad_name in invalid_names:
clean_base = os.path.basename(bad_name).strip()
- # bad_name either gets rejected as empty/dot/null or has basename extracted
if not clean_base or clean_base in (".", "..") or "\x00" in clean_base:
with self.assertRaises(ValueError):
- handle_uploaded_file(self.mock_socket_conn, self.temp_dir, bad_name, dummy_chunk, is_last_chunk=True)
+ handle_uploaded_file(
+ self.mock_socket_conn,
+ self.temp_dir,
+ bad_name,
+ dummy_chunk,
+ is_last_chunk=True,
+ upload_id="sess-valid-1",
+ )
+
+ def test_only_iso_extension_is_accepted(self):
+ chunk = SimpleUploadedFile("image.img", b"not checked")
+ for conn in (self.mock_socket_conn, self.mock_ssh_conn):
+ with self.assertRaisesRegex(ValueError, r"Only \.iso files"):
+ handle_uploaded_file(
+ conn, self.temp_dir, "image.img", chunk,
+ is_last_chunk=True, upload_id="iso-extension-test",
+ )
+ handle_uploaded_file(
+ self.mock_socket_conn, self.temp_dir, "image.ISO",
+ SimpleUploadedFile("image.ISO", b"contents are not inspected"),
+ is_last_chunk=True, upload_id="uppercase-iso-test",
+ )
+ self.assertTrue(os.path.exists(os.path.join(self.temp_dir, "image.ISO")))
def test_handle_uploaded_file_successful_socket_upload(self):
dummy_chunk = SimpleUploadedFile("install.iso", b"ISODATA_BLOCK_123")
file_name = "install.iso"
+ upload_id = "sess-install-1"
- # Chunk 1 (is_last_chunk=False)
- handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, dummy_chunk, is_last_chunk=False)
- part_path = os.path.join(self.temp_dir, f"{file_name}.part")
+ # Chunk 1 (is_last_chunk=False, chunk_index=0, total_chunks=2)
+ handle_uploaded_file(
+ self.mock_socket_conn,
+ self.temp_dir,
+ file_name,
+ dummy_chunk,
+ is_last_chunk=False,
+ chunk_index=0,
+ total_chunks=2,
+ upload_id=upload_id,
+ )
+ part_path = os.path.join(self.temp_dir, f"{file_name}.wvc_upload_{upload_id}.part")
final_path = os.path.join(self.temp_dir, file_name)
self.assertTrue(os.path.exists(part_path))
self.assertFalse(os.path.exists(final_path))
- # Chunk 2 (is_last_chunk=True)
+ # Chunk 2 (is_last_chunk=True, chunk_index=1, total_chunks=2)
dummy_chunk2 = SimpleUploadedFile("install.iso", b"_PART2")
- handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, dummy_chunk2, is_last_chunk=True)
+ handle_uploaded_file(
+ self.mock_socket_conn,
+ self.temp_dir,
+ file_name,
+ dummy_chunk2,
+ is_last_chunk=True,
+ chunk_index=1,
+ total_chunks=2,
+ upload_id=upload_id,
+ )
self.assertFalse(os.path.exists(part_path))
self.assertTrue(os.path.exists(final_path))
with open(final_path, "rb") as f:
self.assertEqual(f.read(), b"ISODATA_BLOCK_123_PART2")
+ def test_handle_uploaded_file_idempotent_retry(self):
+ file_name = "retry_test.iso"
+ upload_id = "sess-retry-123"
+ c0 = SimpleUploadedFile(file_name, b"CHUNK0_")
+ c1 = SimpleUploadedFile(file_name, b"CHUNK1_")
+ c2 = SimpleUploadedFile(file_name, b"CHUNK2")
+
+ # Chunk 0 initial
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, c0, is_last_chunk=False, chunk_index=0, total_chunks=3, upload_id=upload_id)
+ # Chunk 0 retry (idempotent no-op!)
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, c0, is_last_chunk=False, chunk_index=0, total_chunks=3, upload_id=upload_id)
+
+ # Chunk 1 initial
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, c1, is_last_chunk=False, chunk_index=1, total_chunks=3, upload_id=upload_id)
+ # Chunk 1 retry (idempotent no-op!)
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, c1, is_last_chunk=False, chunk_index=1, total_chunks=3, upload_id=upload_id)
+
+ # Chunk 2 initial (final chunk)
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, c2, is_last_chunk=True, chunk_index=2, total_chunks=3, upload_id=upload_id)
+
+ # Chunk 2 retry (idempotent no-op, file already finalized!)
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, c2, is_last_chunk=True, chunk_index=2, total_chunks=3, upload_id=upload_id)
+
+ final_file = os.path.join(self.temp_dir, file_name)
+ self.assertTrue(os.path.exists(final_file))
+ with open(final_file, "rb") as f:
+ self.assertEqual(f.read(), b"CHUNK0_CHUNK1_CHUNK2")
+
+ def test_handle_uploaded_file_interrupted_finalization(self):
+ file_name = "interrupted.iso"
+ upload_id = "sess-interrupted-1"
+ c0 = SimpleUploadedFile(file_name, b"PART1")
+ c1 = SimpleUploadedFile(file_name, b"PART2")
+
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, c0, is_last_chunk=False, chunk_index=0, total_chunks=2, upload_id=upload_id)
+
+ # Simulate state where chunk 1 was recorded in .meta and .part, but finalization was interrupted
+ meta_file = os.path.join(self.temp_dir, f"{file_name}.wvc_upload_{upload_id}.meta")
+ part_file = os.path.join(self.temp_dir, f"{file_name}.wvc_upload_{upload_id}.part")
+ with open(part_file, "ab") as f:
+ f.write(b"PART2")
+ import json
+ with open(meta_file, "w") as f:
+ json.dump({
+ "file_name": file_name,
+ "total_chunks": 2,
+ "expected_chunk": 2,
+ "current_size": 10,
+ "created_at": time.time(),
+ "updated_at": time.time()
+ }, f)
+
+ # Client retries final chunk (chunk 1)
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, c1, is_last_chunk=True, chunk_index=1, total_chunks=2, upload_id=upload_id)
+
+ final_file = os.path.join(self.temp_dir, file_name)
+ self.assertTrue(os.path.exists(final_file))
+ with open(final_file, "rb") as f:
+ self.assertEqual(f.read(), b"PART1PART2")
+
+
+ def test_handle_uploaded_file_inconsistent_total_chunks(self):
+ file_name = "inconsistent.iso"
+ upload_id = "sess-inc-1"
+ c0 = SimpleUploadedFile(file_name, b"part0")
+ c1 = SimpleUploadedFile(file_name, b"part1")
+
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, c0, is_last_chunk=False, chunk_index=0, total_chunks=3, upload_id=upload_id)
+
+ with self.assertRaises(ValueError) as ctx:
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, c1, is_last_chunk=False, chunk_index=1, total_chunks=4, upload_id=upload_id)
+ self.assertIn("Inconsistent total_chunks", str(ctx.exception))
+
+ def test_handle_uploaded_file_missing_previous_chunk(self):
+ dummy_chunk = SimpleUploadedFile("broken.iso", b"data")
+ with self.assertRaises(ValueError):
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, "broken.iso", dummy_chunk, is_last_chunk=False, chunk_index=1, total_chunks=2, upload_id="sess-broken-1")
+
+ def test_handle_uploaded_file_destination_exists_raises(self):
+ file_name = "existing.iso"
+ final_path = os.path.join(self.temp_dir, file_name)
+ with open(final_path, "wb") as f:
+ f.write(b"already exists")
+
+ dummy_chunk = SimpleUploadedFile(file_name, b"new data")
+ with self.assertRaises(Exception):
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, dummy_chunk, is_last_chunk=False, chunk_index=0, total_chunks=2, upload_id="sess-exist-1")
+
+ def test_handle_uploaded_file_symlink_defense(self):
+ file_name = "symlink_test.iso"
+ upload_id = "sess-symlink-1"
+ target_file = os.path.join(self.temp_dir, "real.iso")
+ with open(target_file, "wb") as f:
+ f.write(b"real content")
+
+ symlink_part = os.path.join(self.temp_dir, f"{file_name}.wvc_upload_{upload_id}.part")
+ os.symlink(target_file, symlink_part)
+
+ dummy_chunk = SimpleUploadedFile(file_name, b"malicious chunk")
+ with self.assertRaises(PermissionError):
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, file_name, dummy_chunk, is_last_chunk=False, chunk_index=0, total_chunks=2, upload_id=upload_id)
+
def test_handle_uploaded_file_jail_containment(self):
dummy_chunk = SimpleUploadedFile("test.iso", b"data")
- # Attempt traversal outside sandbox
- with patch("os.path.basename", side_effect=lambda x: x): # simulate raw traversal name
+ with patch("os.path.basename", side_effect=lambda x: x):
with self.assertRaises(PermissionError):
- handle_uploaded_file(self.mock_socket_conn, self.temp_dir, "../../../evil.iso", dummy_chunk, is_last_chunk=True)
+ handle_uploaded_file(self.mock_socket_conn, self.temp_dir, "../../../evil.iso", dummy_chunk, is_last_chunk=True, upload_id="sess-jail-1")
+
+ def test_cleanup_stale_local_uploads_strict_regex(self):
+ unrelated_files = [
+ "unrelated.part",
+ "important.lock",
+ "data.meta",
+ "short.123.part",
+ "backup.abcdefgh.meta",
+ "disk.12345678.part",
+ ]
+ for fname in unrelated_files:
+ p = os.path.join(self.temp_dir, fname)
+ with open(p, "wb") as f:
+ f.write(b"keep me")
+ old_time = time.time() - 172800
+ os.utime(p, (old_time, old_time))
+
+ stale_part = os.path.join(self.temp_dir, "ubuntu.iso.wvc_upload_sess-12345678.part")
+ stale_meta = os.path.join(self.temp_dir, "ubuntu.iso.wvc_upload_sess-12345678.meta")
+ fresh_part = os.path.join(self.temp_dir, "ubuntu.iso.wvc_upload_sess-87654321.part")
+
+ with open(stale_part, "wb") as f:
+ f.write(b"stale part")
+ import json
+ with open(stale_meta, "w") as f:
+ json.dump({"upload_id": "sess-12345678", "expected_chunk": 1}, f)
+ with open(fresh_part, "wb") as f:
+ f.write(b"fresh part")
+
+ old_time = time.time() - 172800
+ os.utime(stale_part, (old_time, old_time))
+ os.utime(stale_meta, (old_time, old_time))
+
+ _cleanup_stale_local_uploads(self.temp_dir, max_age_seconds=86400)
+
+ for fname in unrelated_files:
+ self.assertTrue(os.path.exists(os.path.join(self.temp_dir, fname)), f"{fname} was unexpectedly deleted")
+ self.assertTrue(os.path.exists(fresh_part))
+
+ self.assertFalse(os.path.exists(stale_part))
+ self.assertFalse(os.path.exists(stale_meta))
+
+ def test_sftp_replace_file_requires_atomic_replace(self):
+ mock_sftp = MagicMock()
+ mock_sftp.posix_rename.return_value = None
+ _sftp_replace_file(mock_sftp, "/tmp/src", "/tmp/dst")
+ mock_sftp.posix_rename.assert_called_once_with("/tmp/src", "/tmp/dst")
+ mock_sftp.rename.assert_not_called()
+
+ mock_sftp.reset_mock()
+ mock_sftp.posix_rename.side_effect = IOError("posix_rename unsupported")
+ with self.assertRaises(OSError):
+ _sftp_replace_file(mock_sftp, "/tmp/src", "/tmp/dst")
+ mock_sftp.posix_rename.assert_called_once_with("/tmp/src", "/tmp/dst")
+ mock_sftp.remove.assert_not_called()
+ mock_sftp.rename.assert_not_called()
+
+ # posix_rename with other error must NOT fallback to remove
+ mock_sftp.reset_mock()
+ mock_sftp.posix_rename.side_effect = IOError("permission denied")
+ with self.assertRaises(IOError):
+ _sftp_replace_file(mock_sftp, "/tmp/src", "/tmp/dst")
+ mock_sftp.remove.assert_not_called()
+
+ mock_sftp = MagicMock(spec=["remove", "rename"])
+ with self.assertRaises(OSError):
+ _sftp_replace_file(mock_sftp, "/tmp/src", "/tmp/dst")
+ mock_sftp.remove.assert_not_called()
+ mock_sftp.rename.assert_not_called()
+
+ mock_ssh = MagicMock()
+ stdout = MagicMock()
+ stdout.channel.recv_exit_status.return_value = 0
+ mock_ssh.exec_command.return_value = (MagicMock(), stdout, MagicMock())
+ _sftp_replace_file(mock_sftp, "/tmp/src", "/tmp/dst", ssh=mock_ssh)
+ mock_sftp.remove.assert_not_called()
+
+ @patch("storages.upload.time.sleep")
+ @patch("storages.upload.fcntl.flock", side_effect=BlockingIOError)
+ @patch("storages.upload.LOCK_TIMEOUT_SECONDS", 0)
+ def test_upload_lock_times_out(self, _flock, _sleep):
+ with self.assertRaises(TimeoutError):
+ _acquire_upload_lock(123)
+
+ def test_atomic_finalize_ssh_success_and_overwrite_prevention(self):
+ mock_sftp = MagicMock()
+
+ _atomic_finalize_ssh(mock_sftp, "/tmp/temp.part", "/tmp/final.iso")
+ mock_sftp.rename.assert_called_once_with("/tmp/temp.part", "/tmp/final.iso")
+
+ # A destination created concurrently remains untouched.
+ mock_sftp.rename.side_effect = IOError("File exists")
+ mock_sftp.reset_mock()
+ with self.assertRaises(FileExistsError):
+ _atomic_finalize_ssh(mock_sftp, "/tmp/temp.part", "/tmp/final.iso")
+ mock_sftp.remove.assert_not_called()
+
+ def test_is_session_completed_crash_recovery(self):
+ upload_id = "sess-crash-rec-1"
+ file_name = "recovered.iso"
+ final_file = os.path.join(self.temp_dir, file_name)
+ with open(final_file, "wb") as f:
+ f.write(b"COMPLETE_CONTENT_12345")
+ file_size = os.path.getsize(final_file)
+
+ meta_file = os.path.join(self.temp_dir, f"{file_name}.wvc_upload_{upload_id}.meta")
+ import json
+ with open(meta_file, "w") as mf:
+ json.dump({
+ "file_name": file_name,
+ "expected_chunk": 3,
+ "total_chunks": 3,
+ "current_size": file_size,
+ "created_at": time.time(),
+ "updated_at": time.time(),
+ }, mf)
+
+ # No .done marker exists yet; _is_session_completed should recover the session from .meta
+ lock_dir = _get_upload_lock_directory()
+ marker_path = os.path.join(lock_dir, f"upload_{upload_id}.done")
+ if os.path.exists(marker_path):
+ os.unlink(marker_path)
+
+ recovered = _is_session_completed(upload_id, file_name, path=self.temp_dir)
+ self.assertTrue(recovered)
+ # .meta should now be cleaned up and .done marker created
+ self.assertFalse(os.path.exists(meta_file))
+ self.assertTrue(os.path.exists(marker_path))
+ if os.path.exists(marker_path):
+ os.unlink(marker_path)
+
+ def test_lock_file_not_unlinked_on_completion(self):
+ upload_id = "sess-lock-persist-1"
+ file_name = "lock_persist.iso"
+ dummy_chunk = SimpleUploadedFile(file_name, b"data")
+ handle_uploaded_file(
+ self.mock_socket_conn,
+ self.temp_dir,
+ file_name,
+ dummy_chunk,
+ is_last_chunk=True,
+ chunk_index=0,
+ total_chunks=1,
+ upload_id=upload_id,
+ )
+ lock_dir = _get_upload_lock_directory()
+ lock_path = os.path.join(lock_dir, f"upload_session_{upload_id}.lock")
+ self.assertTrue(os.path.exists(lock_path))
+
+ def test_destination_exists_requires_completed_session(self):
+ file_name = "already_here.iso"
+ dest_path = os.path.join(self.temp_dir, file_name)
+ with open(dest_path, "wb") as f:
+ f.write(b"existing content")
+
+ dummy_chunk = SimpleUploadedFile(file_name, b"new chunk")
+ with self.assertRaises(FileExistsError):
+ handle_uploaded_file(
+ self.mock_socket_conn,
+ self.temp_dir,
+ file_name,
+ dummy_chunk,
+ is_last_chunk=True,
+ chunk_index=0,
+ total_chunks=1,
+ upload_id="sess-rogue-123",
+ )
+
+ @patch("paramiko.SSHClient")
+ def test_sftp_upload_multi_chunk_and_retry(self, mock_ssh_cls):
+ mock_ssh = MagicMock()
+ mock_ssh_cls.return_value = mock_ssh
+ mock_sftp = MagicMock()
+ mock_ssh.open_sftp.return_value = mock_sftp
+
+ remote_files = {}
+
+ def mock_lstat(path):
+ if path in remote_files:
+ stat_obj = MagicMock()
+ stat_obj.st_mode = 0o100644
+ return stat_obj
+ raise FileNotFoundError(f"{path} not found")
+
+ mock_sftp.lstat.side_effect = mock_lstat
+
+ class FakeRemoteFile:
+ def __init__(self, path, mode):
+ self.path = path
+ self.mode = mode
+ if "w" in mode or path not in remote_files:
+ remote_files[path] = b""
+ self.buf = bytearray(remote_files[path])
+ self.pos = 0
+
+ def write(self, data):
+ if isinstance(data, str):
+ data = data.encode("utf-8")
+ self.buf[self.pos:self.pos + len(data)] = data
+ self.pos += len(data)
+ remote_files[self.path] = bytes(self.buf)
+
+ def read(self):
+ return bytes(self.buf)
+
+ def seek(self, pos):
+ self.pos = pos
+
+ def truncate(self, size):
+ del self.buf[size:]
+ remote_files[self.path] = bytes(self.buf)
+
+ def close(self):
+ pass
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ pass
+
+ mock_sftp.open.side_effect = lambda p, m: FakeRemoteFile(p, m)
+
+ def mock_rename(src, dst):
+ if src not in remote_files:
+ raise FileNotFoundError(f"{src} not found")
+ remote_files[dst] = remote_files.pop(src)
+
+ mock_sftp.rename.side_effect = mock_rename
+ mock_sftp.posix_rename.side_effect = mock_rename
+ mock_sftp.remove.side_effect = lambda p: remote_files.pop(p, None)
+
+ def mock_exec_command(command, **kwargs):
+ if "os.replace(" in command:
+ import shlex
+ src, dst = shlex.split(command)[-2:]
+ mock_rename(src, dst)
+ if "os.link(" in command:
+ import shlex
+ src, dst = shlex.split(command)[-2:]
+ if dst in remote_files:
+ raise FileExistsError(dst)
+ remote_files[dst] = remote_files[src]
+ stdout = MagicMock()
+ stdout.channel.recv_exit_status.return_value = 0
+ return MagicMock(), stdout, MagicMock()
+
+ mock_ssh.exec_command.side_effect = mock_exec_command
+
+ upload_id = "sess-sftp-multi-1"
+ file_name = "debian.iso"
+ c0 = SimpleUploadedFile(file_name, b"DEBIAN_CHUNK0_")
+ c1 = SimpleUploadedFile(file_name, b"DEBIAN_CHUNK1")
+
+ handle_uploaded_file(
+ self.mock_ssh_conn,
+ "/var/lib/libvirt/images",
+ file_name,
+ c0,
+ is_last_chunk=False,
+ chunk_index=0,
+ total_chunks=2,
+ upload_id=upload_id,
+ )
+ self.assertEqual(mock_ssh.connect.call_args.kwargs["timeout"], 10)
+ mock_sftp.get_channel.return_value.settimeout.assert_called_with(30)
+
+ # Simulate bytes written before a worker crashed without advancing metadata.
+ part_remote = f"/var/lib/libvirt/images/{file_name}.wvc_upload_{upload_id}.part"
+ remote_files[part_remote] += b"UNCOMMITTED"
+
+ handle_uploaded_file(
+ self.mock_ssh_conn,
+ "/var/lib/libvirt/images",
+ file_name,
+ c1,
+ is_last_chunk=True,
+ chunk_index=1,
+ total_chunks=2,
+ upload_id=upload_id,
+ )
+
+ final_remote = "/var/lib/libvirt/images/debian.iso"
+ self.assertIn(final_remote, remote_files)
+ self.assertEqual(remote_files[final_remote], b"DEBIAN_CHUNK0_DEBIAN_CHUNK1")
+
+ # Retry Chunk 1
+ handle_uploaded_file(
+ self.mock_ssh_conn,
+ "/var/lib/libvirt/images",
+ file_name,
+ c1,
+ is_last_chunk=True,
+ chunk_index=1,
+ total_chunks=2,
+ upload_id=upload_id,
+ )
+ self.assertEqual(remote_files[final_remote], b"DEBIAN_CHUNK0_DEBIAN_CHUNK1")
def test_handle_uploaded_file_ssh_traversal_check(self):
dummy_chunk = SimpleUploadedFile("test.iso", b"data")
with patch("os.path.basename", side_effect=lambda x: x):
with self.assertRaises(PermissionError):
- handle_uploaded_file(self.mock_ssh_conn, "/var/lib/libvirt/images", "../../../evil.iso", dummy_chunk, is_last_chunk=True)
+ handle_uploaded_file(self.mock_ssh_conn, "/var/lib/libvirt/images", "../../../evil.iso", dummy_chunk, is_last_chunk=True, upload_id="sess-ssh-1")
class StorageUploadViewTests(TestCase):
@@ -81,6 +630,44 @@ class StorageUploadViewTests(TestCase):
self.client.login(username="storage_admin", password="password")
self.compute = Compute.objects.create(name="LocalCompute", hostname="localhost", type=1, login="u", password="p")
+ @patch("storages.views.wvmStorage")
+ def test_iso_upload_rejects_other_extensions(self, mock_wvm):
+ mock_conn = mock_wvm.return_value
+ mock_conn.conn = CONN_SOCKET
+ mock_conn.get_target_path.return_value = "/tmp"
+ url = reverse("storage", args=[self.compute.id, "default"])
+ response = self.client.post(url, {
+ "iso_upload": "1", "file": SimpleUploadedFile("notes.txt", b"notes"),
+ "file_name": "notes.txt", "chunk_index": "0", "total_chunks": "1",
+ "upload_id": "invalid-extension-test",
+ })
+ self.assertEqual(response.status_code, 400)
+ self.assertIn("Only .iso files", response.json()["error"])
+
+ @patch("storages.views.handle_uploaded_file")
+ @patch("storages.views.wvmStorage")
+ def test_tcp_upload_passes_declared_size(self, mock_wvm, mock_upload):
+ mock_conn = mock_wvm.return_value
+ mock_conn.conn = CONN_TCP
+ mock_conn.get_target_path.return_value = "/var/lib/libvirt/images"
+ url = reverse("storage", args=[self.compute.id, "default"])
+ payload = {
+ "iso_upload": "1", "file_name": "test.iso", "chunk_index": "0",
+ "total_chunks": "1", "upload_id": "tcp-view-test",
+ }
+ response = self.client.post(url, {
+ **payload, "file": SimpleUploadedFile("test.iso", b"iso"),
+ })
+ self.assertEqual(response.status_code, 400)
+ mock_upload.assert_not_called()
+
+ response = self.client.post(url, {
+ **payload, "file": SimpleUploadedFile("test.iso", b"iso"),
+ "file_size": "3",
+ })
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(mock_upload.call_args.kwargs["file_size"], 3)
+
@patch("storages.views.wvmStorage")
def test_iso_upload_missing_chunk(self, mock_wvm):
mock_conn = MagicMock()
@@ -93,7 +680,7 @@ class StorageUploadViewTests(TestCase):
mock_wvm.return_value = mock_conn
url = reverse("storage", args=[self.compute.id, "default"])
- res = self.client.post(url, {"iso_upload": "1", "file_name": "test.iso"})
+ res = self.client.post(url, {"iso_upload": "1", "file_name": "test.iso", "upload_id": "sess-view-1"})
self.assertEqual(res.status_code, 400)
self.assertIn("No file chunk was submitted", res.json().get("error", ""))
@@ -116,6 +703,109 @@ class StorageUploadViewTests(TestCase):
"file_name": "..\x00..",
"chunk_index": "0",
"total_chunks": "1",
+ "upload_id": "sess-view-2",
})
self.assertEqual(res.status_code, 400)
self.assertIn("Invalid file name", res.json().get("error", ""))
+
+ @patch("storages.views.wvmStorage")
+ def test_iso_upload_missing_upload_id(self, mock_wvm):
+ mock_conn = MagicMock()
+ mock_conn.get_status.return_value = 1
+ mock_conn.get_target_path.return_value = "/tmp"
+ mock_conn.get_type.return_value = "dir"
+ mock_conn.get_autostart.return_value = 1
+ mock_conn.get_size.return_value = [1000, 500]
+ mock_conn.update_volumes.return_value = []
+ mock_wvm.return_value = mock_conn
+
+ url = reverse("storage", args=[self.compute.id, "default"])
+ chunk = SimpleUploadedFile("test.iso", b"chunkdata")
+ res = self.client.post(url, {
+ "iso_upload": "1",
+ "file": chunk,
+ "file_name": "test.iso",
+ "chunk_index": "0",
+ "total_chunks": "1",
+ })
+ self.assertEqual(res.status_code, 400)
+ self.assertIn("Missing upload_id", res.json().get("error", ""))
+
+ @patch("storages.views.wvmStorage")
+ def test_iso_upload_final_chunk_retry_idempotency(self, mock_wvm):
+ mock_conn = MagicMock()
+ mock_conn.conn = CONN_SOCKET
+ mock_conn.get_status.return_value = 1
+ temp_dir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, temp_dir, ignore_errors=True)
+ mock_conn.get_target_path.return_value = temp_dir
+ mock_conn.get_type.return_value = "dir"
+ mock_conn.get_autostart.return_value = 1
+ mock_conn.get_size.return_value = [1000, 500]
+ mock_conn.get_volumes.return_value = []
+ mock_conn.update_volumes.return_value = []
+ mock_wvm.return_value = mock_conn
+
+ url = reverse("storage", args=[self.compute.id, "default"])
+ upload_id = "sess-view-retry-123"
+ file_name = "test_retry.iso"
+ chunk = SimpleUploadedFile(file_name, b"test payload")
+
+ # 1. First upload (single chunk, total_chunks=1)
+ res1 = self.client.post(url, {
+ "iso_upload": "1",
+ "file": chunk,
+ "file_name": file_name,
+ "chunk_index": "0",
+ "total_chunks": "1",
+ "upload_id": upload_id,
+ })
+ self.assertEqual(res1.status_code, 200)
+ self.assertTrue(res1.json().get("success"))
+
+ # Volume now appears in libvirt pool
+ mock_conn.get_volumes.return_value = [file_name]
+
+ # 2. Retrying final chunk with same upload_id returns 200 idempotent success
+ chunk_retry = SimpleUploadedFile(file_name, b"test payload")
+ res2 = self.client.post(url, {
+ "iso_upload": "1",
+ "file": chunk_retry,
+ "file_name": file_name,
+ "chunk_index": "0",
+ "total_chunks": "1",
+ "upload_id": upload_id,
+ })
+ self.assertEqual(res2.status_code, 200)
+ self.assertTrue(res2.json().get("success"))
+ mock_conn.get_storages.assert_not_called()
+ mock_conn.update_volumes.assert_not_called()
+ mock_conn.refresh.assert_not_called()
+ self.assertTrue(res2.json().get("reload"))
+
+ # 3. Upload with different upload_id returns 400 error
+ chunk_attacker = SimpleUploadedFile(file_name, b"evil payload")
+ res3 = self.client.post(url, {
+ "iso_upload": "1",
+ "file": chunk_attacker,
+ "file_name": file_name,
+ "chunk_index": "0",
+ "total_chunks": "1",
+ "upload_id": "sess-attacker-999",
+ })
+ self.assertEqual(res3.status_code, 400)
+ self.assertIn("ISO image already exists", res3.json().get("error", ""))
+
+ # 4. Upload with same upload_id but by a different superuser cannot bypass/claim
+ User.objects.create_superuser(username="other_admin", password="password", email="other@example.com")
+ self.client.login(username="other_admin", password="password")
+ res4 = self.client.post(url, {
+ "iso_upload": "1",
+ "file": chunk_retry,
+ "file_name": file_name,
+ "chunk_index": "0",
+ "total_chunks": "1",
+ "upload_id": upload_id,
+ })
+ self.assertEqual(res4.status_code, 400)
+ self.assertIn("ISO image already exists", res4.json().get("error", ""))
diff --git a/storages/upload.py b/storages/upload.py
new file mode 100644
index 00000000..50aad82d
--- /dev/null
+++ b/storages/upload.py
@@ -0,0 +1,692 @@
+"""Chunked ISO upload operations shared by the storage view."""
+
+import fcntl
+import json
+import logging
+import os
+import posixpath
+import re
+import stat
+import tempfile
+import time
+from xml.sax.saxutils import escape
+
+import paramiko
+from django.utils.translation import gettext_lazy as _
+
+from vrtManager.connection import CONN_SSH, CONN_SOCKET, CONN_TCP, CONN_TLS
+
+logger = logging.getLogger(__name__)
+LOCK_TIMEOUT_SECONDS = 15
+SSH_CONNECT_TIMEOUT_SECONDS = 10
+SSH_IO_TIMEOUT_SECONDS = 30
+
+
+def _fsync_dir(dir_path):
+ if not dir_path or not os.path.isdir(dir_path):
+ return
+ try:
+ dir_fd = os.open(dir_path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+ try:
+ os.fsync(dir_fd)
+ finally:
+ os.close(dir_fd)
+ except OSError:
+ pass
+
+
+def _atomic_finalize_local(target_temp, target_final):
+ parent_dir = os.path.dirname(target_final)
+ # 1. Primary: POSIX os.link creates a hard link atomically and fails with FileExistsError if target_final exists.
+ try:
+ os.link(target_temp, target_final)
+ _fsync_dir(parent_dir)
+ try:
+ os.unlink(target_temp)
+ _fsync_dir(parent_dir)
+ except OSError:
+ pass
+ return
+ except FileExistsError:
+ raise FileExistsError(_("Destination file already exists"))
+ except OSError as link_err:
+ # 2. Secondary: Linux kernel renameat2 with RENAME_NOREPLACE (flag 1)
+ try:
+ import ctypes
+ libc = ctypes.CDLL("libc.so.6", use_errno=True)
+ ret = libc.renameat2(
+ ctypes.c_int(-100),
+ target_temp.encode("utf-8"),
+ ctypes.c_int(-100),
+ target_final.encode("utf-8"),
+ ctypes.c_uint(1),
+ )
+ if ret == 0:
+ _fsync_dir(parent_dir)
+ return
+ err = ctypes.get_errno()
+ if err == 17: # EEXIST
+ raise FileExistsError(_("Destination file already exists"))
+ except (AttributeError, OSError):
+ pass
+
+ # If atomic no-replace cannot be guaranteed, fail safely without destroying target_temp
+ raise OSError(_("Filesystem does not support atomic no-replace finalization: {}").format(link_err))
+
+
+def _atomic_finalize_ssh(sftp, target_temp, target_final):
+ """
+ Finalize with standard SFTP rename, which requires an unused destination.
+ Atomic visibility depends on the remote SFTP server and filesystem.
+ """
+ try:
+ sftp.rename(target_temp, target_final)
+ except OSError as e:
+ if "exist" in str(e).lower() or "already" in str(e).lower():
+ raise FileExistsError(_("Destination file already exists")) from e
+ raise
+
+
+_STALE_UPLOAD_PATTERN = re.compile(r"^.+\.wvc_upload_[a-zA-Z0-9_\-]{8,64}\.(part|meta|meta\.tmp)$")
+
+
+def _sftp_replace_file(sftp, src, dst, ssh=None):
+ """Replace metadata atomically, leaving the old copy intact on failure."""
+ posix_rename = getattr(sftp, "posix_rename", None)
+ if callable(posix_rename):
+ try:
+ posix_rename(src, dst)
+ return
+ except (IOError, OSError) as e:
+ err_str = str(e).lower()
+ if not ("unsupported" in err_str or getattr(e, "code", None) == 8 or isinstance(e, NotImplementedError)):
+ raise
+ if ssh is not None:
+ import shlex
+ cmd = f"python3 -c 'import os, sys; os.replace(sys.argv[1], sys.argv[2])' {shlex.quote(src)} {shlex.quote(dst)}"
+ try:
+ stdin, stdout, stderr = ssh.exec_command(cmd, timeout=SSH_IO_TIMEOUT_SECONDS)
+ if stdout.channel.recv_exit_status() == 0:
+ return
+ except Exception as exc:
+ raise OSError(_("Remote server cannot safely replace upload metadata")) from exc
+ raise OSError(_("Remote server cannot safely replace upload metadata"))
+
+
+def _cleanup_stale_local_uploads(base_dir, max_age_seconds=86400):
+ try:
+ now = time.time()
+ for entry in os.scandir(base_dir):
+ if entry.is_file() and _STALE_UPLOAD_PATTERN.match(entry.name):
+ try:
+ mtime = entry.stat().st_mtime
+ if now - mtime > max_age_seconds:
+ if entry.name.endswith(".meta"):
+ try:
+ with open(entry.path, "r", encoding="utf-8") as f:
+ meta = json.load(f)
+ if not meta.get("upload_id") and not meta.get("expected_chunk"):
+ continue
+ except Exception:
+ pass
+ os.unlink(entry.path)
+ except OSError:
+ pass
+ except OSError:
+ pass
+
+
+def _cleanup_stale_sftp_uploads(sftp, remote_base, max_age_seconds=86400):
+ try:
+ now = time.time()
+ for attr in sftp.listdir_attr(remote_base):
+ name = attr.filename
+ if _STALE_UPLOAD_PATTERN.match(name):
+ try:
+ if attr.st_mtime and (now - attr.st_mtime > max_age_seconds):
+ target_file = posixpath.join(remote_base, name)
+ sftp.remove(target_file)
+ except Exception:
+ pass
+ except Exception:
+ pass
+
+
+def _get_upload_lock_directory():
+ candidates = [
+ "/srv/webvirtcloud/data/locks",
+ os.path.join(tempfile.gettempdir(), f"webvirtcloud_upload_locks_{os.geteuid()}"),
+ ]
+ for candidate in candidates:
+ try:
+ if os.path.islink(candidate):
+ continue
+ if not os.path.exists(candidate):
+ os.makedirs(candidate, mode=0o700, exist_ok=True)
+ if not os.path.isdir(candidate) or os.path.islink(candidate):
+ continue
+ stat_info = os.stat(candidate)
+ if stat_info.st_uid != os.geteuid():
+ continue
+ if (stat.S_IMODE(stat_info.st_mode) & 0o077) != 0:
+ try:
+ os.chmod(candidate, 0o700)
+ except OSError:
+ continue
+ stat_info = os.stat(candidate)
+ if (stat.S_IMODE(stat_info.st_mode) & 0o077) != 0:
+ continue
+ return candidate
+ except (OSError, PermissionError):
+ continue
+ return None
+
+
+def _cleanup_stale_lock_dir(lock_dir, max_age_seconds=86400):
+ if not lock_dir or not os.path.exists(lock_dir):
+ return
+ try:
+ now = time.time()
+ for entry in os.scandir(lock_dir):
+ if entry.is_file():
+ if entry.name.startswith("upload_") and (entry.name.endswith(".done") or entry.name.endswith(".done.tmp")):
+ try:
+ if now - entry.stat().st_mtime > max_age_seconds:
+ os.unlink(entry.path)
+ except OSError:
+ pass
+ except OSError:
+ pass
+
+
+def _mark_session_completed(lock_dir, upload_id, file_name, path=None, user_id=None, compute_id=None, pool=None):
+ if not lock_dir or not upload_id or not file_name:
+ return
+ marker_path = os.path.join(lock_dir, f"upload_{upload_id}.done")
+ if os.path.islink(marker_path):
+ try:
+ os.unlink(marker_path)
+ except OSError:
+ pass
+ try:
+ tmp_marker = f"{marker_path}.tmp"
+ marker_data = {
+ "upload_id": upload_id,
+ "file_name": file_name,
+ "path": posixpath.normpath(path) if path else None,
+ "user_id": user_id,
+ "compute_id": compute_id,
+ "pool": pool,
+ "completed_at": time.time(),
+ }
+ flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0)
+ fd = os.open(tmp_marker, flags, 0o600)
+ with open(fd, "w", encoding="utf-8", closefd=True) as f:
+ json.dump(marker_data, f)
+ f.flush()
+ os.fsync(f.fileno())
+ os.replace(tmp_marker, marker_path)
+ _fsync_dir(lock_dir)
+ except Exception as e:
+ logger.error("Failed to write upload session completion marker %s: %s", marker_path, e)
+ raise
+
+
+def _is_session_completed(upload_id, file_name, path=None, user_id=None, compute_id=None, pool=None, lock_dir=None, check_local_recovery=True):
+ if not upload_id or not file_name:
+ return False
+ if not lock_dir:
+ lock_dir = _get_upload_lock_directory()
+ if not lock_dir:
+ return False
+ marker_path = os.path.join(lock_dir, f"upload_{upload_id}.done")
+ if os.path.exists(marker_path) and not os.path.islink(marker_path):
+ try:
+ with open(marker_path, "r", encoding="utf-8") as f:
+ data = json.load(f)
+ marker_valid = True
+ if data.get("upload_id") != upload_id or data.get("file_name") != file_name:
+ marker_valid = False
+ if path and data.get("path"):
+ if posixpath.normpath(path) != posixpath.normpath(data["path"]):
+ marker_valid = False
+ if user_id is not None and data.get("user_id") is not None:
+ if str(data["user_id"]) != str(user_id):
+ marker_valid = False
+ if compute_id is not None and data.get("compute_id") is not None:
+ if str(data["compute_id"]) != str(compute_id):
+ marker_valid = False
+ if pool is not None and data.get("pool") is not None:
+ if str(data["pool"]) != str(pool):
+ marker_valid = False
+ if time.time() - data.get("completed_at", 0) > 86400:
+ try:
+ os.unlink(marker_path)
+ _fsync_dir(lock_dir)
+ except OSError:
+ pass
+ marker_valid = False
+ if marker_valid:
+ return True
+ except Exception:
+ pass
+
+ # Crash-recovery: check if metadata file exists in path with complete chunks and matching size
+ if path and check_local_recovery:
+ meta_suffix = f".wvc_upload_{upload_id}.meta"
+ meta_path = os.path.join(path, f"{file_name}{meta_suffix}")
+ final_path = os.path.join(path, file_name)
+ if os.path.exists(meta_path) and os.path.exists(final_path) and not os.path.islink(meta_path):
+ try:
+ with open(meta_path, "r", encoding="utf-8") as mf:
+ meta = json.load(mf)
+ if (
+ meta.get("file_name") == file_name
+ and meta.get("expected_chunk") == meta.get("total_chunks")
+ and os.path.getsize(final_path) == meta.get("current_size")
+ ):
+ try:
+ _mark_session_completed(
+ lock_dir, upload_id, file_name, path,
+ user_id=user_id, compute_id=compute_id, pool=pool,
+ )
+ os.unlink(meta_path)
+ _fsync_dir(path)
+ except Exception:
+ pass
+ return True
+ except Exception:
+ pass
+
+ return False
+
+
+class _UploadTarget:
+ """Filesystem differences for one upload; chunk ordering lives in one place."""
+
+ def __init__(self, conn, path, file_name, upload_id):
+ self.conn = conn
+ self.remote = conn.conn == CONN_SSH
+ if not self.remote and conn.conn != CONN_SOCKET:
+ raise ValueError(_("Unsupported connection type for file upload."))
+ self.base = posixpath.normpath(path) if self.remote else os.path.abspath(path)
+ join = posixpath.join if self.remote else os.path.join
+ normalize = posixpath.normpath if self.remote else os.path.abspath
+ self.final = normalize(join(self.base, file_name))
+ self.part = normalize(join(self.base, f"{file_name}.wvc_upload_{upload_id}.part"))
+ self.meta = normalize(join(self.base, f"{file_name}.wvc_upload_{upload_id}.meta"))
+ for target in (self.final, self.part, self.meta):
+ if self.remote:
+ contained = target.startswith(self.base.rstrip("/") + "/")
+ else:
+ contained = os.path.commonpath((self.base, target)) == self.base
+ if not contained:
+ raise PermissionError(_("Security issues with file uploading: path traversal detected"))
+ self.ssh = None
+ self.sftp = None
+
+ def __enter__(self):
+ if self.remote:
+ hostname, separator, port = self.conn.host.partition(":")
+ self.ssh = paramiko.SSHClient()
+ try:
+ self.ssh.load_system_host_keys()
+ self.ssh.set_missing_host_key_policy(paramiko.RejectPolicy())
+ self.ssh.connect(
+ hostname=hostname, port=int(port) if separator else 22,
+ username=self.conn.login, password=self.conn.passwd,
+ timeout=SSH_CONNECT_TIMEOUT_SECONDS,
+ banner_timeout=SSH_CONNECT_TIMEOUT_SECONDS,
+ auth_timeout=SSH_CONNECT_TIMEOUT_SECONDS,
+ channel_timeout=SSH_CONNECT_TIMEOUT_SECONDS,
+ )
+ self.sftp = self.ssh.open_sftp()
+ self.sftp.get_channel().settimeout(SSH_IO_TIMEOUT_SECONDS)
+ except Exception:
+ self.ssh.close()
+ raise
+ return self
+
+ def __exit__(self, *_):
+ try:
+ if self.sftp is not None:
+ self.sftp.close()
+ finally:
+ if self.ssh is not None:
+ self.ssh.close()
+
+ def info(self, path):
+ try:
+ info = self.sftp.lstat(path) if self.remote else os.lstat(path)
+ except FileNotFoundError:
+ return None
+ if stat.S_ISLNK(info.st_mode):
+ raise PermissionError(_("Security issues with file uploading: symlink detected"))
+ return info
+
+ def read_meta(self):
+ if self.info(self.meta) is None:
+ return None
+ try:
+ if self.remote:
+ with self.sftp.open(self.meta, "r") as stream:
+ raw = stream.read()
+ else:
+ with open(self.meta, "r", encoding="utf-8") as stream:
+ raw = stream.read()
+ return json.loads(raw)
+ except Exception as exc:
+ raise ValueError(_("Corrupted upload session metadata")) from exc
+
+ def write_meta(self, data):
+ temporary = self.meta + ".tmp"
+ if self.remote:
+ if self.info(temporary) is not None:
+ self.sftp.remove(temporary)
+ with self.sftp.open(temporary, "w") as stream:
+ stream.write(json.dumps(data))
+ if callable(getattr(stream, "flush", None)):
+ stream.flush()
+ _sftp_replace_file(self.sftp, temporary, self.meta, ssh=self.ssh)
+ else:
+ flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0)
+ fd = os.open(temporary, flags, 0o600)
+ with open(fd, "w", encoding="utf-8", closefd=True) as stream:
+ json.dump(data, stream)
+ stream.flush()
+ os.fsync(stream.fileno())
+ os.replace(temporary, self.meta)
+ _fsync_dir(self.base)
+
+ def write_chunk(self, chunk, offset, new):
+ if self.remote:
+ with self.sftp.open(self.part, "wb" if new else "r+b") as stream:
+ if not new:
+ stream.seek(offset)
+ stream.truncate(offset)
+ for data in chunk.chunks():
+ stream.write(data)
+ offset += len(data)
+ if callable(getattr(stream, "flush", None)):
+ stream.flush()
+ return offset
+ flags = (os.O_WRONLY | os.O_CREAT | os.O_EXCL) if new else os.O_RDWR
+ flags |= getattr(os, "O_NOFOLLOW", 0)
+ fd = os.open(self.part, flags, 0o600)
+ with open(fd, "wb" if new else "r+b", closefd=True) as stream:
+ if not new:
+ stream.seek(offset)
+ stream.truncate()
+ for data in chunk.chunks():
+ stream.write(data)
+ offset += len(data)
+ stream.flush()
+ os.fsync(stream.fileno())
+ return offset
+
+ def finalize(self):
+ if self.remote:
+ _atomic_finalize_ssh(self.sftp, self.part, self.final)
+ else:
+ _atomic_finalize_local(self.part, self.final)
+
+ def remove_meta(self):
+ try:
+ if self.remote:
+ self.sftp.remove(self.meta)
+ else:
+ os.unlink(self.meta)
+ _fsync_dir(self.base)
+ except (OSError, IOError):
+ pass
+
+ def cleanup(self, lock_dir):
+ if self.remote:
+ _cleanup_stale_sftp_uploads(self.sftp, self.base)
+ else:
+ _cleanup_stale_local_uploads(self.base)
+ _cleanup_stale_lock_dir(lock_dir)
+
+
+def _matches_session(meta, file_name, total_chunks, user_id, compute_id, pool):
+ if meta.get("file_name") != file_name:
+ raise ValueError(_("File name does not match upload session"))
+ if meta.get("total_chunks") != total_chunks:
+ raise ValueError(_("Inconsistent total_chunks for session"))
+ for key, expected in (("user_id", user_id), ("compute_id", compute_id), ("pool", pool)):
+ if expected is not None and meta.get(key) is not None and str(meta[key]) != str(expected):
+ raise PermissionError(_("Upload session belongs to another request context"))
+
+
+def _acquire_upload_lock(lock_fd):
+ deadline = time.monotonic() + LOCK_TIMEOUT_SECONDS
+ while True:
+ try:
+ fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ return
+ except BlockingIOError:
+ if time.monotonic() >= deadline:
+ raise TimeoutError(_("Timed out waiting for upload session lock"))
+ time.sleep(0.05)
+
+
+def _write_libvirt_meta(meta_path, data):
+ temporary = meta_path + ".tmp"
+ flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0)
+ fd = os.open(temporary, flags, 0o600)
+ with open(fd, "w", encoding="utf-8", closefd=True) as stream:
+ json.dump(data, stream)
+ stream.flush()
+ os.fsync(stream.fileno())
+ os.replace(temporary, meta_path)
+ _fsync_dir(os.path.dirname(meta_path))
+
+
+def _handle_libvirt_upload(
+ conn, path, file_name, file_chunk, is_last_chunk, chunk_index,
+ total_chunks, upload_id, user_id, compute_id, pool, file_size, lock_dir,
+):
+ """Upload to a remote pool through the existing libvirt connection."""
+ if not isinstance(file_size, int) or file_size <= 0:
+ raise ValueError(_("A positive file_size is required for TCP/TLS uploads"))
+ if file_chunk.size <= 0:
+ raise ValueError(_("Empty upload chunk"))
+
+ meta_path = os.path.join(lock_dir, f"upload_{upload_id}.libvirt.json")
+ if os.path.islink(meta_path):
+ raise PermissionError(_("Invalid upload session metadata"))
+ if os.path.exists(meta_path):
+ try:
+ with open(meta_path, "r", encoding="utf-8") as stream:
+ meta = json.load(stream)
+ except (OSError, ValueError) as exc:
+ raise ValueError(_("Corrupted upload session metadata")) from exc
+ _matches_session(meta, file_name, total_chunks, user_id, compute_id, pool)
+ if meta.get("path") != path or meta.get("file_size") != file_size:
+ raise ValueError(_("Upload session parameters changed"))
+ else:
+ if chunk_index != 0:
+ raise ValueError(_("Upload session not found or expired"))
+ meta = {
+ "file_name": file_name, "total_chunks": total_chunks,
+ "expected_chunk": 0, "current_size": 0, "file_size": file_size,
+ "path": path, "user_id": user_id, "compute_id": compute_id,
+ "pool": pool, "created_at": time.time(),
+ }
+
+ expected = meta["expected_chunk"]
+ if chunk_index < expected:
+ if is_last_chunk and expected == total_chunks:
+ _mark_session_completed(
+ lock_dir, upload_id, file_name, path,
+ user_id=user_id, compute_id=compute_id, pool=pool,
+ )
+ os.unlink(meta_path)
+ _fsync_dir(lock_dir)
+ return
+ if chunk_index != expected:
+ raise ValueError(
+ _("Invalid chunk sequence. Expected chunk %(exp)d, got %(got)d")
+ % {"exp": expected, "got": chunk_index}
+ )
+
+ offset = meta["current_size"]
+ if offset + file_chunk.size > file_size:
+ raise ValueError(_("Upload exceeds declared file size"))
+ if is_last_chunk and offset + file_chunk.size != file_size:
+ raise ValueError(_("Final chunk does not match declared file size"))
+
+ # A volume is visible as soon as libvirt creates it. Keep the session until
+ # the final chunk finishes so retries can resume safely at a known offset.
+ volumes = conn.get_volumes()
+ if meta["expected_chunk"] == 0 and not os.path.exists(meta_path):
+ if file_name in volumes:
+ raise FileExistsError(_("Destination file already exists"))
+ xml = (
+ "
{}{}"
+ ""
+ ).format(escape(file_name), file_size)
+ conn.pool.createXML(xml, 0)
+ _write_libvirt_meta(meta_path, meta)
+ elif file_name not in volumes:
+ raise ValueError(_("Upload volume is missing"))
+
+ volume = conn.get_volume(file_name)
+ stream = conn.wvm.newStream(0)
+ try:
+ volume.upload(stream, offset, file_chunk.size, 0)
+ written = 0
+ for data in file_chunk.chunks():
+ while data:
+ sent = stream.send(data)
+ if sent <= 0:
+ raise OSError(_("Libvirt stream stopped during upload"))
+ written += sent
+ data = data[sent:]
+ if written != file_chunk.size:
+ raise OSError(_("Upload chunk size changed during transfer"))
+ stream.finish()
+ except Exception:
+ stream.abort()
+ raise
+
+ meta["current_size"] = offset + written
+ meta["expected_chunk"] = expected + 1
+ meta["updated_at"] = time.time()
+ _write_libvirt_meta(meta_path, meta)
+ if is_last_chunk:
+ _mark_session_completed(
+ lock_dir, upload_id, file_name, path,
+ user_id=user_id, compute_id=compute_id, pool=pool,
+ )
+ os.unlink(meta_path)
+ _fsync_dir(lock_dir)
+
+
+def handle_uploaded_file(
+ conn, path, file_name, file_chunk, is_last_chunk, chunk_index=0,
+ total_chunks=1, upload_id=None, user_id=None, compute_id=None, pool=None,
+ file_size=None,
+):
+ clean_name = os.path.basename(file_name).strip()
+ if not clean_name or clean_name in (".", "..") or "\x00" in clean_name:
+ raise ValueError(_("Invalid file name"))
+ if len(clean_name) <= 4 or not clean_name.lower().endswith(".iso"):
+ raise ValueError(_("Only .iso files can be uploaded"))
+ if total_chunks < 1 or chunk_index < 0 or chunk_index >= total_chunks:
+ raise ValueError(_("Invalid chunk index or total chunks."))
+ if is_last_chunk != (chunk_index == total_chunks - 1):
+ raise ValueError(_("Invalid final chunk flag"))
+ if not upload_id or not isinstance(upload_id, str):
+ raise ValueError(_("Missing required upload_id"))
+ upload_id = upload_id.strip()
+ if not re.fullmatch(r"[a-zA-Z0-9_-]{1,64}", upload_id):
+ raise ValueError(_("Invalid upload_id format"))
+
+ libvirt_upload = conn.conn in (CONN_TCP, CONN_TLS)
+ target = None if libvirt_upload else _UploadTarget(conn, path, clean_name, upload_id)
+ lock_dir = _get_upload_lock_directory()
+ if not lock_dir:
+ raise OSError(_("Could not acquire secure upload lock directory"))
+ lock_path = os.path.join(lock_dir, f"upload_session_{upload_id}.lock")
+ flags = os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0)
+ lock_fd = os.open(lock_path, flags, 0o600)
+ try:
+ _acquire_upload_lock(lock_fd)
+ if libvirt_upload:
+ if _is_session_completed(
+ upload_id, clean_name, path, user_id=user_id,
+ compute_id=compute_id, pool=pool, lock_dir=lock_dir,
+ check_local_recovery=False,
+ ):
+ if is_last_chunk:
+ if clean_name not in conn.get_volumes():
+ raise ValueError(_("Upload volume is missing"))
+ return
+ raise FileExistsError(_("Destination file already exists"))
+ return _handle_libvirt_upload(
+ conn, path, clean_name, file_chunk, is_last_chunk, chunk_index,
+ total_chunks, upload_id, user_id, compute_id, pool, file_size, lock_dir,
+ )
+ with target:
+ if chunk_index == 0:
+ target.cleanup(lock_dir)
+ final_info = target.info(target.final)
+ if final_info is not None:
+ if is_last_chunk and _is_session_completed(
+ upload_id, clean_name, target.base, user_id=user_id,
+ compute_id=compute_id, pool=pool, lock_dir=lock_dir,
+ ):
+ return
+ # A crash after rename but before the completion marker is recoverable.
+ meta = target.read_meta() if is_last_chunk else None
+ if meta is not None:
+ _matches_session(meta, clean_name, total_chunks, user_id, compute_id, pool)
+ if (meta.get("expected_chunk") == total_chunks
+ and final_info.st_size == meta.get("current_size")):
+ _mark_session_completed(lock_dir, upload_id, clean_name, target.base,
+ user_id=user_id, compute_id=compute_id, pool=pool)
+ target.remove_meta()
+ return
+ raise FileExistsError(_("Destination file already exists"))
+
+ meta = target.read_meta()
+ if meta is None:
+ if chunk_index != 0 or target.info(target.part) is not None:
+ raise ValueError(_("Upload session not found or expired"))
+ meta = {
+ "file_name": clean_name, "total_chunks": total_chunks,
+ "expected_chunk": 0, "current_size": 0,
+ "created_at": time.time(), "user_id": user_id,
+ "compute_id": compute_id, "pool": pool,
+ }
+ else:
+ _matches_session(meta, clean_name, total_chunks, user_id, compute_id, pool)
+ if target.info(target.part) is None:
+ raise ValueError(_("Upload session not found or expired"))
+
+ expected = meta["expected_chunk"]
+ if chunk_index < expected:
+ if is_last_chunk and expected == total_chunks:
+ target.finalize()
+ _mark_session_completed(lock_dir, upload_id, clean_name, target.base,
+ user_id=user_id, compute_id=compute_id, pool=pool)
+ target.remove_meta()
+ return
+ if chunk_index != expected:
+ raise ValueError(
+ _("Invalid chunk sequence. Expected chunk %(exp)d, got %(got)d")
+ % {"exp": expected, "got": chunk_index}
+ )
+
+ meta["current_size"] = target.write_chunk(file_chunk, meta["current_size"], new=expected == 0)
+ meta["expected_chunk"] = expected + 1
+ meta["updated_at"] = time.time()
+ target.write_meta(meta)
+ if is_last_chunk:
+ target.finalize()
+ _mark_session_completed(lock_dir, upload_id, clean_name, target.base,
+ user_id=user_id, compute_id=compute_id, pool=pool)
+ target.remove_meta()
+ finally:
+ os.close(lock_fd)
diff --git a/storages/views.py b/storages/views.py
index ec41cb86..8f18ebc9 100644
--- a/storages/views.py
+++ b/storages/views.py
@@ -1,6 +1,6 @@
import json
import os
-import posixpath
+import re
from admin.decorators import superuser_only
from appsettings.settings import app_settings
@@ -11,12 +11,84 @@ from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from libvirt import libvirtError
-import paramiko
-from vrtManager.connection import CONN_SSH, CONN_SOCKET
+from vrtManager.connection import CONN_TCP, CONN_TLS
from vrtManager.storage import wvmStorage, wvmStorages
from storages.forms import AddStgPool, CloneImage, CreateVolumeForm
+from storages.upload import handle_uploaded_file
+
+
+def _handle_iso_upload(request, conn, path, compute, pool):
+ file_chunk = request.FILES.get("file")
+ if not file_chunk:
+ return JsonResponse({"error": _("No file chunk was submitted.")}, status=400)
+
+ file_name = request.POST.get("file_name", "")
+ clean_file_name = os.path.basename(file_name).strip()
+ if not clean_file_name or clean_file_name in (".", "..") or "\x00" in clean_file_name:
+ return JsonResponse({"error": _("Invalid file name.")}, status=400)
+ file_name = clean_file_name
+
+ try:
+ chunk_index = int(request.POST.get("chunk_index", 0))
+ total_chunks = int(request.POST.get("total_chunks", 1))
+ except (ValueError, TypeError):
+ return JsonResponse({"error": _("Invalid chunk parameters.")}, status=400)
+
+ if total_chunks < 1 or chunk_index < 0 or chunk_index >= total_chunks:
+ return JsonResponse({"error": _("Invalid chunk index or total chunks.")}, status=400)
+
+ is_last_chunk = chunk_index == total_chunks - 1
+
+ file_size = None
+ if conn.conn in (CONN_TCP, CONN_TLS):
+ try:
+ file_size = int(request.POST.get("file_size", ""))
+ except (ValueError, TypeError):
+ return JsonResponse({"error": _("Invalid file size.")}, status=400)
+ if file_size <= 0:
+ return JsonResponse({"error": _("Invalid file size.")}, status=400)
+
+ upload_id = request.POST.get("upload_id")
+ if not upload_id or not isinstance(upload_id, str):
+ return JsonResponse({"error": _("Missing upload_id.")}, status=400)
+ upload_id = upload_id.strip()
+ if not re.match(r"^[a-zA-Z0-9_\-]+$", upload_id) or len(upload_id) > 64:
+ return JsonResponse({"error": _("Invalid upload_id format.")}, status=400)
+
+ try:
+ handle_uploaded_file(
+ conn,
+ path,
+ file_name,
+ file_chunk,
+ is_last_chunk,
+ chunk_index=chunk_index,
+ total_chunks=total_chunks,
+ upload_id=upload_id,
+ user_id=request.user.id,
+ compute_id=compute.id,
+ pool=pool,
+ file_size=file_size,
+ )
+
+ if is_last_chunk:
+ success_msg = _("ISO: %(file)s has been uploaded successfully.") % {"file": file_name}
+ messages.success(request, success_msg)
+ return JsonResponse({"success": True, "message": success_msg, "reload": True})
+ else:
+ return JsonResponse({"success": True, "message": "Chunk received."})
+ except FileExistsError:
+ return JsonResponse({"error": _("ISO image already exists")}, status=400)
+ except TimeoutError:
+ return JsonResponse({"error": _("Upload session is busy. Please retry.")}, status=429)
+ except ValueError as e:
+ return JsonResponse({"error": str(e)}, status=400)
+ except Exception as e:
+ error_msg = str(e)
+ messages.error(request, error_msg)
+ return JsonResponse({"error": error_msg}, status=500)
@superuser_only
@@ -98,74 +170,6 @@ def storages(request, compute_id):
return render(request, "storages.html", locals())
-def handle_uploaded_file(conn, path, file_name, file_chunk, is_last_chunk):
- clean_name = os.path.basename(file_name).strip()
- if not clean_name or clean_name in (".", "..") or "\x00" in clean_name:
- raise ValueError(_("Invalid file name"))
-
- if conn.conn == CONN_SSH:
- remote_base = posixpath.normpath(path)
- target_temp = posixpath.normpath(posixpath.join(remote_base, f"{clean_name}.part"))
- target_final = posixpath.normpath(posixpath.join(remote_base, clean_name))
-
- if not (target_temp.startswith(remote_base + "/") or target_temp == remote_base) or \
- not (target_final.startswith(remote_base + "/") or target_final == remote_base):
- raise PermissionError(_("Security issues with file uploading: path traversal detected"))
-
- try:
- hostname, port = conn.host, 22
- if ":" in hostname:
- hostname, port_str = hostname.split(":")
- port = int(port_str)
-
- ssh = paramiko.SSHClient()
- ssh.load_system_host_keys()
- ssh.set_missing_host_key_policy(paramiko.WarningPolicy())
- ssh.connect(hostname=hostname, port=port, username=conn.login, password=conn.passwd)
- try:
- sftp = ssh.open_sftp()
- try:
- remote_file = sftp.open(target_temp, "ab")
- try:
- remote_file.set_pipelined(True)
- for chunk_data in file_chunk.chunks():
- remote_file.write(chunk_data)
- finally:
- remote_file.close()
-
- if is_last_chunk:
- sftp.rename(target_temp, target_final)
- finally:
- sftp.close()
- finally:
- ssh.close()
- except Exception as e:
- raise Exception(_("SSH upload failed: {}").format(e))
- elif conn.conn == CONN_SOCKET:
- base_dir = os.path.abspath(path)
- target_temp = os.path.abspath(os.path.join(base_dir, f"{clean_name}.part"))
- target_final = os.path.abspath(os.path.join(base_dir, clean_name))
-
- if (
- os.path.commonpath([base_dir, target_temp]) != base_dir
- or os.path.commonpath([base_dir, target_final]) != base_dir
- ):
- raise PermissionError(_("Security issues with file uploading: path traversal detected"))
-
- try:
- with open(target_temp, "ab") as f:
- for chunk_data in file_chunk.chunks():
- f.write(chunk_data)
- if is_last_chunk:
- if os.path.exists(target_final):
- os.remove(target_final)
- os.rename(target_temp, target_final)
- except FileNotFoundError:
- raise Exception(_("File not found. Check the path variable and filename"))
- else:
- raise Exception(_("Unsupported connection type for file upload."))
-
-
@superuser_only
def storage(request, compute_id, pool):
"""
@@ -182,6 +186,12 @@ def storage(request, compute_id, pool):
compute.hostname, compute.login, compute.password, compute.type, pool
)
+ if request.method == "POST" and "iso_upload" in request.POST:
+ try:
+ return _handle_iso_upload(request, conn, conn.get_target_path(), compute, pool)
+ finally:
+ conn.close()
+
storages = conn.get_storages()
state = conn.is_active()
try:
@@ -230,71 +240,6 @@ def storage(request, compute_id, pool):
)
return redirect(reverse("storage", args=[compute.id, pool]))
# return HttpResponseRedirect(request.get_full_path())
- if "iso_upload" in request.POST:
- file_chunk = request.FILES.get("file")
- if not file_chunk:
- return JsonResponse({"error": _("No file chunk was submitted.")}, status=400)
-
- file_name = request.POST.get("file_name", "")
- clean_file_name = os.path.basename(file_name).strip()
- if not clean_file_name or clean_file_name in (".", "..") or "\x00" in clean_file_name:
- return JsonResponse({"error": _("Invalid file name.")}, status=400)
- file_name = clean_file_name
-
- chunk_index = int(request.POST.get("chunk_index", 0))
- total_chunks = int(request.POST.get("total_chunks", 1))
- is_last_chunk = chunk_index == total_chunks - 1
-
- # On first chunk, check if file already exists
- if chunk_index == 0:
- if file_name in conn.get_volumes():
- return JsonResponse({"error": _("ISO image already exists")}, status=400)
- # Clean up any partial files from previous failed uploads
- if conn.conn == CONN_SOCKET:
- base_dir = os.path.abspath(path)
- temp_part_file = os.path.abspath(os.path.join(base_dir, f"{file_name}.part"))
- if os.path.commonpath([base_dir, temp_part_file]) == base_dir and os.path.exists(temp_part_file):
- os.remove(temp_part_file)
- elif conn.conn == CONN_SSH:
- remote_base = posixpath.normpath(path)
- temp_part_file = posixpath.normpath(posixpath.join(remote_base, f"{file_name}.part"))
- if temp_part_file.startswith(remote_base + "/") or temp_part_file == remote_base:
- try:
- hostname, port = conn.host, 22
- if ":" in hostname:
- hostname, port_str = hostname.split(":")
- port = int(port_str)
- ssh = paramiko.SSHClient()
- ssh.load_system_host_keys()
- ssh.set_missing_host_key_policy(paramiko.WarningPolicy())
- ssh.connect(hostname=hostname, port=port, username=conn.login, password=conn.passwd)
- try:
- sftp = ssh.open_sftp()
- try:
- sftp.remove(temp_part_file)
- except FileNotFoundError:
- pass # File doesn't exist, which is fine
- finally:
- sftp.close()
- finally:
- ssh.close()
- except Exception:
- # Best effort to clean up, if it fails, let it be.
- pass
-
- try:
- handle_uploaded_file(conn, path, file_name, file_chunk, is_last_chunk)
-
- if is_last_chunk:
- success_msg = _("ISO: %(file)s has been uploaded successfully.") % {"file": file_name}
- messages.success(request, success_msg)
- return JsonResponse({"success": True, "message": success_msg, "reload": True})
- else:
- return JsonResponse({"success": True, "message": "Chunk received."})
- except Exception as e:
- error_msg = str(e)
- messages.error(request, error_msg)
- return JsonResponse({"error": error_msg}, status=500)
if "cln_volume" in request.POST:
form = CloneImage(request.POST)
if form.is_valid():