From 3f36298716e87131e622d82d8e15e4f896357a96 Mon Sep 17 00:00:00 2001 From: Joona Hoikkala Date: Mon, 21 Oct 2019 22:52:00 +0300 Subject: [PATCH] [Apache v2] find_blocks and find_directives implementation (#7443) * Implement AugeasDirectiveNode and AugeasBlockNode find and create functions * Add tests for _aug_get_block_name --- certbot-apache/certbot_apache/augeasparser.py | 119 ++++++++++++++++-- certbot-apache/certbot_apache/configurator.py | 16 +++ certbot-apache/certbot_apache/dualparser.py | 3 +- certbot-apache/certbot_apache/parser.py | 4 +- .../certbot_apache/tests/augeasnode_test.py | 65 ++++++++++ .../certbot_apache/tests/dualnode_test.py | 39 ++---- 6 files changed, 203 insertions(+), 43 deletions(-) create mode 100644 certbot-apache/certbot_apache/tests/augeasnode_test.py diff --git a/certbot-apache/certbot_apache/augeasparser.py b/certbot-apache/certbot_apache/augeasparser.py index 9325de5de..1bbf1ed77 100644 --- a/certbot-apache/certbot_apache/augeasparser.py +++ b/certbot-apache/certbot_apache/augeasparser.py @@ -1,8 +1,14 @@ """ Augeas implementation of the ParserNode interfaces """ + +from certbot_apache import apache_util from certbot_apache import assertions from certbot_apache import interfaces +from certbot_apache import parser from certbot_apache import parsernode_util as util +from certbot.compat import os +from acme.magic_typing import Set # pylint: disable=unused-import, no-name-in-module + class AugeasParserNode(interfaces.ParserNode): """ Augeas implementation of ParserNode interface """ @@ -14,6 +20,7 @@ class AugeasParserNode(interfaces.ParserNode): self.filepath = filepath self.dirty = dirty self.metadata = metadata + self.parser = self.metadata.get("augeasparser") def save(self, msg): # pragma: no cover pass @@ -85,45 +92,72 @@ class AugeasBlockNode(AugeasDirectiveNode): def add_child_block(self, name, parameters=None, position=None): # pylint: disable=unused-argument """Adds a new BlockNode to the sequence of children""" + new_metadata = {"augeasparser": self.parser} new_block = AugeasBlockNode(name=assertions.PASS, ancestor=self, - filepath=assertions.PASS) + filepath=assertions.PASS, + metadata=new_metadata) self.children += (new_block,) return new_block def add_child_directive(self, name, parameters=None, position=None): # pylint: disable=unused-argument """Adds a new DirectiveNode to the sequence of children""" + new_metadata = {"augeasparser": self.parser} new_dir = AugeasDirectiveNode(name=assertions.PASS, ancestor=self, - filepath=assertions.PASS) + filepath=assertions.PASS, + metadata=new_metadata) self.children += (new_dir,) return new_dir def add_child_comment(self, comment="", position=None): # pylint: disable=unused-argument """Adds a new CommentNode to the sequence of children""" + new_metadata = {"augeasparser": self.parser} new_comment = AugeasCommentNode(comment=assertions.PASS, ancestor=self, - filepath=assertions.PASS) + filepath=assertions.PASS, + metadata=new_metadata) self.children += (new_comment,) return new_comment def find_blocks(self, name, exclude=True): # pylint: disable=unused-argument """Recursive search of BlockNodes from the sequence of children""" - return [AugeasBlockNode(name=assertions.PASS, - ancestor=self, - filepath=assertions.PASS)] + + nodes = list() + paths = self._aug_find_blocks(name) + if exclude: + paths = self.parser.exclude_dirs(paths) + for path in paths: + nodes.append(self._create_blocknode(path)) + + return nodes def find_directives(self, name, exclude=True): # pylint: disable=unused-argument """Recursive search of DirectiveNodes from the sequence of children""" - return [AugeasDirectiveNode(name=assertions.PASS, - ancestor=self, - filepath=assertions.PASS)] + + nodes = list() + ownpath = self.metadata.get("augeaspath") + + directives = self.parser.find_dir(name, start=ownpath, exclude=exclude) + already_parsed = set() # type: Set[str] + for directive in directives: + # Remove the /arg part from the Augeas path + directive = directive.partition("/arg")[0] + # find_dir returns an object for each _parameter_ of a directive + # so we need to filter out duplicates. + if directive not in already_parsed: + nodes.append(self._create_directivenode(directive)) + already_parsed.add(directive) + + return nodes def find_comments(self, comment, exact=False): # pylint: disable=unused-argument """Recursive search of DirectiveNodes from the sequence of children""" + new_metadata = {"augeasparser": self.parser} return [AugeasCommentNode(comment=assertions.PASS, ancestor=self, - filepath=assertions.PASS)] + filepath=assertions.PASS, + metadata=new_metadata)] def delete_child(self, child): # pragma: no cover """Deletes a ParserNode from the sequence of children""" @@ -133,6 +167,71 @@ class AugeasBlockNode(AugeasDirectiveNode): """Returns a list of unsaved filepaths""" return [assertions.PASS] + def _create_directivenode(self, path): + """Helper function to create a DirectiveNode from Augeas path""" + + name = self.parser.get_arg(path) + params = tuple(self._aug_get_params(path)) + metadata = {"augeasparser": self.parser, "augeaspath": path} + + # Because of the dynamic nature, and the fact that we're not populating + # the complete ParserNode tree, we use the search parent as ancestor + return AugeasDirectiveNode(name=name, + parameters=params, + ancestor=self, + filepath=apache_util.get_file_path(path), + metadata=metadata) + + def _create_blocknode(self, path): + """Helper function to create a BlockNode from Augeas path""" + + name = self._aug_get_block_name(path) + params = tuple(self._aug_get_params(path)) + metadata = {"augeasparser": self.parser, "augeaspath": path} + + # Because of the dynamic nature, and the fact that we're not populating + # the complete ParserNode tree, we use the search parent as ancestor + return AugeasBlockNode(name=name, + parameters=params, + ancestor=self, + filepath=apache_util.get_file_path(path), + metadata=metadata) + + def _aug_find_blocks(self, name): + """Helper function to perform a search to Augeas DOM tree to search + configuration blocks with a given name""" + + # The code here is modified from configurator.get_virtual_hosts() + blk_paths = set() + for vhost_path in list(self.parser.parser_paths): + paths = self.parser.aug.match( + ("/files%s//*[label()=~regexp('%s')]" % + (vhost_path, parser.case_i(name)))) + blk_paths.update([path for path in paths if + name.lower() in os.path.basename(path).lower()]) + return blk_paths + + def _aug_get_params(self, path): + """Helper function to get parameters for BlockNodes""" + + arg_paths = self.parser.aug.match(path + "/arg") + return [self.parser.get_arg(apath) for apath in arg_paths] + + def _aug_get_block_name(self, path): + """Helper function to get name of a configuration block from path.""" + + # Remove the ending slash if any + if path[-1] == "/": # pragma: no cover + path = path[:-1] + + # Get the block name + name = path.split("/")[-1] + + # remove [...], it's not allowed in Apache configuration and is used + # for indexing within Augeas + name = name.split("[")[0] + return name + interfaces.CommentNode.register(AugeasCommentNode) interfaces.DirectiveNode.register(AugeasDirectiveNode) diff --git a/certbot-apache/certbot_apache/configurator.py b/certbot-apache/certbot_apache/configurator.py index f7c27bf76..d01b0f5ce 100644 --- a/certbot-apache/certbot_apache/configurator.py +++ b/certbot-apache/certbot_apache/configurator.py @@ -30,8 +30,10 @@ from certbot.plugins.util import path_surgery from certbot.plugins.enhancements import AutoHSTSEnhancement from certbot_apache import apache_util +from certbot_apache import assertions from certbot_apache import constants from certbot_apache import display_ops +from certbot_apache import dualparser from certbot_apache import http_01 from certbot_apache import obj from certbot_apache import parser @@ -204,6 +206,7 @@ class ApacheConfigurator(common.Installer): # These will be set in the prepare function self._prepared = False self.parser = None + self.parser_root = None self.version = version self.vhosts = None self.options = copy.deepcopy(self.OS_DEFAULTS) @@ -253,6 +256,10 @@ class ApacheConfigurator(common.Installer): # Perform the actual Augeas initialization to be able to react self.parser = self.get_parser() + # Set up ParserNode root + pn_meta = {"augeasparser": self.parser} + self.parser_root = self.get_parsernode_root(pn_meta) + # Check for errors in parsing files with Augeas self.parser.check_parsing_errors("httpd.aug") @@ -348,6 +355,15 @@ class ApacheConfigurator(common.Installer): self.option("server_root"), self.conf("vhost-root"), self.version, configurator=self) + def get_parsernode_root(self, metadata): + """Initializes the ParserNode parser root instance.""" + return dualparser.DualBlockNode( + name=assertions.PASS, + ancestor=None, + filepath=assertions.PASS, + metadata=metadata + ) + def _wildcard_domain(self, domain): """ Checks if domain is a wildcard domain diff --git a/certbot-apache/certbot_apache/dualparser.py b/certbot-apache/certbot_apache/dualparser.py index 5fa337a45..d56e07de2 100644 --- a/certbot-apache/certbot_apache/dualparser.py +++ b/certbot-apache/certbot_apache/dualparser.py @@ -17,7 +17,8 @@ class DualNodeBase(object): """ Attribute value assertion """ firstval = getattr(self.primary, aname) secondval = getattr(self.secondary, aname) - assertions.assertEqualSimple(firstval, secondval) + if not callable(firstval): + assertions.assertEqualSimple(firstval, secondval) return firstval diff --git a/certbot-apache/certbot_apache/parser.py b/certbot-apache/certbot_apache/parser.py index b5f0cd81a..7ae9ac386 100644 --- a/certbot-apache/certbot_apache/parser.py +++ b/certbot-apache/certbot_apache/parser.py @@ -613,7 +613,7 @@ class ApacheParser(object): "%s//*[self::directive=~regexp('%s')]" % (start, regex)) if exclude: - matches = self._exclude_dirs(matches) + matches = self.exclude_dirs(matches) if arg is None: arg_suffix = "/arg" @@ -680,7 +680,7 @@ class ApacheParser(object): return value - def _exclude_dirs(self, matches): + def exclude_dirs(self, matches): """Exclude directives that are not loaded into the configuration.""" filters = [("ifmodule", self.modules), ("ifdefine", self.variables)] diff --git a/certbot-apache/certbot_apache/tests/augeasnode_test.py b/certbot-apache/certbot_apache/tests/augeasnode_test.py new file mode 100644 index 000000000..c4631c57c --- /dev/null +++ b/certbot-apache/certbot_apache/tests/augeasnode_test.py @@ -0,0 +1,65 @@ +"""Tests for AugeasParserNode classes""" +import mock + +from certbot_apache import assertions + +from certbot_apache.tests import util + + +class AugeasParserNodeTest(util.ApacheTest): + """Test AugeasParserNode using available test configurations""" + + def setUp(self): # pylint: disable=arguments-differ + super(AugeasParserNodeTest, self).setUp() + + self.config = util.get_apache_configurator( + self.config_path, self.vhost_path, self.config_dir, self.work_dir) + self.vh_truth = util.get_vh_truth( + self.temp_dir, "debian_apache_2_4/multiple_vhosts") + + def test_get_block_node_name(self): + from certbot_apache.augeasparser import AugeasBlockNode + block = AugeasBlockNode( + name=assertions.PASS, + ancestor=None, + filepath=assertions.PASS, + metadata={"augeasparser": mock.Mock()} + ) + testcases = { + "/some/path/FirstNode/SecondNode": "SecondNode", + "/some/path/FirstNode/SecondNode/": "SecondNode", + "OnlyPathItem": "OnlyPathItem", + "/files/etc/apache2/apache2.conf/VirtualHost": "VirtualHost", + "/Anything": "Anything", + } + for test in testcases: + self.assertEqual(block._aug_get_block_name(test), testcases[test]) # pylint: disable=protected-access + + def test_find_blocks(self): + blocks = self.config.parser_root.find_blocks("VirtualHost", exclude=False) + self.assertEqual(len(blocks), 12) + + def test_find_blocks_case_insensitive(self): + vhs = self.config.parser_root.find_blocks("VirtualHost") + vhs2 = self.config.parser_root.find_blocks("viRtuAlHoST") + self.assertEqual(len(vhs), len(vhs2)) + + def test_find_directive_found(self): + directives = self.config.parser_root.find_directives("Listen") + self.assertEqual(len(directives), 1) + self.assertTrue(directives[0].filepath.endswith("/apache2/ports.conf")) + self.assertEqual(directives[0].parameters, (u'80',)) + + def test_find_directive_notfound(self): + directives = self.config.parser_root.find_directives("Nonexistent") + self.assertEqual(len(directives), 0) + + def test_find_directive_from_block(self): + blocks = self.config.parser_root.find_blocks("virtualhost") + found = False + for vh in blocks: + if vh.filepath.endswith("sites-enabled/certbot.conf"): + servername = vh.find_directives("servername") + self.assertEqual(servername[0].parameters[0], "certbot.demo") + found = True + self.assertTrue(found) diff --git a/certbot-apache/certbot_apache/tests/dualnode_test.py b/certbot-apache/certbot_apache/tests/dualnode_test.py index eeea7d151..b37a7fdf9 100644 --- a/certbot-apache/certbot_apache/tests/dualnode_test.py +++ b/certbot-apache/certbot_apache/tests/dualnode_test.py @@ -13,18 +13,23 @@ class DualParserNodeTest(unittest.TestCase): # pylint: disable=too-many-public- """DualParserNode tests""" def setUp(self): # pylint: disable=arguments-differ + metadata = {"augeasparser": mock.Mock()} self.block = dualparser.DualBlockNode(name="block", ancestor=None, - filepath="/tmp/something") + filepath="/tmp/something", + metadata=metadata) self.block_two = dualparser.DualBlockNode(name="block", ancestor=self.block, - filepath="/tmp/something") + filepath="/tmp/something", + metadata=metadata) self.directive = dualparser.DualDirectiveNode(name="directive", ancestor=self.block, - filepath="/tmp/something") + filepath="/tmp/something", + metadata=metadata) self.comment = dualparser.DualCommentNode(comment="comment", ancestor=self.block, - filepath="/tmp/something") + filepath="/tmp/something", + metadata=metadata) def test_create_with_precreated(self): cnode = dualparser.DualCommentNode(comment="comment", @@ -155,32 +160,6 @@ class DualParserNodeTest(unittest.TestCase): # pylint: disable=too-many-public- self.assertEqual(self.block.primary.children[0].ancestor, self.block.primary) - def test_find_blocks(self): - dblks = self.block.find_blocks("block") - p_dblks = [d.primary for d in dblks] - s_dblks = [d.secondary for d in dblks] - p_blks = self.block.primary.find_blocks("block") - s_blks = self.block.secondary.find_blocks("block") - # Check that every block response is represented in the list of - # DualParserNode instances. - for p in p_dblks: - self.assertTrue(p in p_blks) - for s in s_dblks: - self.assertTrue(s in s_blks) - - def test_find_directives(self): - ddirs = self.block.find_directives("directive") - p_ddirs = [d.primary for d in ddirs] - s_ddirs = [d.secondary for d in ddirs] - p_dirs = self.block.primary.find_directives("directive") - s_dirs = self.block.secondary.find_directives("directive") - # Check that every directive response is represented in the list of - # DualParserNode instances. - for p in p_ddirs: - self.assertTrue(p in p_dirs) - for s in s_ddirs: - self.assertTrue(s in s_dirs) - def test_find_comments(self): dcoms = self.block.find_comments("comment") p_dcoms = [d.primary for d in dcoms]