diff --git a/storages/jobs.py b/storages/jobs.py
index 03d0c838..32c03089 100644
--- a/storages/jobs.py
+++ b/storages/jobs.py
@@ -1,15 +1,22 @@
"""Jobs of the storages app (operations framework)."""
+import logging
+import os
+
from django.core.exceptions import PermissionDenied
from django.utils.translation import gettext_noop
from vrtManager import util
from vrtManager.storage import wvmStorage
from computes.models import Compute
-from operations import claims
+from operations import claims, utils
from operations.jobs import Kind, Outcome, Plan, register
from operations.models import Claim, Operation
+from storages import upload
+
+logger = logging.getLogger(__name__)
+
def _storage(params):
c = Compute.objects.get(pk=params["compute"])
@@ -61,3 +68,51 @@ class CloneVolume(Kind):
return Outcome(Operation.UNCERTAIN, error="The copy exists, but the job did not see it finish; "
"check it, then resolve this job.")
return Outcome(Operation.FAILED, error="The copy was not made")
+
+
+@register
+class UploadVolume(Kind):
+ """An ISO upload: its chunk requests run it (storages.upload)."""
+
+ name = "volume.upload"
+ label = gettext_noop("Upload ISO image")
+ cancellable = True
+ progress_bytes = True
+ runs_in_request = True
+
+ def normalize(self, raw):
+ return {"compute": int(raw["compute"]), "pool": str(raw["pool"]),
+ "name": os.path.basename(str(raw["name"])).strip(), "size": int(raw["size"]),
+ "chunks": int(raw["chunks"])}
+
+ def validate(self, user, params):
+ if not user.is_superuser:
+ raise PermissionDenied
+ name = params["name"]
+ if not name or name in (".", "..") or "\x00" in name:
+ raise ValueError("Invalid file name")
+ if len(name) <= 4 or not name.lower().endswith(".iso"):
+ raise ValueError("Only .iso files can be uploaded")
+ if params["size"] <= 0:
+ raise ValueError("Invalid file size")
+ if params["chunks"] < 1:
+ raise ValueError("Invalid chunk index or total chunks")
+
+ def plan(self, user, params):
+ compute, stg = _storage(params)
+ stg.refresh()
+ if params["name"] in stg.get_volumes():
+ raise FileExistsError(f"The pool has a volume named {params['name']}")
+ pool = claims.pool_key(compute.pk, params["pool"])
+ return Plan(claims={claims.compute_key(compute.pk): Claim.SHARED, pool: Claim.SHARED,
+ claims.volume_key(compute.pk, params["pool"], params["name"]): Claim.EXCLUSIVE},
+ resource=pool, label=f"{compute.name}/{params['pool']}", intent={"chunk": 0, "created": False})
+
+ def cancel(self, op):
+ try:
+ upload.end(op, Operation.CANCELLED)
+ except Exception as err: # the next chunk or the expiry ends it
+ logger.error("cancelling upload %s failed: %s", op.pk, utils.safe_text(err))
+
+ def expire(self, op):
+ upload.end(op, Operation.FAILED, upload.IDLE)
diff --git a/storages/templates/storage.html b/storages/templates/storage.html
index 9b0375f1..dd048304 100644
--- a/storages/templates/storage.html
+++ b/storages/templates/storage.html
@@ -304,6 +304,9 @@
processData: false,
contentType: false,
success: function(data) {
+ if (chunkIndex === 0) {
+ window.resourceJobPanel.track(data.job);
+ }
retryCount = 0;
chunkIndex++;
pendingUpload.chunkIndex = chunkIndex;
@@ -311,16 +314,13 @@
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) {
+ if (chunkIndex < totalChunks) {
uploadChunk();
} else {
pendingUpload = null;
uploadInProgress = false;
uploadStatus.removeClass('d-flex').addClass('d-none');
+ progressContainer.hide();
uploadButton.prop('disabled', false);
}
},
diff --git a/storages/tests/test_sftp_pipelined.py b/storages/tests/test_sftp_pipelined.py
index 6d45e2d4..81d2f603 100644
--- a/storages/tests/test_sftp_pipelined.py
+++ b/storages/tests/test_sftp_pipelined.py
@@ -64,7 +64,7 @@ def upload(client):
target = _UploadTarget(MagicMock(conn=CONN_SSH, host="192.0.2.1"), "/pool", "x.iso", "id")
target.sftp = MagicMock(open=lambda path, mode: paramiko.SFTPFile(client, b"handle", "w"))
data = bytes(range(256)) * (SIZE // 256)
- return target.write_chunk(SimpleUploadedFile("x.iso", data), 0, new=True), data
+ return target.write(SimpleUploadedFile("x.iso", data), 0, new=True), data
class PipelinedWriteTestCase(unittest.TestCase):
diff --git a/storages/tests/test_upload_job.py b/storages/tests/test_upload_job.py
new file mode 100644
index 00000000..a589aad9
--- /dev/null
+++ b/storages/tests/test_upload_job.py
@@ -0,0 +1,671 @@
+"""An ISO upload is a job a request runs: each chunk request writes one chunk
+and commits it on the job (spec: the chunk protocol)."""
+
+import errno
+import os
+import shutil
+import tempfile
+from datetime import timedelta
+from unittest.mock import MagicMock, patch
+
+from django.contrib.auth import get_user_model
+from django.core.exceptions import PermissionDenied
+from django.core.files.uploadedfile import SimpleUploadedFile
+from django.test import TestCase
+from django.urls import reverse
+from django.utils import timezone
+from libvirt import VIR_ERR_NO_STORAGE_VOL, VIR_ERR_SYSTEM_ERROR, libvirtError
+
+from computes.models import Compute
+from operations import claims, jobs
+from operations.models import Claim, Operation
+from operations.worker import housekeeping
+from storages.upload import IDLE, _atomic_finalize_local, _ssh_pool, _UploadTarget, _VolumeTarget, write_chunk
+from vrtManager.connection import CONN_SOCKET, CONN_SSH, CONN_TCP, CONN_TLS
+
+User = get_user_model()
+KEY = "upload-key-0001"
+
+
+class UploadTestCase(TestCase):
+ """A pool on conn; the kind and the ending path open the same conn."""
+
+ def setUp(self):
+ self.compute = Compute.objects.create(name="up-c", hostname="192.0.2.30", login="root", password="", type=1)
+ self.admin = User.objects.create_superuser("up_admin", "up_admin@example.com", "pw")
+ self.conn = self.make_conn()
+ self.opened = {}
+ for target in ("storages.jobs.wvmStorage", "storages.upload.wvmStorage"):
+ patcher = patch(target, return_value=self.conn)
+ self.opened[target] = patcher.start()
+ self.addCleanup(patcher.stop)
+
+ def cancel_during_write(self, target_class):
+ """The panel's cancel while this request writes: it finds the upload held."""
+ write = target_class.write
+
+ def cancelled(target, *args):
+ Operation.objects.filter(kind="volume.upload").update(state=Operation.CANCELLING, cancel_requested=True)
+ return write(target, *args)
+
+ return patch.object(target_class, "write", cancelled)
+
+ def send(self, index, data, total=3, size=9, name="disk.iso", key=KEY, user=None):
+ return write_chunk(user or self.admin, self.conn, self.compute, "default", key,
+ {"name": name, "size": size, "chunks": total}, index, SimpleUploadedFile("chunk", data))
+
+ def job(self):
+ return Operation.objects.get(kind="volume.upload")
+
+
+class LocalPool:
+ def make_conn(self):
+ self.pool = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.pool, ignore_errors=True)
+ conn = MagicMock(conn=CONN_SOCKET)
+ conn.get_target_path.return_value = self.pool
+ conn.get_volumes.side_effect = lambda: sorted(os.listdir(self.pool))
+ return conn
+
+ def path(self, name="disk.iso"):
+ return os.path.join(self.pool, name)
+
+ def part(self, op):
+ return self.path(f"disk.iso.wvc_upload_{op.pk.hex}.part")
+
+ def read(self, path):
+ with open(path, "rb") as stream:
+ return stream.read()
+
+
+class LocalUploadTests(LocalPool, UploadTestCase):
+ def test_three_chunks_are_one_job(self):
+ op = self.send(0, b"aaa")
+ op.refresh_from_db()
+ self.assertEqual((op.state, op.worker, op.progress_done, op.progress_total), (Operation.RUNNING, Operation.REQUEST, 3, 9))
+ self.assertEqual(op.intent, {"chunk": 1, "created": True})
+ self.assertEqual(set(Claim.objects.values_list("resource", "mode")), {
+ (f"compute:{self.compute.pk}", "shared"), (f"pool:{self.compute.pk}:default", "shared"),
+ (f"volume:{self.compute.pk}:default/disk.iso", "exclusive")})
+ self.assertEqual(self.read(self.part(op)), b"aaa")
+ self.send(1, b"bbb")
+ op.refresh_from_db()
+ self.assertEqual(op.progress_done, 6)
+ self.send(2, b"ccc")
+ op.refresh_from_db()
+ self.assertEqual(op.state, Operation.SUCCEEDED)
+ self.assertEqual(self.read(self.path()), b"aaabbbccc")
+ self.assertFalse(os.path.exists(self.part(op)))
+ self.assertFalse(Claim.objects.exists())
+ self.assertEqual(Operation.objects.count(), 1)
+
+ def test_a_duplicate_chunk_is_acknowledged_without_a_write(self):
+ op = self.send(0, b"aaa")
+ self.assertEqual(self.send(0, b"zzz").pk, op.pk)
+ self.send(1, b"bbb")
+ self.send(1, b"yyy")
+ self.send(2, b"ccc")
+ # the final answer was lost: the retry is answered success
+ self.assertEqual(self.send(2, b"ccc").state, Operation.SUCCEEDED)
+ self.assertEqual(self.read(self.path()), b"aaabbbccc")
+
+ def test_a_duplicate_last_chunk_finishes_a_committed_upload(self):
+ op = self.send(0, b"aaa", total=1, size=3)
+ Operation.objects.filter(pk=op.pk).update(state=Operation.RUNNING, finished_at=None)
+ os.rename(self.path(), self.part(op)) # the request died before the rename
+ self.assertEqual(self.send(0, b"aaa", total=1, size=3).state, Operation.SUCCEEDED)
+ self.assertEqual(self.read(self.path()), b"aaa")
+
+ def test_a_link_an_interrupted_rename_made_is_completed(self):
+ op = self.committed_but_not_renamed()
+ os.link(self.part(op), self.path()) # the request died after the link, before the unlink
+ self.assertEqual(self.send(1, b"bbb", total=2, size=6).state, Operation.SUCCEEDED)
+ self.assertEqual(os.listdir(self.pool), ["disk.iso"])
+ self.assertEqual(self.read(self.path()), b"aaabbb")
+
+ def test_an_image_gone_before_the_completion_fails_the_job(self):
+ op = self.committed_but_not_renamed()
+ os.unlink(self.part(op))
+ with self.assertRaisesRegex(ValueError, "The uploaded image is missing"):
+ self.send(1, b"bbb", total=2, size=6)
+ op.refresh_from_db()
+ self.assertEqual((op.state, op.error), (Operation.FAILED, "The uploaded image is missing"))
+ self.assertFalse(Claim.objects.exists())
+
+ def test_the_first_write_syncs_the_pool_directory_before_the_commit(self):
+ synced = []
+ with patch("storages.upload._fsync_dir", side_effect=lambda d: synced.append((d, self.job().progress_done))):
+ self.send(0, b"aaa")
+ self.send(1, b"bbb")
+ self.assertEqual(synced, [(self.pool, None)])
+
+ def test_sequence_and_session_are_checked(self):
+ with self.assertRaisesRegex(ValueError, "not found"):
+ self.send(1, b"bbb")
+ self.send(0, b"aaa")
+ with self.assertRaisesRegex(ValueError, "sequence"):
+ self.send(2, b"ccc")
+ for changed in ({"size": 10}, {"total": 4}, {"name": "other.iso"}):
+ with self.assertRaisesRegex(ValueError, "changed"):
+ self.send(1, b"bbb", **changed)
+ with self.assertRaises(jobs.SubmissionConflict):
+ self.send(0, b"aaa", size=10)
+ with self.assertRaisesRegex(ValueError, "declared file size"):
+ self.send(1, b"bbbbbbb")
+ self.assertEqual(self.job().progress_done, 3)
+
+ def test_a_writer_that_died_mid_chunk_is_cut_back(self):
+ op = self.send(0, b"aaa")
+ with open(self.part(op), "ab") as stream:
+ stream.write(b"UNCOMMITTED")
+ self.send(1, b"bbb")
+ self.assertEqual(self.read(self.part(op)), b"aaabbb")
+
+ def test_an_interrupted_first_write_is_retried_from_the_start(self):
+ chunk = SimpleUploadedFile("disk.iso", b"aaa")
+
+ def interrupted():
+ yield b"a"
+ raise OSError("Interrupted first write")
+
+ chunk.chunks = interrupted
+ with self.assertRaises(OSError):
+ write_chunk(self.admin, self.conn, self.compute, "default", KEY,
+ {"name": "disk.iso", "size": 9, "chunks": 3}, 0, chunk)
+ op = self.job()
+ self.assertEqual((op.intent, op.progress_done), ({"chunk": 0, "created": True}, None))
+ self.assertEqual(self.read(self.part(op)), b"a")
+ self.send(0, b"aaa")
+ self.assertEqual(self.read(self.part(op)), b"aaa")
+
+ def test_a_failure_before_the_part_file_is_retried(self):
+ with patch("storages.upload._UploadTarget.write", side_effect=OSError("Cannot open part")):
+ with self.assertRaises(OSError):
+ self.send(0, b"aaa", total=1, size=3)
+ self.send(0, b"aaa", total=1, size=3)
+ self.assertEqual(self.read(self.path()), b"aaa")
+
+ def test_a_name_taken_when_renameat2_finalizes_is_a_taken_name(self):
+ part, final = self.path("disk.iso.part"), self.path()
+ with open(part, "wb") as stream:
+ stream.write(b"aaa")
+ libc = MagicMock()
+ libc.renameat2.return_value = -1
+ with patch("os.link", side_effect=OSError(errno.EPERM, "no hard links")), \
+ patch("ctypes.CDLL", return_value=libc), patch("ctypes.get_errno", return_value=errno.EEXIST):
+ with self.assertRaises(FileExistsError):
+ _atomic_finalize_local(part, final)
+ self.assertEqual(self.read(part), b"aaa")
+
+ def test_a_taken_name_is_refused_without_a_job(self):
+ with open(self.path(), "wb") as stream:
+ stream.write(b"mine")
+ with self.assertRaises(FileExistsError):
+ self.send(0, b"aaa")
+ self.assertFalse(Operation.objects.exists())
+ self.assertFalse(Claim.objects.exists())
+
+ def test_a_destination_that_appears_before_the_rename_fails_the_job(self):
+ op = self.send(0, b"aaa")
+ self.send(1, b"bbb")
+ with open(self.path(), "wb") as stream:
+ stream.write(b"mine")
+ with self.assertRaises(FileExistsError):
+ self.send(2, b"ccc")
+ op.refresh_from_db()
+ self.assertEqual(op.state, Operation.FAILED)
+ self.assertFalse(os.path.exists(self.part(op)))
+ self.assertEqual(self.read(self.path()), b"mine")
+ self.assertFalse(Claim.objects.exists())
+
+ def test_cancel_deletes_the_part_file(self):
+ op = self.send(0, b"aaa")
+ op = jobs.cancel(op.pk, self.admin)
+ self.assertEqual(op.state, Operation.CANCELLED)
+ self.assertFalse(os.path.exists(self.part(op)))
+ self.assertFalse(Claim.objects.exists())
+ with self.assertRaises(ValueError):
+ self.send(1, b"bbb")
+ jobs.get_kind("volume.upload").cancel(op) # already ended: nothing to do
+
+ def test_a_cancel_during_a_chunk_is_ended_by_the_next_chunk(self):
+ op = self.send(0, b"aaa")
+ claims.admit({f"upload:{op.pk}": Claim.EXCLUSIVE}, holder="192.0.2.30:1:1") # a chunk in flight
+ op = jobs.cancel(op.pk, self.admin)
+ self.assertEqual(op.state, Operation.CANCELLING)
+ self.assertTrue(os.path.exists(self.part(op)))
+ with self.assertRaises(claims.Busy):
+ self.send(1, b"bbb")
+ Claim.objects.filter(operation__isnull=True).delete()
+ with self.assertRaisesRegex(ValueError, "cancelled"):
+ self.send(1, b"bbb")
+ op.refresh_from_db()
+ self.assertEqual(op.state, Operation.CANCELLED)
+ self.assertFalse(os.path.exists(self.part(op)))
+ self.assertFalse(Claim.objects.exists())
+
+ def test_a_cancel_during_a_write_ends_the_upload_after_it(self):
+ op = self.send(0, b"aaa")
+ with self.cancel_during_write(_UploadTarget), self.assertRaisesRegex(ValueError, "cancelled"):
+ self.send(1, b"bbb")
+ op.refresh_from_db()
+ self.assertEqual(op.state, Operation.CANCELLED)
+ self.assertEqual(os.listdir(self.pool), [])
+ self.assertFalse(Claim.objects.exists())
+
+ def test_a_cancel_during_the_last_write_comes_too_late(self):
+ with self.cancel_during_write(_UploadTarget):
+ self.send(0, b"aaa", total=1, size=3)
+ self.assertEqual(self.job().state, Operation.SUCCEEDED)
+ self.assertEqual(self.read(self.path()), b"aaa")
+
+ def test_a_repeated_cancel_retries_the_stop(self):
+ op = self.send(0, b"aaa")
+ claims.admit({f"upload:{op.pk}": Claim.EXCLUSIVE}, holder="192.0.2.30:1:1") # a request that died
+ self.assertEqual(jobs.cancel(op.pk, self.admin).state, Operation.CANCELLING)
+ Claim.objects.filter(operation__isnull=True).delete()
+ self.assertEqual(jobs.cancel(op.pk, self.admin).state, Operation.CANCELLED)
+ self.assertFalse(os.path.exists(self.part(op)))
+
+ def committed_but_not_renamed(self):
+ """Every chunk is in; the last request died before the rename."""
+ self.send(0, b"aaa", total=2, size=6)
+ with patch("storages.upload._complete", side_effect=OSError("the request died")):
+ with self.assertRaises(OSError):
+ self.send(1, b"bbb", total=2, size=6)
+ return self.job()
+
+ def test_a_cancel_of_an_upload_with_every_chunk_in_completes_it(self):
+ op = jobs.cancel(self.committed_but_not_renamed().pk, self.admin)
+ self.assertEqual(op.state, Operation.SUCCEEDED)
+ self.assertEqual(os.listdir(self.pool), ["disk.iso"])
+ self.assertEqual(self.read(self.path()), b"aaabbb")
+
+ def test_a_retry_of_the_last_chunk_completes_a_cancelled_upload_with_every_chunk_in(self):
+ op = self.committed_but_not_renamed()
+ Operation.objects.filter(pk=op.pk).update(state=Operation.CANCELLING, cancel_requested=True)
+ self.assertEqual(self.send(1, b"bbb", total=2, size=6).state, Operation.SUCCEEDED)
+ self.assertEqual(self.read(self.path()), b"aaabbb")
+
+ def test_an_idle_upload_with_every_chunk_in_is_completed(self):
+ op = self.committed_but_not_renamed()
+ Operation.objects.filter(pk=op.pk).update(heartbeat_at=timezone.now() - timedelta(seconds=3601))
+ housekeeping()
+ op.refresh_from_db()
+ self.assertEqual(op.state, Operation.SUCCEEDED)
+ self.assertEqual(self.read(self.path()), b"aaabbb")
+
+ def test_a_late_retry_of_the_last_chunk_completes_the_upload(self):
+ op = self.committed_but_not_renamed()
+ Operation.objects.filter(pk=op.pk).update(heartbeat_at=timezone.now() - timedelta(seconds=3601))
+ self.assertEqual(self.send(1, b"bbb", total=2, size=6).state, Operation.SUCCEEDED)
+ self.assertEqual(self.read(self.path()), b"aaabbb")
+
+ def test_ending_without_the_compute(self):
+ op = self.send(0, b"aaa")
+ self.compute.delete()
+ op = jobs.cancel(op.pk, self.admin)
+ self.assertEqual((op.state, op.error), (Operation.FAILED, "The compute was deleted"))
+
+ def test_ending_before_anything_was_created_needs_no_connection(self):
+ op = jobs.submit("volume.upload", self.admin, {"compute": self.compute.pk, "pool": "default",
+ "name": "disk.iso", "size": 9, "chunks": 3}, KEY)
+ self.assertEqual(jobs.cancel(op.pk, self.admin).state, Operation.CANCELLED)
+ self.opened["storages.upload.wvmStorage"].assert_not_called()
+
+ def test_a_non_superuser_is_refused(self):
+ user = User.objects.create_user("up_user", "up_user@example.com", "pw")
+ with self.assertRaises(PermissionDenied):
+ self.send(0, b"aaa", user=user)
+ self.assertFalse(Operation.objects.exists())
+
+ def test_an_idle_upload_expires(self):
+ op = self.send(0, b"aaa")
+ housekeeping() # not idle: nothing is opened
+ op.refresh_from_db()
+ self.assertEqual(op.state, Operation.RUNNING)
+ self.opened["storages.upload.wvmStorage"].assert_not_called()
+ Operation.objects.filter(pk=op.pk).update(heartbeat_at=timezone.now() - timedelta(seconds=3601))
+ housekeeping()
+ op.refresh_from_db()
+ self.assertEqual((op.state, op.error), (Operation.FAILED, IDLE))
+ self.assertFalse(os.path.exists(self.part(op)))
+ self.assertFalse(Claim.objects.exists())
+
+ def test_an_idle_cancelled_upload_expires_cancelled(self):
+ op = self.send(0, b"aaa")
+ Operation.objects.filter(pk=op.pk).update(state=Operation.CANCELLING, cancel_requested=True,
+ heartbeat_at=timezone.now() - timedelta(seconds=3601))
+ housekeeping()
+ op.refresh_from_db()
+ self.assertEqual((op.state, op.error), (Operation.CANCELLED, ""))
+ self.assertFalse(os.path.exists(self.part(op)))
+
+ def test_a_chunk_after_the_idle_timeout_ends_the_job(self):
+ op = self.send(0, b"aaa")
+ Operation.objects.filter(pk=op.pk).update(heartbeat_at=timezone.now() - timedelta(seconds=3601))
+ with self.assertRaisesRegex(ValueError, "idle"):
+ self.send(1, b"bbb")
+ op.refresh_from_db()
+ self.assertEqual((op.state, op.error), (Operation.FAILED, IDLE))
+ self.assertFalse(os.path.exists(self.part(op)))
+
+ def test_a_symlinked_part_file_is_refused(self):
+ op = self.send(0, b"aaa")
+ os.unlink(self.part(op))
+ os.symlink(self.path("real.iso"), self.part(op))
+ with self.assertRaises(PermissionError):
+ self.send(1, b"bbb")
+ self.assertFalse(os.path.exists(self.path("real.iso")))
+
+ def test_path_traversal_is_refused(self):
+ with patch("os.path.basename", side_effect=lambda x: x):
+ with self.assertRaises(PermissionError):
+ self.send(0, b"aaa", total=1, size=3, name="../../../evil.iso")
+ self.assertFalse(os.path.exists(os.path.join(self.pool, "../../../evil.iso")))
+
+ def test_the_name_must_be_an_iso(self):
+ for name in ("image.img", ".iso", "..", "a\x00.iso"):
+ with self.assertRaises(ValueError):
+ self.send(0, b"aaa", name=name)
+ self.send(0, b"aaa", total=1, size=3, name="image.ISO")
+ self.assertTrue(os.path.exists(self.path("image.ISO")))
+
+
+class SshUploadTests(UploadTestCase):
+ base = "/var/lib/libvirt/images"
+
+ def make_conn(self):
+ self.addCleanup(_ssh_pool.close_all)
+ conn = MagicMock(conn=CONN_SSH, host="192.0.2.31", login="root", passwd="pw")
+ conn.get_target_path.return_value = self.base
+ self.files = files = {}
+ conn.get_volumes.side_effect = lambda: [p.rsplit("/", 1)[1] for p in files]
+ patcher = patch("paramiko.SSHClient")
+ self.ssh = patcher.start().return_value
+ self.addCleanup(patcher.stop)
+ sftp = self.ssh.open_sftp.return_value
+
+ def lstat(path):
+ if path not in files:
+ raise FileNotFoundError(path)
+ return MagicMock(st_mode=0o100644)
+
+ class RemoteFile:
+ """The SFTP file of a pipelined write; it is its own client."""
+
+ MAX_REQUEST_SIZE = 32768
+ handle = b"handle"
+ sock = MagicMock(recv_ready=lambda: False)
+
+ def __init__(self, path, mode):
+ self.path = path
+ if "w" in mode or path not in files:
+ files[path] = b""
+ self.pos = 0
+
+ @property
+ def sftp(self):
+ return self
+
+ def _async_request(self, fileobj, kind, handle, offset, data):
+ buf = bytearray(files[self.path])
+ buf[offset:offset + len(data)] = data
+ files[self.path] = bytes(buf)
+
+ def _finish_responses(self, fileobj):
+ pass
+
+ def _check_exception(self):
+ pass
+
+ def seek(self, pos):
+ self.pos = pos
+
+ def truncate(self, size):
+ files[self.path] = files[self.path][:size]
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ pass
+
+ def rename(src, dst):
+ if dst in files:
+ raise OSError("Failure") # what OpenSSH answers
+ files[dst] = files.pop(src)
+
+ sftp.lstat.side_effect = lstat
+ sftp.open.side_effect = RemoteFile
+ sftp.rename.side_effect = rename
+ sftp.remove.side_effect = lambda path: files.pop(path)
+ return conn
+
+ def part(self, op):
+ return f"{self.base}/disk.iso.wvc_upload_{op.pk.hex}.part"
+
+ def test_three_chunks_over_sftp(self):
+ op = self.send(0, b"aaa")
+ self.assertEqual(self.ssh.connect.call_args.kwargs["timeout"], 10)
+ self.files[self.part(op)] += b"UNCOMMITTED" # a writer died after its write
+ self.send(1, b"bbb")
+ self.send(2, b"ccc")
+ self.send(2, b"ccc")
+ op.refresh_from_db()
+ self.assertEqual(op.state, Operation.SUCCEEDED)
+ self.assertEqual(self.files, {f"{self.base}/disk.iso": b"aaabbbccc"})
+ # one connection for the chunks, an SFTP session for each that wrote
+ self.ssh.connect.assert_called_once()
+ self.assertEqual(self.ssh.open_sftp.call_count, 3)
+
+ def test_a_destination_that_appears_before_the_rename_fails_the_job(self):
+ op = self.send(0, b"aaa")
+ self.send(1, b"bbb")
+ self.files[f"{self.base}/disk.iso"] = b"mine"
+ with self.assertRaises(FileExistsError):
+ self.send(2, b"ccc")
+ op.refresh_from_db()
+ self.assertEqual(op.state, Operation.FAILED)
+ self.assertEqual(self.files, {f"{self.base}/disk.iso": b"mine"})
+
+ def test_a_rename_whose_answer_was_lost_is_completed(self):
+ rename = self.ssh.open_sftp.return_value.rename.side_effect
+
+ def lost(src, dst):
+ rename(src, dst)
+ raise OSError("Failure")
+
+ self.ssh.open_sftp.return_value.rename.side_effect = lost
+ self.assertEqual(self.send(0, b"aaa", total=1, size=3).state, Operation.SUCCEEDED)
+ self.assertEqual(self.files, {f"{self.base}/disk.iso": b"aaa"})
+
+ def test_cancel_deletes_the_remote_part_file(self):
+ op = self.send(0, b"aaa")
+ self.assertEqual(jobs.cancel(op.pk, self.admin).state, Operation.CANCELLED)
+ self.assertEqual(self.files, {})
+
+ def test_path_traversal_is_refused(self):
+ with patch("os.path.basename", side_effect=lambda x: x):
+ with self.assertRaises(PermissionError):
+ self.send(0, b"aaa", total=1, size=3, name="../../../evil.iso")
+ self.assertEqual(self.files, {})
+
+
+class LibvirtStreamUploadTests(UploadTestCase):
+ def make_conn(self):
+ conn = MagicMock(conn=CONN_TCP)
+ self.volumes = volumes = {}
+ conn.get_volumes.side_effect = lambda: list(volumes)
+
+ def get_volume(name):
+ if name not in volumes:
+ error = libvirtError("Storage volume not found")
+ error.err = (VIR_ERR_NO_STORAGE_VOL,)
+ raise error
+ return volumes[name]
+
+ conn.get_volume.side_effect = get_volume
+
+ def create(xml, flags):
+ from xml.etree import ElementTree
+
+ root = ElementTree.fromstring(xml)
+ name = root.findtext("name")
+ self.assertNotIn(name, volumes)
+ self.assertEqual((root.findtext("capacity"), root.find("target/format").get("type")), ("9", "raw"))
+ volumes[name] = MagicMock()
+ volumes[name].delete.side_effect = lambda flags: volumes.pop(name)
+
+ conn.pool.createXML.side_effect = create
+ conn.wvm.newStream.return_value.send.side_effect = len
+ return conn
+
+ def test_chunks_stream_at_their_offsets(self):
+ self.send(0, b"aaa")
+ self.send(0, b"aaa")
+ volume = self.volumes["disk.iso"]
+ self.send(1, b"bbb")
+ self.send(2, b"ccc")
+ self.send(2, b"ccc")
+ self.conn.pool.createXML.assert_called_once()
+ self.assertEqual([c.args[1:] for c in volume.upload.call_args_list], [(0, 3, 0), (3, 3, 0), (6, 3, 0)])
+ self.assertEqual(self.job().state, Operation.SUCCEEDED)
+
+ def test_tls_uses_the_stream(self):
+ self.conn.conn = CONN_TLS
+ self.send(0, b"aaaaaaaaa", total=1)
+ self.volumes["disk.iso"].upload.assert_called_once()
+
+ def test_a_failed_stream_aborts_and_the_chunk_is_retried(self):
+ stream = self.conn.wvm.newStream.return_value
+ stream.send.side_effect = [1, OSError("connection lost")]
+ with self.assertRaises(OSError):
+ self.send(0, b"aaa")
+ stream.abort.assert_called_once()
+ stream.send.side_effect = len
+ self.send(0, b"aaa")
+ self.conn.pool.createXML.assert_called_once()
+ self.assertEqual(self.job().progress_done, 3)
+
+ def test_an_idle_upload_with_every_chunk_in_is_kept(self):
+ self.send(0, b"aaa")
+ self.send(1, b"bbb")
+ with patch("storages.upload._complete", side_effect=OSError("the request died")):
+ with self.assertRaises(OSError):
+ self.send(2, b"ccc")
+ Operation.objects.filter(pk=self.job().pk).update(heartbeat_at=timezone.now() - timedelta(seconds=3601))
+ housekeeping()
+ self.assertEqual(self.job().state, Operation.SUCCEEDED)
+ self.assertEqual(list(self.volumes), ["disk.iso"])
+
+ def test_a_cancel_during_the_last_write_keeps_the_volume(self):
+ with self.cancel_during_write(_VolumeTarget):
+ self.send(0, b"aaaaaaaaa", total=1)
+ self.assertEqual((self.job().state, list(self.volumes)), (Operation.SUCCEEDED, ["disk.iso"]))
+
+ def test_a_volume_gone_before_the_completion_fails_the_job(self):
+ self.send(0, b"aaa")
+ self.send(1, b"bbb")
+ with patch("storages.upload._complete", side_effect=OSError("the request died")):
+ with self.assertRaises(OSError):
+ self.send(2, b"ccc")
+ del self.volumes["disk.iso"]
+ Operation.objects.filter(pk=self.job().pk).update(heartbeat_at=timezone.now() - timedelta(seconds=3601))
+ housekeeping()
+ self.assertEqual((self.job().state, self.job().error), (Operation.FAILED, "The uploaded image is missing"))
+
+ def test_a_failed_volume_lookup_is_raised(self):
+ error = libvirtError("connection lost")
+ error.err = (VIR_ERR_SYSTEM_ERROR,)
+ self.conn.get_volume.side_effect = error
+ with self.assertRaises(libvirtError):
+ _VolumeTarget(self.conn, "disk.iso", 9).present()
+
+ def test_cancel_deletes_the_volume(self):
+ op = self.send(0, b"aaa")
+ self.assertEqual(jobs.cancel(op.pk, self.admin).state, Operation.CANCELLED)
+ self.assertEqual(self.volumes, {})
+
+ def test_an_existing_volume_is_kept(self):
+ self.volumes["disk.iso"] = MagicMock()
+ with self.assertRaises(FileExistsError):
+ self.send(0, b"aaa")
+ self.conn.pool.createXML.assert_not_called()
+
+
+class UploadViewTests(LocalPool, UploadTestCase):
+ def setUp(self):
+ super().setUp()
+ patcher = patch("storages.views.wvmStorage", return_value=self.conn)
+ patcher.start()
+ self.addCleanup(patcher.stop)
+ self.client.force_login(self.admin)
+ self.url = reverse("storage", args=[self.compute.pk, "default"])
+
+ def post(self, index=0, data=b"aaa", total=1, size="3", name="disk.iso", key=KEY, **fields):
+ payload = {"iso_upload": "1", "file": SimpleUploadedFile(name, data), "file_name": name,
+ "chunk_index": str(index), "total_chunks": str(total), "file_size": size, "upload_id": key}
+ payload.update(fields)
+ return self.client.post(self.url, {k: v for k, v in payload.items() if v is not None})
+
+ def test_the_answer_carries_the_job(self):
+ first = self.post(0, total=2, size="6")
+ self.assertEqual(first.status_code, 200)
+ op = self.job()
+ self.assertEqual(first.json(), {"success": True, "job": str(op.pk)})
+ self.assertEqual(self.client.session["operations_shown"],
+ [{"resource": f"pool:{self.compute.pk}:default", "id": str(op.pk)}])
+ last = self.post(1, data=b"bbb", total=2, size="6").json()
+ self.assertEqual((last["success"], last["job"]), (True, str(op.pk)))
+ self.assertEqual(last, {"success": True, "job": str(op.pk)})
+
+ def test_field_errors(self):
+ for fields, status, error in (
+ ({"name": "notes.txt"}, 400, "Only .iso files"),
+ ({"file": None}, 400, "No file chunk was submitted"),
+ ({"file_name": "..\x00.."}, 400, "Invalid file name"),
+ ({"key": None}, 400, "no valid submission key"),
+ ({"size": None}, 400, "Invalid file size"),
+ ({"size": "0"}, 400, "Invalid file size"),
+ ({"index": 1}, 400, "Invalid chunk index"),
+ ):
+ with self.subTest(fields=fields):
+ kwargs = {k: v for k, v in fields.items() if k in ("name", "size", "key", "index")}
+ extra = {k: v for k, v in fields.items() if k not in kwargs}
+ response = self.post(**kwargs, **extra)
+ self.assertEqual(response.status_code, status)
+ self.assertIn(error, response.json()["error"])
+ self.assertFalse(Operation.objects.exists())
+
+ def test_a_taken_name(self):
+ with open(self.path(), "wb") as stream:
+ stream.write(b"mine")
+ response = self.post()
+ self.assertEqual((response.status_code, response.json()["error"]), (400, "ISO image already exists"))
+
+ def test_a_busy_session_is_retried(self):
+ self.post(0, total=2, size="6")
+ claims.admit({f"upload:{self.job().pk}": Claim.EXCLUSIVE}, holder="192.0.2.30:1:1")
+ response = self.post(1, data=b"bbb", total=2, size="6")
+ self.assertEqual((response.status_code, response.json()["error"]),
+ (429, "Upload session is busy. Please retry."))
+
+ def test_a_name_another_upload_holds_is_refused_with_the_reason(self):
+ self.post(0, total=2, size="6")
+ response = self.post(0, total=2, size="6", key="upload-key-0002")
+ self.assertEqual(response.status_code, 400)
+ self.assertIn("Upload ISO image is running on this resource", response.json()["error"])
+
+ def test_submission_keys_are_per_user(self):
+ self.post(0, total=2, size="6")
+ other = User.objects.create_superuser("up_other", "up_other@example.com", "pw")
+ self.client.force_login(other)
+ response = self.post(1, data=b"bbb", total=2, size="6")
+ self.assertEqual(response.status_code, 400)
+ self.assertIn("not found", response.json()["error"])
+ response = self.post(0, total=1, size="3", name="other.iso")
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(Operation.objects.filter(submission_key=KEY).count(), 2)
+ self.assertNotEqual(response.json()["job"], str(Operation.objects.get(user=self.admin).pk))
+
diff --git a/storages/tests/test_views.py b/storages/tests/test_views.py
index 5c8847f6..8ef08e5c 100644
--- a/storages/tests/test_views.py
+++ b/storages/tests/test_views.py
@@ -1,874 +1,16 @@
-import os
-import shutil
-import tempfile
-import time
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
-from django.core.files.uploadedfile import SimpleUploadedFile
-from django.test import RequestFactory, TestCase
-from django.contrib.messages import get_messages
-from django.urls import reverse
from django.contrib.auth import get_user_model
+from django.contrib.messages import get_messages
+from django.test import RequestFactory, TestCase
+from django.urls import reverse
from computes.models import Compute
from operations import claims
-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,
- _ssh_pool,
-)
-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()
- self.mock_socket_conn = MagicMock()
- self.mock_socket_conn.conn = CONN_SOCKET
-
- self.mock_ssh_conn = MagicMock()
- self.mock_ssh_conn.conn = CONN_SSH
- self.mock_ssh_conn.host = "127.0.0.1"
- self.mock_ssh_conn.login = "test"
- self.mock_ssh_conn.passwd = "test"
- self.addCleanup(_ssh_pool.close_all)
-
- 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()
- 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,
- 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, 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_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,
- 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")
- 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, 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)
-
- # the SFTP client calls of a pipelined write: the file is its own client
- MAX_REQUEST_SIZE = 32768
- handle = b"handle"
- sock = MagicMock(recv_ready=lambda: False)
-
- @property
- def sftp(self):
- return self
-
- def _async_request(self, fileobj, kind, handle, offset, data):
- self.seek(offset)
- self.write(data)
-
- def _finish_responses(self, fileobj):
- pass
-
- def _check_exception(self):
- pass
-
- 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")
- # one connection for the three requests, an SFTP session for each
- mock_ssh.connect.assert_called_once()
- self.assertEqual(mock_ssh.open_sftp.call_count, 3)
-
- 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, upload_id="sess-ssh-1")
-
-
- def test_interrupted_first_chunk_can_be_retried_without_appending_partial_data(self):
- file_name = "first-chunk-retry.iso"
- chunk = SimpleUploadedFile(file_name, b"complete-data")
-
- def interrupted():
- yield b"partial"
- raise OSError("Interrupted first write")
-
- chunk.chunks = interrupted
- with self.assertRaises(OSError):
- handle_uploaded_file(
- self.mock_socket_conn, self.temp_dir, file_name, chunk, True,
- upload_id="first-chunk-retry",
- )
- handle_uploaded_file(
- self.mock_socket_conn, self.temp_dir, file_name,
- SimpleUploadedFile(file_name, b"complete-data"), True,
- upload_id="first-chunk-retry",
- )
- with open(os.path.join(self.temp_dir, file_name), "rb") as stream:
- self.assertEqual(stream.read(), b"complete-data")
-
- def test_first_chunk_can_retry_after_failure_before_part_file_creation(self):
- chunk = SimpleUploadedFile("start-retry.iso", b"complete-data")
- with patch("storages.upload._UploadTarget.write_chunk", side_effect=OSError("Cannot open part")):
- with self.assertRaises(OSError):
- handle_uploaded_file(self.mock_socket_conn, self.temp_dir, chunk.name, chunk, True, upload_id="start-retry")
- handle_uploaded_file(self.mock_socket_conn, self.temp_dir, chunk.name, chunk, True, upload_id="start-retry")
- with open(os.path.join(self.temp_dir, chunk.name), "rb") as stream:
- self.assertEqual(stream.read(), b"complete-data")
-
-
-class StorageUploadViewTests(TestCase):
- def setUp(self):
- self.admin = User.objects.create_superuser(username="storage_admin", password="password", email="storage_admin@example.com")
- self.client.login(request=RequestFactory().get("/"), 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()
- 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"])
- 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", ""))
-
- @patch("storages.views.wvmStorage")
- def test_iso_upload_invalid_filename(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": "..\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(request=RequestFactory().get("/"), 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", ""))
-
-
class GetVolumesViewTests(TestCase):
"""get_volumes must not answer 200 with an empty list when the pool cannot
be read: the caller could not tell that apart from a pool with no volumes."""
diff --git a/storages/upload.py b/storages/upload.py
index 34b5cb8b..aa8ef184 100644
--- a/storages/upload.py
+++ b/storages/upload.py
@@ -1,25 +1,28 @@
-"""Chunked ISO upload operations shared by the storage view."""
+"""A chunked ISO upload: a job its chunk requests run (the chunk protocol)."""
import atexit
-import fcntl
-import json
-import logging
+import errno
import os
import posixpath
-import re
import stat
-import tempfile
import threading
import time
+from datetime import timedelta
from xml.sax.saxutils import escape
import paramiko
+from django.utils import timezone
from django.utils.translation import gettext_lazy as _
+from libvirt import VIR_ERR_NO_STORAGE_VOL, libvirtError
+from computes.models import Compute
+from operations import claims, jobs
+from operations.models import Claim, Operation
from vrtManager.connection import CONN_SSH, CONN_SOCKET, CONN_TCP, CONN_TLS
+from vrtManager.storage import wvmStorage
-logger = logging.getLogger(__name__)
-LOCK_TIMEOUT_SECONDS = 15
+IDLE = "The upload was idle for over an hour; its partial file was deleted"
+IDLE_SECONDS = 3600
SSH_CONNECT_TIMEOUT_SECONDS = 10
SSH_IO_TIMEOUT_SECONDS = 30
SSH_IDLE_SECONDS = 60
@@ -122,17 +125,13 @@ def _atomic_finalize_local(target_temp, 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"))
+ # an earlier attempt linked the part file and died before unlinking it
+ if not os.path.samestat(os.lstat(target_temp), os.lstat(target_final)):
+ raise FileExistsError(_("Destination file already exists")) from None
except OSError as link_err:
# 2. Secondary: Linux kernel renameat2 with RENAME_NOREPLACE (flag 1)
+ ret, err = -1, 0
try:
import ctypes
libc = ctypes.CDLL("libc.so.6", use_errno=True)
@@ -143,17 +142,22 @@ def _atomic_finalize_local(target_temp, target_final):
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):
+ except (AttributeError, OSError): # no libc or no renameat2
pass
-
+ if ret == 0:
+ _fsync_dir(parent_dir)
+ return
+ if err == errno.EEXIST:
+ raise FileExistsError(_("Destination file already exists"))
# 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))
+ _fsync_dir(parent_dir)
+ try:
+ os.unlink(target_temp)
+ _fsync_dir(parent_dir)
+ except OSError:
+ pass
def _atomic_finalize_ssh(sftp, target_temp, target_final):
@@ -163,36 +167,17 @@ def _atomic_finalize_ssh(sftp, target_temp, target_final):
"""
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):
+ except OSError as err:
+ # OpenSSH answers a taken destination with a generic "Failure"
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)}"
+ sftp.lstat(target_final)
+ except OSError:
+ raise err from None
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"))
+ sftp.lstat(target_temp)
+ except FileNotFoundError:
+ return # the rename was done; its answer was lost
+ raise FileExistsError(_("Destination file already exists")) from err
def _write_pipelined(stream, chunk, offset):
@@ -215,209 +200,21 @@ def _write_pipelined(stream, chunk, offset):
return offset
-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."""
+ """The part file of an upload to a local (socket) or SSH pool, renamed to
+ the image when the last chunk is written."""
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.base = normalize(path)
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):
+ for target in (self.final, self.part):
if self.remote:
contained = target.startswith(self.base.rstrip("/") + "/")
else:
@@ -505,41 +302,13 @@ class _UploadTarget:
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 present(self):
+ return self.info(self.part) is not None
- 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 done(self):
+ return self.info(self.final) is not None
- def write_chunk(self, chunk, offset, new):
+ def write(self, chunk, offset, new):
if self.remote:
with self.sftp.open(self.part, "wb" if new else "r+b") as stream:
if not new:
@@ -560,6 +329,8 @@ class _UploadTarget:
offset += len(data)
stream.flush()
os.fsync(stream.fileno())
+ if new: # the part file's name survives a host crash once its offset is committed
+ _fsync_dir(self.base)
return offset
def finalize(self):
@@ -568,268 +339,223 @@ class _UploadTarget:
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)
+ def remove(self):
+ if not self.present():
return
- except BlockingIOError:
- if time.monotonic() >= deadline:
- raise TimeoutError(_("Timed out waiting for upload session lock"))
- time.sleep(0.05)
+ if self.remote:
+ self.sftp.remove(self.part)
+ else:
+ os.unlink(self.part)
+ _fsync_dir(self.base)
-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))
+class _VolumeTarget:
+ """The volume of an upload to a TCP/TLS pool, written through libvirt
+ streams; it has the image's name from the first chunk."""
+ def __init__(self, conn, name, size):
+ self.conn = conn
+ self.name = name
+ self.size = size
-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"))
+ def __enter__(self):
+ return self
- 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):
+ def __exit__(self, *_):
+ pass
+
+ def present(self):
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(),
- }
+ self.conn.get_volume(self.name)
+ except libvirtError as err:
+ if err.get_error_code() == VIR_ERR_NO_STORAGE_VOL:
+ return False
+ raise
+ return True
- 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}
- )
+ done = present
- 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"))
+ def write(self, chunk, offset, new):
+ if new:
+ xml = (
+ "{}{}"
+ ""
+ ).format(escape(self.name), self.size)
+ self.conn.pool.createXML(xml, 0)
+ volume = self.conn.get_volume(self.name)
+ stream = self.conn.wvm.newStream(0)
+ try:
+ volume.upload(stream, offset, chunk.size, 0)
+ written = 0
+ for data in 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 != chunk.size:
+ raise OSError(_("Upload chunk size changed during transfer"))
+ stream.finish()
+ except Exception:
+ stream.abort()
+ raise
+ return offset + written
- # A volume is visible as soon as libvirt creates it. Keep the session until
- # the final chunk finishes so retries can resume safely at a known offset.
- volumes = conn.get_volumes()
- if meta["expected_chunk"] == 0 and not os.path.exists(meta_path):
- if file_name in volumes:
- raise FileExistsError(_("Destination file already exists"))
- xml = (
- "{}{}"
- ""
- ).format(escape(file_name), file_size)
- conn.pool.createXML(xml, 0)
- _write_libvirt_meta(meta_path, meta)
- elif file_name not in volumes:
- raise ValueError(_("Upload volume is missing"))
+ def finalize(self):
+ pass
- volume = conn.get_volume(file_name)
- stream = conn.wvm.newStream(0)
+ def remove(self):
+ if self.present():
+ self.conn.get_volume(self.name).delete(0)
+
+
+def _target(conn, op):
+ if conn.conn in (CONN_TCP, CONN_TLS):
+ return _VolumeTarget(conn, op.params["name"], op.params["size"])
+ return _UploadTarget(conn, conn.get_target_path(), op.params["name"], op.pk.hex)
+
+
+def _session_key(op):
+ """Held by a chunk request, and by the ending path, for the whole of its work."""
+ return f"upload:{op.pk}"
+
+
+def _idle(op):
+ return op.heartbeat_at < timezone.now() - timedelta(seconds=IDLE_SECONDS)
+
+
+def _commit(op, **fields):
+ if not Operation.objects.filter(pk=op.pk, state__in=Operation.ACTIVE).update(**fields):
+ raise ValueError(_("The upload has ended"))
+ for name, value in fields.items():
+ setattr(op, name, value)
+
+
+def _complete(target, op):
+ """Rename the part file to the image, unless a retry finds it done, and
+ succeed; a name taken meanwhile, or an image gone, fails the upload."""
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()
+ if target.present():
+ target.finalize()
+ elif not target.done():
+ jobs.finish(op, jobs.Outcome(Operation.FAILED, error="The uploaded image is missing"))
+ raise ValueError(_("The uploaded image is missing"))
+ except FileExistsError:
+ target.remove()
+ jobs.finish(op, jobs.Outcome(Operation.FAILED, error="An image of this name appeared before the upload "
+ "finished; the upload was deleted"))
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)
+ jobs.finish(op, jobs.Outcome(Operation.SUCCEEDED))
-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"))
+def _end(target, op, state, error=""):
+ if op.intent.get("created"):
+ target.remove()
+ jobs.finish(op, jobs.Outcome(state, error=error))
- 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)
+
+def _refuse(target, op, state, message, error=""):
+ """End the upload and refuse the chunk."""
+ _end(target, op, state, error)
+ raise ValueError(message)
+
+
+def end(op, state, error=""):
+ """Delete what the upload made and end it, unless a chunk request holds
+ it: that request sees the cancel, or the next expiry pass retries. A
+ requested cancel ends it cancelled. An upload whose every chunk is in is
+ completed instead: a cancel then comes too late."""
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, "upload_id": upload_id,
- }
- # Record ownership before creating the part file, so an interrupted
- # first write can be retried without adopting an unrelated file.
- target.write_meta(meta)
- else:
- _matches_session(meta, clean_name, total_chunks, user_id, compute_id, pool)
- if meta["expected_chunk"] > 0 and 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()
+ with claims.hold({_session_key(op): Claim.EXCLUSIVE}):
+ op.refresh_from_db()
+ if op.state not in Operation.ACTIVE or (error == IDLE and not _idle(op)):
+ return # ended already, or expiry of an upload that is not idle
+ if op.cancel_requested:
+ state, error = Operation.CANCELLED, ""
+ if not op.intent.get("created"):
+ jobs.finish(op, jobs.Outcome(state, error=error))
return
- if chunk_index != expected:
- raise ValueError(
- _("Invalid chunk sequence. Expected chunk %(exp)d, got %(got)d")
- % {"exp": expected, "got": chunk_index}
- )
+ compute = Compute.objects.filter(pk=op.params["compute"]).first()
+ if compute is None:
+ jobs.finish(op, jobs.Outcome(Operation.FAILED, error="The compute was deleted"))
+ return
+ conn = wvmStorage(compute.hostname, compute.login, compute.password, compute.type, op.params["pool"])
+ try:
+ with _target(conn, op) as target:
+ # its last request died before the rename
+ if op.intent["chunk"] == op.params["chunks"]:
+ try:
+ _complete(target, op)
+ except (FileExistsError, ValueError): # _complete failed the upload
+ pass
+ else:
+ _end(target, op, state, error)
+ finally:
+ conn.close()
+ except claims.Busy:
+ pass
- meta["current_size"] = target.write_chunk(
- file_chunk, meta["current_size"], new=target.info(target.part) is None
- )
- 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)
+
+def write_chunk(user, conn, compute, pool, upload_id, raw, chunk_index, chunk, request=None):
+ """Write chunk chunk_index of the upload upload_id (the job's submission
+ key) and commit it on the job; the last chunk renames the image."""
+ kind = jobs.get_kind("volume.upload")
+ raw = {**raw, "compute": compute.pk, "pool": pool}
+ if chunk_index == 0:
+ op = jobs.submit(kind.name, user, raw, upload_id)
+ if request is not None:
+ jobs.remember_job(request, op)
+ else:
+ op = Operation.objects.filter(user=user, submission_key=upload_id).first()
+ if op is None:
+ raise ValueError(_("Upload session not found or expired"))
+ if op.kind != kind.name or op.params != kind.normalize(raw):
+ raise ValueError(_("Upload session parameters changed"))
+
+ with claims.hold({_session_key(op): Claim.EXCLUSIVE}):
+ op.refresh_from_db()
+ params = op.params
+ last = chunk_index == params["chunks"] - 1
+ if op.state == Operation.SUCCEEDED and last:
+ return op # the answer to the last chunk was lost
+ if op.state not in Operation.ACTIVE:
+ raise ValueError(op.error or _("The upload has ended"))
+ with _target(conn, op) as target:
+ expected = op.intent["chunk"]
+ # once every chunk is in, a cancel comes too late: a retry completes it below
+ if op.cancel_requested and expected < params["chunks"]:
+ _refuse(target, op, Operation.CANCELLED, _("The upload was cancelled"))
+ if chunk_index > expected:
+ raise ValueError(_("Invalid chunk sequence. Expected chunk %(exp)d, got %(got)d")
+ % {"exp": expected, "got": chunk_index})
+ if chunk_index < expected: # a retry of a committed chunk
+ if last and expected == params["chunks"]:
+ _complete(target, op)
+ return op
+ if _idle(op):
+ _refuse(target, op, Operation.FAILED, IDLE, IDLE)
+ offset = op.progress_done or 0
+ if chunk.size <= 0:
+ raise ValueError(_("Empty upload chunk"))
+ if offset + chunk.size > params["size"]:
+ raise ValueError(_("Upload exceeds declared file size"))
+ if last and offset + chunk.size != params["size"]:
+ raise ValueError(_("Final chunk does not match declared file size"))
+ present = target.present() # refuses a symlinked part file
+ # a first write that failed before creating it is retried as new
+ new = not (op.intent["created"] and present)
+ if new and offset:
+ jobs.finish(op, jobs.Outcome(Operation.FAILED, error="The partial upload is missing"))
+ raise ValueError(_("The partial upload is missing"))
+ if not op.intent["created"]:
+ _commit(op, intent={"chunk": expected, "created": True})
+ written = target.write(chunk, offset, new)
+ _commit(op, progress_done=written, progress_total=params["size"],
+ intent={"chunk": expected + 1, "created": True}, heartbeat_at=timezone.now())
+ if last: # a cancel during the write comes too late
+ _complete(target, op)
+ # a cancel that came during the write found this request holding the upload
+ elif Operation.objects.filter(pk=op.pk, cancel_requested=True).exists():
+ _refuse(target, op, Operation.CANCELLED, _("The upload was cancelled"))
+ return op
diff --git a/storages/views.py b/storages/views.py
index 5942680a..cf9a69e9 100644
--- a/storages/views.py
+++ b/storages/views.py
@@ -1,6 +1,3 @@
-import os
-import re
-
from admin.decorators import superuser_only
from appsettings.settings import app_settings
from computes.models import Compute
@@ -14,84 +11,46 @@ from libvirt import libvirtError
from operations import claims, jobs
from operations.utils import busy_message
-from vrtManager.connection import CONN_TCP, CONN_TLS
from vrtManager.util import OperationError
from vrtManager.storage import wvmStorage, wvmStorages
from storages.forms import AddStgPool, CloneImage, CreateVolumeForm
-from storages.upload import handle_uploaded_file
+from storages.upload import write_chunk
-def _handle_iso_upload(request, conn, path, compute, pool):
+def _handle_iso_upload(request, conn, 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:
+ if 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:
+ file_size = int(request.POST.get("file_size", ""))
+ except (ValueError, TypeError):
+ return JsonResponse({"error": _("Invalid file size.")}, 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."})
+ op = write_chunk(request.user, conn, compute, pool, request.POST.get("upload_id", ""),
+ {"name": request.POST.get("file_name", ""), "size": file_size, "chunks": total_chunks},
+ chunk_index, file_chunk, request=request)
+ except claims.Busy as busy:
+ if busy.blocker is not None: # another job holds the pool or the name
+ return JsonResponse({"error": busy_message(busy, request.user)}, status=400)
+ return JsonResponse({"error": _("Upload session is busy. Please retry.")}, status=429)
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:
+ except (ValueError, jobs.SubmissionConflict, OperationError) 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)
+ return JsonResponse({"error": str(e)}, status=500)
+ return JsonResponse({"success": True, "job": str(op.pk)})
@superuser_only
@@ -191,7 +150,7 @@ def storage(request, compute_id, pool):
if request.method == "POST" and "iso_upload" in request.POST:
try:
- return _handle_iso_upload(request, conn, conn.get_target_path(), compute, pool)
+ return _handle_iso_upload(request, conn, compute, pool)
finally:
conn.close()