diff --git a/instances/test_missing_fields.py b/instances/test_missing_fields.py new file mode 100644 index 00000000..83b721fb --- /dev/null +++ b/instances/test_missing_fields.py @@ -0,0 +1,95 @@ +"""A mutation view whose form field the UI always sends must not crash with a +500 when the field is missing (a malformed or direct request): it reports the +problem and returns to the VM page without changing anything.""" + +from unittest.mock import patch + +from django.contrib.auth import get_user_model +from django.contrib.messages import get_messages +from django.test import TestCase +from django.urls import reverse + +from computes.models import Compute +from instances.models import Instance +from logs.models import Logs + + +class MissingFieldTestCase(TestCase): + def setUp(self): + self.client.force_login(get_user_model().objects.create_superuser("mf", "mf@example.com", "pw")) + self.compute = Compute.objects.create(name="mf-c", hostname="127.0.0.1", login="", password="", type=4) + self.instance = Instance.objects.create( + compute=self.compute, name="mf-vm", uuid="11111111-2222-3333-4444-555555555555" + ) + + def post(self, name, data=None): + with patch("instances.models.wvmInstance"), patch("instances.views.wvmInterface"): + return self.client.post( + reverse(f"instances:{name}", args=[self.instance.id]), + data or {}, + HTTP_REFERER=f"http://testserver/instances/{self.instance.id}/", + ) + + def assert_reported(self, response): + self.assertEqual(response.status_code, 302) + levels = [m.level_tag for m in get_messages(response.wsgi_request)] + self.assertIn("error", levels, "an error message is shown") + self.assertFalse(Logs.objects.exists(), "nothing is logged as done") + + def test_add_network_without_a_network(self): + self.assert_reported(self.post("add_network")) + + def test_add_network_mutation_is_not_attempted(self): + with patch("instances.models.wvmInstance") as wvm, patch("instances.views.wvmInterface"): + self.client.post(reverse("instances:add_network", args=[self.instance.id]), {}) + wvm.return_value.add_network.assert_not_called() + + def test_set_qos_without_a_mac(self): + self.assert_reported(self.post("set_qos", {"qos_direction": "inbound"})) + + def test_set_qos_without_a_direction(self): + self.assert_reported(self.post("set_qos", {"net-mac-0": "52:54:00:00:00:01"})) + + def test_resize_cpu_without_a_value(self): + self.assert_reported(self.post("resizevm_cpu")) + + def test_resize_cpu_with_a_non_numeric_value(self): + self.assert_reported(self.post("resizevm_cpu", {"vcpu": "lots", "cur_vcpu": "1"})) + + def test_resize_memory_without_a_value(self): + self.assert_reported(self.post("resize_memory")) + + def test_resize_cpu_with_a_unicode_digit(self): + # "²".isdigit() is True but int("²") raises; the guard must still hold + self.assert_reported(self.post("resizevm_cpu", {"vcpu": "²", "cur_vcpu": "1"})) + + def test_resize_memory_with_a_unicode_digit(self): + self.assert_reported(self.post("resize_memory", {"memory_custom": "²", "cur_memory": "512"})) + + def test_set_qos_with_non_numeric_rate(self): + self.assert_reported(self.post("set_qos", {"qos_direction": "inbound", "net-mac-0": "52:54:00:00:00:01", "qos_average": "fast"})) + + def test_set_qos_with_an_invalid_direction(self): + self.assert_reported(self.post("set_qos", {"qos_direction": "sideways", "net-mac-0": "52:54:00:00:00:01"})) + + def test_unset_qos_without_a_direction(self): + self.assert_reported(self.post("unset_qos", {"net-mac": "52:54:00:00:00:01"})) + + def test_add_public_key_without_a_key(self): + response = self.post("add_public_key") + self.assertEqual(response.status_code, 302) + levels = [m.level_tag for m in get_messages(response.wsgi_request)] + self.assertIn("error", levels) + + def test_mutations_are_not_attempted(self): + cases = { + "set_qos": ({"qos_direction": "inbound"}, "set_qos"), + "unset_qos": ({"net-mac": "52:54:00:00:00:01"}, "unset_qos"), + "resizevm_cpu": ({}, "resize_cpu"), + "resize_memory": ({}, "resize_mem"), + } + for view, (data, method) in cases.items(): + with self.subTest(view=view): + with patch("instances.models.wvmInstance") as wvm, patch("instances.views.wvmInterface"): + self.client.post(reverse(f"instances:{view}", args=[self.instance.id]), data) + getattr(wvm.return_value, method).assert_not_called() diff --git a/instances/views.py b/instances/views.py index e442a57e..0441f383 100755 --- a/instances/views.py +++ b/instances/views.py @@ -366,6 +366,14 @@ def get_safe_redirect(request, default=None): return redirect(_same_origin_referer(request, default or reverse("instances:index"))) +def _is_int(value): + try: + int(value) + except (TypeError, ValueError): + return False + return True + + 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.""" @@ -744,6 +752,11 @@ def add_public_key(request, pk): if request.method == "POST": sshkeyid = request.POST.get("sshkeyid", "") + if not _is_int(sshkeyid): + messages.error(request, _("Select an SSH public key")) + return get_safe_redirect( + request, default=reverse("instances:instance", args=[instance.id]) + "#access" + ) publickey = get_object_or_404(UserSSHKey, id=sshkeyid, user=request.user) data = { "action": "publickey", @@ -782,6 +795,9 @@ def resizevm_cpu(request, pk): new_vcpu = request.POST.get("vcpu", "") new_cur_vcpu = request.POST.get("cur_vcpu", "") + if not _is_int(new_vcpu) or not _is_int(new_cur_vcpu): + messages.error(request, _("Enter a whole number of vCPUs")) + return redirect(reverse("instances:instance", args=[instance.id]) + "#resize") quota_msg = utils.check_user_quota( request.user, 0, int(new_vcpu) - vcpu, 0, 0 @@ -830,6 +846,9 @@ def resize_memory(request, pk): # shuts down meanwhile does not get an unchecked one. if instance.proxy.get_status() == VIR_DOMAIN_RUNNING: new_memory = memory + if not _is_int(new_memory) or not _is_int(new_cur_memory): + messages.error(request, _("Enter a whole number for memory")) + return redirect(reverse("instances:instance", args=[instance.id]) + "#resize") quota_msg = utils.check_user_quota(request.user, 0, 0, int(new_memory) - memory, 0) if not request.user.is_superuser and quota_msg: msg = quota_refused(quota_msg, _( @@ -1592,9 +1611,13 @@ def change_network(request, pk): def add_network(request, pk): instance = get_instance(request.user, pk) + network = request.POST.get("add-net-network") + if not network: + messages.error(request, _("No network selected")) + return _back(request, pk, "network") mac = request.POST.get("add-net-mac") nwfilter = request.POST.get("add-net-nwfilter") - (source, source_type) = utils.get_network_tuple(request.POST.get("add-net-network")) + (source, source_type) = utils.get_network_tuple(network) model = request.POST.get("add-net-model") if source_type == "iface": @@ -1651,10 +1674,14 @@ def set_qos(request, pk): average = request.POST.get("qos_average") or 0 peak = request.POST.get("qos_peak") or 0 burst = request.POST.get("qos_burst") or 0 - keys = request.POST.keys() - mac_key = [key for key in keys if "mac" in key] - if mac_key: - mac = request.POST.get(mac_key[0]) + mac_key = [key for key in request.POST if "mac" in key] + mac = request.POST.get(mac_key[0]) if mac_key else None + if not mac or qos_dir not in ("inbound", "outbound"): + messages.error(request, _("Select a network interface and a QoS direction")) + return _back(request, pk, "network") + if not all(_is_int(rate) for rate in (average, peak, burst)): + messages.error(request, _("QoS rates must be whole numbers")) + return _back(request, pk, "network") instance.proxy.set_qos(mac, qos_dir, average, peak, burst) if instance.proxy.get_status() == 5: @@ -1681,6 +1708,9 @@ def unset_qos(request, pk): instance = get_instance(request.user, pk) qos_dir = request.POST.get("qos_direction", "") mac = request.POST.get("net-mac") + if not mac or qos_dir not in ("inbound", "outbound"): + messages.error(request, _("Select a network interface and a QoS direction")) + return _back(request, pk, "network") instance.proxy.unset_qos(mac, qos_dir) if instance.proxy.get_status() == 5: