j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-11 12:46:49 +00:00
Code Issues Projects Releases Wiki Activity

feat(storages): an ISO upload is a job

Browse source 
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.
This commit is contained in:
catborise 2026-10-16 09:21:07 +03:00
parent f2138e54d8
commit 56b4b03500
7 changed files with 1003 additions and 1450 deletions
Download patch file Download diff file Expand all files Collapse all files

57
storages/jobs.py
View file

@ -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)

10
storages/templates/storage.html
View file

@ -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);
}
},

2
storages/tests/test_sftp_pipelined.py
View file

@ -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):

671
storages/tests/test_upload_job.py Normal file
View file

@ -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))

866
storages/tests/test_views.py
View file

@ -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."""

770
storages/upload.py
View file

@ -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 = (
"<volume><name>{}</name><capacity unit='bytes'>{}</capacity>"
"<target><format type='raw'/></target></volume>"
).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 = (
"<volume><name>{}</name><capacity unit='bytes'>{}</capacity>"
"<target><format type='raw'/></target></volume>"
).format(escape(file_name), file_size)
conn.pool.createXML(xml, 0)
_write_libvirt_meta(meta_path, meta)
elif file_name not in volumes:
raise ValueError(_("Upload volume is missing"))
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

77
storages/views.py
View file

@ -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()