path: root/certbot-postfix/certbot_postfix/installer.py

"""certbot installer plugin for postfix."""
import logging
import os

import zope.interface
import zope.component
import six

from certbot import errors
from certbot import interfaces
from certbot import util as certbot_util
from certbot.plugins import common as plugins_common

from certbot_postfix import constants
from certbot_postfix import postconf
from certbot_postfix import util

# pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Callable, Dict, List
# pylint: enable=unused-import, no-name-in-module

logger = logging.getLogger(__name__)

@zope.interface.implementer(interfaces.IInstaller)
@zope.interface.provider(interfaces.IPluginFactory)
class Installer(plugins_common.Installer):
    """Certbot installer plugin for Postfix.

    :ivar str config_dir: Postfix configuration directory to modify
    :ivar list save_notes: documentation for proposed changes. This is
        cleared and stored in Certbot checkpoints when save() is called

    :ivar postconf: Wrapper for Postfix configuration command-line tool.
    :type postconf: :class: `certbot_postfix.postconf.ConfigMain`
    :ivar postfix: Wrapper for Postfix command-line tool.
    :type postfix: :class: `certbot_postfix.util.PostfixUtil`
    """

    description = "Configure TLS with the Postfix MTA"

    @classmethod
    def add_parser_arguments(cls, add):
        add("ctl", default=constants.CLI_DEFAULTS["ctl"],
            help="Path to the 'postfix' control program.")
        # This directory points to Postfix's configuration directory.
        add("config-dir", default=constants.CLI_DEFAULTS["config_dir"],
            help="Path to the directory containing the "
            "Postfix main.cf file to modify instead of using the "
            "default configuration paths.")
        add("config-utility", default=constants.CLI_DEFAULTS["config_utility"],
            help="Path to the 'postconf' executable.")
        add("tls-only", action="store_true", default=constants.CLI_DEFAULTS["tls_only"],
            help="Only set params to enable opportunistic TLS and install certificates.")
        add("server-only", action="store_true", default=constants.CLI_DEFAULTS["server_only"],
            help="Only set server params (prefixed with smtpd*)")
        add("ignore-master-overrides", action="store_true",
            default=constants.CLI_DEFAULTS["ignore_master_overrides"],
            help="Ignore errors reporting overridden TLS parameters in master.cf.")

    def __init__(self, *args, **kwargs):
        super(Installer, self).__init__(*args, **kwargs)
        # Wrapper around postconf commands
        self.postfix = None
        self.postconf = None

        # Files to save
        self.save_notes = [] # type: List[str]

        self._enhance_func = {} # type: Dict[str, Callable[[str, str], None]]
        # Since we only need to enable TLS once for all domains,
        # keep track of whether this enhancement was already called.
        self._tls_enabled = False

    def prepare(self):
        """Prepare the installer.

        :raises errors.PluginError: when an unexpected error occurs
        :raises errors.MisconfigurationError: when the config is invalid
        :raises errors.NoInstallationError: when can't find installation
        :raises errors.NotSupportedError: when version is not supported
        """
        # Verify postfix and postconf are installed
        for param in ("ctl", "config_utility",):
            util.verify_exe_exists(self.conf(param),
                    "Cannot find executable '{0}'. You can provide the "
                    "path to this command with --{1}".format(
                        self.conf(param),
                        self.option_name(param)))

        # Set up CLI tools
        self.postfix = util.PostfixUtil(self.conf('config-dir'))
        self.postconf = postconf.ConfigMain(self.conf('config-utility'),
                                            self.conf('ignore-master-overrides'),
                                            self.conf('config-dir'))

        # Ensure current configuration is valid.
        self.config_test()

        # Check Postfix version
        self._check_version()
        self._lock_config_dir()
        self.install_ssl_dhparams()

    def config_test(self):
        """Test to see that the current Postfix configuration is valid.

        :raises errors.MisconfigurationError: If the configuration is invalid.
        """
        self.postfix.test()

    def _check_version(self):
        """Verifies that the installed Postfix version is supported.

        :raises errors.NotSupportedError: if the version is unsupported
        """
        if self._get_version() < constants.MINIMUM_VERSION:
            version_string = '.'.join([str(n) for n in constants.MINIMUM_VERSION])
            raise errors.NotSupportedError('Postfix version must be at least %s' % version_string)

    def _lock_config_dir(self):
        """Stop two Postfix plugins from modifying the config at once.

        :raises .PluginError: if unable to acquire the lock
        """
        try:
            certbot_util.lock_dir_until_exit(self.conf('config-dir'))
        except (OSError, errors.LockError):
            logger.debug("Encountered error:", exc_info=True)
            raise errors.PluginError(
                "Unable to lock %s" % self.conf('config-dir'))

    def more_info(self):
        """Human-readable string to help the user. Describes steps taken and any relevant
        info to help the user decide which plugin to use.

        :rtype: str
        """
        return (
            "Configures Postfix to try to authenticate mail servers, use "
            "installed certificates and disable weak ciphers and protocols.{0}"
            "Server root: {root}{0}"
            "Version: {version}".format(
                os.linesep,
                root=self.conf('config-dir'),
                version='.'.join([str(i) for i in self._get_version()]))
        )

    def _get_version(self):
        """Return the version of Postfix, as a tuple. (e.g. '2.11.3' is (2, 11, 3))

        :returns: version
        :rtype: tuple

        :raises errors.PluginError: Unable to find Postfix version.
        """
        mail_version = self.postconf.get_default("mail_version")
        return tuple(int(i) for i in mail_version.split('.'))

    def get_all_names(self):
        """Returns all names that may be authenticated.

        :rtype: `set` of `str`

        """
        return certbot_util.get_filtered_names(self.postconf.get(var)
                   for var in ('mydomain', 'myhostname', 'myorigin',))

    def _set_vars(self, var_dict):
        """Sets all parameters in var_dict to config file. If current value is already set
        as more secure (acceptable), then don't set/overwrite it.
        """
        for param, acceptable in six.iteritems(var_dict):
            if not util.is_acceptable_value(param, self.postconf.get(param), acceptable):
                self.postconf.set(param, acceptable[0], acceptable)

    def _confirm_changes(self):
        """Confirming outstanding updates for configuration parameters.

        :raises errors.PluginError: when user rejects the configuration changes.
        """
        updates = self.postconf.get_changes()
        output_string = "Postfix TLS configuration parameters to update in main.cf:\n"
        for name, value in six.iteritems(updates):
            output_string += "{0} = {1}\n".format(name, value)
        output_string += "Is this okay?\n"
        if not zope.component.getUtility(interfaces.IDisplay).yesno(output_string,
            force_interactive=True, default=True):
            raise errors.PluginError(
                "Manually rejected configuration changes.\n"
                "Try using --tls-only or --server-only to change a particular"
                "subset of configuration parameters.")

    def deploy_cert(self, domain, cert_path,
                    key_path, chain_path, fullchain_path):
        """Configure the Postfix SMTP server to use the given TLS cert.

        :param str domain: domain to deploy certificate file
        :param str cert_path: absolute path to the certificate file
        :param str key_path: absolute path to the private key file
        :param str chain_path: absolute path to the certificate chain file
        :param str fullchain_path: absolute path to the certificate fullchain
            file (cert plus chain)

        :raises .PluginError: when cert cannot be deployed

        """
        # pylint: disable=unused-argument
        if self._tls_enabled:
            return
        self._tls_enabled = True
        self.save_notes.append("Configuring TLS for {0}".format(domain))
        self.postconf.set("smtpd_tls_cert_file", cert_path)
        self.postconf.set("smtpd_tls_key_file", key_path)
        self._set_vars(constants.TLS_SERVER_VARS)
        if not self.conf('server_only'):
            self._set_vars(constants.TLS_CLIENT_VARS)
        if not self.conf('tls_only'):
            self._set_vars(constants.DEFAULT_SERVER_VARS)
            if not self.conf('server_only'):
                self._set_vars(constants.DEFAULT_CLIENT_VARS)
            # Despite the name, this option also supports 2048-bit DH params.
            # http://www.postfix.org/FORWARD_SECRECY_README.html#server_fs
            self.postconf.set("smtpd_tls_dh1024_param_file", self.ssl_dhparams)
        self._confirm_changes()

    def enhance(self, domain, enhancement, options=None):
        """Raises an exception since this installer doesn't support any enhancements.
        """
        # pylint: disable=unused-argument
        raise errors.PluginError(
            "Unsupported enhancement: {0}".format(enhancement))

    def supported_enhancements(self):
        """Returns a list of supported enhancements.

        :rtype: list

        """
        return []

    def save(self, title=None, temporary=False):
        """Creates backups and writes changes to configuration files.

        :param str title: The title of the save. If a title is given, the
            configuration will be saved as a new checkpoint and put in a
            timestamped directory. `title` has no effect if temporary is true.

        :param bool temporary: Indicates whether the changes made will
            be quickly reversed in the future (challenges)

        :raises errors.PluginError: when save is unsuccessful
        """
        save_files = set((os.path.join(self.conf('config-dir'), "main.cf"),))
        self.add_to_checkpoint(save_files,
                               "\n".join(self.save_notes), temporary)
        self.postconf.flush()

        del self.save_notes[:]

        if title and not temporary:
            self.finalize_checkpoint(title)

    def recovery_routine(self):
        super(Installer, self).recovery_routine()
        self.postconf = postconf.ConfigMain(self.conf('config-utility'),
                                            self.conf('ignore-master-overrides'),
                                            self.conf('config-dir'))

    def rollback_checkpoints(self, rollback=1):
        """Rollback saved checkpoints.

        :param int rollback: Number of checkpoints to revert

        :raises .errors.PluginError: If there is a problem with the input or
            the function is unable to correctly revert the configuration
        """
        super(Installer, self).rollback_checkpoints(rollback)
        self.postconf = postconf.ConfigMain(self.conf('config-utility'),
                                            self.conf('ignore-master-overrides'),
                                            self.conf('config-dir'))

    def restart(self):
        """Restart or refresh the server content.

        :raises .PluginError: when server cannot be restarted
        """
        self.postfix.restart()