fix(instances): missing or malformed mutation form fields are reported, not a 500

Several instance mutation views crashed with a bare HTTP 500 when a form field
the UI always sends was absent or malformed (a direct or crafted request);
ExceptionMiddleware turns only libvirt and timeout errors into 4xx, so these
reached a 500. Fixed:

- add_network parsed a None network source ("'NoneType' has no attribute
  'split'"); it now requires a selected network.
- set_qos read an unbound local 'mac' when no field name contained "mac", and
  passed an invalid direction or non-numeric rate to the proxy (ValueError); it
  now requires a MAC, an inbound/outbound direction and whole-number rates.
- unset_qos passed an empty direction into the proxy's XPath (XPathEvalError);
  it now requires a MAC and a valid direction.
- resizevm_cpu and resize_memory ran int() over an empty or non-numeric value;
  they now validate with int() (not str.isdigit(), which accepts "²" that
  int() rejects).
- add_public_key passed a missing/non-numeric sshkeyid into an integer-id
  lookup (ValueError); it now requires a numeric id.

Each reports the problem and returns to the VM page without calling libvirt or
logging, as change_network already does.

Tests: each view with a field missing, non-numeric or a unicode digit returns
302 with an error message, writes no log, and does not attempt the mutation.
The tests fail on the previous code.

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

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: