diff options
Diffstat (limited to 'certbot-nginx/certbot_nginx')
| -rw-r--r-- | certbot-nginx/certbot_nginx/configurator.py | 46 | ||||
| -rw-r--r-- | certbot-nginx/certbot_nginx/http_01.py | 7 | ||||
| -rw-r--r-- | certbot-nginx/certbot_nginx/parser_obj.py | 392 | ||||
| -rw-r--r-- | certbot-nginx/certbot_nginx/tests/configurator_test.py | 77 | ||||
| -rw-r--r-- | certbot-nginx/certbot_nginx/tests/http_01_test.py | 36 | ||||
| -rw-r--r-- | certbot-nginx/certbot_nginx/tests/parser_obj_test.py | 253 | ||||
| -rw-r--r-- | certbot-nginx/certbot_nginx/tls_sni_01.py | 8 |
7 files changed, 768 insertions, 51 deletions
diff --git a/certbot-nginx/certbot_nginx/configurator.py b/certbot-nginx/certbot_nginx/configurator.py index d526381a2..dd0bf9e8b 100644 --- a/certbot-nginx/certbot_nginx/configurator.py +++ b/certbot-nginx/certbot_nginx/configurator.py @@ -8,7 +8,6 @@ import tempfile import time import OpenSSL -import six import zope.interface from acme import challenges @@ -32,6 +31,12 @@ from certbot_nginx import obj # pylint: disable=unused-import from acme.magic_typing import List, Dict, Set # pylint: disable=unused-import, no-name-in-module +NAME_RANK = 0 +START_WILDCARD_RANK = 1 +END_WILDCARD_RANK = 2 +REGEX_RANK = 3 +NO_SSL_MODIFIER = 4 + logger = logging.getLogger(__name__) @@ -405,7 +410,8 @@ class NginxConfigurator(common.Installer): """ if not matches: return None - elif matches[0]['rank'] in six.moves.range(2, 6): + elif matches[0]['rank'] in [START_WILDCARD_RANK, END_WILDCARD_RANK, + START_WILDCARD_RANK + NO_SSL_MODIFIER, END_WILDCARD_RANK + NO_SSL_MODIFIER]: # Wildcard match - need to find the longest one rank = matches[0]['rank'] wildcards = [x for x in matches if x['rank'] == rank] @@ -414,10 +420,9 @@ class NginxConfigurator(common.Installer): # Exact or regex match return matches[0]['vhost'] - - def _rank_matches_by_name_and_ssl(self, vhost_list, target_name): + def _rank_matches_by_name(self, vhost_list, target_name): """Returns a ranked list of vhosts from vhost_list that match target_name. - The ranking gives preference to SSL vhosts. + This method should always be followed by a call to _select_best_name_match. :param list vhost_list: list of vhosts to filter and rank :param str target_name: The name to match @@ -437,21 +442,37 @@ class NginxConfigurator(common.Installer): if name_type == 'exact': matches.append({'vhost': vhost, 'name': name, - 'rank': 0 if vhost.ssl else 1}) + 'rank': NAME_RANK}) elif name_type == 'wildcard_start': matches.append({'vhost': vhost, 'name': name, - 'rank': 2 if vhost.ssl else 3}) + 'rank': START_WILDCARD_RANK}) elif name_type == 'wildcard_end': matches.append({'vhost': vhost, 'name': name, - 'rank': 4 if vhost.ssl else 5}) + 'rank': END_WILDCARD_RANK}) elif name_type == 'regex': matches.append({'vhost': vhost, 'name': name, - 'rank': 6 if vhost.ssl else 7}) + 'rank': REGEX_RANK}) return sorted(matches, key=lambda x: x['rank']) + def _rank_matches_by_name_and_ssl(self, vhost_list, target_name): + """Returns a ranked list of vhosts from vhost_list that match target_name. + The ranking gives preference to SSLishness before name match level. + + :param list vhost_list: list of vhosts to filter and rank + :param str target_name: The name to match + :returns: list of dicts containing the vhost, the matching name, and + the numerical rank + :rtype: list + + """ + matches = self._rank_matches_by_name(vhost_list, target_name) + for match in matches: + if not match['vhost'].ssl: + match['rank'] += NO_SSL_MODIFIER + return sorted(matches, key=lambda x: x['rank']) def choose_redirect_vhosts(self, target_name, port, create_if_no_match=False): """Chooses a single virtual host for redirect enhancement. @@ -531,9 +552,7 @@ class NginxConfigurator(common.Installer): matching_vhosts = [vhost for vhost in all_vhosts if _vhost_matches(vhost, port)] - # We can use this ranking function because sslishness doesn't matter to us, and - # there shouldn't be conflicting plaintextish servers listening on 80. - return self._rank_matches_by_name_and_ssl(matching_vhosts, target_name) + return self._rank_matches_by_name(matching_vhosts, target_name) def get_all_names(self): """Returns all names found in the Nginx Configuration. @@ -568,6 +587,7 @@ class NginxConfigurator(common.Installer): return util.get_filtered_names(all_names) def _get_snakeoil_paths(self): + """Generate invalid certs that let us create ssl directives for Nginx""" # TODO: generate only once tmp_dir = os.path.join(self.config.work_dir, "snakeoil") le_key = crypto_util.init_save_key( @@ -1019,7 +1039,7 @@ class NginxConfigurator(common.Installer): ########################################################################### def get_chall_pref(self, unused_domain): # pylint: disable=no-self-use """Return list of challenge preferences.""" - return [challenges.TLSSNI01, challenges.HTTP01] + return [challenges.HTTP01, challenges.TLSSNI01] # Entry point in main.py for performing challenges def perform(self, achalls): diff --git a/certbot-nginx/certbot_nginx/http_01.py b/certbot-nginx/certbot_nginx/http_01.py index 677ce0737..e46d7b9b9 100644 --- a/certbot-nginx/certbot_nginx/http_01.py +++ b/certbot-nginx/certbot_nginx/http_01.py @@ -40,8 +40,6 @@ class NginxHttp01(common.ChallengePerformer): super(NginxHttp01, self).__init__(configurator) self.challenge_conf = os.path.join( configurator.config.config_dir, "le_http_01_cert_challenge.conf") - self._ipv6 = None - self._ipv6only = None def perform(self): """Perform a challenge on Nginx. @@ -102,6 +100,7 @@ class NginxHttp01(common.ChallengePerformer): config = [self._make_or_mod_server_block(achall) for achall in self.achalls] config = [x for x in config if x is not None] config = nginxparser.UnspacedList(config) + logger.debug("Generated server block:\n%s", str(config)) self.configurator.reverter.register_file_creation( True, self.challenge_conf) @@ -120,9 +119,7 @@ class NginxHttp01(common.ChallengePerformer): self.configurator.config.http01_port) port = self.configurator.config.http01_port - if self._ipv6 is None or self._ipv6only is None: - self._ipv6, self._ipv6only = self.configurator.ipv6_info(port) - ipv6, ipv6only = self._ipv6, self._ipv6only + ipv6, ipv6only = self.configurator.ipv6_info(port) if ipv6: # If IPv6 is active in Nginx configuration diff --git a/certbot-nginx/certbot_nginx/parser_obj.py b/certbot-nginx/certbot_nginx/parser_obj.py new file mode 100644 index 000000000..f01cb2fd3 --- /dev/null +++ b/certbot-nginx/certbot_nginx/parser_obj.py @@ -0,0 +1,392 @@ +""" This file contains parsing routines and object classes to help derive meaning from +raw lists of tokens from pyparsing. """ + +import abc +import logging +import six + +from certbot import errors + +from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module + +logger = logging.getLogger(__name__) +COMMENT = " managed by Certbot" +COMMENT_BLOCK = ["#", COMMENT] + +class Parsable(object): + """ Abstract base class for "Parsable" objects whose underlying representation + is a tree of lists. + + :param .Parsable parent: This object's parsed parent in the tree + """ + + __metaclass__ = abc.ABCMeta + + def __init__(self, parent=None): + self._data = [] # type: List[object] + self._tabs = None + self.parent = parent + + @classmethod + def parsing_hooks(cls): + """Returns object types that this class should be able to `parse` recusrively. + The order of the objects indicates the order in which the parser should + try to parse each subitem. + :returns: A list of Parsable classes. + :rtype list: + """ + return (Block, Sentence, Statements) + + @staticmethod + @abc.abstractmethod + def should_parse(lists): + """ Returns whether the contents of `lists` can be parsed into this object. + + :returns: Whether `lists` can be parsed as this object. + :rtype bool: + """ + raise NotImplementedError() + + @abc.abstractmethod + def parse(self, raw_list, add_spaces=False): + """ Loads information into this object from underlying raw_list structure. + Each Parsable object might make different assumptions about the structure of + raw_list. + + :param list raw_list: A list or sublist of tokens from pyparsing, containing whitespace + as separate tokens. + :param bool add_spaces: If set, the method can and should manipulate and insert spacing + between non-whitespace tokens and lists to delimit them. + :raises .errors.MisconfigurationError: when the assumptions about the structure of + raw_list are not met. + """ + raise NotImplementedError() + + @abc.abstractmethod + def iterate(self, expanded=False, match=None): + """ Iterates across this object. If this object is a leaf object, only yields + itself. If it contains references other parsing objects, and `expanded` is set, + this function should first yield itself, then recursively iterate across all of them. + :param bool expanded: Whether to recursively iterate on possible children. + :param callable match: If provided, an object is only iterated if this callable + returns True when called on that object. + + :returns: Iterator over desired objects. + """ + raise NotImplementedError() + + @abc.abstractmethod + def get_tabs(self): + """ Guess at the tabbing style of this parsed object, based on whitespace. + + If this object is a leaf, it deducts the tabbing based on its own contents. + Other objects may guess by calling `get_tabs` recursively on child objects. + + :returns: Guess at tabbing for this object. Should only return whitespace strings + that does not contain newlines. + :rtype str: + """ + raise NotImplementedError() + + @abc.abstractmethod + def set_tabs(self, tabs=" "): + """This tries to set and alter the tabbing of the current object to a desired + whitespace string. Primarily meant for objects that were constructed, so they + can conform to surrounding whitespace. + + :param str tabs: A whitespace string (not containing newlines). + """ + raise NotImplementedError() + + def dump(self, include_spaces=False): + """ Dumps back to pyparsing-like list tree. The opposite of `parse`. + + Note: if this object has not been modified, `dump` with `include_spaces=True` + should always return the original input of `parse`. + + :param bool include_spaces: If set to False, magically hides whitespace tokens from + dumped output. + + :returns: Pyparsing-like list tree. + :rtype list: + """ + return [elem.dump(include_spaces) for elem in self._data] + +class Statements(Parsable): + """ A group or list of "Statements". A Statement is either a Block or a Sentence. + + The underlying representation is simply a list of these Statement objects, with + an extra `_trailing_whitespace` string to keep track of the whitespace that does not + precede any more statements. + """ + def __init__(self, parent=None): + super(Statements, self).__init__(parent) + self._trailing_whitespace = None + + # ======== Begin overridden functions + + @staticmethod + def should_parse(lists): + return isinstance(lists, list) + + def set_tabs(self, tabs=" "): + """ Sets the tabbing for this set of statements. Does this by calling `set_tabs` + on each of the child statements. + + Then, if a parent is present, sets trailing whitespace to parent tabbing. This + is so that the trailing } of any Block that contains Statements lines up + with parent tabbing. + """ + for statement in self._data: + statement.set_tabs(tabs) + if self.parent is not None: + self._trailing_whitespace = "\n" + self.parent.get_tabs() + + def parse(self, parse_this, add_spaces=False): + """ Parses a list of statements. + Expects all elements in `parse_this` to be parseable by `type(self).parsing_hooks`, + with an optional whitespace string at the last index of `parse_this`. + """ + if not isinstance(parse_this, list): + raise errors.MisconfigurationError("Statements parsing expects a list!") + # If there's a trailing whitespace in the list of statements, keep track of it. + if len(parse_this) > 0 and isinstance(parse_this[-1], six.string_types) \ + and parse_this[-1].isspace(): + self._trailing_whitespace = parse_this[-1] + parse_this = parse_this[:-1] + self._data = [parse_raw(elem, self, add_spaces) for elem in parse_this] + + def get_tabs(self): + """ Takes a guess at the tabbing of all contained Statements by retrieving the + tabbing of the first Statement.""" + if len(self._data) > 0: + return self._data[0].get_tabs() + return "" + + def dump(self, include_spaces=False): + """ Dumps this object by first dumping each statement, then appending its + trailing whitespace (if `include_spaces` is set) """ + data = super(Statements, self).dump(include_spaces) + if include_spaces and self._trailing_whitespace is not None: + return data + [self._trailing_whitespace] + return data + + def iterate(self, expanded=False, match=None): + """ Combines each statement's iterator. """ + for elem in self._data: + for sub_elem in elem.iterate(expanded, match): + yield sub_elem + + # ======== End overridden functions + +def _space_list(list_): + """ Inserts whitespace between adjacent non-whitespace tokens. """ + spaced_statement = [] # type: List[str] + for i in reversed(six.moves.xrange(len(list_))): + spaced_statement.insert(0, list_[i]) + if i > 0 and not list_[i].isspace() and not list_[i-1].isspace(): + spaced_statement.insert(0, " ") + return spaced_statement + +class Sentence(Parsable): + """ A list of words. Non-whitespace words are typically separated with whitespace tokens. """ + + # ======== Begin overridden functions + + @staticmethod + def should_parse(lists): + """ Returns True if `lists` can be parseable as a `Sentence`-- that is, + every element is a string type. + + :param list lists: The raw unparsed list to check. + + :returns: whether this lists is parseable by `Sentence`. + """ + return isinstance(lists, list) and len(lists) > 0 and \ + all([isinstance(elem, six.string_types) for elem in lists]) + + def parse(self, parse_this, add_spaces=False): + """ Parses a list of string types into this object. + If add_spaces is set, adds whitespace tokens between adjacent non-whitespace tokens.""" + if add_spaces: + parse_this = _space_list(parse_this) + if not isinstance(parse_this, list) or \ + any([not isinstance(elem, six.string_types) for elem in parse_this]): + raise errors.MisconfigurationError("Sentence parsing expects a list of string types.") + self._data = parse_this + + def iterate(self, expanded=False, match=None): + """ Simply yields itself. """ + if match is None or match(self): + yield self + + def set_tabs(self, tabs=" "): + """ Sets the tabbing on this sentence. Inserts a newline and `tabs` at the + beginning of `self._data`. """ + if self._data[0].isspace(): + return + self._data.insert(0, "\n" + tabs) + + def dump(self, include_spaces=False): + """ Dumps this sentence. If include_spaces is set, includes whitespace tokens.""" + if not include_spaces: + return self.words + return self._data + + def get_tabs(self): + """ Guesses at the tabbing of this sentence. If the first element is whitespace, + returns the whitespace after the rightmost newline in the string. """ + first = self._data[0] + if not first.isspace(): + return "" + rindex = first.rfind("\n") + return first[rindex+1:] + + # ======== End overridden functions + + @property + def words(self): + """ Iterates over words, but without spaces. Like Unspaced List. """ + return [word.strip("\"\'") for word in self._data if not word.isspace()] + + def __getitem__(self, index): + return self.words[index] + + def __contains__(self, word): + return word in self.words + +class Block(Parsable): + """ Any sort of bloc, denoted by a block name and curly braces, like so: + The parsed block: + block name { + content 1; + content 2; + } + might be represented with the list [names, contents], where + names = ["block", " ", "name", " "] + contents = [["\n ", "content", " ", "1"], ["\n ", "content", " ", "2"], "\n"] + """ + def __init__(self, parent=None): + super(Block, self).__init__(parent) + self.names = None # type: Sentence + self.contents = None # type: Block + + @staticmethod + def should_parse(lists): + """ Returns True if `lists` can be parseable as a `Block`-- that is, + it's got a length of 2, the first element is a `Sentence` and the second can be + a `Statements`. + + :param list lists: The raw unparsed list to check. + + :returns: whether this lists is parseable by `Block`. """ + return isinstance(lists, list) and len(lists) == 2 and \ + Sentence.should_parse(lists[0]) and isinstance(lists[1], list) + + def set_tabs(self, tabs=" "): + """ Sets tabs by setting equivalent tabbing on names, then adding tabbing + to contents.""" + self.names.set_tabs(tabs) + self.contents.set_tabs(tabs + " ") + + def iterate(self, expanded=False, match=None): + """ Iterator over self, and if expanded is set, over its contents. """ + if match is None or match(self): + yield self + if expanded: + for elem in self.contents.iterate(expanded, match): + yield elem + + def parse(self, parse_this, add_spaces=False): + """ Parses a list that resembles a block. + + The assumptions that this routine makes are: + 1. the first element of `parse_this` is a valid Sentence. + 2. the second element of `parse_this` is a valid Statement. + If add_spaces is set, we call it recursively on `names` and `contents`, and + add an extra trailing space to `names` (to separate the block's opening bracket + and the block name). + """ + if not Block.should_parse(parse_this): + raise errors.MisconfigurationError("Block parsing expects a list of length 2. " + "First element should be a list of string types (the bloc names), " + "and second should be another list of statements (the bloc content).") + self.names = Sentence(self) + if add_spaces: + parse_this[0].append(" ") + self.names.parse(parse_this[0], add_spaces) + self.contents = Statements(self) + self.contents.parse(parse_this[1], add_spaces) + self._data = [self.names, self.contents] + + def get_tabs(self): + """ Guesses tabbing by retrieving tabbing guess of self.names. """ + return self.names.get_tabs() + +def _is_comment(parsed_obj): + """ Checks whether parsed_obj is a comment. + + :param .Parsable parsed_obj: + + :returns: whether parsed_obj represents a comment sentence. + :rtype bool: + """ + if not isinstance(parsed_obj, Sentence): + return False + return parsed_obj.words[0] == "#" + +def _is_certbot_comment(parsed_obj): + """ Checks whether parsed_obj is a "managed by Certbot" comment. + + :param .Parsable parsed_obj: + + :returns: whether parsed_obj is a "managed by Certbot" comment. + :rtype bool: + """ + if not _is_comment(parsed_obj): + return False + if len(parsed_obj.words) != len(COMMENT_BLOCK): + return False + for i, word in enumerate(parsed_obj.words): + if word != COMMENT_BLOCK[i]: + return False + return True + +def _certbot_comment(parent, preceding_spaces=4): + """ A "Managed by Certbot" comment. + :param int preceding_spaces: Number of spaces between the end of the previous + statement and the comment. + :returns: Sentence containing the comment. + :rtype: .Sentence + """ + result = Sentence(parent) + result.parse([" " * preceding_spaces] + COMMENT_BLOCK) + return result + +def _choose_parser(parent, list_): + """ Choose a parser from type(parent).parsing_hooks, depending on whichever hook + returns True first. """ + hooks = Parsable.parsing_hooks() + if parent: + hooks = type(parent).parsing_hooks() + for type_ in hooks: + if type_.should_parse(list_): + return type_(parent) + raise errors.MisconfigurationError( + "None of the parsing hooks succeeded, so we don't know how to parse this set of lists.") + +def parse_raw(lists_, parent=None, add_spaces=False): + """ Primary parsing factory function. + + :param list lists_: raw lists from pyparsing to parse. + :param .Parent parent: The parent containing this object. + :param bool add_spaces: Whether to pass add_spaces to the parser. + + :returns .Parsable: The parsed object. + + :raises errors.MisconfigurationError: If no parsing hook passes, and we can't + determine which type to parse the raw lists into. + """ + parser = _choose_parser(parent, lists_) + parser.parse(lists_, add_spaces) + return parser diff --git a/certbot-nginx/certbot_nginx/tests/configurator_test.py b/certbot-nginx/certbot_nginx/tests/configurator_test.py index 4d23f3518..2814cbb8c 100644 --- a/certbot-nginx/certbot_nginx/tests/configurator_test.py +++ b/certbot-nginx/certbot_nginx/tests/configurator_test.py @@ -103,7 +103,7 @@ class NginxConfiguratorTest(util.NginxTest): errors.PluginError, self.config.enhance, 'myhost', 'unknown_enhancement') def test_get_chall_pref(self): - self.assertEqual([challenges.TLSSNI01, challenges.HTTP01], + self.assertEqual([challenges.HTTP01, challenges.TLSSNI01], self.config.get_chall_pref('myhost')) def test_save(self): @@ -128,22 +128,39 @@ class NginxConfiguratorTest(util.NginxTest): ['#', parser.COMMENT]]]], parsed[0]) - def test_choose_vhosts(self): - localhost_conf = set(['localhost', r'~^(www\.)?(example|bar)\.']) - server_conf = set(['somename', 'another.alias', 'alias']) - example_conf = set(['.example.com', 'example.*']) - foo_conf = set(['*.www.foo.com', '*.www.example.com']) - ipv6_conf = set(['ipv6.com']) - - results = {'localhost': localhost_conf, - 'alias': server_conf, - 'example.com': example_conf, - 'example.com.uk.test': example_conf, - 'www.example.com': example_conf, - 'test.www.example.com': foo_conf, - 'abc.www.foo.com': foo_conf, - 'www.bar.co.uk': localhost_conf, - 'ipv6.com': ipv6_conf} + def test_choose_vhosts_alias(self): + self._test_choose_vhosts_common('alias', 'server_conf') + + def test_choose_vhosts_example_com(self): + self._test_choose_vhosts_common('example.com', 'example_conf') + + def test_choose_vhosts_localhost(self): + self._test_choose_vhosts_common('localhost', 'localhost_conf') + + def test_choose_vhosts_example_com_uk_test(self): + self._test_choose_vhosts_common('example.com.uk.test', 'example_conf') + + def test_choose_vhosts_www_example_com(self): + self._test_choose_vhosts_common('www.example.com', 'example_conf') + + def test_choose_vhosts_test_www_example_com(self): + self._test_choose_vhosts_common('test.www.example.com', 'foo_conf') + + def test_choose_vhosts_abc_www_foo_com(self): + self._test_choose_vhosts_common('abc.www.foo.com', 'foo_conf') + + def test_choose_vhosts_www_bar_co_uk(self): + self._test_choose_vhosts_common('www.bar.co.uk', 'localhost_conf') + + def test_choose_vhosts_ipv6_com(self): + self._test_choose_vhosts_common('ipv6.com', 'ipv6_conf') + + def _test_choose_vhosts_common(self, name, conf): + conf_names = {'localhost_conf': set(['localhost', r'~^(www\.)?(example|bar)\.']), + 'server_conf': set(['somename', 'another.alias', 'alias']), + 'example_conf': set(['.example.com', 'example.*']), + 'foo_conf': set(['*.www.foo.com', '*.www.example.com']), + 'ipv6_conf': set(['ipv6.com'])} conf_path = {'localhost': "etc_nginx/nginx.conf", 'alias': "etc_nginx/nginx.conf", @@ -155,22 +172,22 @@ class NginxConfiguratorTest(util.NginxTest): 'www.bar.co.uk': "etc_nginx/nginx.conf", 'ipv6.com': "etc_nginx/sites-enabled/ipv6.com"} + vhost = self.config.choose_vhosts(name)[0] + path = os.path.relpath(vhost.filep, self.temp_dir) + + self.assertEqual(conf_names[conf], vhost.names) + self.assertEqual(conf_path[name], path) + # IPv6 specific checks + if name == "ipv6.com": + self.assertTrue(vhost.ipv6_enabled()) + # Make sure that we have SSL enabled also for IPv6 addr + self.assertTrue( + any([True for x in vhost.addrs if x.ssl and x.ipv6])) + + def test_choose_vhosts_bad(self): bad_results = ['www.foo.com', 'example', 't.www.bar.co', '69.255.225.155'] - for name in results: - vhost = self.config.choose_vhosts(name)[0] - path = os.path.relpath(vhost.filep, self.temp_dir) - - self.assertEqual(results[name], vhost.names) - self.assertEqual(conf_path[name], path) - # IPv6 specific checks - if name == "ipv6.com": - self.assertTrue(vhost.ipv6_enabled()) - # Make sure that we have SSL enabled also for IPv6 addr - self.assertTrue( - any([True for x in vhost.addrs if x.ssl and x.ipv6])) - for name in bad_results: self.assertRaises(errors.MisconfigurationError, self.config.choose_vhosts, name) diff --git a/certbot-nginx/certbot_nginx/tests/http_01_test.py b/certbot-nginx/certbot_nginx/tests/http_01_test.py index 0f764e92e..ed3c257ee 100644 --- a/certbot-nginx/certbot_nginx/tests/http_01_test.py +++ b/certbot-nginx/certbot_nginx/tests/http_01_test.py @@ -12,6 +12,7 @@ from certbot import achallenges from certbot.plugins import common_test from certbot.tests import acme_util +from certbot_nginx.obj import Addr from certbot_nginx.tests import util @@ -108,6 +109,41 @@ class HttpPerformTest(util.NginxTest): # self.assertEqual(vhost.addrs, set(v_addr2_print)) # self.assertEqual(vhost.names, set([response.z_domain.decode('ascii')])) + @mock.patch("certbot_nginx.configurator.NginxConfigurator.ipv6_info") + def test_default_listen_addresses_no_memoization(self, ipv6_info): + # pylint: disable=protected-access + ipv6_info.return_value = (True, True) + self.http01._default_listen_addresses() + self.assertEqual(ipv6_info.call_count, 1) + ipv6_info.return_value = (False, False) + self.http01._default_listen_addresses() + self.assertEqual(ipv6_info.call_count, 2) + + @mock.patch("certbot_nginx.configurator.NginxConfigurator.ipv6_info") + def test_default_listen_addresses_t_t(self, ipv6_info): + # pylint: disable=protected-access + ipv6_info.return_value = (True, True) + addrs = self.http01._default_listen_addresses() + http_addr = Addr.fromstring("80") + http_ipv6_addr = Addr.fromstring("[::]:80") + self.assertEqual(addrs, [http_addr, http_ipv6_addr]) + + @mock.patch("certbot_nginx.configurator.NginxConfigurator.ipv6_info") + def test_default_listen_addresses_t_f(self, ipv6_info): + # pylint: disable=protected-access + ipv6_info.return_value = (True, False) + addrs = self.http01._default_listen_addresses() + http_addr = Addr.fromstring("80") + http_ipv6_addr = Addr.fromstring("[::]:80 ipv6only=on") + self.assertEqual(addrs, [http_addr, http_ipv6_addr]) + + @mock.patch("certbot_nginx.configurator.NginxConfigurator.ipv6_info") + def test_default_listen_addresses_f_f(self, ipv6_info): + # pylint: disable=protected-access + ipv6_info.return_value = (False, False) + addrs = self.http01._default_listen_addresses() + http_addr = Addr.fromstring("80") + self.assertEqual(addrs, [http_addr]) if __name__ == "__main__": unittest.main() # pragma: no cover diff --git a/certbot-nginx/certbot_nginx/tests/parser_obj_test.py b/certbot-nginx/certbot_nginx/tests/parser_obj_test.py new file mode 100644 index 000000000..c9c9dd440 --- /dev/null +++ b/certbot-nginx/certbot_nginx/tests/parser_obj_test.py @@ -0,0 +1,253 @@ +""" Tests for functions and classes in parser_obj.py """ + +import unittest +import mock + +from certbot_nginx.parser_obj import parse_raw +from certbot_nginx.parser_obj import COMMENT_BLOCK + +class CommentHelpersTest(unittest.TestCase): + def test_is_comment(self): + from certbot_nginx.parser_obj import _is_comment + self.assertTrue(_is_comment(parse_raw(['#']))) + self.assertTrue(_is_comment(parse_raw(['#', ' literally anything else']))) + self.assertFalse(_is_comment(parse_raw(['not', 'even', 'a', 'comment']))) + + def test_is_certbot_comment(self): + from certbot_nginx.parser_obj import _is_certbot_comment + self.assertTrue(_is_certbot_comment( + parse_raw(COMMENT_BLOCK))) + self.assertFalse(_is_certbot_comment( + parse_raw(['#', ' not a certbot comment']))) + self.assertFalse(_is_certbot_comment( + parse_raw(['#', ' managed by Certbot', ' also not a certbot comment']))) + self.assertFalse(_is_certbot_comment( + parse_raw(['not', 'even', 'a', 'comment']))) + + def test_certbot_comment(self): + from certbot_nginx.parser_obj import _certbot_comment, _is_certbot_comment + comment = _certbot_comment(None) + self.assertTrue(_is_certbot_comment(comment)) + self.assertEqual(comment.dump(), COMMENT_BLOCK) + self.assertEqual(comment.dump(True), [' '] + COMMENT_BLOCK) + self.assertEqual(_certbot_comment(None, 2).dump(True), + [' '] + COMMENT_BLOCK) + +class ParsingHooksTest(unittest.TestCase): + def test_is_sentence(self): + from certbot_nginx.parser_obj import Sentence + self.assertFalse(Sentence.should_parse([])) + self.assertTrue(Sentence.should_parse([''])) + self.assertTrue(Sentence.should_parse(['word'])) + self.assertTrue(Sentence.should_parse(['two', 'words'])) + self.assertFalse(Sentence.should_parse([[]])) + self.assertFalse(Sentence.should_parse(['word', []])) + + def test_is_block(self): + from certbot_nginx.parser_obj import Block + self.assertFalse(Block.should_parse([])) + self.assertFalse(Block.should_parse([''])) + self.assertFalse(Block.should_parse(['two', 'words'])) + self.assertFalse(Block.should_parse([[[]], []])) + self.assertFalse(Block.should_parse([['block_name'], ['hi', []], []])) + self.assertFalse(Block.should_parse([['block_name'], 'lol'])) + self.assertTrue(Block.should_parse([['block_name'], ['hi', []]])) + self.assertTrue(Block.should_parse([['hello'], []])) + self.assertTrue(Block.should_parse([['block_name'], [['many'], ['statements'], 'here']])) + self.assertTrue(Block.should_parse([['if', ' ', '(whatever)'], ['hi']])) + + def test_parse_raw(self): + fake_parser1 = mock.Mock() + fake_parser1.should_parse = lambda x: True + fake_parser2 = mock.Mock() + fake_parser2.should_parse = lambda x: False + # First encountered "match" should parse. + parse_raw([]) + fake_parser1.called_once() + fake_parser2.not_called() + fake_parser1.reset_mock() + # "match" that returns False shouldn't parse. + parse_raw([]) + fake_parser1.not_called() + fake_parser2.called_once() + + @mock.patch("certbot_nginx.parser_obj.Parsable.parsing_hooks") + def test_parse_raw_no_match(self, parsing_hooks): + from certbot import errors + fake_parser1 = mock.Mock() + fake_parser1.should_parse = lambda x: False + parsing_hooks.return_value = (fake_parser1,) + self.assertRaises(errors.MisconfigurationError, parse_raw, []) + parsing_hooks.return_value = tuple() + self.assertRaises(errors.MisconfigurationError, parse_raw, []) + + def test_parse_raw_passes_add_spaces(self): + fake_parser1 = mock.Mock() + fake_parser1.should_parse = lambda x: True + parse_raw([]) + fake_parser1.parse.called_with([None]) + parse_raw([], add_spaces=True) + fake_parser1.parse.called_with([None, True]) + +class SentenceTest(unittest.TestCase): + def setUp(self): + from certbot_nginx.parser_obj import Sentence + self.sentence = Sentence(None) + + def test_parse_bad_sentence_raises_error(self): + from certbot import errors + self.assertRaises(errors.MisconfigurationError, self.sentence.parse, 'lol') + self.assertRaises(errors.MisconfigurationError, self.sentence.parse, [[]]) + self.assertRaises(errors.MisconfigurationError, self.sentence.parse, [5]) + + def test_parse_sentence_words_hides_spaces(self): + og_sentence = ['\r\n', 'hello', ' ', ' ', '\t\n ', 'lol', ' ', 'spaces'] + self.sentence.parse(og_sentence) + self.assertEquals(self.sentence.words, ['hello', 'lol', 'spaces']) + self.assertEquals(self.sentence.dump(), ['hello', 'lol', 'spaces']) + self.assertEquals(self.sentence.dump(True), og_sentence) + + def test_parse_sentence_with_add_spaces(self): + self.sentence.parse(['hi', 'there'], add_spaces=True) + self.assertEquals(self.sentence.dump(True), ['hi', ' ', 'there']) + self.sentence.parse(['one', ' ', 'space', 'none'], add_spaces=True) + self.assertEquals(self.sentence.dump(True), ['one', ' ', 'space', ' ', 'none']) + + def test_iterate(self): + expected = [['1', '2', '3']] + self.sentence.parse(['1', ' ', '2', ' ', '3']) + for i, sentence in enumerate(self.sentence.iterate()): + self.assertEquals(sentence.dump(), expected[i]) + + def test_set_tabs(self): + self.sentence.parse(['tabs', 'pls'], add_spaces=True) + self.sentence.set_tabs() + self.assertEquals(self.sentence.dump(True)[0], '\n ') + self.sentence.parse(['tabs', 'pls'], add_spaces=True) + + def test_get_tabs(self): + self.sentence.parse(['no', 'tabs']) + self.assertEquals(self.sentence.get_tabs(), '') + self.sentence.parse(['\n \n ', 'tabs']) + self.assertEquals(self.sentence.get_tabs(), ' ') + self.sentence.parse(['\n\t ', 'tabs']) + self.assertEquals(self.sentence.get_tabs(), '\t ') + self.sentence.parse(['\n\t \n', 'tabs']) + self.assertEquals(self.sentence.get_tabs(), '') + +class BlockTest(unittest.TestCase): + def setUp(self): + from certbot_nginx.parser_obj import Block + self.bloc = Block(None) + self.name = ['server', 'name'] + self.contents = [['thing', '1'], ['thing', '2'], ['another', 'one']] + self.bloc.parse([self.name, self.contents]) + + def test_iterate(self): + # Iterates itself normally + self.assertEquals(self.bloc, next(self.bloc.iterate())) + # Iterates contents while expanded + expected = [self.bloc.dump()] + self.contents + for i, elem in enumerate(self.bloc.iterate(expanded=True)): + self.assertEquals(expected[i], elem.dump()) + + def test_iterate_match(self): + # can match on contents while expanded + from certbot_nginx.parser_obj import Block, Sentence + expected = [['thing', '1'], ['thing', '2']] + for i, elem in enumerate(self.bloc.iterate(expanded=True, + match=lambda x: isinstance(x, Sentence) and 'thing' in x.words)): + self.assertEquals(expected[i], elem.dump()) + # can match on self + self.assertEquals(self.bloc, next(self.bloc.iterate( + expanded=True, + match=lambda x: isinstance(x, Block) and 'server' in x.names))) + + def test_parse_with_added_spaces(self): + import copy + self.bloc.parse([copy.copy(self.name), self.contents], add_spaces=True) + self.assertEquals(self.bloc.dump(), [self.name, self.contents]) + self.assertEquals(self.bloc.dump(True), [ + ['server', ' ', 'name', ' '], + [['thing', ' ', '1'], + ['thing', ' ', '2'], + ['another', ' ', 'one']]]) + + def test_bad_parse_raises_error(self): + from certbot import errors + self.assertRaises(errors.MisconfigurationError, self.bloc.parse, [[[]], [[]]]) + self.assertRaises(errors.MisconfigurationError, self.bloc.parse, ['lol']) + self.assertRaises(errors.MisconfigurationError, self.bloc.parse, ['fake', 'news']) + + def test_set_tabs(self): + self.bloc.set_tabs() + self.assertEquals(self.bloc.names.dump(True)[0], '\n ') + for elem in self.bloc.contents.dump(True)[:-1]: + self.assertEquals(elem[0], '\n ') + self.assertEquals(self.bloc.contents.dump(True)[-1][0], '\n') + + def test_get_tabs(self): + self.bloc.parse([[' \n \t', 'lol'], []]) + self.assertEquals(self.bloc.get_tabs(), ' \t') + +class StatementsTest(unittest.TestCase): + def setUp(self): + from certbot_nginx.parser_obj import Statements + self.statements = Statements(None) + self.raw = [ + ['sentence', 'one'], + ['sentence', 'two'], + ['and', 'another'] + ] + self.raw_spaced = [ + ['\n ', 'sentence', ' ', 'one'], + ['\n ', 'sentence', ' ', 'two'], + ['\n ', 'and', ' ', 'another'], + '\n\n' + ] + + def test_set_tabs(self): + self.statements.parse(self.raw) + self.statements.set_tabs() + for statement in self.statements.iterate(): + self.assertEquals(statement.dump(True)[0], '\n ') + + def test_set_tabs_with_parent(self): + # Trailing whitespace should inherit from parent tabbing. + self.statements.parse(self.raw) + self.statements.parent = mock.Mock() + self.statements.parent.get_tabs.return_value = '\t\t' + self.statements.set_tabs() + for statement in self.statements.iterate(): + self.assertEquals(statement.dump(True)[0], '\n ') + self.assertEquals(self.statements.dump(True)[-1], '\n\t\t') + + def test_get_tabs(self): + self.raw[0].insert(0, '\n \n \t') + self.statements.parse(self.raw) + self.assertEquals(self.statements.get_tabs(), ' \t') + self.statements.parse([]) + self.assertEquals(self.statements.get_tabs(), '') + + def test_parse_with_added_spaces(self): + self.statements.parse(self.raw, add_spaces=True) + self.assertEquals(self.statements.dump(True)[0], ['sentence', ' ', 'one']) + + def test_parse_bad_list_raises_error(self): + from certbot import errors + self.assertRaises(errors.MisconfigurationError, self.statements.parse, 'lol not a list') + + def test_parse_hides_trailing_whitespace(self): + self.statements.parse(self.raw + ['\n\n ']) + self.assertTrue(isinstance(self.statements.dump()[-1], list)) + self.assertTrue(self.statements.dump(True)[-1].isspace()) + self.assertEquals(self.statements.dump(True)[-1], '\n\n ') + + def test_iterate(self): + self.statements.parse(self.raw) + expected = [['sentence', 'one'], ['sentence', 'two']] + for i, elem in enumerate(self.statements.iterate(match=lambda x: 'sentence' in x)): + self.assertEquals(expected[i], elem.dump()) + +if __name__ == "__main__": + unittest.main() # pragma: no cover diff --git a/certbot-nginx/certbot_nginx/tls_sni_01.py b/certbot-nginx/certbot_nginx/tls_sni_01.py index 0fd37e0cb..60ec1ed1a 100644 --- a/certbot-nginx/certbot_nginx/tls_sni_01.py +++ b/certbot-nginx/certbot_nginx/tls_sni_01.py @@ -51,9 +51,6 @@ class NginxTlsSni01(common.TLSSNI01): default_addr = "{0} ssl".format( self.configurator.config.tls_sni_01_port) - ipv6, ipv6only = self.configurator.ipv6_info( - self.configurator.config.tls_sni_01_port) - for achall in self.achalls: vhosts = self.configurator.choose_vhosts(achall.domain, create_if_no_match=True) @@ -61,6 +58,9 @@ class NginxTlsSni01(common.TLSSNI01): if vhosts and vhosts[0].addrs: addresses.append(list(vhosts[0].addrs)) else: + # choose_vhosts might have modified vhosts, so put this after + ipv6, ipv6only = self.configurator.ipv6_info( + self.configurator.config.tls_sni_01_port) if ipv6: # If IPv6 is active in Nginx configuration ipv6_addr = "[::]:{0} ssl".format( @@ -141,6 +141,8 @@ class NginxTlsSni01(common.TLSSNI01): with open(self.challenge_conf, "w") as new_conf: nginxparser.dump(config, new_conf) + logger.debug("Generated server block:\n%s", str(config)) + def _make_server_block(self, achall, addrs): """Creates a server block for a challenge. |