1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 19:26:48 +00:00

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.
This commit is contained in:
catborise 2026-10-20 21:56:27 +03:00
parent 4ece22230e
commit 90c148d287
3 changed files with 39 additions and 20 deletions

View file

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

View file

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