Unittests and revisions

This commit is contained in:
James Kasten
2015-07-23 01:34:51 -07:00
parent 8b96264576
commit c927149185
9 changed files with 316 additions and 48 deletions
@@ -117,7 +117,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
# Verify that all directories and files exist with proper permissions
if os.geteuid() == 0:
self.verify_setup()
self.verify_setup() # pragma: no cover
# Add name_server association dict
self.assoc = dict()
@@ -147,7 +147,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
# Set Version
if self.version is None:
self.version = self.get_version()
self.version = self.get_version() # pragma: no cover
# Get all of the available vhosts
self.vhosts = self.get_virtual_hosts()
@@ -265,6 +265,9 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
def _find_best_vhost(self, target_name):
"""Finds the best vhost for a target_name.
This does not upgrade a vhost to HTTPS... it only finds the most
appropriate vhost for the given target_name.
:returns: VHost or None
"""
@@ -281,6 +284,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
elif any(addr.get_addr() == target_name for addr in vhost.addrs):
points = 1
else:
# No points given if names can't be found.
continue
if vhost.ssl:
@@ -290,8 +294,21 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
best_points = points
best_candidate = vhost
# No winners here... is there only one reasonable vhost?
if best_candidate is None:
# reasonable == Not all _default_ addrs
reasonable_vhosts = self._non_default_vhosts()
if len(reasonable_vhosts) == 1:
best_candidate = reasonable_vhosts[0]
return best_candidate
def _non_default_vhosts(self):
"""Return all non _default_ only vhosts."""
return [vh for vh in self.vhosts if not all(
addr.get_addr() == "_default_" for addr in vh.addrs
)]
def create_dn_server_assoc(self, domain, vhost):
"""Create an association between a domain name and virtual host.
@@ -887,17 +904,13 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
key_path = self.parser.find_dir(
"SSLCertificateKeyFile", None, vhost.path)
# Can be removed once find directive can return ordered results
if len(cert_path) != 1 or len(key_path) != 1:
logger.error("Too many cert or key directives in vhost %s",
vhost.filep)
errors.MisconfigurationError(
"Too many cert/key directives in vhost")
cert = os.path.abspath(self.parser.get_arg(cert_path[0]))
key = os.path.abspath(self.parser.get_arg(key_path[0]))
c_k.add((cert, key, get_file_path(cert_path[0])))
if cert_path and key_path:
cert = os.path.abspath(self.parser.get_arg(cert_path[-1]))
key = os.path.abspath(self.parser.get_arg(key_path[-1]))
c_k.add((cert, key, get_file_path(cert_path[-1])))
else:
logger.warning(
"Invalid VirtualHost configuration - %s", vhost.filep)
return c_k
def is_site_enabled(self, avail_fp):
@@ -117,7 +117,6 @@ class ApacheDvsni(common.Dvsni):
default_addr = obj.Addr(("*", self.configurator.config.dvsni_port))
for addr in vhost.addrs:
# I don't think there can be two _default_ namebasedvhosts
if "_default_" == addr.get_addr():
dvsni_addrs.add(default_addr)
else:
+51 -12
View File
@@ -21,6 +21,44 @@ class Addr(common.Addr):
def __ne__(self, other):
return not self.__eq__(other)
def _addr_less_specific(self, addr):
"""Returns if addr.get_addr() is more specific than self.get_addr()."""
return addr._rank_specific_addr() > self._rank_specific_addr()
def _rank_specific_addr(self):
"""Returns numerical rank for get_addr()"""
if self.get_addr() == "_default_":
return 0
elif self.get_addr() == "*":
return 1
else:
return 2
def conflicts(self, addr):
"""Returns if address could conflict with correct function of self.
Could addr take away service provided by self within Apache?
.. note::IP Address is more important than wildcard.
Connection from 127.0.0.1:80 with choices of *:80 and 127.0.0.1:*
chooses 127.0.0.1:*
.. todo:: Handle domain name addrs...
Examples:
127.0.0.1:*.conflicts(127.0.0.1:443) - True
127.0.0.1:443.conflicts(127.0.0.1:*) - False
*:443.conflicts(*:80) - False
_default_:443.conflicts(*:443) - True
"""
if self._addr_less_specific(addr):
return True
elif self.get_addr() == addr.get_addr():
if self.is_wildcard() or self.get_port() == addr.get_port():
return True
return False
def is_wildcard(self):
"""Returns if address has a wildcard port."""
return self.tup[1] == "*" or not self.tup[1]
@@ -55,7 +93,9 @@ class VirtualHost(object): # pylint: disable=too-few-public-methods
:ivar bool ssl: SSLEngine on in vhost
:ivar bool enabled: Virtual host is enabled
.. todo:: Handle ServerNames appropriately...
https://httpd.apache.org/docs/2.4/vhosts/details.html
.. todo:: Any vhost that includes the magic _default_ wildcard is given the
same ServerName as the main server.
"""
# ?: is used for not returning enclosed characters
@@ -124,16 +164,14 @@ class VirtualHost(object): # pylint: disable=too-few-public-methods
:param addrs: Iterable Addresses
:type addrs: Iterable :class:~obj.Addr
:returns: If addresses conflict with vhost
:returns: If addresses conflicts with vhost
:rtype: bool
"""
# TODO: Handle domain name addrs...
for addr in addrs:
if (addr in self.addrs or addr.get_addr_obj("") in self.addrs or
addr.get_addr_obj("*") in self.addrs):
return True
for pot_addr in addrs:
for addr in self.addrs:
if addr.conflicts(pot_addr):
return True
return False
def same_server(self, vhost):
@@ -150,7 +188,7 @@ class VirtualHost(object): # pylint: disable=too-few-public-methods
return False
# If equal and set is not empty... assume same server
if self.name is not None:
if self.name is not None or self.aliases:
return True
# Both sets of names are empty.
@@ -167,11 +205,12 @@ class VirtualHost(object): # pylint: disable=too-few-public-methods
for addr in vhost.addrs:
for local_addr in self.addrs:
if (local_addr.get_addr() == addr.get_addr() and
local_addr != addr and
local_addr.get_addr() not in already_found):
local_addr != addr and
local_addr.get_addr() not in already_found):
# This intends to make sure we aren't double counting...
# e.g. 127.0.0.1:*
# e.g. 127.0.0.1:* - We require same number of addrs
# currently
already_found.add(local_addr.get_addr())
break
else:
@@ -19,12 +19,18 @@ class ApacheParser(object):
:ivar str root: Normalized absolute path to the server root
directory. Without trailing slash.
:ivar str root: Server root
:ivar set modules: All module names that are currently enabled.
:ivar dict loc: Location to place directives, root - configuration origin,
default - user config file, name - NameVirtualHost,
"""
arg_var_interpreter = re.compile(r"\$\{[^ \}]*}")
fnmatch_chars = set(["*", "?", "\\", "[", "]"])
def __init__(self, aug, root, ctl):
# Note: Order is important here.
# This uses the binary, so it can be done first.
# https://httpd.apache.org/docs/2.4/mod/core.html#define
# https://httpd.apache.org/docs/2.4/mod/core.html#ifdefine
@@ -47,7 +53,8 @@ class ApacheParser(object):
self.modules = set()
self._init_modules()
self.loc.update(self._set_locations(self.loc["root"]))
# Set up rest of locations
self.loc.update(self._set_locations())
# Must also attempt to parse sites-available or equivalent
# Sites-available is not included naturally in configuration
@@ -89,7 +96,10 @@ class ApacheParser(object):
variables = dict()
matches = re.compile(r"Define: ([^ \n]*)").findall(stdout)
matches.remove("DUMP_RUN_CFG")
try:
matches.remove("DUMP_RUN_CFG")
except ValueError:
raise errors.PluginError("Unable to parse runtime variables")
for match in matches:
if match.count("=") > 1:
@@ -183,7 +193,7 @@ class ApacheParser(object):
self.aug.set(nvh_path + "/arg", args[0])
else:
for i, arg in enumerate(args):
self.aug.set("%s/arg[%d]" % (nvh_path, i), arg)
self.aug.set("%s/arg[%d]" % (nvh_path, i+1), arg)
def _get_ifmod(self, aug_conf_path, mod):
@@ -497,14 +507,14 @@ class ApacheParser(object):
self.aug.load()
def _set_locations(self, root):
def _set_locations(self):
"""Set default location for directives.
Locations are given as file_paths
.. todo:: Make sure that files are included
"""
default = self._set_user_config_file(root)
default = self._set_user_config_file()
temp = os.path.join(self.root, "ports.conf")
if os.path.isfile(temp):
@@ -526,7 +536,7 @@ class ApacheParser(object):
raise errors.NoInstallationError("Could not find configuration root")
def _set_user_config_file(self, root):
def _set_user_config_file(self):
"""Set the appropriate user configuration file
.. todo:: This will have to be updated for other distros versions
@@ -538,7 +548,7 @@ class ApacheParser(object):
# in hierarchy via direct include
# httpd.conf was very common as a user file in Apache 2.2
if (os.path.isfile(os.path.join(self.root, "httpd.conf")) and
self.find_dir("Include", "httpd.conf", root)):
self.find_dir("Include", "httpd.conf", self.loc["root"])):
return os.path.join(self.root, "httpd.conf")
else:
return os.path.join(self.root, "apache2.conf")
@@ -3,6 +3,8 @@ import os
import shutil
import unittest
from letsencrypt import errors
from letsencrypt_apache.tests import util
@@ -46,6 +48,14 @@ class ComplexParserTest(util.ParserTest):
self.assertEqual(len(matches), 1)
self.assertEqual(self.parser.get_arg(matches[0]), "1234")
def test_invalid_variable_parsing(self):
del self.parser.variables["tls_port"]
matches = self.parser.find_dir("TestVariablePort")
self.assertRaises(
errors.PluginError, self.parser.get_arg, matches[0])
def test_basic_ifdefine(self):
self.assertEqual(len(self.parser.find_dir("VAR_DIRECTIVE")), 2)
self.assertEqual(len(self.parser.find_dir("INVALID_VAR_DIRECTIVE")), 0)
@@ -36,6 +36,11 @@ class TwoVhost80Test(util.ApacheTest):
shutil.rmtree(self.config_dir)
shutil.rmtree(self.work_dir)
def test_add_parser_arguments(self):
from letsencrypt_apache.configurator import ApacheConfigurator
# Weak test..
ApacheConfigurator.add_parser_arguments(mock.MagicMock())
def test_get_all_names(self):
names = self.config.get_all_names()
self.assertEqual(names, set(
@@ -58,10 +63,44 @@ class TwoVhost80Test(util.ApacheTest):
found += 1
break
else:
raise Exception("Missed: %s" % vhost)
raise Exception("Missed: %s" % vhost) # pragma: no cover
self.assertEqual(found, 4)
@mock.patch("letsencrypt_apache.display_ops.select_vhost")
def test_choose_vhost_none_avail(self, mock_select):
mock_select.return_value = None
self.assertRaises(
errors.PluginError, self.config.choose_vhost, "none.com")
@mock.patch("letsencrypt_apache.display_ops.select_vhost")
def test_choose_vhost_select_vhost(self, mock_select):
mock_select.return_value = self.vh_truth[3]
self.assertEqual(
self.vh_truth[3], self.config.choose_vhost("none.com"))
def test_find_best_vhost(self):
self.assertEqual(
self.vh_truth[3], self.config._find_best_vhost("letsencrypt.demo"))
self.assertEqual(
self.vh_truth[0],
self.config._find_best_vhost("encryption-example.demo"))
self.assertTrue(
self.config._find_best_vhost("does-not-exist.com") is None)
def test_find_best_vhost_default(self):
# Assume only the two default vhosts.
self.config.vhosts = [vh for vh in self.config.vhosts
if vh.name not in
["letsencrypt.demo", "encryption-example.demo"]]
self.assertEqual(
self.config._find_best_vhost("example.demo"), self.vh_truth[2])
def test_non_default_vhosts(self):
# pylint: disable=protected-access
self.assertEqual(len(self.config._non_default_vhosts()), 3)
def test_is_site_enabled(self):
"""Test if site is enabled.
@@ -137,6 +176,14 @@ class TwoVhost80Test(util.ApacheTest):
self.assertEqual(configurator.get_file_path(loc_chain[0]),
self.vh_truth[1].filep)
# One more time for chain directive setting
self.config.deploy_cert(
"random.demo",
"two/cert.pem", "two/key.pem", "two/cert_chain.pem")
self.assertTrue(self.config.parser.find_dir(
"SSLCertificateChainFile", "two/cert_chain.pem",
self.vh_truth[1].path))
def test_deploy_cert_invalid_vhost(self):
self.config.parser.modules.add("ssl_module")
mock_find = mock.MagicMock()
@@ -1,27 +1,73 @@
"""Tests for letsencrypt_apache.obj."""
import unittest
from letsencrypt.plugins import common
class VirtualHostTest(unittest.TestCase):
"""Test the VirtualHost class."""
def setUp(self):
from letsencrypt_apache.obj import Addr
from letsencrypt_apache.obj import VirtualHost
self.addr1 = Addr.fromstring("127.0.0.1")
self.addr2 = Addr.fromstring("127.0.0.1:443")
self.addr_default = Addr.fromstring("_default_:443")
self.vhost1 = VirtualHost(
"filep", "vh_path",
set([common.Addr.fromstring("localhost")]), False, False)
"filep", "vh_path", set([self.addr1]), False, False, "localhost")
self.vhost1b = VirtualHost(
"filep", "vh_path", set([self.addr1]), False, False, "localhost")
self.vhost2 = VirtualHost(
"fp", "vhp", set([self.addr2]), False, False, "localhost")
def test_eq(self):
from letsencrypt_apache.obj import VirtualHost
vhost1b = VirtualHost(
"filep", "vh_path",
set([common.Addr.fromstring("localhost")]), False, False)
self.assertTrue(self.vhost1b == self.vhost1)
self.assertFalse(self.vhost1 == self.vhost2)
self.assertEqual(str(self.vhost1b), str(self.vhost1))
self.assertFalse(self.vhost1b == 1234)
self.assertEqual(vhost1b, self.vhost1)
self.assertEqual(str(vhost1b), str(self.vhost1))
self.assertFalse(vhost1b == 1234)
def test_ne(self):
self.assertTrue(self.vhost1 != self.vhost2)
self.assertFalse(self.vhost1 != self.vhost1b)
def test_conflicts(self):
from letsencrypt_apache.obj import Addr
from letsencrypt_apache.obj import VirtualHost
complex_vh = VirtualHost(
"fp", "vhp",
set([Addr.fromstring("*:443"), Addr.fromstring("1.2.3.4:443")]),
False, False)
self.assertTrue(complex_vh.conflicts([self.addr1]))
self.assertTrue(complex_vh.conflicts([self.addr2]))
self.assertFalse(complex_vh.conflicts([self.addr_default]))
self.assertTrue(self.vhost1.conflicts([self.addr2]))
self.assertFalse(self.vhost1.conflicts([self.addr_default]))
self.assertFalse(self.vhost2.conflicts([self.addr1, self.addr_default]))
def test_same_server(self):
from letsencrypt_apache.obj import VirtualHost
no_name1 = VirtualHost(
"fp", "vhp", set([self.addr1]), False, False, None)
no_name2 = VirtualHost(
"fp", "vhp", set([self.addr2]), False, False, None)
no_name3 = VirtualHost(
"fp", "vhp", set([self.addr_default]),
False, False, None)
no_name4 = VirtualHost(
"fp", "vhp", set([self.addr2, self.addr_default]),
False, False, None)
self.assertTrue(self.vhost1.same_server(self.vhost2))
self.assertTrue(no_name1.same_server(no_name2))
self.assertFalse(self.vhost1.same_server(no_name1))
self.assertFalse(no_name1.same_server(no_name3))
self.assertFalse(no_name1.same_server(no_name4))
class AddrTest(unittest.TestCase):
@@ -33,6 +79,9 @@ class AddrTest(unittest.TestCase):
self.addr1 = Addr.fromstring("127.0.0.1")
self.addr2 = Addr.fromstring("127.0.0.1:*")
self.addr_defined = Addr.fromstring("127.0.0.1:443")
self.addr_default = Addr.fromstring("_default_:443")
def test_wildcard(self):
self.assertFalse(self.addr.is_wildcard())
self.assertTrue(self.addr1.is_wildcard())
@@ -47,9 +96,36 @@ class AddrTest(unittest.TestCase):
self.assertEqual(
self.addr1.get_sni_addr("443"), Addr.fromstring("127.0.0.1"))
def test_conflicts(self):
# Note: Defined IP is more important than defined port in match
self.assertTrue(self.addr.conflicts(self.addr1))
self.assertTrue(self.addr.conflicts(self.addr2))
self.assertTrue(self.addr.conflicts(self.addr_defined))
self.assertFalse(self.addr.conflicts(self.addr_default))
self.assertFalse(self.addr1.conflicts(self.addr))
self.assertTrue(self.addr1.conflicts(self.addr_defined))
self.assertFalse(self.addr1.conflicts(self.addr_default))
self.assertFalse(self.addr_defined.conflicts(self.addr1))
self.assertFalse(self.addr_defined.conflicts(self.addr2))
self.assertFalse(self.addr_defined.conflicts(self.addr))
self.assertFalse(self.addr_defined.conflicts(self.addr_default))
self.assertTrue(self.addr_default.conflicts(self.addr))
self.assertTrue(self.addr_default.conflicts(self.addr1))
self.assertTrue(self.addr_default.conflicts(self.addr_defined))
# Self test
self.assertTrue(self.addr.conflicts(self.addr))
self.assertTrue(self.addr1.conflicts(self.addr1))
# This is a tricky one...
self.assertTrue(self.addr1.conflicts(self.addr2))
def test_equal(self):
self.assertTrue(self.addr1 == self.addr2)
self.assertFalse(self.addr == self.addr1)
self.assertFalse(self.addr == 123)
def test_not_equal(self):
self.assertFalse(self.addr1 != self.addr2)
@@ -6,6 +6,8 @@ import unittest
import augeas
import mock
from letsencrypt import errors
from letsencrypt_apache.tests import util
@@ -77,6 +79,20 @@ class BasicParserTest(util.ParserTest):
self.assertEqual(len(matches), 1)
self.assertTrue("IfModule" in matches[0])
def test_add_dir_to_ifmodssl_multiple(self):
from letsencrypt_apache.parser import get_aug_path
# This makes sure that find_dir will work
self.parser.modules.add("mod_ssl.c")
self.parser.add_dir_to_ifmodssl(
get_aug_path(self.parser.loc["default"]),
"FakeDirective", ["123", "456", "789"])
matches = self.parser.find_dir("FakeDirective")
self.assertEqual(len(matches), 3)
self.assertTrue("IfModule" in matches[0])
def test_get_aug_path(self):
from letsencrypt_apache.parser import get_aug_path
self.assertEqual("/files/etc/apache", get_aug_path("/etc/apache"))
@@ -87,11 +103,69 @@ class BasicParserTest(util.ParserTest):
mock_path.isfile.side_effect = [True, False, False]
# pylint: disable=protected-access
results = self.parser._set_locations("root")
results = self.parser._set_locations()
self.assertEqual(results["default"], results["listen"])
self.assertEqual(results["default"], results["name"])
def test_set_user_config_file(self):
path = os.path.join(self.parser.root, "httpd.conf")
open(path, 'w').close()
self.parser.add_dir(self.parser.loc["default"], "Include", "httpd.conf")
self.assertEqual(
path, self.parser._set_user_config_file())
@mock.patch("letsencrypt_apache.parser.ApacheParser._get_runtime_cfg")
def test_update_runtime_variables(self, mock_cfg):
mock_cfg.return_value = (
'ServerRoot: "/etc/apache2"\n'
'Main DocumentRoot: "/var/www"\n'
'Main ErrorLog: "/var/log/apache2/error.log"\n'
'Mutex ssl-stapling: using_defaults\n'
'Mutex ssl-cache: using_defaults\n'
'Mutex default: dir="/var/lock/apache2" mechanism=fcntl\n'
'Mutex watchdog-callback: using_defaults\n'
'PidFile: "/var/run/apache2/apache2.pid"\n'
'Define: TEST\n'
'Define: DUMP_RUN_CFG\n'
'Define: U_MICH\n'
'Define: TLS=443\n'
'Define: example_path=Documents/path\n'
'User: name="www-data" id=33 not_used\n'
'Group: name="www-data" id=33 not_used\n'
)
expected_vars = {"TEST": "", "U_MICH": "", "TLS": "443",
"example_path":"Documents/path"}
self.parser.update_runtime_variables("ctl")
self.assertEqual(self.parser.variables, expected_vars)
@mock.patch("letsencrypt_apache.parser.ApacheParser._get_runtime_cfg")
def test_update_runtime_vars_bad_output(self, mock_cfg):
mock_cfg.return_value = "Define: TLS=443=24"
self.assertRaises(
errors.PluginError, self.parser.update_runtime_variables, "ctl")
mock_cfg.return_value = "Define: DUMP_RUN_CFG\nDefine: TLS=443=24"
self.assertRaises(
errors.PluginError, self.parser.update_runtime_variables, "ctl")
@mock.patch("letsencrypt_apache.parser.subprocess.Popen")
def test_update_runtime_vars_bad_ctl(self, mock_popen):
mock_popen.side_effect = OSError
self.assertRaises(
errors.MisconfigurationError,
self.parser.update_runtime_variables, "ctl")
@mock.patch("letsencrypt_apache.parser.subprocess.Popen")
def test_update_runtime_vars_bad_exit(self, mock_popen):
mock_popen().communicate.return_value = ("", "")
mock_popen.returncode = -1
self.assertRaises(
errors.MisconfigurationError,
self.parser.update_runtime_variables, "ctl")
class ParserInitTest(util.ApacheTest):
def setUp(self): # pylint: disable=arguments-differ
@@ -87,7 +87,7 @@ def get_apache_configurator(
version=version)
# This allows testing scripts to set it a bit more quickly
if conf is not None:
config.conf = conf
config.conf = conf # pragma: no cover
config.prepare()
@@ -123,4 +123,4 @@ def get_vh_truth(temp_dir, config_name):
]
return vh_truth
return None
return None # pragma: no cover