starred/certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-28 16:54:56 +02:00
Code Issues Packages Projects Releases Wiki Activity

[Apache v2] find_blocks and find_directives implementation (#7443)

Browse Source 
* Implement AugeasDirectiveNode and AugeasBlockNode find and create functions

* Add tests for _aug_get_block_name
This commit is contained in:
Joona Hoikkala
2019-10-21 12:52:00 -07:00
committed by sydneyli
parent 63c7dd109c
commit 3f36298716
6 changed files with 203 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
certbot-apache/certbot_apache/augeasparser.py
+109 -10
View File
@@ -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)
certbot-apache/certbot_apache/configurator.py
+16
View File
@@ -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
certbot-apache/certbot_apache/dualparser.py
+2 -1
View File
@@ -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
certbot-apache/certbot_apache/parser.py
+2 -2
View File
@@ -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)]
certbot-apache/certbot_apache/tests/augeasnode_test.py
+65
View File
@@ -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)
certbot-apache/certbot_apache/tests/dualnode_test.py
+9 -30
View File
@@ -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]