mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-09 03:36:48 +00:00
feat(storage): support resilient ISO uploads across connection types
This commit is contained in:
catborise
parent
73d8fd71fb
commit
2441b21c0c
5 changed files with 1537 additions and 154 deletions
|
|
@ -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>
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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", ""))
|
||||
|
|
|
|||
692
storages/upload.py
Normal file
692
storages/upload.py
Normal file
|
|
@ -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)
|
||||
|
|
@ -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():
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue