summaryrefslogtreecommitdiffstats
path: root/testsuite
diff options
context:
space:
mode:
authorChris St. Pierre <chris.a.st.pierre@gmail.com>2013-02-12 07:48:33 -0500
committerChris St. Pierre <chris.a.st.pierre@gmail.com>2013-02-12 09:18:38 -0500
commit5363e6d9a53146333da0d109aae170befc1b9481 (patch)
tree22f1180360c6844f3ca1f77a7cee59a01c05ad9b /testsuite
parentd0cb9264234851ad65ec8502a56c3afefd39fbad (diff)
downloadbcfg2-5363e6d9a53146333da0d109aae170befc1b9481.tar.gz
bcfg2-5363e6d9a53146333da0d109aae170befc1b9481.tar.bz2
bcfg2-5363e6d9a53146333da0d109aae170befc1b9481.zip
Added client ACLs:
* IP and CIDR-based ACLs * Metadata (group/hostname)-based ACLs * Documentation * Unit tests
Diffstat (limited to 'testsuite')
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py161
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testinterfaces.py18
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestACL.py222
3 files changed, 348 insertions, 53 deletions
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py
index ba837f0c9..4cb148bb0 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py
@@ -812,32 +812,44 @@ class TestStructFile(TestXMLFileBacked):
sf._include_element = Mock()
metadata = Mock()
- (xdata, groups, subgroups, children, subchildren, standalone) = \
- self._get_test_data()
-
sf._include_element.side_effect = \
lambda x, _: (x.tag not in sf._include_tests.keys() or
x.get("include") == "true")
- for i, group in groups.items():
- actual = sf._match(group, metadata)
- expected = children[i] + subchildren[i]
- self.assertEqual(len(actual), len(expected))
- # easiest way to compare the values is actually to make
- # them into an XML document and let assertXMLEqual compare
- # them
- xactual = lxml.etree.Element("Container")
- xactual.extend(actual)
- xexpected = lxml.etree.Element("Container")
- xexpected.extend(expected)
- self.assertXMLEqual(xactual, xexpected)
+ for test_data in [self._get_test_data(),
+ self._get_template_test_data()]:
+ (xdata, groups, subgroups, children, subchildren, standalone) = \
+ test_data
- for el in standalone:
- self.assertXMLEqual(el, sf._match(el, metadata)[0])
+ for i, group in groups.items():
+ actual = sf._match(group, metadata)
+ expected = children[i] + subchildren[i]
+ self.assertEqual(len(actual), len(expected))
+ # easiest way to compare the values is actually to make
+ # them into an XML document and let assertXMLEqual compare
+ # them
+ xactual = lxml.etree.Element("Container")
+ xactual.extend(actual)
+ xexpected = lxml.etree.Element("Container")
+ xexpected.extend(expected)
+ self.assertXMLEqual(xactual, xexpected)
+
+ for el in standalone:
+ self.assertXMLEqual(el, sf._match(el, metadata)[0])
def test_do_match(self):
sf = self.get_obj()
sf._match = Mock()
+
+ def match_rv(el, _):
+ if el.tag not in sf._include_tests.keys():
+ return [el]
+ elif el.get("include") == "true":
+ return el.getchildren()
+ else:
+ return []
+ sf._match.side_effect = match_rv
+
metadata = Mock()
for test_data in [self._get_test_data(),
@@ -847,14 +859,6 @@ class TestStructFile(TestXMLFileBacked):
sf.data = lxml.etree.tostring(xdata)
sf.Index()
- def match_rv(el, _):
- if el.tag not in sf._include_tests.keys():
- return [el]
- elif el.get("include") == "true":
- return el.getchildren()
- else:
- return []
- sf._match.side_effect = match_rv
actual = sf._do_match(metadata)
expected = reduce(lambda x, y: x + y,
list(children.values()) + \
@@ -874,45 +878,96 @@ class TestStructFile(TestXMLFileBacked):
sf._include_element = Mock()
metadata = Mock()
- (xdata, groups, subgroups, children, subchildren, standalone) = \
- self._get_test_data()
-
sf._include_element.side_effect = \
lambda x, _: (x.tag not in sf._include_tests.keys() or
x.get("include") == "true")
- actual = copy.deepcopy(xdata)
- for el in actual.getchildren():
- sf._xml_match(el, metadata)
- expected = lxml.etree.Element(xdata.tag, **dict(xdata.attrib))
- expected.text = xdata.text
- expected.extend(reduce(lambda x, y: x + y,
- list(children.values()) + list(subchildren.values())))
- expected.extend(standalone)
- self.assertXMLEqual(actual, expected)
+ for test_data in [self._get_test_data(),
+ self._get_template_test_data()]:
+ (xdata, groups, subgroups, children, subchildren, standalone) = \
+ test_data
+
+ actual = copy.deepcopy(xdata)
+ for el in actual.getchildren():
+ sf._xml_match(el, metadata)
+ expected = lxml.etree.Element(xdata.tag, **dict(xdata.attrib))
+ expected.text = xdata.text
+ expected.extend(reduce(lambda x, y: x + y,
+ list(children.values()) + \
+ list(subchildren.values())))
+ expected.extend(standalone)
+ self.assertXMLEqual(actual, expected)
def test_do_xmlmatch(self):
sf = self.get_obj()
sf._xml_match = Mock()
metadata = Mock()
- (sf.xdata, groups, subgroups, children, subchildren, standalone) = \
- self._get_test_data()
+ for data_type, test_data in \
+ [("", self._get_test_data()),
+ ("templated ", self._get_template_test_data())]:
+ (xdata, groups, subgroups, children, subchildren, standalone) = \
+ test_data
+ sf.xdata = xdata
+ sf._xml_match.reset_mock()
+
+ sf._do_xmlmatch(metadata)
+ actual = []
+ for call in sf._xml_match.call_args_list:
+ actual.append(call[0][0])
+ self.assertEqual(call[0][1], metadata)
+ expected = list(groups.values()) + standalone
+ # easiest way to compare the values is actually to make
+ # them into an XML document and let assertXMLEqual compare
+ # them
+ xactual = lxml.etree.Element("Container")
+ xactual.extend(actual)
+ xexpected = lxml.etree.Element("Container")
+ xexpected.extend(expected)
+ self.assertXMLEqual(xactual, xexpected,
+ "XMLMatch() calls were incorrect for "
+ "%stest data" % data_type)
+
+ def test_match_ordering(self):
+ """ Match() returns elements in document order """
+ sf = self.get_obj()
+ sf._match = Mock()
+
+ def match_rv(el, _):
+ if el.tag not in sf._include_tests.keys():
+ return [el]
+ elif el.get("include") == "true":
+ return el.getchildren()
+ else:
+ return []
+ sf._match.side_effect = match_rv
- sf._do_xmlmatch(metadata)
- actual = []
- for call in sf._xml_match.call_args_list:
- actual.append(call[0][0])
- self.assertEqual(call[0][1], metadata)
- expected = list(groups.values()) + standalone
- # easiest way to compare the values is actually to make
- # them into an XML document and let assertXMLEqual compare
- # them
- xactual = lxml.etree.Element("Container")
- xactual.extend(actual)
- xexpected = lxml.etree.Element("Container")
- xexpected.extend(expected)
- self.assertXMLEqual(xactual, xexpected)
+ metadata = Mock()
+
+ test_data = lxml.etree.Element("Test")
+ group = lxml.etree.SubElement(test_data, "Group", name="group",
+ include="true")
+ first = lxml.etree.SubElement(group, "Element", name="first")
+ second = lxml.etree.SubElement(test_data, "Element", name="second")
+
+ # sanity check to ensure that first and second are in the
+ # correct document order
+ if test_data.xpath("//Element") != [first, second]:
+ skip("lxml.etree does not construct documents in a reliable order")
+
+ sf.data = lxml.etree.tostring(test_data)
+ sf.Index()
+ rv = sf._do_match(metadata)
+ self.assertEqual(len(rv), 2,
+ "Match() seems to be broken, cannot test ordering")
+ msg = "Match() does not return elements in document order:\n" + \
+ "Expected: [%s, %s]\n" % (first, second) + \
+ "Actual: %s" % rv
+ self.assertXMLEqual(rv[0], first, msg)
+ self.assertXMLEqual(rv[1], second, msg)
+
+ # TODO: add tests to ensure that XMLMatch() returns elements
+ # in document order
class TestINode(Bcfg2TestCase):
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testinterfaces.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testinterfaces.py
index 35f4e0700..6effe05de 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testinterfaces.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testinterfaces.py
@@ -368,3 +368,21 @@ class TestVersion(TestPlugin):
class TestClientRunHooks(Bcfg2TestCase):
""" placeholder for future tests """
pass
+
+
+class TestClientACLs(Bcfg2TestCase):
+ test_obj = ClientACLs
+
+ def get_obj(self):
+ return self.test_obj()
+
+ def test_check_acl_ip(self):
+ ca = self.get_obj()
+ self.assertIn(ca.check_acl_ip(Mock(), Mock()),
+ [True, False, None])
+
+ def test_check_acl_metadata(self):
+ ca = self.get_obj()
+ self.assertIn(ca.check_acl_metadata(Mock(), Mock()),
+ [True, False])
+
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestACL.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestACL.py
new file mode 100644
index 000000000..e457ca7c1
--- /dev/null
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestACL.py
@@ -0,0 +1,222 @@
+import os
+import sys
+import lxml.etree
+import Bcfg2.Server.Plugin
+from mock import Mock, MagicMock, patch
+from Bcfg2.Server.Plugins.ACL import *
+
+# add all parent testsuite directories to sys.path to allow (most)
+# relative imports in python 2.4
+path = os.path.dirname(__file__)
+while path != "/":
+ if os.path.basename(path).lower().startswith("test"):
+ sys.path.append(path)
+ if os.path.basename(path) == "testsuite":
+ break
+ path = os.path.dirname(path)
+from common import *
+from TestPlugin import TestXMLFileBacked, TestStructFile, TestPlugin, \
+ TestClientACLs
+
+
+class TestFunctions(Bcfg2TestCase):
+ def test_rmi_names_equal(self):
+ good_cases = [('*', 'foo'),
+ ('foo', 'foo'),
+ ('foo.*', 'foo.bar'),
+ ('*.*', 'foo.bar'),
+ ('foo.bar', 'foo.bar'),
+ ('*.bar', 'foo.bar'),
+ ('foo.*.bar', 'foo.baz.bar')]
+ bad_cases = [('foo', 'bar'),
+ ('*', 'foo.bar'),
+ ('*.*', 'foo'),
+ ('*.*', 'foo.bar.baz'),
+ ('foo.*', 'bar.foo'),
+ ('*.bar', 'bar.foo'),
+ ('foo.*', 'foobar')]
+ for first, second in good_cases:
+ self.assertTrue(rmi_names_equal(first, second),
+ "rmi_names_equal(%s, %s) unexpectedly False" %
+ (first, second))
+ self.assertTrue(rmi_names_equal(second, first),
+ "rmi_names_equal(%s, %s) unexpectedly False" %
+ (second, first))
+ for first, second in bad_cases:
+ self.assertFalse(rmi_names_equal(first, second),
+ "rmi_names_equal(%s, %s) unexpectedly True" %
+ (first, second))
+ self.assertFalse(rmi_names_equal(second, first),
+ "rmi_names_equal(%s, %s) unexpectedly True" %
+ (second, first))
+
+ def test_ip_matches(self):
+ good_cases = [
+ ("192.168.1.1", lxml.etree.Element("test", address="192.168.1.1")),
+ ("192.168.1.17", lxml.etree.Element("test", address="192.168.1.0",
+ netmask="24")),
+ ("192.168.1.17", lxml.etree.Element("test", address="192.168.1.0",
+ netmask="255.255.255.0")),
+ ("192.168.1.31", lxml.etree.Element("test", address="192.168.1.0",
+ netmask="255.255.255.224")),
+ ("192.168.1.31", lxml.etree.Element("test", address="192.168.1.0",
+ netmask="27")),
+ ("10.55.67.191", lxml.etree.Element("test", address="10.55.0.0",
+ netmask="16"))]
+ bad_cases = [
+ ("192.168.1.1", lxml.etree.Element("test", address="192.168.1.2")),
+ ("192.168.2.17", lxml.etree.Element("test", address="192.168.1.0",
+ netmask="24")),
+ ("192.168.2.17", lxml.etree.Element("test", address="192.168.1.0",
+ netmask="255.255.255.0")),
+ ("192.168.1.35", lxml.etree.Element("test", address="192.168.1.0",
+ netmask="255.255.255.224")),
+ ("192.168.1.35", lxml.etree.Element("test", address="192.168.1.0",
+ netmask="27")),
+ ("10.56.67.191", lxml.etree.Element("test", address="10.55.0.0",
+ netmask="16"))]
+ for ip, entry in good_cases:
+ self.assertTrue(ip_matches(ip, entry),
+ "ip_matches(%s, %s) unexpectedly False" %
+ (ip, lxml.etree.tostring(entry)))
+ for ip, entry in bad_cases:
+ self.assertFalse(ip_matches(ip, entry),
+ "ip_matches(%s, %s) unexpectedly True" %
+ (ip, lxml.etree.tostring(entry)))
+
+
+class TestIPACLFile(TestXMLFileBacked):
+ test_obj = IPACLFile
+
+ @patch("Bcfg2.Server.Plugins.ACL.ip_matches")
+ @patch("Bcfg2.Server.Plugins.ACL.rmi_names_equal")
+ def test_check_acl(self, mock_rmi_names_equal, mock_ip_matches):
+ af = self.get_obj()
+ ip = "10.0.0.8"
+ rmi = "ACL.test"
+
+ def reset():
+ mock_rmi_names_equal.reset_mock()
+ mock_ip_matches.reset_mock()
+
+ # test default defer with no entries
+ af.entries = []
+ self.assertIsNone(af.check_acl(ip, rmi))
+
+ # test explicit allow, deny, and defer
+ entries = dict(Allow=lxml.etree.Element("Allow", method=rmi),
+ Deny=lxml.etree.Element("Deny", method=rmi),
+ Defer=lxml.etree.Element("Defer", method=rmi))
+ af.entries = list(entries.values())
+
+ def get_ip_matches(tag):
+ def ip_matches(ip, entry):
+ return entry.tag == tag
+
+ return ip_matches
+
+ mock_rmi_names_equal.return_value = True
+
+ reset()
+ mock_ip_matches.side_effect = get_ip_matches("Allow")
+ self.assertTrue(af.check_acl(ip, rmi))
+ mock_ip_matches.assert_called_with(ip, entries['Allow'])
+ mock_rmi_names_equal.assert_called_with(rmi, rmi)
+
+ reset()
+ mock_ip_matches.side_effect = get_ip_matches("Deny")
+ self.assertFalse(af.check_acl(ip, rmi))
+ mock_ip_matches.assert_called_with(ip, entries['Deny'])
+ mock_rmi_names_equal.assert_called_with(rmi, rmi)
+
+ reset()
+ mock_ip_matches.side_effect = get_ip_matches("Defer")
+ self.assertIsNone(af.check_acl(ip, rmi))
+ mock_ip_matches.assert_called_with(ip, entries['Defer'])
+ mock_rmi_names_equal.assert_called_with(rmi, rmi)
+
+ # test matching RMI names
+ reset()
+ mock_ip_matches.side_effect = lambda i, e: True
+ mock_rmi_names_equal.side_effect = lambda a, b: a == b
+ rmi = "ACL.test2"
+ matching = lxml.etree.Element("Allow", method=rmi)
+ af.entries.append(matching)
+ self.assertTrue(af.check_acl(ip, rmi))
+ mock_ip_matches.assert_called_with(ip, matching)
+ self.assertTrue(
+ call('ACL.test', rmi) in mock_rmi_names_equal.call_args_list or
+ call(rmi, 'ACL.test') in mock_rmi_names_equal.call_args_list)
+
+ # test implicit allow for localhost, defer for others
+ reset()
+ mock_ip_matches.side_effect = lambda i, e: False
+ self.assertIsNone(af.check_acl(ip, rmi))
+
+ reset()
+ self.assertTrue(af.check_acl("127.0.0.1", rmi))
+
+
+class TestMetadataACLFile(TestStructFile):
+ test_obj = MetadataACLFile
+
+ @patch("Bcfg2.Server.Plugins.ACL.rmi_names_equal")
+ def test_check_acl(self, mock_rmi_names_equal):
+ af = self.get_obj()
+ af.Match = Mock()
+ metadata = Mock()
+ mock_rmi_names_equal.side_effect = lambda a, b: a == b
+
+ def reset():
+ af.Match.reset_mock()
+ mock_rmi_names_equal.reset_mock()
+
+ # test default allow
+ af.entries = []
+ self.assertTrue(af.check_acl(metadata, 'ACL.test'))
+
+ # test explicit allow and deny
+ reset()
+ af.entries = [lxml.etree.Element("Allow", method='ACL.test'),
+ lxml.etree.Element("Deny", method='ACL.test2')]
+ af.Match.return_value = af.entries
+ self.assertTrue(af.check_acl(metadata, 'ACL.test'))
+ af.Match.assert_called_with(metadata)
+ self.assertIn(call('ACL.test', 'ACL.test'),
+ mock_rmi_names_equal.call_args_list)
+
+ reset()
+ self.assertFalse(af.check_acl(metadata, 'ACL.test2'))
+ af.Match.assert_called_with(metadata)
+ self.assertIn(call('ACL.test2', 'ACL.test2'),
+ mock_rmi_names_equal.call_args_list)
+
+ # test default deny for non-localhost
+ reset()
+ self.assertFalse(af.check_acl(metadata, 'ACL.test3'))
+ af.Match.assert_called_with(metadata)
+
+ # test default allow for localhost
+ reset()
+ metadata.hostname = 'localhost'
+ self.assertTrue(af.check_acl(metadata, 'ACL.test3'))
+ af.Match.assert_called_with(metadata)
+
+
+class TestACL(TestPlugin, TestClientACLs):
+ test_obj = ACL
+
+ def test_check_acl_ip(self):
+ acl = self.get_obj()
+ acl.ip_acls = Mock()
+ self.assertEqual(acl.check_acl_ip("192.168.1.10", "ACL.test"),
+ acl.ip_acls.check_acl.return_value)
+ acl.ip_acls.check_acl.assert_called_with("192.168.1.10", "ACL.test")
+
+ def test_check_acl_metadata(self):
+ acl = self.get_obj()
+ acl.metadata_acls = Mock()
+ metadata = Mock()
+ self.assertEqual(acl.check_acl_metadata(metadata, "ACL.test"),
+ acl.metadata_acls.check_acl.return_value)
+ acl.metadata_acls.check_acl.assert_called_with(metadata, "ACL.test")