From 56b4b03500bd81f47ff028d0ddb5623e04768029 Mon Sep 17 00:00:00 2001 From: catborise Date: Fri, 16 Oct 2026 09:21:07 +0300 Subject: [PATCH] feat(storages): an ISO upload is a job The upload session lives in a job instead of flocks and .meta/.done files; the job panel shows and cancels it; an upload idle for an hour is ended and its partial file deleted; the page no longer reloads after an upload. --- storages/jobs.py | 57 +- storages/templates/storage.html | 10 +- storages/tests/test_sftp_pipelined.py | 2 +- storages/tests/test_upload_job.py | 671 ++++++++++++++++++++ storages/tests/test_views.py | 866 +------------------------- storages/upload.py | 770 ++++++++--------------- storages/views.py | 77 +-- 7 files changed, 1003 insertions(+), 1450 deletions(-) create mode 100644 storages/tests/test_upload_job.py 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()