mirror of
https://github.com/certbot/certbot.git
synced 2026-07-28 08:45:22 +02:00
Add certbot-postfix to tools
pep8ify Delint cover++ test more_info() Refactor get_config_var Don't duplicate changes to Postfix config document instance variables Always clear save_notes on save Test deploy_cert and save and add MockPostfix. Move mock and call to InstallerTest Add getters and setters Use postfix getters and setters protect get_config_var bump cover to 100% bump required coverage to 100 s/config_dir/config_utility Decrease minimum version to Postfix 2.6. This is the minimum version that allows us to set ciphers to be used with opportunistic TLS and is the oldest version packaged in any major distro. Use tls_security_level instead of use_tls. smtpd_tls_security_level should be used instead according to Postfix documentation. Test smtpd_tls_security_level conditional make dunder method an under method refactor postconf usage add check_all_output test check_all_output Add and test verify_exe_exists Add PostfixUtilBase Add ReadOnlyMainMap Use _get_output instead of _call Fix split strip typo
This commit is contained in:
Brad Warren
@@ -1,9 +1,7 @@
|
||||
"""Certbot installer plugin for Postfix."""
|
||||
import logging
|
||||
import os
|
||||
import string
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import zope.interface
|
||||
|
||||
@@ -22,7 +20,15 @@ logger = logging.getLogger(__name__)
|
||||
@zope.interface.implementer(interfaces.IInstaller)
|
||||
@zope.interface.provider(interfaces.IPluginFactory)
|
||||
class Installer(plugins_common.Installer):
|
||||
"""Certbot installer plugin for Postfix."""
|
||||
"""Certbot installer plugin for Postfix.
|
||||
|
||||
:ivar str config_dir: Postfix configuration directory to modify
|
||||
:ivar dict proposed_changes: configuration parameters and values to
|
||||
be written to the Postfix config when save() is called
|
||||
:ivar list save_notes: documentation for proposed changes. This is
|
||||
cleared and stored in Certbot checkpoints when save() is called
|
||||
|
||||
"""
|
||||
|
||||
description = "Configure TLS with the Postfix MTA"
|
||||
|
||||
@@ -53,7 +59,7 @@ class Installer(plugins_common.Installer):
|
||||
:raises errors.NotSupportedError: when version is not supported
|
||||
|
||||
"""
|
||||
for param in ("ctl", "config_dir",):
|
||||
for param in ("ctl", "config_utility",):
|
||||
self._verify_executable_is_available(param)
|
||||
self._set_config_dir()
|
||||
self._check_version()
|
||||
@@ -85,7 +91,7 @@ class Installer(plugins_common.Installer):
|
||||
|
||||
"""
|
||||
if self.conf("config-dir") is None:
|
||||
self.config_dir = self.get_config_var("config_directory")
|
||||
self.config_dir = self._get_config_var("config_directory")
|
||||
else:
|
||||
self.config_dir = self.conf("config-dir")
|
||||
|
||||
@@ -95,7 +101,7 @@ class Installer(plugins_common.Installer):
|
||||
:raises errors.NotSupportedError: if the version is unsupported
|
||||
|
||||
"""
|
||||
if self._get_version() < (2, 11, 0):
|
||||
if self._get_version() < (2, 6,):
|
||||
raise errors.NotSupportedError('Postfix version is too old')
|
||||
|
||||
def _lock_config_dir(self):
|
||||
@@ -138,7 +144,7 @@ class Installer(plugins_common.Installer):
|
||||
:raises .PluginError: Unable to find Postfix version.
|
||||
|
||||
"""
|
||||
mail_version = self.get_config_var("mail_version", default=True)
|
||||
mail_version = self._get_config_default("mail_version")
|
||||
return tuple(int(i) for i in mail_version.split('.'))
|
||||
|
||||
def get_all_names(self):
|
||||
@@ -147,7 +153,7 @@ class Installer(plugins_common.Installer):
|
||||
:rtype: `set` of `str`
|
||||
|
||||
"""
|
||||
return set(self.get_config_var(var)
|
||||
return set(self._get_config_var(var)
|
||||
for var in ('mydomain', 'myhostname', 'myorigin',))
|
||||
|
||||
def deploy_cert(self, domain, cert_path,
|
||||
@@ -164,12 +170,16 @@ class Installer(plugins_common.Installer):
|
||||
:raises .PluginError: when cert cannot be deployed
|
||||
|
||||
"""
|
||||
# pylint: disable=unused-argument
|
||||
self.save_notes.append("Configuring TLS for {0}".format(domain))
|
||||
self._set_config_var("smtpd_tls_cert_file", fullchain_path)
|
||||
self._set_config_var("smtpd_tls_key_file", key_path)
|
||||
self._set_config_var("smtpd_tls_mandatory_protocols", "!SSLv2, !SSLv3")
|
||||
self._set_config_var("smtpd_tls_protocols", "!SSLv2, !SSLv3")
|
||||
self._set_config_var("smtpd_use_tls", "yes")
|
||||
|
||||
# Don't configure opportunistic TLS if it's currently mandatory
|
||||
if self._get_config_var("smtpd_tls_security_level") != "encrypt":
|
||||
self._set_config_var("smtpd_tls_security_level", "may")
|
||||
|
||||
def enhance(self, domain, enhancement, options=None):
|
||||
"""Raises an exception for request for unsupported enhancement.
|
||||
@@ -178,6 +188,7 @@ class Installer(plugins_common.Installer):
|
||||
are currently supported
|
||||
|
||||
"""
|
||||
# pylint: disable=unused-argument
|
||||
raise errors.PluginError(
|
||||
"Unsupported enhancement: {0}".format(enhancement))
|
||||
|
||||
@@ -202,14 +213,13 @@ class Installer(plugins_common.Installer):
|
||||
:raises errors.PluginError: when save is unsuccessful
|
||||
|
||||
"""
|
||||
if not self.proposed_changes:
|
||||
return
|
||||
if self.proposed_changes:
|
||||
save_files = set((os.path.join(self.config_dir, "main.cf"),))
|
||||
self.add_to_checkpoint(save_files,
|
||||
"\n".join(self.save_notes), temporary)
|
||||
self._write_config_changes()
|
||||
self.proposed_changes.clear()
|
||||
|
||||
self.add_to_checkpoint(os.path.join(self.config_dir, "main.cf"),
|
||||
"\n".join(self.save_notes), temporary)
|
||||
self._write_config_changes()
|
||||
|
||||
self.proposed_changes.clear()
|
||||
del self.save_notes[:]
|
||||
|
||||
if title and not temporary:
|
||||
@@ -298,70 +308,67 @@ class Installer(plugins_common.Installer):
|
||||
|
||||
util.check_call(cmd)
|
||||
|
||||
def get_config_var(self, name, default=False):
|
||||
def _get_config_default(self, name):
|
||||
"""Return the default value of the specified config parameter.
|
||||
|
||||
:param str name: name of the Postfix config default to return
|
||||
|
||||
:returns: default for the specified configuration parameter if it
|
||||
exists, otherwise, None
|
||||
:rtype: str or types.NoneType
|
||||
|
||||
:raises errors.PluginError: if an error occurs while running postconf
|
||||
or parsing its output
|
||||
|
||||
"""
|
||||
try:
|
||||
return self._get_value_from_postconf(("-d", name,))
|
||||
except (subprocess.CalledProcessError, errors.PluginError):
|
||||
raise errors.PluginError("Unable to determine the default value of"
|
||||
" the Postfix parameter {0}".format(name))
|
||||
|
||||
def _get_config_var(self, name):
|
||||
"""Return the value of the specified Postfix config parameter.
|
||||
|
||||
:param str name: name of the Postfix config parameter to return
|
||||
:param bool default: whether or not to return the default value
|
||||
instead of the actual value
|
||||
If there is an unsaved change modifying the value of the
|
||||
specified config parameter, the value after this proposed change
|
||||
is returned rather than the current value. If the value is
|
||||
unset, `None` is returned.
|
||||
|
||||
:returns: value of the specified configuration parameter
|
||||
:rtype: str
|
||||
:param str name: name of the Postfix config parameter to return
|
||||
|
||||
:returns: value of the parameter included in postconf_args
|
||||
:rtype: str or types.NoneType
|
||||
|
||||
:raises errors.PluginError: if an error occurs while running postconf
|
||||
or parsing its output
|
||||
|
||||
"""
|
||||
cmd = self._build_cmd_for_config_var(name, default)
|
||||
if name in self.proposed_changes:
|
||||
return self.proposed_changes[name]
|
||||
|
||||
try:
|
||||
output = util.check_output(cmd)
|
||||
except subprocess.CalledProcessError:
|
||||
logger.debug("Encountered an error when running 'postconf'",
|
||||
exc_info=True)
|
||||
raise errors.PluginError(
|
||||
"Unable to determine the value "
|
||||
"of Postfix parameter {0}".format(name))
|
||||
|
||||
expected_prefix = name + " ="
|
||||
if not output.startswith(expected_prefix):
|
||||
raise errors.PluginError(
|
||||
"Unexpected output '{0}' from '{1}'".format(output,
|
||||
' '.join(cmd)))
|
||||
|
||||
return output[len(expected_prefix):].strip()
|
||||
|
||||
def _build_cmd_for_config_var(self, name, default):
|
||||
"""Return a command to run to get a Postfix config parameter.
|
||||
|
||||
:param str name: name of the Postfix config parameter to return
|
||||
:param bool default: whether or not to return the default value
|
||||
instead of the actual value
|
||||
|
||||
:returns: command to run
|
||||
:rtype: list
|
||||
|
||||
"""
|
||||
cmd = self._postconf_command_base()
|
||||
|
||||
if default:
|
||||
cmd.append("-d")
|
||||
|
||||
cmd.append(name)
|
||||
|
||||
return cmd
|
||||
return self._get_value_from_postconf((name,))
|
||||
except (subprocess.CalledProcessError, errors.PluginError):
|
||||
raise errors.PluginError("Unable to determine the value of"
|
||||
" the Postfix parameter {0}".format(name))
|
||||
|
||||
def _set_config_var(self, name, value):
|
||||
"""Set the Postfix config parameter name to value.
|
||||
|
||||
This method only stores the requested change in memory. The
|
||||
Postfix configuration is not modified until save() is called.
|
||||
If there's already an identical in progress change or the
|
||||
Postfix configuration parameter already has the specified value,
|
||||
no changes are made.
|
||||
|
||||
:param str name: name of the Postfix config parameter
|
||||
:param str value: value to set the Postfix config parameter to
|
||||
|
||||
"""
|
||||
assert isinstance(name, str), "Invalid name value"
|
||||
assert isinstance(value, str), "Invalid key value"
|
||||
self.proposed_changes[name] = value
|
||||
self.save_notes.append("\t* Set {0} to {1}".format(name, value))
|
||||
if self._get_config_var(name) != value:
|
||||
self.proposed_changes[name] = value
|
||||
self.save_notes.append("\t* Set {0} to {1}".format(name, value))
|
||||
|
||||
def _write_config_changes(self):
|
||||
"""Write proposed changes to the Postfix config.
|
||||
@@ -369,21 +376,83 @@ class Installer(plugins_common.Installer):
|
||||
:raises errors.PluginError: if an error occurs
|
||||
|
||||
"""
|
||||
cmd = self._postconf_command_base()
|
||||
cmd.extend("{0}={1}".format(name, value)
|
||||
for name, value in self.proposed_changes.items())
|
||||
|
||||
try:
|
||||
util.check_call(cmd)
|
||||
self._run_postconf_command(
|
||||
"{0}={1}".format(name, value)
|
||||
for name, value in self.proposed_changes.items())
|
||||
except subprocess.CalledProcessError:
|
||||
raise errors.PluginError(
|
||||
"An error occurred while updating your Postfix config.")
|
||||
|
||||
def _postconf_command_base(self):
|
||||
"""Builds start of a postconf command using the selected config."""
|
||||
cmd = [self.conf("config-utility")]
|
||||
def _get_value_from_postconf(self, postconf_args):
|
||||
"""Runs postconf and extracts the specified config value.
|
||||
|
||||
It is assumed that the name of the Postfix config parameter to
|
||||
parse from the output is the last value in postconf_args. If the
|
||||
value is unset, `None` is returned. If an error occurs, the
|
||||
relevant information is logged before an exception is raised.
|
||||
|
||||
:param collections.Iterable args: arguments to postconf
|
||||
|
||||
:returns: value of the parameter included in postconf_args
|
||||
:rtype: str or types.NoneType
|
||||
|
||||
:raises errors.PluginError: if unable to parse postconf output
|
||||
:raises subprocess.CalledProcessError: if postconf fails
|
||||
|
||||
"""
|
||||
name = postconf_args[-1]
|
||||
output = self._run_postconf_command(postconf_args)
|
||||
|
||||
try:
|
||||
return self._parse_postconf_output(output, name)
|
||||
except errors.PluginError:
|
||||
logger.debug("An error occurred while parsing postconf output",
|
||||
exc_info=True)
|
||||
raise
|
||||
|
||||
def _run_postconf_command(self, args):
|
||||
"""Runs a postconf command using the selected config.
|
||||
|
||||
If postconf exits with a nonzero status, the error is logged
|
||||
before an exception is raised.
|
||||
|
||||
:param collections.Iterable args: additional arguments to postconf
|
||||
|
||||
:returns: stdout output of postconf
|
||||
:rtype: str
|
||||
|
||||
:raises subprocess.CalledProcessError: if the command fails
|
||||
|
||||
"""
|
||||
|
||||
cmd = [self.conf("config-utility")]
|
||||
if self.conf("config-dir") is not None:
|
||||
cmd.extend(("-c", self.conf("config-dir"),))
|
||||
cmd.extend(args)
|
||||
|
||||
return cmd
|
||||
return util.check_output(cmd)
|
||||
|
||||
def _parse_postconf_output(self, output, name):
|
||||
"""Parses postconf output and returns the specified value.
|
||||
|
||||
If the specified Postfix parameter is unset, `None` is returned.
|
||||
It is assumed that most one configuration parameter will be
|
||||
included in the given output.
|
||||
|
||||
:param str output: output from postconf
|
||||
:param str name: name of the Postfix config parameter to obtain
|
||||
|
||||
:returns: value of the parameter included in postconf_args
|
||||
:rtype: str or types.NoneType
|
||||
|
||||
:raises errors.PluginError: if unable to parse postconf ouput
|
||||
|
||||
"""
|
||||
expected_prefix = name + " ="
|
||||
if output.count("\n") != 1 or not output.startswith(expected_prefix):
|
||||
raise errors.PluginError(
|
||||
"Unexpected output '{0}' from postconf".format(output))
|
||||
|
||||
value = output[len(expected_prefix):].strip()
|
||||
return value if value else None
|
||||
|
||||
@@ -1,36 +1,25 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from __future__ import absolute_import
|
||||
from __future__ import division
|
||||
from __future__ import print_function
|
||||
from __future__ import unicode_literals
|
||||
|
||||
"""Tests for certbot_postfix.installer."""
|
||||
import functools
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import unittest
|
||||
|
||||
import mock
|
||||
import six
|
||||
|
||||
from certbot import errors
|
||||
from certbot.tests import util as certbot_test_util
|
||||
|
||||
# Fake Postfix Configs
|
||||
names_only_config = """mydomain = fubard.org
|
||||
myhostname = mail.fubard.org
|
||||
myorigin = fubard.org"""
|
||||
|
||||
|
||||
class InstallerTest(certbot_test_util.ConfigTestCase):
|
||||
# pylint: disable=too-many-public-methods
|
||||
|
||||
def setUp(self):
|
||||
super(InstallerTest, self).setUp()
|
||||
self.config.postfix_ctl = "postfix"
|
||||
self.config.postfix_config_dir = self.tempdir
|
||||
self.config.postfix_config_utility = "postconf"
|
||||
self.mock_postfix = MockPostfix(self.tempdir,
|
||||
{"mail_version": "3.1.4"})
|
||||
|
||||
def test_add_parser_arguments(self):
|
||||
options = set(('ctl', 'config-dir', 'config-utility',))
|
||||
@@ -51,7 +40,29 @@ class InstallerTest(certbot_test_util.ConfigTestCase):
|
||||
|
||||
with mock.patch(path_surgery_path, return_value=False):
|
||||
with mock.patch(exe_exists_path, return_value=False):
|
||||
self.assertRaises(errors.NoInstallationError, installer.prepare)
|
||||
self.assertRaises(errors.NoInstallationError,
|
||||
installer.prepare)
|
||||
|
||||
def test_postconf_error(self):
|
||||
installer = self._create_installer()
|
||||
|
||||
check_output_path = "certbot_postfix.installer.util.check_output"
|
||||
exe_exists_path = "certbot_postfix.installer.certbot_util.exe_exists"
|
||||
with mock.patch(check_output_path) as mock_check_output:
|
||||
mock_check_output.side_effect = subprocess.CalledProcessError(42,
|
||||
"a")
|
||||
with mock.patch(exe_exists_path, return_value=True):
|
||||
self.assertRaises(errors.PluginError, installer.prepare)
|
||||
|
||||
def test_unexpected_postconf(self):
|
||||
installer = self._create_installer()
|
||||
|
||||
check_output_path = "certbot_postfix.installer.util.check_output"
|
||||
exe_exists_path = "certbot_postfix.installer.certbot_util.exe_exists"
|
||||
with mock.patch(check_output_path) as mock_check_output:
|
||||
mock_check_output.return_value = "foobar"
|
||||
with mock.patch(exe_exists_path, return_value=True):
|
||||
self.assertRaises(errors.PluginError, installer.prepare)
|
||||
|
||||
def test_set_config_dir(self):
|
||||
self.config.postfix_config_dir = os.path.join(self.tempdir, "subdir")
|
||||
@@ -61,32 +72,100 @@ class InstallerTest(certbot_test_util.ConfigTestCase):
|
||||
expected = self.config.postfix_config_dir
|
||||
self.config.postfix_config_dir = None
|
||||
|
||||
check_call_path = "certbot_postfix.installer.subprocess.check_call"
|
||||
check_output_path = "certbot_postfix.installer.util.check_output"
|
||||
self.mock_postfix.set_value("config_directory", expected)
|
||||
exe_exists_path = "certbot_postfix.installer.certbot_util.exe_exists"
|
||||
with mock.patch(check_output_path) as mock_check_output:
|
||||
mock_check_output.side_effect = [
|
||||
"config_directory = " + expected, "mail_version = 3.1.4"
|
||||
]
|
||||
with mock.patch(exe_exists_path, return_value=True):
|
||||
with mock.patch(check_call_path):
|
||||
installer.prepare()
|
||||
with mock.patch(exe_exists_path, return_value=True):
|
||||
self._mock_postfix_and_call(installer.prepare)
|
||||
self.assertEqual(installer.config_dir, expected)
|
||||
|
||||
@mock.patch("certbot_postfix.installer.certbot_util.exe_exists")
|
||||
def test_old_version(self, mock_exe_exists):
|
||||
installer = self._create_installer()
|
||||
mock_exe_exists.return_value = True
|
||||
self.mock_postfix.set_value("mail_version", "0.0.1")
|
||||
self._mock_postfix_and_call(
|
||||
self.assertRaises, errors.NotSupportedError, installer.prepare)
|
||||
|
||||
def test_lock_error(self):
|
||||
assert_raises = functools.partial(self.assertRaises,
|
||||
errors.PluginError,
|
||||
self._create_prepared_installer)
|
||||
certbot_test_util.lock_and_call(assert_raises, self.tempdir)
|
||||
|
||||
@mock.patch("certbot_postfix.installer.util.check_output")
|
||||
def test_get_all_names(self, mock_check_output):
|
||||
def test_more_info(self):
|
||||
installer = self._create_prepared_installer()
|
||||
mock_check_output.side_effect = names_only_config.splitlines()
|
||||
version = "3.1.2"
|
||||
self.mock_postfix.set_value("mail_version", version)
|
||||
|
||||
result = installer.get_all_names()
|
||||
self.assertTrue("fubard.org" in result)
|
||||
self.assertTrue("mail.fubard.org" in result)
|
||||
output = self._mock_postfix_and_call(installer.more_info)
|
||||
self.assertTrue("Postfix" in output)
|
||||
self.assertTrue(self.tempdir in output)
|
||||
self.assertTrue(version in output)
|
||||
|
||||
def test_get_all_names(self):
|
||||
config = {"mydomain": "example.org",
|
||||
"myhostname": "mail.example.org",
|
||||
"myorigin": "example.org"}
|
||||
for name, value in config.items():
|
||||
self.mock_postfix.set_value(name, value)
|
||||
|
||||
installer = self._create_prepared_installer()
|
||||
result = self._mock_postfix_and_call(installer.get_all_names)
|
||||
self.assertEqual(result, set(config.values()))
|
||||
|
||||
def test_deploy(self):
|
||||
installer = self._create_prepared_installer()
|
||||
|
||||
def deploy_cert(domain):
|
||||
"""Calls deploy_cert for the given domain.
|
||||
|
||||
:param str domain: domain to deploy cert for
|
||||
|
||||
"""
|
||||
installer.deploy_cert(domain, "foo", "bar", "baz", "qux")
|
||||
|
||||
self._mock_postfix_and_call(deploy_cert, "example.org")
|
||||
# No calls to postconf are expected so mock isn't needed
|
||||
deploy_cert("mail.example.org")
|
||||
|
||||
def test_deploy_and_save(self):
|
||||
self._test_deploy_and_save_common({"smtpd_tls_security_level": "may"})
|
||||
|
||||
def test_deploy_and_save2(self):
|
||||
self.mock_postfix.set_value("smtpd_tls_security_level", "encrypt")
|
||||
self._test_deploy_and_save_common({"smtpd_tls_security_level":
|
||||
"encrypt"})
|
||||
|
||||
def _test_deploy_and_save_common(self, expected_config):
|
||||
key_path = "key_path"
|
||||
fullchain_path = "fullchain_path"
|
||||
installer = self._create_prepared_installer()
|
||||
|
||||
for i, domain in enumerate(("example.org", "mail.example.org",)):
|
||||
self._mock_postfix_and_call(
|
||||
installer.deploy_cert, domain, "unused",
|
||||
key_path, "unused", fullchain_path)
|
||||
if i:
|
||||
# No mock because Postfix utilities aren't expected to be used
|
||||
installer.save("noop")
|
||||
else:
|
||||
self._mock_postfix_and_call(installer.save, "real save")
|
||||
|
||||
expected_config.setdefault("smtpd_tls_cert_file", fullchain_path)
|
||||
expected_config.setdefault("smtpd_tls_key_file", key_path)
|
||||
for key, value in expected_config.items():
|
||||
self.assertEqual(self.mock_postfix.get_value(key), value)
|
||||
|
||||
def test_save_error(self):
|
||||
installer = self._create_prepared_installer()
|
||||
self._mock_postfix_and_call(
|
||||
installer.deploy_cert, "example.org", "foo", "bar", "baz", "qux")
|
||||
|
||||
check_call_path = "certbot_postfix.installer.util.check_output"
|
||||
with mock.patch(check_call_path) as mock_check_call:
|
||||
mock_check_call.side_effect = subprocess.CalledProcessError(42,
|
||||
"foo")
|
||||
self.assertRaises(errors.PluginError, installer.save)
|
||||
|
||||
def test_enhance(self):
|
||||
self.assertRaises(errors.PluginError,
|
||||
@@ -111,10 +190,10 @@ class InstallerTest(certbot_test_util.ConfigTestCase):
|
||||
]
|
||||
self.assertRaises(errors.PluginError, installer.restart)
|
||||
|
||||
@mock.patch("certbot_postfix.installer.subprocess.check_call")
|
||||
def test_postfix_reload_success(self, mock_check_call):
|
||||
installer = self._create_prepared_installer()
|
||||
installer.restart()
|
||||
def test_postfix_reload_success(self):
|
||||
with mock.patch("certbot_postfix.installer.subprocess.check_call"):
|
||||
installer = self._create_prepared_installer()
|
||||
installer.restart()
|
||||
|
||||
@mock.patch("certbot_postfix.installer.subprocess.check_call")
|
||||
def test_postfix_start_failure(self, mock_check_call):
|
||||
@@ -130,52 +209,6 @@ class InstallerTest(certbot_test_util.ConfigTestCase):
|
||||
]
|
||||
installer.restart()
|
||||
|
||||
def test_get_config_var_success(self):
|
||||
self.config.postfix_config_dir = None
|
||||
|
||||
command = self._test_get_config_var_success_common('foo', False)
|
||||
self.assertFalse("-c" in command)
|
||||
self.assertFalse("-d" in command)
|
||||
|
||||
def test_get_config_var_success_with_config(self):
|
||||
command = self._test_get_config_var_success_common('foo', False)
|
||||
self.assertTrue("-c" in command)
|
||||
self.assertFalse("-d" in command)
|
||||
|
||||
def test_get_config_var_success_with_default(self):
|
||||
self.config.postfix_config_dir = None
|
||||
|
||||
command = self._test_get_config_var_success_common('foo', True)
|
||||
self.assertFalse("-c" in command)
|
||||
self.assertTrue("-d" in command)
|
||||
|
||||
@mock.patch("certbot_postfix.installer.logger")
|
||||
@mock.patch("certbot_postfix.installer.util.check_output")
|
||||
def test_get_config_var_failure(self, mock_check_output, mock_logger):
|
||||
mock_check_output.side_effect = subprocess.CalledProcessError(42, "foo")
|
||||
installer = self._create_installer()
|
||||
self.assertRaises(errors.PluginError, installer.get_config_var, "foo")
|
||||
self.assertTrue(mock_logger.debug.call_args[1]["exc_info"])
|
||||
|
||||
@mock.patch("certbot_postfix.installer.util.check_output")
|
||||
def test_get_config_var_unexpected_output(self, mock_check_output):
|
||||
self.config.postfix_config_dir = None
|
||||
mock_check_output.return_value = "foo"
|
||||
|
||||
installer = self._create_installer()
|
||||
self.assertRaises(errors.PluginError, installer.get_config_var, "foo")
|
||||
|
||||
def _test_get_config_var_success_common(self, name, default):
|
||||
installer = self._create_installer()
|
||||
|
||||
check_output_path = "certbot_postfix.installer.util.check_output"
|
||||
with mock.patch(check_output_path) as mock_check_output:
|
||||
value = "bar"
|
||||
mock_check_output.return_value = name + " = " + value
|
||||
self.assertEqual(installer.get_config_var(name, default), value)
|
||||
|
||||
return mock_check_output.call_args[0][0]
|
||||
|
||||
def _create_prepared_installer(self):
|
||||
"""Creates and returns a new prepared Postfix Installer.
|
||||
|
||||
@@ -188,14 +221,9 @@ class InstallerTest(certbot_test_util.ConfigTestCase):
|
||||
"""
|
||||
installer = self._create_installer()
|
||||
|
||||
check_call_path = "certbot_postfix.installer.subprocess.check_call"
|
||||
check_output_path = "certbot_postfix.installer.util.check_output"
|
||||
exe_exists_path = "certbot_postfix.installer.certbot_util.exe_exists"
|
||||
with mock.patch(check_output_path) as mock_check_output:
|
||||
with mock.patch(exe_exists_path, return_value=True):
|
||||
with mock.patch(check_call_path):
|
||||
mock_check_output.return_value = "mail_version = 3.1.4"
|
||||
installer.prepare()
|
||||
with mock.patch(exe_exists_path, return_value=True):
|
||||
self._mock_postfix_and_call(installer.prepare)
|
||||
|
||||
return installer
|
||||
|
||||
@@ -211,6 +239,113 @@ class InstallerTest(certbot_test_util.ConfigTestCase):
|
||||
from certbot_postfix import installer
|
||||
return installer.Installer(self.config, name)
|
||||
|
||||
def _mock_postfix_and_call(self, func, *args, **kwargs):
|
||||
"""Calls func with mocked responses from Postfix utilities.
|
||||
|
||||
:param callable func: function to call with mocked args
|
||||
:param tuple args: positional arguments to func
|
||||
:param dict kwargs: keyword arguments to func
|
||||
|
||||
:returns: the return value of func
|
||||
|
||||
"""
|
||||
check_call_path = "certbot_postfix.installer.subprocess.check_call"
|
||||
check_output_path = "certbot_postfix.installer.util.check_output"
|
||||
|
||||
with mock.patch(check_call_path) as mock_check_call:
|
||||
mock_check_call.side_effect = self.mock_postfix
|
||||
with mock.patch(check_output_path) as mock_check_output:
|
||||
mock_check_output.side_effect = self.mock_postfix
|
||||
return func(*args, **kwargs)
|
||||
|
||||
|
||||
class MockPostfix(object):
|
||||
"""A callable to mimic Postfix command line utilities.
|
||||
|
||||
This is best used a side effect to a mock object. All calls to
|
||||
'postfix' are noops. For calls to 'postconf', values that are set in
|
||||
the constructor or through mocked out runs of postconf are
|
||||
remembered and properly returned if the installer attempts to fetch
|
||||
the value. If the Postfix installer attempts to obtain a value that
|
||||
hasn't yet been set, a dummy value is returned.
|
||||
|
||||
:ivar str config_path: path to Postfix main.cf file
|
||||
|
||||
"""
|
||||
def __init__(self, config_dir, initial_values):
|
||||
"""Create Postfix configuration.
|
||||
|
||||
:param str config_dir: path for Postfix config dir
|
||||
:param dict initial_values: initial Postfix config values
|
||||
|
||||
"""
|
||||
initial_values["config_directory"] = config_dir
|
||||
|
||||
self.config_path = os.path.join(config_dir, "main.cf")
|
||||
self._write_config(initial_values)
|
||||
|
||||
def __call__(self, args, *unused_args, **unused_kwargs):
|
||||
cmd = os.path.basename(args[0])
|
||||
if cmd == "postfix":
|
||||
return
|
||||
elif cmd != "postconf": # pragma: no cover
|
||||
assert False, "Unexpected command '{0}'".format(''.join(args))
|
||||
|
||||
output = []
|
||||
|
||||
skip = False
|
||||
for arg in args[1:]:
|
||||
if skip:
|
||||
skip = False
|
||||
elif arg[0] == "-":
|
||||
if arg == "-c":
|
||||
skip = True
|
||||
elif "=" in arg:
|
||||
name, _, value = arg.partition("=")
|
||||
self.set_value(name, value)
|
||||
else:
|
||||
output.append("{0} = {1}\n".format(arg, self.get_value(arg)))
|
||||
|
||||
return "\n".join(output)
|
||||
|
||||
def get_value(self, name):
|
||||
"""Returns the value for the Postfix config parameter name.
|
||||
|
||||
If the value isn't set, an empty string is returned.
|
||||
|
||||
:param str name: name of the Postfix config parameter
|
||||
|
||||
:returns: value of the named parameter
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
return self._read_config().get(name, "")
|
||||
|
||||
def set_value(self, name, value):
|
||||
"""Sets the value for a Postfix config parameter.
|
||||
|
||||
:param str name: name of the Postfix config parameter
|
||||
:param str value: value ot set the parameter to
|
||||
|
||||
"""
|
||||
config = self._read_config()
|
||||
config[name] = value
|
||||
self._write_config(config)
|
||||
|
||||
def _read_config(self):
|
||||
config = {}
|
||||
with open(self.config_path) as f:
|
||||
for line in f:
|
||||
key, _, value = line.strip().partition(" = ")
|
||||
config[key] = value
|
||||
|
||||
return config
|
||||
|
||||
def _write_config(self, config):
|
||||
with open(self.config_path, "w") as f:
|
||||
f.writelines("{0} = {1}\n".format(key, value)
|
||||
for key, value in config.items())
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
unittest.main() # pragma: no cover
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
"""Classes that wrap the postconf command line utility.
|
||||
|
||||
These classes allow you to interact with a Postfix config like it is a
|
||||
dictionary, with the getting and setting of values in the config being
|
||||
handled automatically by the class.
|
||||
|
||||
"""
|
||||
import collections
|
||||
|
||||
from certbot_postfix import util
|
||||
|
||||
|
||||
class ReadOnlyMainMap(util.PostfixUtilBase, collections.Mapping):
|
||||
"""A read-only view of a Postfix main.cf file."""
|
||||
|
||||
_modifiers = None
|
||||
"""An iterable containing additional CLI flags for postconf."""
|
||||
|
||||
def __getitem__(self, name):
|
||||
return next(_parse_main_output(self._get_output([name])))[1]
|
||||
|
||||
def __iter__(self):
|
||||
for name, _ in _parse_main_output(self._get_output()):
|
||||
yield name
|
||||
|
||||
def __len__(self):
|
||||
return sum(1 for _ in _parse_main_output(self._get_output()))
|
||||
|
||||
def _call(self, extra_args=None):
|
||||
"""Runs Postconf and returns the result.
|
||||
|
||||
If self._modifiers is set, it is provided on the command line to
|
||||
postconf before any values in extra_args.
|
||||
|
||||
:param list extra_args: additional arguments for the command
|
||||
|
||||
:returns: data written to stdout and stderr
|
||||
:rtype: `tuple` of `str`
|
||||
|
||||
:raises subprocess.CalledProcessError: if the command fails
|
||||
|
||||
"""
|
||||
all_extra_args = []
|
||||
for args_list in (self._modifiers, extra_args,):
|
||||
if args_list is not None:
|
||||
all_extra_args.extend(args_list)
|
||||
|
||||
return super(ReadOnlyMainMap, self)._call(all_extra_args)
|
||||
|
||||
|
||||
def _parse_main_output(output):
|
||||
"""Parses the raw output from Postconf about main.cf.
|
||||
|
||||
:param str output: data postconf wrote to stdout about main.cf
|
||||
|
||||
:returns: generator providing key-value pairs from main.cf
|
||||
:rtype: generator
|
||||
|
||||
"""
|
||||
for line in output.splitlines():
|
||||
name, _, value = line.partition(" =")
|
||||
yield name, value.strip()
|
||||
@@ -1,12 +1,129 @@
|
||||
"""Utility functions for use in the Postfix installer."""
|
||||
|
||||
import logging
|
||||
import subprocess
|
||||
|
||||
from certbot import errors
|
||||
from certbot import util as certbot_util
|
||||
from certbot.plugins import util as plugins_util
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PostfixUtilBase(object):
|
||||
"""A base class for wrapping Postfix command line utilities."""
|
||||
|
||||
def __init__(self, executable, config_dir=None):
|
||||
"""Sets up the Postfix utility class.
|
||||
|
||||
:param str executable: name or path of the Postfix utility
|
||||
:param str config_dir: path to an alternative Postfix config
|
||||
|
||||
:raises .NoInstallationError: when the executable isn't found
|
||||
|
||||
"""
|
||||
verify_exe_exists(executable)
|
||||
|
||||
self._base_command = [executable]
|
||||
if config_dir is not None:
|
||||
self._base_command.extend(('-c', config_dir,))
|
||||
|
||||
def _call(self, extra_args=None):
|
||||
"""Runs the Postfix utility and returns the result.
|
||||
|
||||
:param list extra_args: additional arguments for the command
|
||||
|
||||
:returns: data written to stdout and stderr
|
||||
:rtype: `tuple` of `str`
|
||||
|
||||
:raises subprocess.CalledProcessError: if the command fails
|
||||
|
||||
"""
|
||||
args = list(self._base_command)
|
||||
if extra_args is not None:
|
||||
args.extend(extra_args)
|
||||
return check_all_output(args)
|
||||
|
||||
def _get_output(self, extra_args=None):
|
||||
"""Runs the Postfix utility and returns only stdout output.
|
||||
|
||||
This function relies on self._call for running the utility.
|
||||
|
||||
:param list extra_args: additional arguments for the command
|
||||
|
||||
:returns: data written to stdout
|
||||
:rtype: str
|
||||
|
||||
:raises subprocess.CalledProcessError: if the command fails
|
||||
|
||||
"""
|
||||
return self._call(extra_args)[0]
|
||||
|
||||
|
||||
def check_all_output(*args, **kwargs):
|
||||
"""A version of subprocess.check_output that also captures stderr.
|
||||
|
||||
This is the same as :func:`subprocess.check_output` except output
|
||||
written to stderr is also captured and returned to the caller. The
|
||||
return value is a tuple of two strings (rather than byte strings).
|
||||
To accomplish this, the caller cannot set the stdout, stderr, or
|
||||
universal_newlines parameters to :class:`subprocess.Popen`.
|
||||
|
||||
Additionally, if the command exits with a nonzero status, output is
|
||||
not included in the raised :class:`subprocess.CalledProcessError`
|
||||
because Python 2.6 does not support this. Instead, the failure
|
||||
including the output is logged.
|
||||
|
||||
:param tuple args: positional arguments for Popen
|
||||
:param dict kwargs: keyword arguments for Popen
|
||||
|
||||
:returns: data written to stdout and stderr
|
||||
:rtype: `tuple` of `str`
|
||||
|
||||
:raises ValueError: if arguments are invalid
|
||||
:raises subprocess.CalledProcessError: if the command fails
|
||||
|
||||
"""
|
||||
for keyword in ('stdout', 'stderr', 'universal_newlines',):
|
||||
if keyword in kwargs:
|
||||
raise ValueError(
|
||||
keyword + ' argument not allowed, it will be overridden.')
|
||||
|
||||
kwargs['stdout'] = subprocess.PIPE
|
||||
kwargs['stderr'] = subprocess.PIPE
|
||||
kwargs['universal_newlines'] = True
|
||||
|
||||
process = subprocess.Popen(*args, **kwargs)
|
||||
output, err = process.communicate()
|
||||
retcode = process.poll()
|
||||
if retcode:
|
||||
cmd = kwargs.get('args')
|
||||
if cmd is None:
|
||||
cmd = args[0]
|
||||
logger.debug(
|
||||
"'%s' exited with %d. stdout output was:\n%s\nstderr output was:\n%s",
|
||||
cmd, retcode, output, err)
|
||||
raise subprocess.CalledProcessError(retcode, cmd)
|
||||
return (output, err)
|
||||
|
||||
|
||||
def verify_exe_exists(exe):
|
||||
"""Ensures an executable with the given name is available.
|
||||
|
||||
If an executable isn't found for the given path or name, extra
|
||||
directories are added to the user's PATH to help find system
|
||||
utilities that may not be available in the default cron PATH.
|
||||
|
||||
:param str exe: executable path or name
|
||||
|
||||
:raises .NoInstallationError: when the executable isn't found
|
||||
|
||||
"""
|
||||
if not (certbot_util.exe_exists(exe) or plugins_util.path_surgery(exe)):
|
||||
raise errors.NoInstallationError(
|
||||
"Cannot find executable '{0}'.".format(exe))
|
||||
|
||||
|
||||
def check_call(*args, **kwargs):
|
||||
"""A simple wrapper of subprocess.check_call that logs errors.
|
||||
|
||||
@@ -63,7 +180,8 @@ def check_output(*args, **kwargs):
|
||||
if retcode:
|
||||
cmd = _get_cmd(*args, **kwargs)
|
||||
logger.debug(
|
||||
"'%s' exited with %d. Output was:\n%s", cmd, retcode, output)
|
||||
"'%s' exited with %d. Output was:\n%s",
|
||||
cmd, retcode, output, exc_info=True)
|
||||
raise subprocess.CalledProcessError(retcode, cmd)
|
||||
return output
|
||||
|
||||
|
||||
@@ -5,6 +5,98 @@ import unittest
|
||||
|
||||
import mock
|
||||
|
||||
from certbot import errors
|
||||
|
||||
|
||||
class PostfixUtilBaseTest(unittest.TestCase):
|
||||
"""Tests for certbot_postfix.util.PostfixUtilBase."""
|
||||
|
||||
@classmethod
|
||||
def _create_object(cls, *args, **kwargs):
|
||||
from certbot_postfix.util import PostfixUtilBase
|
||||
return PostfixUtilBase(*args, **kwargs)
|
||||
|
||||
@mock.patch('certbot_postfix.util.verify_exe_exists')
|
||||
def test_no_exe(self, mock_verify):
|
||||
expected_error = errors.NoInstallationError
|
||||
mock_verify.side_effect = expected_error
|
||||
self.assertRaises(expected_error, self._create_object, 'nonexistent')
|
||||
|
||||
def test_object_creation(self):
|
||||
with mock.patch('certbot_postfix.util.verify_exe_exists'):
|
||||
self._create_object('existent')
|
||||
|
||||
|
||||
class CheckAllOutputTest(unittest.TestCase):
|
||||
"""Tests for certbot_postfix.util.check_all_output."""
|
||||
|
||||
@classmethod
|
||||
def _call(cls, *args, **kwargs):
|
||||
from certbot_postfix.util import check_all_output
|
||||
return check_all_output(*args, **kwargs)
|
||||
|
||||
@mock.patch('certbot_postfix.util.logger')
|
||||
@mock.patch('certbot_postfix.util.subprocess.Popen')
|
||||
def test_command_error(self, mock_popen, mock_logger):
|
||||
command = 'foo'
|
||||
retcode = 42
|
||||
output = 'bar'
|
||||
err = 'baz'
|
||||
|
||||
mock_popen().communicate.return_value = (output, err)
|
||||
mock_popen().poll.return_value = 42
|
||||
|
||||
self.assertRaises(subprocess.CalledProcessError, self._call, command)
|
||||
log_args = mock_logger.debug.call_args[0]
|
||||
for value in (command, retcode, output, err,):
|
||||
self.assertTrue(value in log_args)
|
||||
|
||||
@mock.patch('certbot_postfix.util.subprocess.Popen')
|
||||
def test_success(self, mock_popen):
|
||||
command = 'foo'
|
||||
expected = ('bar', '')
|
||||
mock_popen().communicate.return_value = expected
|
||||
mock_popen().poll.return_value = 0
|
||||
|
||||
self.assertEqual(self._call(command), expected)
|
||||
|
||||
def test_stdout_error(self):
|
||||
self.assertRaises(ValueError, self._call, stdout=None)
|
||||
|
||||
def test_stderr_error(self):
|
||||
self.assertRaises(ValueError, self._call, stderr=None)
|
||||
|
||||
def test_universal_newlines_error(self):
|
||||
self.assertRaises(ValueError, self._call, universal_newlines=False)
|
||||
|
||||
|
||||
class VerifyExeExistsTest(unittest.TestCase):
|
||||
"""Tests for certbot_postfix.util.verify_exe_exists."""
|
||||
|
||||
@classmethod
|
||||
def _call(cls, *args, **kwargs):
|
||||
from certbot_postfix.util import verify_exe_exists
|
||||
return verify_exe_exists(*args, **kwargs)
|
||||
|
||||
@mock.patch('certbot_postfix.util.certbot_util.exe_exists')
|
||||
@mock.patch('certbot_postfix.util.plugins_util.path_surgery')
|
||||
def test_failure(self, mock_exe_exists, mock_path_surgery):
|
||||
mock_exe_exists.return_value = mock_path_surgery.return_value = False
|
||||
self.assertRaises(errors.NoInstallationError, self._call, 'foo')
|
||||
|
||||
@mock.patch('certbot_postfix.util.certbot_util.exe_exists')
|
||||
def test_simple_success(self, mock_exe_exists):
|
||||
mock_exe_exists.return_value = True
|
||||
self._call('foo')
|
||||
|
||||
@mock.patch('certbot_postfix.util.certbot_util.exe_exists')
|
||||
@mock.patch('certbot_postfix.util.plugins_util.path_surgery')
|
||||
def test_successful_surgery(self, mock_exe_exists, mock_path_surgery):
|
||||
mock_exe_exists.return_value = False
|
||||
mock_path_surgery.return_value = True
|
||||
self._call('foo')
|
||||
|
||||
|
||||
class CheckCallTest(unittest.TestCase):
|
||||
"""Tests for certbot_postfix.util.check_call."""
|
||||
|
||||
|
||||
@@ -25,5 +25,6 @@ fi
|
||||
-e certbot-dns-rfc2136 \
|
||||
-e certbot-dns-route53 \
|
||||
-e certbot-nginx \
|
||||
-e certbot-postfix \
|
||||
-e letshelp-certbot \
|
||||
-e certbot-compatibility-test
|
||||
|
||||
@@ -23,5 +23,6 @@ fi
|
||||
-e certbot-dns-nsone \
|
||||
-e certbot-dns-route53 \
|
||||
-e certbot-nginx \
|
||||
-e certbot-postfix \
|
||||
-e letshelp-certbot \
|
||||
-e certbot-compatibility-test
|
||||
|
||||
+3
-1
@@ -9,7 +9,7 @@
|
||||
# -e makes sure we fail fast and don't submit coveralls submit
|
||||
|
||||
if [ "xxx$1" = "xxx" ]; then
|
||||
pkgs="certbot acme certbot_apache certbot_dns_cloudflare certbot_dns_cloudxns certbot_dns_digitalocean certbot_dns_dnsimple certbot_dns_dnsmadeeasy certbot_dns_google certbot_dns_luadns certbot_dns_nsone certbot_dns_rfc2136 certbot_dns_route53 certbot_nginx letshelp_certbot"
|
||||
pkgs="certbot acme certbot_apache certbot_dns_cloudflare certbot_dns_cloudxns certbot_dns_digitalocean certbot_dns_dnsimple certbot_dns_dnsmadeeasy certbot_dns_google certbot_dns_luadns certbot_dns_nsone certbot_dns_rfc2136 certbot_dns_route53 certbot_nginx certbot_postfix letshelp_certbot"
|
||||
else
|
||||
pkgs="$@"
|
||||
fi
|
||||
@@ -43,6 +43,8 @@ cover () {
|
||||
min=99
|
||||
elif [ "$1" = "certbot_nginx" ]; then
|
||||
min=97
|
||||
elif [ "$1" = "certbot_postfix" ]; then
|
||||
min=100
|
||||
elif [ "$1" = "letshelp_certbot" ]; then
|
||||
min=100
|
||||
else
|
||||
|
||||
@@ -28,6 +28,7 @@ py26_packages =
|
||||
certbot-dns-rfc2136 \
|
||||
certbot-dns-route53 \
|
||||
certbot-nginx \
|
||||
certbot-postfix \
|
||||
letshelp-certbot
|
||||
non_py26_packages =
|
||||
certbot-dns-cloudxns \
|
||||
@@ -55,6 +56,7 @@ source_paths =
|
||||
certbot-dns-rfc2136/certbot_dns_rfc2136
|
||||
certbot-dns-route53/certbot_dns_route53
|
||||
certbot-nginx/certbot_nginx
|
||||
certbot-postfix/certbot_postfix
|
||||
letshelp-certbot/letshelp_certbot
|
||||
tests/lock_test.py
|
||||
|
||||
|
||||
Reference in New Issue
Block a user