mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
fix(nwfilters): a filter in use is not deleted, and a change replaces no other filter
Plan reviewed with Codex; libvirt's behaviour checked on a host. - A filter is not deleted while a VM's interface references it, as the VM runs or as saved for its next boot (a stopped VM too), or while another filter references it: libvirt refuses only one a running VM uses, and a VM using a filter that references a deleted one does not start. The VMs are read from one listing, one or two XML reads each, instead of a connection per VM; the refusal names them. - Create refuses a name or UUID the host has, clone a name in use or one outside the panel's names, and editing the XML must keep the filter's name and UUID: each would replace another filter or make a copy. Entered XML is read for these checks without expanding entities or taking a DTD (util.parse_xml, shared with the network editor), and a rule is one <rule> element. - A rule or reference is deleted only while it reads as the page showed it, not by index and three attributes: a rule alike in those or a filter changed meanwhile loses nothing. - A reference is added only to another filter of the host not referenced yet; the picker offers those. - References keep their parameters, in order (a name may repeat); a rule shows every protocol element it has, a rule without one too. - One change a POST (two would both start from the page's reading, the second undoing the first); every change redirects and reports its refusal as a message, a POST for a filter gone meanwhile going to the list; the catch-all that showed any error goes, and so does wvmInstance.get_filterrefs, which only the old delete check used. The list reads each filter's XML once from one listing; a filter's page reads its XML once and only the other filters' names; the per-filter XML dialog of the list goes (a filter's page shows its XML). The XML editors send their text untouched too.
This commit is contained in:
catborise
parent
65c5a251ba
commit
bc2b70544a
13 changed files with 502 additions and 352 deletions
|
|
@ -142,11 +142,11 @@ class NetworkChangeTests(SimpleTestCase):
|
|||
net.net.update.assert_not_called()
|
||||
|
||||
def test_the_xml_editor_expands_no_entity(self):
|
||||
# a file of the panel's host named by an external entity is not read into the check
|
||||
# a file of the panel's host named by an external entity is not read: no DTD is taken
|
||||
net = self.network()
|
||||
xml = ('<!DOCTYPE network [<!ENTITY name SYSTEM "file:///etc/hostname">]>'
|
||||
"<network><name>&name;</name></network>")
|
||||
with self.assertRaisesRegex(OperationError, "must define network lab"):
|
||||
with self.assertRaisesRegex(OperationError, "may not declare a DTD"):
|
||||
net.edit_network(xml)
|
||||
|
||||
def test_the_xml_editor_takes_an_encoding_declaration(self):
|
||||
|
|
|
|||
|
|
@ -50,8 +50,8 @@
|
|||
<label class="input-group-text" for="nwfilter_select">{% trans "Filter" %}</label>
|
||||
<select id="nwfilter_select" name="nwfilters_select" class="form-select">
|
||||
<option value="" selected>{% trans 'None' %}</option>
|
||||
{% for nwf in nwfilters_all %}
|
||||
<option value="{{ nwf.name }}">{{ nwf.name }}</option>
|
||||
{% for nwf in ref_choices %}
|
||||
<option value="{{ nwf }}">{{ nwf }}</option>
|
||||
{% endfor %}
|
||||
</select>
|
||||
<button type="submit" class="btn btn-success float-end" name="add_nwfilter_ref" title="{% trans "Add" %}">
|
||||
|
|
@ -78,11 +78,12 @@
|
|||
{% for ref in refs %}
|
||||
<tr>
|
||||
<th scope="row">{{ forloop.counter }}</th>
|
||||
<td>{{ ref }}</td>
|
||||
<td>{{ ref.filter }}{% for param, value in ref.parameters %} <span class="font-monospace small">{{ param }}={{ value }}</span>{% endfor %}</td>
|
||||
<td style="width:30px;">
|
||||
<form action="" method="post" role="form" aria-label="Delete NWFilter form">{% csrf_token %}
|
||||
<input type="hidden" name="ref" value="{{ ref }}">
|
||||
<button type="submit" class="btn btn-sm btn-danger" name="del_nwfilter_ref" title="{% trans "Delete" %}" data-confirm="{% blocktrans with ref=ref %}Remove the reference to {{ ref }}? Its rules no longer apply through this filter.{% endblocktrans %}" onclick="return confirm(this.dataset.confirm)">
|
||||
<input type="hidden" name="index" value="{{ forloop.counter0 }}">
|
||||
<input type="hidden" name="ref" value="{{ ref.text }}">
|
||||
<button type="submit" class="btn btn-sm btn-danger" name="del_nwfilter_ref" title="{% trans "Delete" %}" data-confirm="{% blocktrans with ref=ref.filter %}Remove the reference to {{ ref }}? Its rules no longer apply through this filter.{% endblocktrans %}" onclick="return confirm(this.dataset.confirm)">
|
||||
{% bs_icon 'trash' %}
|
||||
</button>
|
||||
</form>
|
||||
|
|
@ -120,13 +121,11 @@
|
|||
<td>{{ rule.direction }}</td>
|
||||
<td>{{ rule.priority }}</td>
|
||||
<td>{{ rule.statematch }}</td>
|
||||
<td>{{ rule.directives }}</td>
|
||||
<td class="font-monospace small">{% for protocol in rule.protocols %}<div>{{ protocol.name }}{% for attribute, value in protocol.attributes %} {{ attribute }}={{ value }}{% endfor %}</div>{% endfor %}</td>
|
||||
<td>
|
||||
<form action="" method="post" role="form" aria-label="Delete NWFilter rule">{% csrf_token %}
|
||||
<input type="hidden" name="index" value="{{ rule.index }}">
|
||||
<input type="hidden" name="action" value="{{ rule.action }}">
|
||||
<input type="hidden" name="direction" value="{{ rule.direction }}">
|
||||
<input type="hidden" name="priority" value="{{ rule.priority }}">
|
||||
<input type="hidden" name="index" value="{{ forloop.counter0 }}">
|
||||
<input type="hidden" name="rule" value="{{ rule.text }}">
|
||||
<button type="submit" class="btn btn-sm btn-danger" name="del_nwfilter_rule" title="{% trans 'Delete' %}" data-confirm="{% trans "Delete this rule? Interfaces that use the filter no longer get it." %}" onclick="return confirm(this.dataset.confirm)">
|
||||
{% bs_icon 'trash' %}
|
||||
</button>
|
||||
|
|
@ -146,6 +145,7 @@
|
|||
editor.getSession().setMode("ace/mode/xml");
|
||||
|
||||
var edit_input = $('input[name="edit_xml"]');
|
||||
edit_input.val(editor.getSession().getValue());
|
||||
editor.getSession().on("change",function () {
|
||||
edit_input.val(editor.getSession().getValue());
|
||||
});
|
||||
|
|
@ -154,6 +154,7 @@
|
|||
rule_editor.getSession().setMode("ace/mode/xml");
|
||||
|
||||
var rule_input = $('input[name="nwfilterrule_xml"]');
|
||||
rule_input.val(rule_editor.getSession().getValue());
|
||||
rule_editor.getSession().on("change",function () {
|
||||
rule_input.val(rule_editor.getSession().getValue());
|
||||
});
|
||||
|
|
|
|||
|
|
@ -29,7 +29,7 @@
|
|||
<th scope="col" data-sort-method="number">#</th>
|
||||
<th scope="col">{% trans "UUID" %}</th>
|
||||
<th scope="col">{% trans "Name" %}</th>
|
||||
<th scope="colgroup" data-sort-method="none" colspan="3">{% trans "Action" %}</th>
|
||||
<th scope="colgroup" data-sort-method="none" colspan="2">{% trans "Action" %}</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody class="searchable">
|
||||
|
|
@ -38,29 +38,6 @@
|
|||
<th scope="row">{{ forloop.counter }}</th>
|
||||
<td><a class="link-primary" href="{% url 'nwfilter' compute.id nwfilter.name %} ">{{ nwfilter.uuid }}</a></td>
|
||||
<td>{{ nwfilter.name }}</td>
|
||||
<td>
|
||||
<div class="modal fade" id="Show{{ forloop.counter }}" tabindex="-1" role="dialog" aria-labelledby="showNWFilter" aria-hidden="true">
|
||||
<div class="modal-dialog">
|
||||
<div class="modal-content">
|
||||
<div class="modal-header">
|
||||
<h5 class="modal-title">{% trans "Details of NWFilter" %}: <span class="text-danger">{{ nwfilter.name }}</span></h5>
|
||||
<button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
|
||||
</div>
|
||||
<div class="modal-body">
|
||||
<textarea rows="10" class="w-100" aria-label="{% trans "Details of NWFilter" %}">
|
||||
{{ nwfilter.xml }}
|
||||
</textarea>
|
||||
</div>
|
||||
<div class="modal-footer">
|
||||
<button type="button" class="btn btn-secondary" data-bs-dismiss="modal">{% trans "Close" %}</button>
|
||||
</div>
|
||||
</div> <!-- /.modal-content -->
|
||||
</div> <!-- /.modal-dialog -->
|
||||
</div> <!-- /.modal -->
|
||||
<button class="btn btn-sm btn-secondary" data-bs-toggle="modal" data-bs-target="#Show{{ forloop.counter }}" title="{% trans "Show" %}">
|
||||
{% bs_icon 'eye' %}
|
||||
</button>
|
||||
</td>
|
||||
<td>
|
||||
<div class="modal fade" id="Clone{{ forloop.counter }}" tabindex="-1" role="dialog" aria-labelledby="addNwFilterLabel" aria-hidden="true">
|
||||
<div class="modal-dialog">
|
||||
|
|
@ -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());
|
||||
})
|
||||
|
|
|
|||
|
|
@ -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"<filter name='web' chain='root'><uuid>{UUID}</uuid><filterref filter='clean-traffic'/>"
|
||||
|
|
@ -31,16 +33,16 @@ class FilterRowTests(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"]
|
||||
self.wvm.listAllNWFilters.return_value = [nwfilter]
|
||||
self.wvm.listAllDomains.return_value = []
|
||||
self.compute = Compute.objects.create(name="c1", hostname="localhost", login="", password="", type=4)
|
||||
self.client.force_login(get_user_model().objects.create_superuser("audit-admin", "a@example.com", "pw"))
|
||||
Logs.objects.all().delete() # the login's row
|
||||
self.list_url = reverse("nwfilters", args=[self.compute.id])
|
||||
self.url = reverse("nwfilter", args=[self.compute.id, "web"])
|
||||
|
||||
@patch("nwfilters.views.wvmInstances")
|
||||
def test_filter_list_actions(self, instances):
|
||||
instances.return_value.get_instances.return_value = []
|
||||
def test_filter_list_actions(self):
|
||||
# the list is read again, so the new filter shows
|
||||
self.assertRedirects(self.client.post(self.list_url, {"create_nwfilter": "", "nwfilter_xml": NEW}),
|
||||
self.list_url, fetch_redirect_response=False)
|
||||
|
|
@ -61,34 +63,35 @@ class FilterRowTests(TestCase):
|
|||
self.client.post(self.list_url, {"create_nwfilter": "", "nwfilter_xml": other})
|
||||
self.wvm.nwfilterDefineXML.assert_not_called()
|
||||
self.assertEqual(rows(), [
|
||||
("Network filter web: add: name already in use", False),
|
||||
("Network filter other: add: UUID already in use", False),
|
||||
("Network filter web: add: A network filter with this name already exists", False),
|
||||
("Network filter other: add: A network filter with this UUID already exists", False),
|
||||
])
|
||||
|
||||
def test_an_unparsable_filter(self):
|
||||
self.wvm.nwfilterDefineXML.side_effect = libvirtError("XML error")
|
||||
response = self.client.post(self.list_url, {"create_nwfilter": "", "nwfilter_xml": "<filter"})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertRedirects(response, self.list_url, fetch_redirect_response=False)
|
||||
[(message, success)] = rows()
|
||||
self.assertFalse(success)
|
||||
self.assertTrue(message.startswith("Network filter (no name): add: "), message)
|
||||
self.assertTrue(message.startswith("Network filter (no name): add: The XML is not valid"), message)
|
||||
self.wvm.nwfilterDefineXML.assert_not_called()
|
||||
|
||||
@patch("nwfilters.views.wvmInstance")
|
||||
@patch("nwfilters.views.wvmInstances")
|
||||
def test_a_filter_in_use_is_not_deleted(self, instances, instance):
|
||||
instances.return_value.get_instances.return_value = ["web-1"]
|
||||
instance.return_value.get_filterrefs.return_value = ["web"]
|
||||
def test_a_filter_in_use_is_not_deleted(self):
|
||||
vm = MagicMock(**{"name.return_value": "web-1", "isActive.return_value": False, "XMLDesc.return_value":
|
||||
"<domain><devices><interface type='network'><filterref filter='web'/></interface></devices></domain>"})
|
||||
self.wvm.listAllDomains.return_value = [vm]
|
||||
self.client.post(self.list_url, {"del_nwfilter": "", "nwfiltername": "web"})
|
||||
self.wvm.nwfilterLookupByName.return_value.undefine.assert_not_called()
|
||||
self.assertEqual(rows(), [("Network filter web: delete: in use by web-1", False)])
|
||||
self.assertEqual(rows(), [("Network filter web: delete: Filter web is used by VM web-1", False)])
|
||||
|
||||
def test_filter_page_actions(self):
|
||||
self.client.post(self.url, {"edit_nwfilter": "", "edit_xml": XML})
|
||||
self.client.post(self.url, {"add_nwfilter_rule": "",
|
||||
"nwfilterrule_xml": "<rule action='accept' direction='in' priority='500'><all/></rule>"})
|
||||
self.client.post(self.url, {"del_nwfilter_rule": "", "index": "0", "action": "drop", "direction": "in", "priority": "900"})
|
||||
tree = etree.fromstring(XML)
|
||||
self.client.post(self.url, {"del_nwfilter_rule": "", "index": "0", "rule": element_text(tree.find("rule"))})
|
||||
self.client.post(self.url, {"add_nwfilter_ref": "", "nwfilters_select": "no-spoofing"})
|
||||
self.client.post(self.url, {"del_nwfilter_ref": "", "ref": "clean-traffic"})
|
||||
self.client.post(self.url, {"del_nwfilter_ref": "", "index": "0", "ref": element_text(tree.find("filterref"))})
|
||||
self.assertEqual(rows(), [
|
||||
("Network filter web: change XML", True),
|
||||
("Network filter web: add rule", True),
|
||||
|
|
|
|||
182
nwfilters/tests/test_filters.py
Normal file
182
nwfilters/tests/test_filters.py
Normal file
|
|
@ -0,0 +1,182 @@
|
|||
"""What the panel reads of the host's network filters and what it refuses
|
||||
to change: which VMs use a filter, as they run and as saved; a filter in use
|
||||
or referenced is not deleted; a create, clone or edit replaces no other
|
||||
filter; references and rules as libvirt keeps them. libvirt's test driver has
|
||||
no filters, so the connection is a mock."""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import libvirt
|
||||
from django.test import SimpleTestCase
|
||||
from lxml import etree
|
||||
from vrtManager.nwfilters import wvmNWFilter, wvmNWFilters
|
||||
from vrtManager.util import OperationError
|
||||
|
||||
UUID = "6d1f7a10-0000-4000-8000-00000000c0de"
|
||||
WEB = (f"<filter name='web' chain='root'><uuid>{UUID}</uuid>"
|
||||
"<filterref filter='clean-traffic'><parameter name='IP' value='192.0.2.10'/>"
|
||||
"<parameter name='IP' value='192.0.2.11'/></filterref>"
|
||||
"<rule action='accept' direction='inout' priority='500'><arp match='no' arpsrcipaddr='192.0.2.1'/>"
|
||||
"<arp opcode='request'/></rule>"
|
||||
"<rule action='drop' direction='in'/></filter>")
|
||||
|
||||
|
||||
def nwfilter(name, xml, uuid=UUID):
|
||||
return MagicMock(**{"name.return_value": name, "UUIDString.return_value": uuid, "XMLDesc.return_value": xml})
|
||||
|
||||
|
||||
def vm(name, xml, saved=None, active=False, persistent=True):
|
||||
def describe(flags=0):
|
||||
return saved if flags else xml
|
||||
return MagicMock(**{"name.return_value": name, "XMLDesc.side_effect": describe,
|
||||
"isActive.return_value": active, "isPersistent.return_value": persistent})
|
||||
|
||||
|
||||
def using(*names):
|
||||
refs = "".join(f"<interface type='network'><filterref filter='{name}'/></interface>" for name in names)
|
||||
return f"<domain><devices>{refs}</devices></domain>"
|
||||
|
||||
|
||||
def error(code):
|
||||
err = libvirt.libvirtError("refused")
|
||||
err.get_error_code = lambda: code
|
||||
return err
|
||||
|
||||
|
||||
def filters(*listed, domains=()):
|
||||
conn = wvmNWFilters.__new__(wvmNWFilters)
|
||||
conn.wvm = MagicMock(**{"listAllNWFilters.return_value": list(listed), "listAllDomains.return_value": list(domains),
|
||||
"listNWFilters.return_value": [f.name() for f in listed]})
|
||||
return conn
|
||||
|
||||
|
||||
def one(xml=WEB, names=("web", "clean-traffic", "no-spoofing")):
|
||||
conn = wvmNWFilter.__new__(wvmNWFilter)
|
||||
conn.nwfilter = nwfilter("web", xml)
|
||||
conn.wvm = MagicMock(**{"listNWFilters.return_value": list(names)})
|
||||
return conn
|
||||
|
||||
|
||||
def defined(conn):
|
||||
return etree.fromstring(conn.wvm.nwfilterDefineXML.call_args.args[0])
|
||||
|
||||
|
||||
class UsersTests(SimpleTestCase):
|
||||
def test_a_filter_is_used_as_the_vm_runs_and_as_saved(self):
|
||||
conn = filters(domains=[
|
||||
vm("stopped", using("a")),
|
||||
vm("running", using("b", "b"), saved=using("c"), active=True), # c only from its next boot
|
||||
vm("transient", using("d"), saved=using("x"), active=True, persistent=False),
|
||||
])
|
||||
self.assertEqual(conn.get_users(), {"a": ["stopped"], "b": ["running"], "c": ["running"], "d": ["transient"]})
|
||||
stopped, running, transient = conn.wvm.listAllDomains.return_value
|
||||
self.assertEqual((stopped.XMLDesc.call_count, running.XMLDesc.call_count, transient.XMLDesc.call_count), (1, 2, 1))
|
||||
|
||||
def test_a_vm_gone_meanwhile_uses_nothing_other_errors_are_raised(self):
|
||||
gone = vm("gone", "")
|
||||
gone.XMLDesc.side_effect = error(libvirt.VIR_ERR_NO_DOMAIN)
|
||||
self.assertEqual(filters(domains=[gone, vm("web-1", using("a"))]).get_users(), {"a": ["web-1"]})
|
||||
gone.XMLDesc.side_effect = error(libvirt.VIR_ERR_OPERATION_FAILED)
|
||||
with self.assertRaises(libvirt.libvirtError):
|
||||
filters(domains=[gone]).get_users()
|
||||
|
||||
|
||||
class DeleteTests(SimpleTestCase):
|
||||
def test_a_filter_a_vm_or_another_filter_uses_is_kept(self):
|
||||
target = nwfilter("clean-traffic", "<filter name='clean-traffic'/>", uuid="other")
|
||||
for conn, why in ((filters(target, domains=[vm("web-1", using("clean-traffic"))]), "used by VM web-1"),
|
||||
(filters(target, nwfilter("web", WEB)), "referenced by filter web")):
|
||||
with self.subTest(why=why), self.assertRaisesRegex(OperationError, why):
|
||||
conn.delete_nwfilter("clean-traffic")
|
||||
conn.wvm.nwfilterLookupByName.return_value.undefine.assert_not_called()
|
||||
|
||||
def test_an_unused_filter_is_undefined_libvirt_may_still_refuse(self):
|
||||
conn = filters(nwfilter("web", WEB))
|
||||
conn.delete_nwfilter("web")
|
||||
conn.wvm.nwfilterLookupByName.return_value.undefine.assert_called_once_with()
|
||||
conn.wvm.nwfilterLookupByName.return_value.undefine.side_effect = error(libvirt.VIR_ERR_OPERATION_INVALID)
|
||||
with self.assertRaises(libvirt.libvirtError): # a VM started meanwhile: libvirt refuses itself
|
||||
conn.delete_nwfilter("web")
|
||||
|
||||
|
||||
class IdentityTests(SimpleTestCase):
|
||||
def test_a_create_replaces_no_filter(self):
|
||||
conn = filters(nwfilter("web", WEB))
|
||||
for xml, why in ((WEB, "name already exists"),
|
||||
(WEB.replace("name='web'", "name='other'"), "UUID already exists"),
|
||||
("<rule/>", "one <filter>"),
|
||||
("<filter name='a b'/>", "letters, digits"),
|
||||
('<!DOCTYPE filter [<!ENTITY n "web">]><filter name="&n;"/>', "DTD"),
|
||||
("", "not valid")):
|
||||
with self.subTest(why=why), self.assertRaisesRegex(OperationError, why):
|
||||
conn.create_nwfilter(xml)
|
||||
conn.wvm.nwfilterDefineXML.assert_not_called()
|
||||
conn.create_nwfilter("<filter name='other' chain='root'/>")
|
||||
conn.wvm.nwfilterDefineXML.assert_called_once_with("<filter name='other' chain='root'/>")
|
||||
|
||||
def test_a_clone_keeps_rules_and_parameters_with_a_uuid_of_its_own(self):
|
||||
conn = filters(nwfilter("web", WEB))
|
||||
conn.wvm.nwfilterLookupByName.return_value.XMLDesc.return_value = WEB
|
||||
for name, why in (("web", "already exists"), ("web copy", "letters, digits"), ("", "letters, digits")):
|
||||
with self.subTest(name=name), self.assertRaisesRegex(OperationError, why):
|
||||
conn.clone_nwfilter("web", name)
|
||||
conn.clone_nwfilter("web", "web-copy")
|
||||
clone = defined(conn)
|
||||
self.assertEqual((clone.get("name"), clone.find("uuid")), ("web-copy", None))
|
||||
self.assertEqual(clone.xpath("filterref/parameter/@value"), ["192.0.2.10", "192.0.2.11"])
|
||||
self.assertEqual(len(clone.findall("rule")), 2)
|
||||
|
||||
def test_an_edit_is_of_the_same_filter(self):
|
||||
conn = one()
|
||||
for xml, why in ((WEB.replace("name='web'", "name='other'"), "must define filter web"),
|
||||
(WEB.replace(f"<uuid>{UUID}</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 <rule>"):
|
||||
one().add_rule("<filter name='web'/>")
|
||||
|
||||
def test_a_change_keeps_the_rest_of_the_filter(self):
|
||||
conn = one()
|
||||
conn.add_rule("<rule action='accept' direction='in'><udp dstportstart='53'/></rule>")
|
||||
tree = defined(conn)
|
||||
self.assertEqual((tree.findtext("uuid"), len(tree.findall("rule"))), (UUID, 3))
|
||||
self.assertEqual(tree.xpath("filterref/parameter/@value"), ["192.0.2.10", "192.0.2.11"])
|
||||
self.assertEqual(len(etree.fromstring(conn.get_xml()).findall("rule")), 2, "the page's own copy is not changed")
|
||||
|
||||
|
||||
class ReadTests(SimpleTestCase):
|
||||
def test_rules_and_references_as_libvirt_keeps_them(self):
|
||||
conn = one()
|
||||
arp, drop = conn.get_rules()
|
||||
# two elements of one protocol, a match and a negated one; a rule of no protocol
|
||||
self.assertEqual([p["attributes"] for p in arp["protocols"]], [[("match", "no"), ("arpsrcipaddr", "192.0.2.1")],
|
||||
[("opcode", "request")]])
|
||||
self.assertEqual((drop["protocols"], drop["priority"]), ([], None))
|
||||
[ref] = conn.get_refs()
|
||||
self.assertEqual(ref["parameters"], [("IP", "192.0.2.10"), ("IP", "192.0.2.11")])
|
||||
self.assertEqual(conn.nwfilter.XMLDesc.call_count, 1)
|
||||
|
||||
def test_the_list_reads_each_filter_once(self):
|
||||
conn = filters(nwfilter("web", WEB), nwfilter("clean-traffic", "<filter name='clean-traffic' chain='root'/>", "x"))
|
||||
info = conn.get_nwfilters_info()
|
||||
self.assertEqual([(f["name"], f["rules"], f["refs"]) for f in info],
|
||||
[("clean-traffic", 0, []), ("web", 2, ["clean-traffic"])])
|
||||
for listed in conn.wvm.listAllNWFilters.return_value:
|
||||
self.assertEqual(listed.XMLDesc.call_count, 1)
|
||||
|
|
@ -13,6 +13,7 @@ from django.test import TestCase
|
|||
from django.urls import reverse
|
||||
from libvirt import libvirtError
|
||||
from lxml import etree, html
|
||||
from vrtManager.nwfilters import element_text
|
||||
|
||||
UUID = "6d1f7a10-0000-4000-8000-00000000c0de"
|
||||
XML = (f"<filter name='web' chain='root'><uuid>{UUID}</uuid><filterref filter='clean-traffic'/>"
|
||||
|
|
@ -20,6 +21,11 @@ XML = (f"<filter name='web' chain='root'><uuid>{UUID}</uuid><filterref filter='c
|
|||
"<rule action='drop' direction='in' priority='900'><all/></rule></filter>")
|
||||
|
||||
|
||||
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 = "<rule action='accept' direction='in' priority='500'><udp dstportstart='53'/></rule>"
|
||||
self.defined({"add_nwfilter_rule": "", "nwfilterrule_xml": rule, "add_nwfilter_ref": "", "nwfilters_select": "no-spoofing"})
|
||||
|
||||
def test_a_post_for_a_filter_gone_meanwhile_goes_to_the_list(self):
|
||||
self.wvm.nwfilterLookupByName.side_effect = libvirtError("Network filter not found")
|
||||
response = self.client.post(self.url, {"del_nwfilter_rule": "", "index": "0", "rule": ""})
|
||||
self.assertRedirects(response, self.list_url, fetch_redirect_response=False)
|
||||
self.wvm.nwfilterDefineXML.assert_not_called()
|
||||
|
||||
def test_a_refused_definition_is_reported_not_undone(self):
|
||||
self.wvm.nwfilterDefineXML.side_effect = libvirtError("XML error: Invalid value for attribute 'action'")
|
||||
_, response = self.defined({"del_nwfilter_rule": "", "index": "1", "action": "drop", "direction": "in", "priority": "900"})
|
||||
_, response = self.defined({"del_nwfilter_rule": "", "index": "1", "rule": text(XML, "rule", 1)})
|
||||
self.assertIn("Invalid value for attribute 'action'", " ".join(str(m) for m in get_messages(response.wsgi_request)))
|
||||
|
||||
|
||||
ALIKE = (f"<filter name='web' chain='root'><uuid>{UUID}</uuid>"
|
||||
"<rule action='accept' direction='in' priority='500'><tcp dstportstart='22'/></rule>"
|
||||
"<rule action='accept' direction='in' priority='500'><tcp dstportstart='80'/></rule></filter>")
|
||||
|
||||
|
||||
class RuleDeleteTests(TestCase):
|
||||
"""Rules sharing action, direction and priority: the one clicked goes."""
|
||||
"""Rules sharing action, direction and priority: the one clicked goes,
|
||||
and only while it reads as the page showed it."""
|
||||
|
||||
def setUp(self):
|
||||
patcher = patch("vrtManager.connection.connection_manager.get_connection")
|
||||
self.wvm = patcher.start().return_value
|
||||
self.addCleanup(patcher.stop)
|
||||
nwfilter = self.wvm.nwfilterLookupByName.return_value
|
||||
nwfilter.XMLDesc.return_value = (
|
||||
f"<filter name='web' chain='root'><uuid>{UUID}</uuid>"
|
||||
"<rule action='accept' direction='in' priority='500'><tcp dstportstart='22'/></rule>"
|
||||
"<rule action='accept' direction='in' priority='500'><tcp dstportstart='80'/></rule></filter>")
|
||||
nwfilter.XMLDesc.return_value = ALIKE
|
||||
nwfilter.name.return_value = "web"
|
||||
nwfilter.UUIDString.return_value = UUID
|
||||
self.wvm.listNWFilters.return_value = ["web"]
|
||||
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"])
|
||||
|
||||
def delete(self, index, priority="500"):
|
||||
return self.client.post(self.url, {"del_nwfilter_rule": "", "index": index, "action": "accept",
|
||||
"direction": "in", "priority": priority})
|
||||
def delete(self, index, rule):
|
||||
return self.client.post(self.url, {"del_nwfilter_rule": "", "index": index, "rule": rule})
|
||||
|
||||
def test_the_second_of_two_alike_rules_is_deleted(self):
|
||||
self.delete("1")
|
||||
# sent back by a browser, with other line breaks between its tags
|
||||
self.delete("1", text(ALIKE, "rule", 1).replace("><", ">\r\n <"))
|
||||
tree = etree.fromstring(self.wvm.nwfilterDefineXML.call_args.args[0])
|
||||
self.assertEqual(tree.xpath("rule/*/@dstportstart"), ["22"])
|
||||
|
||||
def test_a_rule_that_is_no_longer_there_is_not_deleted(self):
|
||||
for index, priority in (("2", "500"), ("1", "900")):
|
||||
# gone, another one at its place (reordered), or not an index
|
||||
for index, rule in (("2", text(ALIKE, "rule", 1)), ("0", text(ALIKE, "rule", 1)),
|
||||
("-1", text(ALIKE, "rule", 1)), ("x", text(ALIKE, "rule", 0)),
|
||||
("²", text(ALIKE, "rule", 0)), ("9" * 5000, text(ALIKE, "rule", 0))):
|
||||
with self.subTest(index=index):
|
||||
response = self.delete(index, priority)
|
||||
response = self.delete(index, rule)
|
||||
self.assertIn("reload the page", " ".join(str(m) for m in get_messages(response.wsgi_request)))
|
||||
self.wvm.nwfilterDefineXML.assert_not_called()
|
||||
|
|
|
|||
|
|
@ -1,7 +1,10 @@
|
|||
from unittest.mock import patch
|
||||
|
||||
from django.contrib.auth.models import User
|
||||
from django.test import TestCase
|
||||
from django.urls import reverse
|
||||
from computes.models import Compute
|
||||
from libvirt import libvirtError
|
||||
|
||||
|
||||
class NWFiltersTestCase(TestCase):
|
||||
|
|
@ -28,3 +31,10 @@ class NWFiltersTestCase(TestCase):
|
|||
reverse("nwfilter", args=[self.compute.id, "clean-traffic"])
|
||||
)
|
||||
self.assertEqual(response.status_code, 403)
|
||||
|
||||
def test_a_list_post_to_a_host_not_reached_is_redirected(self):
|
||||
self.client.force_login(User.objects.create_superuser("nwfilter-admin", "a@example.com", "pwd"))
|
||||
url = reverse("nwfilters", args=[self.compute.id])
|
||||
with patch("nwfilters.views.wvmNWFilters", side_effect=libvirtError("unable to connect")):
|
||||
response = self.client.post(url, {"del_nwfilter": "", "nwfiltername": "web"})
|
||||
self.assertRedirects(response, url, fetch_redirect_response=False)
|
||||
|
|
|
|||
|
|
@ -3,209 +3,93 @@ from computes.models import Compute
|
|||
from django.contrib import messages
|
||||
from django.http import HttpResponseRedirect
|
||||
from django.shortcuts import get_object_or_404, render
|
||||
from django.utils.translation import gettext_noop as _
|
||||
from django.urls import reverse
|
||||
from libvirt import libvirtError
|
||||
from logs.audit import audit, failed
|
||||
from logs.audit import audit
|
||||
from vrtManager import util
|
||||
from vrtManager.instance import wvmInstance, wvmInstances
|
||||
from vrtManager.nwfilters import wvmNWFilter, wvmNWFilters
|
||||
|
||||
|
||||
@superuser_only
|
||||
def nwfilters(request, compute_id):
|
||||
"""
|
||||
:param request:
|
||||
:param compute_id:
|
||||
:return:
|
||||
"""
|
||||
|
||||
nwfilters_all = []
|
||||
"""The host's network filters; one created, cloned or deleted, after
|
||||
which the list is read again."""
|
||||
compute = get_object_or_404(Compute, pk=compute_id)
|
||||
nwfilters_all = []
|
||||
|
||||
try:
|
||||
conn = wvmNWFilters(
|
||||
compute.hostname,
|
||||
compute.login,
|
||||
compute.password,
|
||||
compute.type
|
||||
)
|
||||
|
||||
for nwf in conn.get_nwfilters():
|
||||
nwfilters_all.append(conn.get_nwfilter_info(nwf))
|
||||
conn = wvmNWFilters(compute.hostname, compute.login, compute.password, compute.type)
|
||||
|
||||
if request.method == "POST":
|
||||
name = request.POST.get("nwfiltername", "")
|
||||
if "create_nwfilter" in request.POST:
|
||||
xml = request.POST.get("nwfilter_xml", "")
|
||||
if xml:
|
||||
name = uuid = None
|
||||
try:
|
||||
util.etree.fromstring(xml)
|
||||
name = util.get_xml_path(xml, "/filter/@name")
|
||||
uuid = util.get_xml_path(xml, "/filter/uuid")
|
||||
except util.etree.ParseError:
|
||||
pass
|
||||
subject = f"Network filter {name or '(no name)'}"
|
||||
try: # the row names the filter where the XML does
|
||||
new_name = util.parse_xml(xml).get("name")
|
||||
except util.OperationError:
|
||||
new_name = None
|
||||
with audit(request, compute, f"Network filter {new_name or '(no name)'}", "add"):
|
||||
conn.create_nwfilter(xml)
|
||||
elif "cln_nwfilter" in request.POST:
|
||||
clone_name = request.POST.get("cln_name", "")
|
||||
with audit(request, compute, f"Network filter {name}", f"clone as {clone_name}"):
|
||||
conn.clone_nwfilter(name, clone_name)
|
||||
elif "del_nwfilter" in request.POST:
|
||||
with audit(request, compute, f"Network filter {name}", "delete"):
|
||||
conn.delete_nwfilter(name)
|
||||
return HttpResponseRedirect(request.get_full_path())
|
||||
|
||||
for nwf in nwfilters_all:
|
||||
if name == nwf["name"]:
|
||||
failed(request, compute, subject, "add", "name already in use")
|
||||
messages.error(request, _("A network filter with this name already exists"))
|
||||
break
|
||||
if uuid == nwf["uuid"]:
|
||||
failed(request, compute, subject, "add", "UUID already in use")
|
||||
messages.error(request, _("A network filter with this UUID already exists"))
|
||||
break
|
||||
else:
|
||||
try:
|
||||
with audit(request, compute, subject, "add"):
|
||||
conn.create_nwfilter(xml)
|
||||
except libvirtError as lib_err:
|
||||
messages.error(request, lib_err)
|
||||
else:
|
||||
return HttpResponseRedirect(request.get_full_path())
|
||||
|
||||
if "del_nwfilter" in request.POST:
|
||||
name = request.POST.get("nwfiltername", "")
|
||||
in_use = False
|
||||
nwfilter = conn.get_nwfilter(name)
|
||||
nwfilter_info = conn.get_nwfilter_info(name)
|
||||
|
||||
is_conn = wvmInstances(
|
||||
compute.hostname,
|
||||
compute.login,
|
||||
compute.password,
|
||||
compute.type
|
||||
)
|
||||
instances = is_conn.get_instances()
|
||||
for inst in instances:
|
||||
i_conn = wvmInstance(
|
||||
compute.hostname,
|
||||
compute.login,
|
||||
compute.password,
|
||||
compute.type,
|
||||
inst,
|
||||
)
|
||||
dom_filterrefs = i_conn.get_filterrefs()
|
||||
|
||||
if name in dom_filterrefs:
|
||||
in_use = True
|
||||
msg = _(
|
||||
"NWFilter is in use by %(instance)s. Cannot be deleted."
|
||||
) % {"instance": inst}
|
||||
messages.error(request, msg)
|
||||
failed(request, compute, f"Network filter {name}", "delete", f"in use by {inst}")
|
||||
i_conn.close()
|
||||
break
|
||||
|
||||
is_conn.close()
|
||||
if nwfilter and not in_use:
|
||||
with audit(request, compute, f"Network filter {name}", "delete"):
|
||||
nwfilter.undefine()
|
||||
nwfilters_all.remove(nwfilter_info)
|
||||
|
||||
if "cln_nwfilter" in request.POST:
|
||||
name = request.POST.get("nwfiltername", "")
|
||||
cln_name = request.POST.get("cln_name", name + "-clone")
|
||||
|
||||
with audit(request, compute, f"Network filter {name}", f"clone as {cln_name}"):
|
||||
conn.clone_nwfilter(name, cln_name)
|
||||
nwfilters_all.append(conn.get_nwfilter_info(cln_name))
|
||||
|
||||
conn.close()
|
||||
except libvirtError as lib_err:
|
||||
nwfilters_all = conn.get_nwfilters_info()
|
||||
except libvirtError as lib_err: # OperationError too: a name in use, a filter in use
|
||||
messages.error(request, lib_err)
|
||||
except Exception as err:
|
||||
messages.error(request, err)
|
||||
if request.method == "POST": # a host not reached as well: nothing was changed
|
||||
return HttpResponseRedirect(request.get_full_path())
|
||||
|
||||
return render(
|
||||
request,
|
||||
"nwfilters.html",
|
||||
{
|
||||
"nwfilters": nwfilters_all,
|
||||
"compute": compute,
|
||||
},
|
||||
)
|
||||
return render(request, "nwfilters.html", {"nwfilters": nwfilters_all, "compute": compute})
|
||||
|
||||
|
||||
@superuser_only
|
||||
def nwfilter(request, compute_id, nwfltr):
|
||||
"""
|
||||
:param request:
|
||||
:param compute_id:
|
||||
:param nwfltr:
|
||||
:return:
|
||||
"""
|
||||
nwfilters_all = []
|
||||
"""A network filter of the host: its rules and references changed one at
|
||||
a time, or its XML as a whole."""
|
||||
compute = get_object_or_404(Compute, pk=compute_id)
|
||||
|
||||
try:
|
||||
nwfilter = wvmNWFilter(
|
||||
compute.hostname,
|
||||
compute.login,
|
||||
compute.password,
|
||||
compute.type,
|
||||
nwfltr
|
||||
)
|
||||
conn = wvmNWFilters(
|
||||
compute.hostname,
|
||||
compute.login,
|
||||
compute.password,
|
||||
compute.type
|
||||
)
|
||||
|
||||
for nwf in conn.get_nwfilters():
|
||||
nwfilters_all.append(conn.get_nwfilter_info(nwf))
|
||||
|
||||
uuid = nwfilter.get_uuid()
|
||||
name = nwfilter.get_name()
|
||||
xml = nwfilter.get_xml()
|
||||
rules = nwfilter.get_rules()
|
||||
refs = nwfilter.get_filter_refs()
|
||||
nwfilter = wvmNWFilter(compute.hostname, compute.login, compute.password, compute.type, nwfltr)
|
||||
|
||||
if request.method == "POST":
|
||||
|
||||
subject = f"Network filter {name}"
|
||||
if "edit_nwfilter" in request.POST:
|
||||
new_xml = request.POST.get("edit_xml", "")
|
||||
|
||||
if new_xml:
|
||||
post = request.POST
|
||||
subject = f"Network filter {nwfltr}"
|
||||
try:
|
||||
if "edit_nwfilter" in post:
|
||||
with audit(request, compute, subject, "change XML"):
|
||||
conn.create_nwfilter(new_xml)
|
||||
|
||||
if "del_nwfilter_rule" in request.POST:
|
||||
index = int(request.POST.get("index", ""))
|
||||
action = request.POST.get("action", "")
|
||||
direction = request.POST.get("direction", "")
|
||||
priority = request.POST.get("priority", "")
|
||||
|
||||
with audit(request, compute, subject, "delete rule"):
|
||||
conn.create_nwfilter(nwfilter.delete_rule(index, action, direction, priority))
|
||||
|
||||
if "del_nwfilter_ref" in request.POST:
|
||||
ref_name = request.POST.get("ref")
|
||||
with audit(request, compute, subject, f"delete reference {ref_name}"):
|
||||
conn.create_nwfilter(nwfilter.delete_ref(ref_name))
|
||||
|
||||
if "add_nwfilter_rule" in request.POST:
|
||||
rule_xml = request.POST.get("nwfilterrule_xml", "")
|
||||
if not rule_xml:
|
||||
return HttpResponseRedirect(request.get_full_path())
|
||||
with audit(request, compute, subject, "add rule"):
|
||||
conn.create_nwfilter(nwfilter.add_rule(rule_xml))
|
||||
|
||||
if "add_nwfilter_ref" in request.POST:
|
||||
ref_name = request.POST.get("nwfilters_select", "")
|
||||
if not ref_name:
|
||||
return HttpResponseRedirect(request.get_full_path())
|
||||
with audit(request, compute, subject, f"add reference {ref_name}"):
|
||||
conn.create_nwfilter(nwfilter.add_ref(ref_name))
|
||||
|
||||
nwfilter.edit(post.get("edit_xml", ""))
|
||||
elif "add_nwfilter_rule" in post:
|
||||
with audit(request, compute, subject, "add rule"):
|
||||
nwfilter.add_rule(post.get("nwfilterrule_xml", ""))
|
||||
elif "del_nwfilter_rule" in post:
|
||||
with audit(request, compute, subject, "delete rule"):
|
||||
nwfilter.delete_rule(post.get("index", ""), post.get("rule", ""))
|
||||
elif "add_nwfilter_ref" in post:
|
||||
ref_name = post.get("nwfilters_select", "")
|
||||
with audit(request, compute, subject, f"add reference {ref_name}"):
|
||||
nwfilter.add_ref(ref_name)
|
||||
elif "del_nwfilter_ref" in post:
|
||||
with audit(request, compute, subject, "delete reference") as entry:
|
||||
entry.action = f"delete reference {nwfilter.delete_ref(post.get('index', ''), post.get('ref', ''))}"
|
||||
except libvirtError as err: # OperationError too: a filter changed meanwhile, XML refused
|
||||
messages.error(request, err)
|
||||
return HttpResponseRedirect(request.get_full_path())
|
||||
conn.close()
|
||||
nwfilter.close()
|
||||
|
||||
name = nwfilter.get_name()
|
||||
uuid = nwfilter.get_uuid()
|
||||
xml = nwfilter.get_xml()
|
||||
rules = nwfilter.get_rules()
|
||||
refs = nwfilter.get_refs()
|
||||
ref_choices = nwfilter.get_ref_choices()
|
||||
except libvirtError as lib_err:
|
||||
messages.error(request, lib_err)
|
||||
except Exception as error_msg:
|
||||
messages.error(request, error_msg)
|
||||
if request.method == "POST": # a filter gone meanwhile, a host not reached: nothing was changed
|
||||
return HttpResponseRedirect(reverse("nwfilters", args=[compute_id]))
|
||||
|
||||
return render(request, "nwfilter.html", locals())
|
||||
|
|
|
|||
|
|
@ -522,17 +522,6 @@ class wvmInstance(wvmConnect):
|
|||
title = util.get_xml_path(self._XMLDesc(PERSISTENT_XML if config else 0), "/domain/title")
|
||||
return title or ""
|
||||
|
||||
def get_filterrefs(self):
|
||||
def filterrefs(ctx):
|
||||
result = []
|
||||
for net in ctx.xpath("/domain/devices/interface"):
|
||||
filterref = net.xpath("filterref/@filter")
|
||||
if filterref:
|
||||
result.append(filterref[0])
|
||||
return result
|
||||
|
||||
return util.get_xml_path(self._XMLDesc(0), func=filterrefs)
|
||||
|
||||
def get_description(self, config=False):
|
||||
description = util.get_xml_path(self._XMLDesc(PERSISTENT_XML if config else 0), "/domain/description")
|
||||
return description or ""
|
||||
|
|
|
|||
|
|
@ -442,12 +442,7 @@ class wvmNetwork(wvmConnect):
|
|||
def edit_network(self, new_xml):
|
||||
"""Defines this network anew from new_xml: of the same name and, if it
|
||||
names one, the same UUID, not another network."""
|
||||
# read for its name and UUID only, its entities not expanded: libvirt gets the text itself
|
||||
parser = etree.XMLParser(resolve_entities=False, no_network=True)
|
||||
try:
|
||||
tree = etree.fromstring(new_xml.encode(), parser) # bytes: the XML may declare its encoding
|
||||
except etree.XMLSyntaxError as err:
|
||||
raise util.OperationError(f"The XML is not valid: {err}") from None
|
||||
tree = util.parse_xml(new_xml)
|
||||
if tree.tag != "network" or tree.findtext("name") != self.net.name():
|
||||
raise util.OperationError(f"The XML must define network {self.net.name()}")
|
||||
if tree.findtext("uuid") not in (None, self.net.UUIDString()):
|
||||
|
|
|
|||
|
|
@ -1,35 +1,119 @@
|
|||
from xml.etree import ElementTree
|
||||
import copy
|
||||
import re
|
||||
from functools import cached_property
|
||||
|
||||
from libvirt import VIR_DOMAIN_XML_INACTIVE, VIR_ERR_NO_DOMAIN, libvirtError
|
||||
from lxml import etree
|
||||
|
||||
from vrtManager import util
|
||||
from vrtManager.connection import wvmConnect
|
||||
|
||||
# the names the panel gives filters: letters, digits, dots, dashes and underscores
|
||||
NAME = re.compile(r"[A-Za-z0-9._-]+")
|
||||
|
||||
|
||||
def element_text(element):
|
||||
"""An element as the page shows it and sends it back: a rule or reference
|
||||
is changed only while it still reads the same."""
|
||||
return same_text(etree.tostring(element, encoding="unicode", with_tail=False))
|
||||
|
||||
|
||||
def same_text(text):
|
||||
"""text without the space between its tags, which a browser may send
|
||||
back with other line breaks."""
|
||||
return re.sub(r">\s+<", "><", text.strip())
|
||||
|
||||
|
||||
def filter_info(nwfilter, tree):
|
||||
"""What a filter and its XML tell: name, UUID, chain, priority, how many
|
||||
rules it has and the filters it references."""
|
||||
return {"name": nwfilter.name(), "uuid": nwfilter.UUIDString(), "chain": tree.get("chain"),
|
||||
"priority": tree.get("priority"), "rules": len(tree.findall("rule")),
|
||||
"refs": [ref.get("filter") for ref in tree.findall("filterref")]}
|
||||
|
||||
|
||||
class wvmNWFilters(wvmConnect):
|
||||
def get_nwfilter_info(self, name):
|
||||
nwfilter = self.get_nwfilter(name)
|
||||
xml = nwfilter.XMLDesc(0)
|
||||
uuid = nwfilter.UUIDString()
|
||||
return {"name": name, "uuid": uuid, "xml": xml}
|
||||
def get_nwfilters_info(self):
|
||||
"""Every filter of the host (filter_info) by name: one listing, one XML
|
||||
read per filter."""
|
||||
filters = [filter_info(nwfilter, etree.fromstring(nwfilter.XMLDesc(0)))
|
||||
for nwfilter in self.wvm.listAllNWFilters(0)]
|
||||
return sorted(filters, key=lambda nwfilter: nwfilter["name"])
|
||||
|
||||
def get_users(self):
|
||||
"""The VMs whose interfaces reference each filter, by filter name: as
|
||||
they run and, for a persistent VM that runs, as saved for its next
|
||||
boot. One listing, one or two XML reads per VM."""
|
||||
users = {}
|
||||
for dom in self.wvm.listAllDomains(0):
|
||||
try:
|
||||
xmls = [dom.XMLDesc(0)]
|
||||
if dom.isActive() and dom.isPersistent():
|
||||
xmls.append(dom.XMLDesc(VIR_DOMAIN_XML_INACTIVE))
|
||||
except libvirtError as err: # a VM undefined meanwhile uses none
|
||||
if err.get_error_code() != VIR_ERR_NO_DOMAIN:
|
||||
raise
|
||||
continue
|
||||
names = set()
|
||||
for xml in xmls:
|
||||
names.update(etree.fromstring(xml).xpath("devices/interface/filterref/@filter"))
|
||||
for name in names:
|
||||
users.setdefault(name, []).append(dom.name())
|
||||
return users
|
||||
|
||||
def create_nwfilter(self, xml):
|
||||
"""Define a filter from the XML entered: one whose name or UUID the
|
||||
host has is refused, as it would replace that filter."""
|
||||
tree = util.parse_xml(xml)
|
||||
if tree.tag != "filter":
|
||||
raise util.OperationError("Enter one <filter> 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 <rule> element")
|
||||
raise util.OperationError("Enter one <rule> element")
|
||||
tree = copy.deepcopy(self.tree)
|
||||
tree.append(rule)
|
||||
return ElementTree.tostring(tree).decode()
|
||||
self._define(tree)
|
||||
|
|
|
|||
|
|
@ -10,16 +10,11 @@ if not settings.configured:
|
|||
|
||||
from vrtManager.connection import wvmConnect
|
||||
from vrtManager.network import wvmNetwork
|
||||
from vrtManager.nwfilters import wvmNWFilter
|
||||
|
||||
CAPS = """<capabilities><guest><arch name='x86_64'><wordsize>64</wordsize><emulator>/usr/bin/qemu-x86</emulator>
|
||||
<machine canonical='pc-q35'>q35</machine></arch>
|
||||
<arch name='aarch64'><emulator>/usr/bin/qemu-arm</emulator><machine>virt</machine></arch></guest></capabilities>"""
|
||||
|
||||
FILTER = """<filter name='f'><rule action='drop' direction='in' priority='1'><tcp/></rule>
|
||||
<rule action='drop' direction='in' priority='2'><udp/></rule></filter>"""
|
||||
|
||||
|
||||
class FakeConn:
|
||||
def __init__(self):
|
||||
self.defined = None
|
||||
|
|
@ -41,20 +36,6 @@ class XPathValueTests(unittest.TestCase):
|
|||
self.assertEqual(self.conn.get_machine_types(injected), [])
|
||||
self.assertEqual(self.conn.get_capabilities(injected), {})
|
||||
|
||||
def test_nwfilter_rule_is_matched_by_its_attributes(self):
|
||||
nwf = wvmNWFilter.__new__(wvmNWFilter)
|
||||
nwf._XMLDesc = lambda flags: FILTER
|
||||
# values with quotes or brackets are compared, not parsed as XPath
|
||||
for action in ("drop'][@priority='2", "it's"):
|
||||
with self.subTest(action=action), self.assertRaises(ValueError):
|
||||
nwf.delete_rule(0, action, "in", "1")
|
||||
kept = nwf.delete_rule(1, "drop", "in", "2")
|
||||
self.assertEqual(kept.count("<rule "), 1)
|
||||
self.assertIn("priority=\"1\"", kept)
|
||||
# add_rule adds a rule of its own
|
||||
added = nwf.add_rule("<rule action='drop' direction='in' priority='2'><icmp/></rule>")
|
||||
self.assertEqual(added.count("<rule "), 3)
|
||||
|
||||
def test_qos_direction_must_be_known(self):
|
||||
net = wvmNetwork.__new__(wvmNetwork)
|
||||
net.wvm = FakeConn()
|
||||
|
|
|
|||
|
|
@ -102,6 +102,19 @@ def compareMAC(p, q):
|
|||
return 0
|
||||
|
||||
|
||||
def parse_xml(xml):
|
||||
"""XML a user entered, read for the panel's checks: its entities not
|
||||
expanded, nothing fetched and no DTD; libvirt gets the text itself."""
|
||||
parser = etree.XMLParser(resolve_entities=False, no_network=True)
|
||||
try:
|
||||
tree = etree.fromstring(xml.encode(), parser) # bytes: the XML may declare its encoding
|
||||
except etree.XMLSyntaxError as err:
|
||||
raise OperationError(f"The XML is not valid: {err}") from None
|
||||
if tree.getroottree().docinfo.internalDTD is not None:
|
||||
raise OperationError("The XML may not declare a DTD")
|
||||
return tree
|
||||
|
||||
|
||||
def get_xml_path(xml, path=None, func=None):
|
||||
"""
|
||||
Return the content from the passed xml xpath, or return the result
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue