From 90c148d28761ea6f9586debd6de1c4fc4e9687bd Mon Sep 17 00:00:00 2001 From: catborise Date: Tue, 20 Oct 2026 21:56:27 +0300 Subject: [PATCH] fix: a form libvirt refuses goes back to its page with the reason A libvirt error that a view left to ExceptionMiddleware showed the "500 Internal Server Error" page, which blames the server, at the form's action URL; its only way on was the browser's Back. Adding a NIC, a new or existing disk, editing a disk, setting a NIC's QoS and creating a volume end there when libvirt refuses, as do the other VM actions. For a form (any method but GET or HEAD) the middleware now goes back to the page the form was on, with the error in the messages, as it already does for a busy compute; a page that cannot be shown keeps the error page. same_origin_referer moves from the instances views to the middleware module, next to error_text, for both to use. --- instances/views.py | 23 +++++--------------- webvirtcloud/middleware.py | 14 ++++++++++++ webvirtcloud/tests/test_connection_errors.py | 22 ++++++++++++++++--- 3 files changed, 39 insertions(+), 20 deletions(-) diff --git a/instances/views.py b/instances/views.py index 66e25a1d..e4ed2119 100755 --- a/instances/views.py +++ b/instances/views.py @@ -28,7 +28,6 @@ from django.http import Http404, HttpResponse, JsonResponse from django.db.models import prefetch_related_objects from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse -from django.utils.http import url_has_allowed_host_and_scheme from django.utils.translation import gettext_noop as _ from django.views.decorators.http import require_POST from libvirt import (VIR_DOMAIN_UNDEFINE_KEEP_NVRAM, @@ -40,7 +39,7 @@ from libvirt import (VIR_DOMAIN_UNDEFINE_KEEP_NVRAM, from logs.views import addlogmsg from operations import claims, jobs from operations.utils import busy_message, summary -from webvirtcloud.middleware import error_text +from webvirtcloud.middleware import error_text, same_origin_referer from vrtManager import util from vrtManager.create import wvmCreate from vrtManager.instance import USAGE_MAX_WINDOW @@ -377,18 +376,8 @@ def sshkeys(request, pk): return JsonResponse(instance_keys, safe=False) -def _same_origin_referer(request, default): - """The Referer without its fragment if it is on this site, else default.""" - referer = request.META.get("HTTP_REFERER") - if referer and url_has_allowed_host_and_scheme( - url=referer, allowed_hosts={request.get_host()}, require_https=request.is_secure() - ): - return referer.split("#")[0] - return default - - def get_safe_redirect(request, default=None): - return redirect(_same_origin_referer(request, default or reverse("instances:index"))) + return redirect(same_origin_referer(request, default or reverse("instances:index"))) def _is_int(value): @@ -402,7 +391,7 @@ def _is_int(value): def _back(request, pk, tab): """Back to the page the form was on, at the given tab; the VM page if the Referer is missing or from another site.""" - return redirect(_same_origin_referer(request, reverse("instances:instance", args=[pk])) + "#" + tab) + return redirect(same_origin_referer(request, reverse("instances:instance", args=[pk])) + "#" + tab) def get_instance(user, pk, perm_type="view"): @@ -1318,7 +1307,7 @@ def unmount_iso(request, pk): else: msg = _("Unmount media: %(dev)s") % {"dev": dev} addlogmsg(request.user.username, instance.compute.name, instance.name, msg) - return redirect(_same_origin_referer(request, reverse("instances:instance", args=[pk])) + "#disks") + return redirect(same_origin_referer(request, reverse("instances:instance", args=[pk])) + "#disks") @require_POST @@ -1335,7 +1324,7 @@ def mount_iso(request, pk): else: msg = _("Mount media: %(dev)s") % {"dev": dev} addlogmsg(request.user.username, instance.compute.name, instance.name, msg) - return redirect(_same_origin_referer(request, reverse("instances:instance", args=[pk])) + "#disks") + return redirect(same_origin_referer(request, reverse("instances:instance", args=[pk])) + "#disks") @require_POST @@ -1562,7 +1551,7 @@ def set_video_model(request, pk): @serialize_instance_mutation def change_network(request, pk): instance = get_instance(request.user, pk) - back = _same_origin_referer(request, reverse("instances:instance", args=[pk])) + "#network" + back = same_origin_referer(request, reverse("instances:instance", args=[pk])) + "#network" # Each NIC has its own form; its fields end with that NIC's index. nums = [key[len("net-old-mac-"):] for key in request.POST if key.startswith("net-old-mac-")] diff --git a/webvirtcloud/middleware.py b/webvirtcloud/middleware.py index f9db7196..f029476a 100644 --- a/webvirtcloud/middleware.py +++ b/webvirtcloud/middleware.py @@ -1,6 +1,7 @@ from django.contrib import messages from django.http import JsonResponse from django.shortcuts import redirect, render +from django.utils.http import url_has_allowed_host_and_scheme from django.utils.translation import gettext_lazy as _ from libvirt import libvirtError from operations import claims, jobs @@ -11,6 +12,16 @@ API_PREFIX = "/api/" CONTENT_SECURITY_POLICY = "object-src 'none'; base-uri 'self'; form-action 'self'; frame-ancestors 'none'" +def same_origin_referer(request, default): + """The Referer without its fragment if it is on this site, else default.""" + referer = request.META.get("HTTP_REFERER") + if referer and url_has_allowed_host_and_scheme( + url=referer, allowed_hosts={request.get_host()}, require_https=request.is_secure() + ): + return referer.split("#")[0] + return default + + def error_text(request, exception): """ The libvirt error to show. Administrators (superusers and the global @@ -63,5 +74,8 @@ class ExceptionMiddleware: request, _("libvirt Error - %(exception)s") % {"exception": error_text(request, exception)}, ) + if request.method not in ("GET", "HEAD"): + # a refused form: back to the page it was on, with the reason + return redirect(same_origin_referer(request, "/")) return render(request, "500.html", {"libvirt_error": True}, status=500) # TODO: check connecting to host via VPN diff --git a/webvirtcloud/tests/test_connection_errors.py b/webvirtcloud/tests/test_connection_errors.py index ebe53a0b..84ca5e3d 100644 --- a/webvirtcloud/tests/test_connection_errors.py +++ b/webvirtcloud/tests/test_connection_errors.py @@ -33,8 +33,8 @@ class ConnectionErrorPageTests(TestCase): self.viewer.user_permissions.add(Permission.objects.get(codename="view_instances")) self.viewer = User.objects.get(pk=self.viewer.pk) - def handle(self, user, exception, path="/instances/1/poweron/"): - request = RequestFactory().post(path) + def handle(self, user, exception, path="/instances/1/poweron/", method="post", **headers): + request = getattr(RequestFactory(), method)(path, **headers) SessionMiddleware(lambda r: None).process_request(request) request._messages = FallbackStorage(request) request.user = user @@ -48,10 +48,26 @@ class ConnectionErrorPageTests(TestCase): ): with self.subTest(error=str(error)): response, messages = self.handle(self.owner, error) - self.assertEqual(response.status_code, 500) self.assertNotIn("10.0.0.9", " ".join(messages)) self.assertTrue(any("cannot be reached" in m for m in messages), messages) + def test_a_refused_form_goes_back_to_its_page_with_the_error(self): + error = libvirt_error(libvirt.VIR_FROM_QEMU, "Requested operation is not valid: domain is not running", + libvirt.VIR_ERR_OPERATION_INVALID) + response, messages = self.handle(self.admin, error, HTTP_REFERER="http://testserver/instances/1/#network") + self.assertEqual((response.status_code, response["Location"]), (302, "http://testserver/instances/1/")) + self.assertTrue(any("domain is not running" in m for m in messages), messages) + for referer in ({}, {"HTTP_REFERER": "https://evil.example/"}): + with self.subTest(referer=referer): + response, _ = self.handle(self.admin, error, **referer) + self.assertEqual(response["Location"], "/") + + def test_a_page_that_cannot_be_shown_is_the_error_page(self): + response, messages = self.handle(self.admin, libvirt_error(libvirt.VIR_FROM_RPC, HOST_DETAIL), + path="/instances/1/", method="get") + self.assertEqual(response.status_code, 500) + self.assertTrue(messages) + def test_administrators_see_connection_details(self): for user in (self.admin, self.viewer): with self.subTest(user=user.username):