From 2441b21c0c2506f6ad175681260ed3f1031ea79e Mon Sep 17 00:00:00 2001
From: catborise <catborise@gmail.com>
Date: Mon, 28 Sep 2026 14:57:31 +0300
Subject: [PATCH] feat(storage): support resilient ISO uploads across
 connection types

---
 storages/templates/create_stg_vol_block.html |   6 +-
 storages/templates/storage.html              |  58 +-
 storages/tests.py                            | 718 ++++++++++++++++++-
 storages/upload.py                           | 692 ++++++++++++++++++
 storages/views.py                            | 217 +++---
 5 files changed, 1537 insertions(+), 154 deletions(-)
 create mode 100644 storages/upload.py

diff --git a/storages/templates/create_stg_vol_block.html b/storages/templates/create_stg_vol_block.html
index 020b9c17..f418b8ba 100644
--- a/storages/templates/create_stg_vol_block.html
+++ b/storages/templates/create_stg_vol_block.html
@@ -21,9 +21,13 @@
                                 <div class="row mb-3">
                                     <label for="id_file" class="col-sm-3 col-form-label">{% trans "Name" %}</label>
                                     <div class="col-sm-9">
-                                        <input type="file" name="file" id="id_file" class="form-control" required>
+                                        <input type="file" name="file" id="id_file" class="form-control" accept=".iso" required>
                                     </div>
                                 </div>
+                                <div id="upload-status" class="d-none align-items-center gap-2 text-primary mt-3" role="status" aria-live="polite">
+                                    <span class="spinner-border spinner-border-sm" aria-hidden="true"></span>
+                                    <span id="upload-status-text">{% trans "Starting upload..." %}</span>
+                                </div>
                                 <div id="upload-progress-container" class="mt-3" style="display: none;">
                                     <div class="progress">
                                         <div id="upload-progress-bar" class="progress-bar" role="progressbar" style="width: 0%;" aria-valuenow="0" aria-valuemin="0" aria-valuemax="100">0%</div>
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 = (
+            "<volume><name>{}</name><capacity unit='bytes'>{}</capacity>"
+            "<target><format type='raw'/></target></volume>"
+        ).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():