@@ -38,29 +38,6 @@
{{ forloop.counter }}
{{ nwfilter.uuid }} {{ nwfilter.name }}
-
-
-
- {% bs_icon 'eye' %}
-
-
@@ -123,6 +100,7 @@
editor.getSession().setMode("ace/mode/xml");
var input = $('input[name="nwfilter_xml"]');
+ input.val(editor.getSession().getValue());
editor.getSession().on("change",function () {
input.val(editor.getSession().getValue());
})
diff --git a/nwfilters/tests/test_audit_rows.py b/nwfilters/tests/test_audit_rows.py
index 27380f67..9b60b57a 100644
--- a/nwfilters/tests/test_audit_rows.py
+++ b/nwfilters/tests/test_audit_rows.py
@@ -1,7 +1,7 @@
"""Network filter changes write rows on the Logs page. libvirt's test driver
has no filters, so the connection is a mock (as in test_redefine.py)."""
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
from computes.models import Compute
from django.contrib.auth import get_user_model
@@ -11,6 +11,8 @@ from django.urls import reverse
from libvirt import libvirtError
from logs.models import Logs
+from lxml import etree
+from vrtManager.nwfilters import element_text
UUID = "6d1f7a10-0000-4000-8000-00000000c0de"
XML = (f"
{UUID} {UUID} "
+ "{refs} "),
+ ("{UUID} ", ""), "keep the filter's UUID"),
+ (WEB.replace(UUID, "11111111-2222-3333-4444-555555555555"), "keep the filter's UUID")):
+ with self.subTest(why=why), self.assertRaisesRegex(OperationError, why):
+ conn.edit(xml)
+ conn.wvm.nwfilterDefineXML.assert_not_called()
+ conn.edit(WEB.replace("priority='500'", "priority='400'"))
+ conn.wvm.nwfilterDefineXML.assert_called_once()
+
+
+class ChangeTests(SimpleTestCase):
+ def test_a_reference_is_to_another_filter_of_the_host_not_yet_referenced(self):
+ for name in ("web", "clean-traffic", "missing"):
+ with self.subTest(name=name), self.assertRaisesRegex(OperationError, "cannot be referenced"):
+ one().add_ref(name)
+ conn = one()
+ conn.add_ref("no-spoofing")
+ self.assertEqual(defined(conn).xpath("filterref/@filter"), ["clean-traffic", "no-spoofing"])
+ self.assertEqual(one().get_ref_choices(), ["no-spoofing"])
+
+ def test_a_rule_is_one_rule_element(self):
+ with self.assertRaisesRegex(OperationError, "one "):
+ one().add_rule("{UUID} {UUID} ")
+def text(xml, tag, index):
+ """The text the page sends back for the tag element at index."""
+ return element_text(etree.fromstring(xml).findall(tag)[index])
+
+
class FilterRedefineTests(TestCase):
def setUp(self):
patcher = patch("vrtManager.connection.connection_manager.get_connection")
@@ -29,10 +35,11 @@ class FilterRedefineTests(TestCase):
nwfilter.XMLDesc.return_value = XML
nwfilter.name.return_value = "web"
nwfilter.UUIDString.return_value = UUID
- self.wvm.listNWFilters.return_value = ["web"]
+ self.wvm.listNWFilters.return_value = ["web", "clean-traffic", "no-spoofing"]
compute = Compute.objects.create(name="nwf", hostname="localhost", login="", password="", type=4)
self.client.force_login(get_user_model().objects.create_superuser("nwf", "nwf@example.com", "pw"))
self.url = reverse("nwfilter", args=[compute.id, "web"])
+ self.list_url = reverse("nwfilters", args=[compute.id])
def defined(self, data):
response = self.client.post(self.url, data)
@@ -41,12 +48,12 @@ class FilterRedefineTests(TestCase):
return etree.fromstring(self.wvm.nwfilterDefineXML.call_args.args[0]), response
def test_a_rule_is_removed_in_place(self):
- tree, _ = self.defined({"del_nwfilter_rule": "", "index": "1", "action": "drop", "direction": "in", "priority": "900"})
+ tree, _ = self.defined({"del_nwfilter_rule": "", "index": "1", "rule": text(XML, "rule", 1)})
self.assertEqual(tree.findtext("uuid"), UUID)
self.assertEqual([r.get("action") for r in tree.findall("rule")], ["accept"])
def test_a_reference_is_removed_in_place(self):
- tree, _ = self.defined({"del_nwfilter_ref": "", "ref": "clean-traffic"})
+ tree, _ = self.defined({"del_nwfilter_ref": "", "index": "0", "ref": text(XML, "filterref", 0)})
self.assertEqual((tree.findtext("uuid"), tree.findall("filterref")), (UUID, []))
def test_a_rule_is_added_in_place(self):
@@ -70,43 +77,59 @@ class FilterRedefineTests(TestCase):
tree, _ = self.defined({"edit_nwfilter": "", "edit_xml": edited})
self.assertEqual(tree.findtext("uuid"), UUID)
+ def test_one_change_a_post(self):
+ # two at once would both start from the page's reading, the second undoing the first
+ rule = "{UUID} "
+ "{UUID} "
- " element")
+ if not NAME.fullmatch(tree.get("name") or ""):
+ raise util.OperationError("The filter's name may have letters, digits, dots, dashes and underscores")
+ for nwfilter in self.wvm.listAllNWFilters(0):
+ if nwfilter.name() == tree.get("name"):
+ raise util.OperationError("A network filter with this name already exists")
+ if nwfilter.UUIDString() == tree.findtext("uuid"):
+ raise util.OperationError("A network filter with this UUID already exists")
self.wvm.nwfilterDefineXML(xml)
- def clone_nwfilter(self, name, cln_name):
- nwfilter = self.get_nwfilter(name)
- if nwfilter:
- tree = ElementTree.fromstring(nwfilter.XMLDesc(0))
- tree.set("name", cln_name)
- uuid = tree.find("uuid")
- tree.remove(uuid)
- self.create_nwfilter(ElementTree.tostring(tree).decode())
+ def clone_nwfilter(self, name, clone_name):
+ """Define a copy of filter name as clone_name, with its rules,
+ references and their parameters, and a UUID of its own."""
+ if not NAME.fullmatch(clone_name):
+ raise util.OperationError("The filter's name may have letters, digits, dots, dashes and underscores")
+ if clone_name in self.get_nwfilters():
+ raise util.OperationError(f"A network filter named {clone_name} already exists")
+ tree = etree.fromstring(self.get_nwfilter(name).XMLDesc(0))
+ tree.set("name", clone_name)
+ tree.remove(tree.find("uuid"))
+ self.wvm.nwfilterDefineXML(etree.tostring(tree, encoding="unicode"))
+
+ def delete_nwfilter(self, name):
+ """Undefine a filter that no VM's interface and no other filter
+ references. libvirt refuses only one a running VM uses; a stopped VM,
+ or a VM using a filter that references it, would not start."""
+ vms = self.get_users().get(name, [])
+ if vms:
+ raise util.OperationError(f"Filter {name} is used by VM {', '.join(vms)}")
+ referencing = [nwfilter["name"] for nwfilter in self.get_nwfilters_info() if name in nwfilter["refs"]]
+ if referencing:
+ raise util.OperationError(f"Filter {name} is referenced by filter {', '.join(referencing)}")
+ self.get_nwfilter(name).undefine()
class wvmNWFilter(wvmConnect):
+ """A filter of the host. A page reads its XML once (tree); a change is
+ the filter defined again with its UUID, which libvirt does in place, also
+ while a VM uses it, and refuses for XML it cannot parse or apply to the
+ running VMs' interfaces."""
+
def __init__(self, host, login, passwd, conn, nwfiltername):
wvmConnect.__init__(self, host, login, passwd, conn)
self.nwfilter = self.get_nwfilter(nwfiltername)
- def _XMLDesc(self, flags):
- return self.nwfilter.XMLDesc(flags)
+ @cached_property
+ def tree(self):
+ return etree.fromstring(self.nwfilter.XMLDesc(0))
def get_uuid(self):
return self.nwfilter.UUIDString()
@@ -38,75 +122,82 @@ class wvmNWFilter(wvmConnect):
return self.nwfilter.name()
def get_xml(self):
- """With the UUID: defined again with it, libvirt changes the filter in
- place, also while a VM uses it. It refuses XML it cannot parse or
- apply to the running VMs' interfaces."""
- return self._XMLDesc(0)
+ return etree.tostring(self.tree, encoding="unicode")
- def get_filter_refs(self):
- tree = ElementTree.fromstring(self._XMLDesc(0))
- return [ref.get("filter") for ref in tree.findall("./filterref")]
+ def get_refs(self):
+ """The filters it references, each with its parameters in order (one
+ name may repeat, giving a list) and its text."""
+ return [{"filter": ref.get("filter"), "text": element_text(ref),
+ "parameters": [(param.get("name"), param.get("value")) for param in ref.findall("parameter")]}
+ for ref in self.tree.findall("filterref")]
def get_rules(self):
+ """Its rules: action, direction, priority, statematch, each protocol
+ element with its attributes (one rule may have several of one
+ protocol, a match and a negated one) and the rule's text."""
rules = []
-
- tree = ElementTree.fromstring(self._XMLDesc(0))
- for index, r in enumerate(tree.findall("./rule")):
- rule_action = r.get("action")
- rule_direction = r.get("direction")
- rule_priority = r.get("priority")
- rule_statematch = r.get("statematch")
-
- rule_directives = r.find("./")
- if rule_directives is not None:
- rule_directives = ElementTree.tostring(rule_directives).decode()
-
- rule_info = {
- "index": index,
- "action": rule_action,
- "direction": rule_direction,
- "priority": rule_priority,
- "statematch": rule_statematch,
- "directives": rule_directives,
- }
-
- rules.append(rule_info)
-
+ for rule in self.tree.findall("rule"):
+ protocols = [{"name": protocol.tag, "attributes": list(protocol.attrib.items())} for protocol in rule]
+ rules.append({"action": rule.get("action"), "direction": rule.get("direction"),
+ "priority": rule.get("priority"), "statematch": rule.get("statematch"),
+ "protocols": protocols, "text": element_text(rule)})
return rules
- def delete_ref(self, name):
- tree = ElementTree.fromstring(self._XMLDesc(0))
- for ref in tree.findall("./filterref"):
- if name == ref.get("filter"):
- tree.remove(ref)
- break
- return ElementTree.tostring(tree).decode()
+ def get_ref_choices(self):
+ """The filters a reference can be added to: not this one, nor one it
+ references already."""
+ taken = {self.get_name(), *(ref.get("filter") for ref in self.tree.findall("filterref"))}
+ return [name for name in sorted(self.get_nwfilters()) if name not in taken]
- def delete_rule(self, index, action, direction, priority):
- """The filter without its rule at index (several rules may share
- action, direction and priority); those must still match, so a filter
- changed since the page was read loses no other rule."""
- tree = ElementTree.fromstring(self._XMLDesc(0))
- rules = tree.findall("./rule")
- if index >= len(rules) or (rules[index].get("action"), rules[index].get("direction"),
- rules[index].get("priority")) != (action, direction, priority):
- raise ValueError("The filter has changed; reload the page and try again")
- tree.remove(rules[index])
- return ElementTree.tostring(tree).decode()
+ def _define(self, tree):
+ self.wvm.nwfilterDefineXML(etree.tostring(tree, encoding="unicode"))
+
+ def _without(self, tag, index, text):
+ """The filter without its tag element at index, which must still read
+ as text: a filter changed since the page was read loses nothing else.
+ Returns the element removed."""
+ tree = copy.deepcopy(self.tree)
+ elements = tree.findall(tag)
+ try:
+ index = int(index)
+ except ValueError:
+ index = -1
+ if not 0 <= index < len(elements) or element_text(elements[index]) != same_text(text):
+ raise util.OperationError("The filter has changed; reload the page and try again")
+ tree.remove(elements[index])
+ self._define(tree)
+ return elements[index]
+
+ def edit(self, xml):
+ """Define the filter anew from the XML entered: the same filter, of
+ its name and UUID."""
+ tree = util.parse_xml(xml)
+ if tree.tag != "filter" or tree.get("name") != self.get_name():
+ raise util.OperationError(f"The XML must define filter {self.get_name()}")
+ if tree.findtext("uuid") != self.get_uuid():
+ raise util.OperationError("The XML must keep the filter's UUID")
+ self.wvm.nwfilterDefineXML(xml)
+
+ def delete_rule(self, index, text):
+ self._without("rule", index, text)
+
+ def delete_ref(self, index, text):
+ """Returns the name of the filter it referenced."""
+ return self._without("filterref", index, text).get("filter")
def add_ref(self, name):
- tree = ElementTree.fromstring(self._XMLDesc(0))
- element = ElementTree.Element("filterref")
- element.attrib["filter"] = name
- tree.append(element)
- return ElementTree.tostring(tree).decode()
+ if name not in self.get_ref_choices():
+ raise util.OperationError(f"Filter {name} cannot be referenced: it is this one, referenced already or not on the host")
+ tree = copy.deepcopy(self.tree)
+ etree.SubElement(tree, "filterref", filter=name)
+ self._define(tree)
def add_rule(self, xml):
"""The filter with the rule added as entered; libvirt keeps one
protocol per rule, so it is never merged into another rule."""
- tree = ElementTree.fromstring(self._XMLDesc(0))
- rule = ElementTree.fromstring(xml)
+ rule = util.parse_xml(xml)
if rule.tag != "rule":
- raise ValueError("Enter one element")
+ raise util.OperationError("Enter one element")
+ tree = copy.deepcopy(self.tree)
tree.append(rule)
- return ElementTree.tostring(tree).decode()
+ self._define(tree)
diff --git a/vrtManager/tests/test_xpath_values.py b/vrtManager/tests/test_xpath_values.py
index 78947f70..884c7b33 100644
--- a/vrtManager/tests/test_xpath_values.py
+++ b/vrtManager/tests/test_xpath_values.py
@@ -10,16 +10,11 @@ if not settings.configured:
from vrtManager.connection import wvmConnect
from vrtManager.network import wvmNetwork
-from vrtManager.nwfilters import wvmNWFilter
CAPS = """64 /usr/bin/qemu-x86
q35 /usr/bin/qemu-arm virt