starred/certbot-certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-31 10:26:06 +02:00
Code Issues Packages Projects Releases Wiki Activity

More unit tests... better configurator

Browse Source
This commit is contained in:
James Kasten
2014-12-05 17:31:36 -08:00
parent 48ae79e8c4
commit 779dd755ca
2 changed files with 145 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
letsencrypt/client/apache_configurator.py
+81 -58
View File
@@ -116,13 +116,11 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
and the typical directories are parsed by the Augeas configuration
parser automatically.
.. todo:: Add support for config file variables Define rootDir /var/www/
The API of this class will change in the coming weeks as the exact
needs of client's are clarified with the new and developing protocol.
This class will eventually derive from a generic Configurator class
so that other Configurators (like Nginx) can be developed and interoperate
with the client.
:ivar str server_root: Path to Apache root directory
:ivar float version: version of Apache
@@ -180,9 +178,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
# TODO: attempt to make the check faster... this enable should
# be asynchronous as it shouldn't be that time sensitive
# on initialization
if not check_ssl_loaded():
logger.info("Loading mod_ssl into Apache Server")
enable_mod("ssl")
self._prepare_server_https()
# Note: initialization doesn't check to see if the config is correct
# by Apache's standards. This should be done by the client (client.py)
@@ -238,6 +234,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
return False
logger.info("Deploying Certificate to VirtualHost %s" % vhost.file)
print path
self.aug.set(path["cert_file"][0], cert)
self.aug.set(path["cert_key"][0], key)
@@ -443,7 +440,9 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
def is_name_vhost(self, target_addr):
"""Returns if vhost is a name based vhost
Checks if addr has a NameVirtualHost directive in the Apache config
NameVirtualHost was deprecated in Apache 2.4 as all VirtualHosts are
now NameVirtualHosts. If version is earlier than 2.4, check if addr
has a NameVirtualHost directive in the Apache config
:param str addr: vhost address ie. \*:443
@@ -451,49 +450,67 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
:rtype: bool
"""
# search for NameVirtualHost directive for ip_addr
# check httpd.conf, ports.conf,
# note ip_addr can be FQDN although Apache does not recommend it
paths = self.find_directive(case_i("NameVirtualHost"), None)
name_vh = []
for path in paths:
name_vh.append(self.aug.get(path))
# Mixed and matched wildcard NameVirtualHost with VirtualHost
# behavior is undefined. Make sure that an exact match exists
# Check for exact match
for addr in name_vh:
if addr == target_addr:
return True
# search for NameVirtualHost directive for ip_addr
# check httpd.conf, ports.conf,
# note ip_addr can be FQDN although Apache does not recommend it
if (self.version >= (2, 4) or
self.find_directive(
case_i("NameVirtualHost"), case_i(target_addr))):
return True
return False
def add_name_vhost(self, addr):
"""Adds NameVirtualHost directive for given address.
Directive is added to ports.conf unless the file doesn't exist
It is added to httpd.conf as a backup
It is added to user_config_file as a backup
:param str addr: Address that will be added as NameVirtualHost directive
"""
aug_file_path = "/files%sports.conf" % self.server_root
self.add_dir_to_ifmodssl(aug_file_path, "NameVirtualHost", addr)
aug_path = "/files%sports.conf" % self.server_root
path = self.add_gen_ssl_dir_config(aug_path, "NameVirtualHost", addr)
# TODO: Check to see if len(find_dir) can just be if find_dir()
if len(self.find_directive(
case_i("NameVirtualHost"), case_i(addr))) == 0:
logger.warn("ports.conf is not included in your Apache config...")
logger.warn("Adding NameVirtualHost directive to httpd.conf")
self.save_notes += "Setting %s to be NameBasedVirtualHost\n" % addr
self.save_notes += "\tDirective added to %s\n" % path
self.add_dir_to_ifmodssl("/files%shttpd.conf" % self.server_root,
"NameVirtualHost",
addr)
def add_gen_ssl_dir_config(self, aug_path, directive, val):
"""Adds directive to ifmodssl somewhere in config.
self.save_notes += 'Setting %s to be NameBasedVirtualHost\n' % addr
First the function tries to place the ssl directive along the
aug_path, if that is not available it places the directive
in the self.user_config_file which is expected to be valid.
def add_dir_to_ifmodssl(self, aug_conf_path, directive, val):
:param str aug_path: Augeas configuration path
:param str directive: Directive you would like to add to config
:param str val: Value of the directive
:returns: Path to file that directive was added
:rtype: str
"""
# First try original path
self._add_dir_to_ifmodssl(
aug_path, re.escape(directive), re.escape(val))
if self.find_directive(case_i(directive), case_i(val), start=aug_path):
return get_file_path(aug_path)
# Try default path
logger.debug(
"%s is not included in your Apache config..." %
os.path.basename(get_file_path(aug_path)))
logger.debug(
"Adding %s directive to %s" % (directive, self.user_config_file))
self._add_dir_to_ifmodssl(
"/files%s" % self.user_config_file, directive, val)
return self.user_config_file
def _add_dir_to_ifmodssl(self, aug_conf_path, directive, val):
"""Adds directive and value to IfMod ssl block.
Adds given directive and value along configuration path within
@@ -507,13 +524,34 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
"""
# TODO: Add error checking code... does the path given even exist?
# Does it throw exceptions?
if_mod_path = self.get_ifmod(aug_conf_path, "mod_ssl.c")
if_mod_path = self._get_ifmod(aug_conf_path, "mod_ssl.c")
# IfModule can have only one valid argument, so append after
self.aug.insert(if_mod_path + "arg", "directive", False)
nvh_path = if_mod_path + "directive[1]"
self.aug.set(nvh_path, directive)
self.aug.set(nvh_path + "/arg", val)
def _prepare_server_https(self):
"""Prepare the server for HTTPS.
Make sure that the ssl_module is loaded and that the server
is appropriately listening on port 443.
"""
if not check_ssl_loaded():
logger.info("Loading mod_ssl into Apache Server")
enable_mod("ssl")
# Check for Listen 443
# Note: This could be made to also look for ip:443 combo
# TODO: Need to search only open directives and IfMod mod_ssl.c
if len(self.find_directive(case_i("Listen"), "443")) == 0:
logger.debug("No Listen 443 directive found")
logger.debug("Setting the Apache Server to Listen on port 443")
path = self.add_gen_ssl_dir_config(
"/files%sports.conf" % self.server_root, "Listen", "443")
self.save_notes += "Added Listen 443 directive to %s\n" % path
def make_server_sni_ready(self, vhost, default_addr="*:443"):
"""Checks to see if the server is ready for SNI challenges.
@@ -525,21 +563,6 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
:param str default_addr: TODO - investigate function further
"""
# Check if mod_ssl is loaded
if not check_ssl_loaded():
logger.error("Please load the SSL module with Apache")
return False
# Check for Listen 443
# TODO: This could be made to also look for ip:443 combo
# TODO: Need to search only open directives and IfMod mod_ssl.c
if len(self.find_directive(case_i("Listen"), "443")) == 0:
logger.debug("No Listen 443 directive found")
logger.debug("Setting the Apache Server to Listen on port 443")
self.add_dir_to_ifmodssl("/files%sports.conf" % self.server_root,
"Listen", "443")
self.save_notes += "Added Listen 443 directive to ports.conf\n"
# Check for NameVirtualHost
# First see if any of the vhost addresses is a _default_ addr
for addr in vhost.addrs:
@@ -560,7 +583,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
return True
def get_ifmod(self, aug_conf_path, mod):
def _get_ifmod(self, aug_conf_path, mod):
"""Returns the path to <IfMod mod> and creates one if it doesn't exist.
:param str aug_conf_path: Augeas configuration path
@@ -586,7 +609,6 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
:param str directive: Directive to add
:param str arg: Value of the directive. ie. Listen 443, 443 is arg
"""
self.aug.set(aug_conf_path + "/directive[last() + 1]", directive)
if type(arg) is not list:
@@ -1256,7 +1278,7 @@ LogLevel warn \n\
:rtype: bool
"""
return apache_restart(quiet)
return apache_restart()
def _add_httpd_transform(self, incl):
"""Add a transform to Augeas.
@@ -1326,7 +1348,7 @@ LogLevel warn \n\
raise errors.LetsEncryptConfiguratorError(
"Unable to find Apache version")
num_decimal = tuple(matches[0].split('.'))
return tuple(matches[0].split('.'))
###########################################################################
# Challenges Section
@@ -1508,7 +1530,7 @@ def enable_mod(mod_name):
stdout=open("/dev/null", 'w'),
stderr=open("/dev/null", 'w'))
# Hopefully this waits for output
subprocess.check_call(["sudo", APACHE2, "restart"],
subprocess.check_call(["sudo", CONFIG.APACHE2, "restart"],
stdout=open("/dev/null", 'w'),
stderr=open("/dev/null", 'w'))
except (OSError, subprocess.CalledProcessError) as err:
@@ -1531,12 +1553,13 @@ def check_ssl_loaded():
try:
# p=subprocess.check_output(['sudo', '/usr/sbin/apache2ctl', '-M'],
# stderr=open("/dev/null", 'w'))
proc = subprocess.Popen(['sudo', '/usr/sbin/apache2ctl', '-M'],
proc = subprocess.Popen([CONFIG.APACHE_CTL, '-M'],
stdout=subprocess.PIPE,
stderr=open(
"/dev/null", 'w')).communicate()[0]
except (OSError, ValueError):
logger.error("Error accessing apache2ctl for loaded modules!")
logger.error(
"Error accessing %s for loaded modules!" % CONFIG.APACHE_CTL)
logger.error("This may be caused by an Apache Configuration Error")
return False
@@ -1553,7 +1576,7 @@ def apache_restart():
"""
try:
proc = subprocess.Popen([APACHE2, 'restart'],
proc = subprocess.Popen([CONFIG.APACHE2, 'restart'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
text = proc.communicate()
letsencrypt/client/tests/apache_configurator_test.py
+64 -30
View File
@@ -8,11 +8,12 @@ still running smoothly.
"""
import re
import os
import shutil
import sys
import tarfile
import unittest
import sys
from letsencrypt.client import apache_configurator
from letsencrypt.client import CONFIG
@@ -49,9 +50,14 @@ class TwoVhosts_80(unittest.TestCase):
def setUp(self):
# Final slash is currently important
self.config_path = os.path.join(TEMP_DIR, "two_vhost_80/apache2/")
# Using a new configurator every time allows the Configurator to clean
# up after itself
self.config = apache_configurator.ApacheConfigurator(
self.config_path, (2, 4, 7))
self.aug_path = "/files" + self.config_path
prefix = os.path.join(TEMP_DIR, "two_vhost_80/apache2/sites-available/")
aug_pre = "/files" + prefix
self.vh_truth = []
@@ -128,44 +134,68 @@ class TwoVhosts_80(unittest.TestCase):
self.assertTrue(self.config.is_site_enabled(self.vh_truth[2].file))
self.assertTrue(self.config.is_site_enabled(self.vh_truth[3].file))
# def test_deploy_cert(self):
# """test deploy_cert."""
# self.config.deploy_cert(
# self.vh_truth[1],
# "example/cert.pem", "example/key.pem", "example/cert_chain.pem")
def test_deploy_cert(self):
"""test deploy_cert.
# loc_cert = self.config.find_directive(
# apache_configurator.case_i(
# "sslcertificatefile"), "example/cert.pem")
# loc_key = self.config.find_directive(
# apache_configurator.case_i(
# "sslcertificateKeyfile"), "example/key.pem")
# loc_chain = self.config.find_directive(
# apache_configurator.case_i(
# "SSLCertificateChainFile"), "example/chain.pem")
This test modifies the default-ssl vhost SSL directives.
# self.assertTrue(len(loc_cert) == 1 and
# apache_configurator.get_file_path(
# loc_cert[0]) == self.vh_truth[1].file)
"""
self.config.deploy_cert(
self.vh_truth[1],
"example/cert.pem", "example/key.pem", "example/cert_chain.pem")
# self.assertTrue(len(loc_key) == 1 and
# apache_configurator.get_file_path(
# loc_key[0]) == self.vh_truth[1].file)
loc_cert = self.config.find_directive(
apache_configurator.case_i("sslcertificatefile"),
re.escape("example/cert.pem"), self.vh_truth[1].path)
loc_key = self.config.find_directive(
apache_configurator.case_i("sslcertificateKeyfile"),
re.escape("example/key.pem"), self.vh_truth[1].path)
loc_chain = self.config.find_directive(
apache_configurator.case_i("SSLCertificateChainFile"),
re.escape("example/cert_chain.pem"), self.vh_truth[1].path)
# self.assertTrue(len(loc_chain) == 1 and
# apache_configurator.get_file_path(
# loc_chain[0]) == self.vh_truth[1].file)
# debug_file(self.vh_truth[1].file)
# Verify one directive was found in the correct file
self.assertTrue(len(loc_cert) == 1 and
apache_configurator.get_file_path(loc_cert[0]) ==
self.vh_truth[1].file)
self.assertTrue(len(loc_key) == 1 and
apache_configurator.get_file_path(loc_key[0]) ==
self.vh_truth[1].file)
self.assertTrue(len(loc_chain) == 1 and
apache_configurator.get_file_path(loc_chain[0]) ==
self.vh_truth[1].file)
def test_is_name_vhost(self):
"""test is_name_vhost."""
self.assertTrue(not self.config.is_name_vhost("*:80"))
self.assertTrue(self.config.is_name_vhost("*:80"))
self.config.version = (2, 2)
self.assertFalse(self.config.is_name_vhost("*:80"))
# def test_add_name_vhost(self):
# """test add_name_vhost."""
# self.config.add_name_vhost("*:443")
# self.config.save(temporary=True)
def test_add_name_vhost(self):
"""test add_name_vhost."""
self.config.add_name_vhost("*:443")
self.config.save(temporary=True)
self.assertTrue(self.config.find_directive(
"NameVirtualHost", re.escape("*:443")))
# self.assertTrue(self.config.is_name_vhost("*:443"))
def test_add_dir_to_ifmodssl(self):
"""test _add_dir_to_ifmodssl.
.. todo:: test what happens when a bad path is given... ie. ports.conf
doesn't exist
"""
self.config._add_dir_to_ifmodssl(
self.aug_path + "ports.conf", "FakeDirective", "123")
matches = self.config.find_directive("FakeDirective", "123")
print matches
self.assertTrue(len(matches) == 1)
self.assertTrue("IfModule" in matches[0])
def _verify_redirect(self, config_path):
with open(config_path, 'r') as config_fd:
@@ -174,6 +204,10 @@ class TwoVhosts_80(unittest.TestCase):
return CONFIG.REWRITE_HTTPS_ARGS[1] in conf
def debug_file(filepath):
with open(filepath, 'r')as file_d:
print file_d.read()
if __name__ == '__main__':
unittest.main()
# def download_unpack_tests(url=CONFIG_TGZ_URL):