certbot/certbot/cert_manager.py

"""Tools for managing certificates."""
import datetime
import logging
import os
import pytz
import re
import traceback
import zope.component

from acme.magic_typing import List  # pylint: disable=unused-import, no-name-in-module
from certbot import crypto_util
from certbot import errors
from certbot import interfaces
from certbot import ocsp
from certbot import storage
from certbot import util

from certbot.display import util as display_util

logger = logging.getLogger(__name__)

###################
# Commands
###################

def update_live_symlinks(config):
    """Update the certificate file family symlinks to use archive_dir.

    Use the information in the config file to make symlinks point to
    the correct archive directory.

    .. note:: This assumes that the installation is using a Reverter object.

    :param config: Configuration.
    :type config: :class:`certbot.configuration.NamespaceConfig`

    """
    for renewal_file in storage.renewal_conf_files(config):
        storage.RenewableCert(renewal_file, config, update_symlinks=True)

def rename_lineage(config):
    """Rename the specified lineage to the new name.

    :param config: Configuration.
    :type config: :class:`certbot.configuration.NamespaceConfig`

    """
    disp = zope.component.getUtility(interfaces.IDisplay)

    certname = get_certnames(config, "rename")[0]

    new_certname = config.new_certname
    if not new_certname:
        code, new_certname = disp.input(
            "Enter the new name for certificate {0}".format(certname),
            flag="--updated-cert-name", force_interactive=True)
        if code != display_util.OK or not new_certname:
            raise errors.Error("User ended interaction.")

    lineage = lineage_for_certname(config, certname)
    if not lineage:
        raise errors.ConfigurationError("No existing certificate with name "
            "{0} found.".format(certname))
    storage.rename_renewal_config(certname, new_certname, config)
    disp.notification("Successfully renamed {0} to {1}."
        .format(certname, new_certname), pause=False)

def certificates(config):
    """Display information about certs configured with Certbot

    :param config: Configuration.
    :type config: :class:`certbot.configuration.NamespaceConfig`
    """
    parsed_certs = []
    parse_failures = []
    for renewal_file in storage.renewal_conf_files(config):
        try:
            renewal_candidate = storage.RenewableCert(renewal_file, config)
            crypto_util.verify_renewable_cert(renewal_candidate)
            parsed_certs.append(renewal_candidate)
        except Exception as e:  # pylint: disable=broad-except
            logger.warning("Renewal configuration file %s produced an "
                           "unexpected error: %s. Skipping.", renewal_file, e)
            logger.debug("Traceback was:\n%s", traceback.format_exc())
            parse_failures.append(renewal_file)

    # Describe all the certs
    _describe_certs(config, parsed_certs, parse_failures)

def delete(config):
    """Delete Certbot files associated with a certificate lineage."""
    certnames = get_certnames(config, "delete", allow_multiple=True)
    for certname in certnames:
        storage.delete_files(config, certname)
        disp = zope.component.getUtility(interfaces.IDisplay)
        disp.notification("Deleted all files relating to certificate {0}."
            .format(certname), pause=False)

###################
# Public Helpers
###################

def lineage_for_certname(cli_config, certname):
    """Find a lineage object with name certname."""
    configs_dir = cli_config.renewal_configs_dir
    # Verify the directory is there
    util.make_or_verify_dir(configs_dir, mode=0o755, uid=os.geteuid())
    try:
        renewal_file = storage.renewal_file_for_certname(cli_config, certname)
    except errors.CertStorageError:
        return None
    try:
        return storage.RenewableCert(renewal_file, cli_config)
    except (errors.CertStorageError, IOError):
        logger.debug("Renewal conf file %s is broken.", renewal_file)
        logger.debug("Traceback was:\n%s", traceback.format_exc())
        return None

def domains_for_certname(config, certname):
    """Find the domains in the cert with name certname."""
    lineage = lineage_for_certname(config, certname)
    return lineage.names() if lineage else None

def find_duplicative_certs(config, domains):
    """Find existing certs that match the given domain names.

    This function searches for certificates whose domains are equal to
    the `domains` parameter and certificates whose domains are a subset
    of the domains in the `domains` parameter. If multiple certificates
    are found whose names are a subset of `domains`, the one whose names
    are the largest subset of `domains` is returned.

    If multiple certificates' domains are an exact match or equally
    sized subsets, which matching certificates are returned is
    undefined.

    :param config: Configuration.
    :type config: :class:`certbot.configuration.NamespaceConfig`
    :param domains: List of domain names
    :type domains: `list` of `str`

    :returns: lineages representing the identically matching cert and the
        largest subset if they exist
    :rtype: `tuple` of `storage.RenewableCert` or `None`

    """
    def update_certs_for_domain_matches(candidate_lineage, rv):
        """Return cert as identical_names_cert if it matches,
           or subset_names_cert if it matches as subset
        """
        # TODO: Handle these differently depending on whether they are
        #       expired or still valid?
        identical_names_cert, subset_names_cert = rv
        candidate_names = set(candidate_lineage.names())
        if candidate_names == set(domains):
            identical_names_cert = candidate_lineage
        elif candidate_names.issubset(set(domains)):
            # This logic finds and returns the largest subset-names cert
            # in the case where there are several available.
            if subset_names_cert is None:
                subset_names_cert = candidate_lineage
            elif len(candidate_names) > len(subset_names_cert.names()):
                subset_names_cert = candidate_lineage
        return (identical_names_cert, subset_names_cert)

    return _search_lineages(config, update_certs_for_domain_matches, (None, None))

def _archive_files(candidate_lineage, filetype):
    """ In order to match things like:
        /etc/letsencrypt/archive/example.com/chain1.pem.

        Anonymous functions which call this function are eventually passed (in a list) to
        `match_and_check_overlaps` to help specify the acceptable_matches.

        :param `.storage.RenewableCert` candidate_lineage: Lineage whose archive dir is to
            be searched.
        :param str filetype: main file name prefix e.g. "fullchain" or "chain".

        :returns: Files in candidate_lineage's archive dir that match the provided filetype.
        :rtype: list of str or None
    """
    archive_dir = candidate_lineage.archive_dir
    pattern = [os.path.join(archive_dir, f) for f in os.listdir(archive_dir)
                    if re.match("{0}[0-9]*.pem".format(filetype), f)]
    if len(pattern) > 0:
        return pattern
    else:
        return None

def _acceptable_matches():
    """ Generates the list that's passed to match_and_check_overlaps. Is its own function to
    make unit testing easier.

    :returns: list of functions
    :rtype: list
    """
    return [lambda x: x.fullchain_path, lambda x: x.cert_path,
            lambda x: _archive_files(x, "cert"), lambda x: _archive_files(x, "fullchain")]

def cert_path_to_lineage(cli_config):
    """ If config.cert_path is defined, try to find an appropriate value for config.certname.

    :param `configuration.NamespaceConfig` cli_config: parsed command line arguments

    :returns: a lineage name
    :rtype: str

    :raises `errors.Error`: If the specified cert path can't be matched to a lineage name.
    :raises `errors.OverlappingMatchFound`: If the matched lineage's archive is shared.
    """
    acceptable_matches = _acceptable_matches()
    match = match_and_check_overlaps(cli_config, acceptable_matches,
            lambda x: cli_config.cert_path[0], lambda x: x.lineagename)
    return match[0]

def match_and_check_overlaps(cli_config, acceptable_matches, match_func, rv_func):
    """ Searches through all lineages for a match, and checks for duplicates.
    If a duplicate is found, an error is raised, as performing operations on lineages
    that have their properties incorrectly duplicated elsewhere is probably a bad idea.

    :param `configuration.NamespaceConfig` cli_config: parsed command line arguments
    :param list acceptable_matches: a list of functions that specify acceptable matches
    :param function match_func: specifies what to match
    :param function rv_func: specifies what to return

    """
    def find_matches(candidate_lineage, return_value, acceptable_matches):
        """Returns a list of matches using _search_lineages."""
        acceptable_matches = [func(candidate_lineage) for func in acceptable_matches]
        acceptable_matches_rv = []  # type: List[str]
        for item in acceptable_matches:
            if isinstance(item, list):
                acceptable_matches_rv += item
            else:
                acceptable_matches_rv.append(item)
        match = match_func(candidate_lineage)
        if match in acceptable_matches_rv:
            return_value.append(rv_func(candidate_lineage))
        return return_value

    matched = _search_lineages(cli_config, find_matches, [], acceptable_matches)
    if not matched:
        raise errors.Error("No match found for cert-path {0}!".format(cli_config.cert_path[0]))
    elif len(matched) > 1:
        raise errors.OverlappingMatchFound()
    else:
        return matched

def human_readable_cert_info(config, cert, skip_filter_checks=False):
    """ Returns a human readable description of info about a RenewableCert object"""
    certinfo = []
    checker = ocsp.RevocationChecker()

    if config.certname and cert.lineagename != config.certname and not skip_filter_checks:
        return ""
    if config.domains and not set(config.domains).issubset(cert.names()):
        return ""
    now = pytz.UTC.fromutc(datetime.datetime.utcnow())

    reasons = []
    if cert.is_test_cert:
        reasons.append('TEST_CERT')
    if cert.target_expiry <= now:
        reasons.append('EXPIRED')
    if checker.ocsp_revoked(cert.cert, cert.chain):
        reasons.append('REVOKED')

    if reasons:
        status = "INVALID: " + ", ".join(reasons)
    else:
        diff = cert.target_expiry - now
        if diff.days == 1:
            status = "VALID: 1 day"
        elif diff.days < 1:
            status = "VALID: {0} hour(s)".format(diff.seconds // 3600)
        else:
            status = "VALID: {0} days".format(diff.days)

    valid_string = "{0} ({1})".format(cert.target_expiry, status)
    certinfo.append("  Certificate Name: {0}\n"
                    "    Domains: {1}\n"
                    "    Expiry Date: {2}\n"
                    "    Certificate Path: {3}\n"
                    "    Private Key Path: {4}".format(
                         cert.lineagename,
                         " ".join(cert.names()),
                         valid_string,
                         cert.fullchain,
                         cert.privkey))
    return "".join(certinfo)

def get_certnames(config, verb, allow_multiple=False, custom_prompt=None):
    """Get certname from flag, interactively, or error out.
    """
    certname = config.certname
    if certname:
        certnames = [certname]
    else:
        disp = zope.component.getUtility(interfaces.IDisplay)
        filenames = storage.renewal_conf_files(config)
        choices = [storage.lineagename_for_filename(name) for name in filenames]
        if not choices:
            raise errors.Error("No existing certificates found.")
        if allow_multiple:
            if not custom_prompt:
                prompt = "Which certificate(s) would you like to {0}?".format(verb)
            else:
                prompt = custom_prompt
            code, certnames = disp.checklist(
                prompt, choices, cli_flag="--cert-name", force_interactive=True)
            if code != display_util.OK:
                raise errors.Error("User ended interaction.")
        else:
            if not custom_prompt:
                prompt = "Which certificate would you like to {0}?".format(verb)
            else:
                prompt = custom_prompt

            code, index = disp.menu(
                prompt, choices, cli_flag="--cert-name", force_interactive=True)

            if code != display_util.OK or index not in range(0, len(choices)):
                raise errors.Error("User ended interaction.")
            certnames = [choices[index]]
    return certnames

###################
# Private Helpers
###################

def _report_lines(msgs):
    """Format a results report for a category of single-line renewal outcomes"""
    return "  " + "\n  ".join(str(msg) for msg in msgs)

def _report_human_readable(config, parsed_certs):
    """Format a results report for a parsed cert"""
    certinfo = []
    for cert in parsed_certs:
        certinfo.append(human_readable_cert_info(config, cert))
    return "\n".join(certinfo)

def _describe_certs(config, parsed_certs, parse_failures):
    """Print information about the certs we know about"""
    out = []  # type: List[str]

    notify = out.append

    if not parsed_certs and not parse_failures:
        notify("No certs found.")
    else:
        if parsed_certs:
            match = "matching " if config.certname or config.domains else ""
            notify("Found the following {0}certs:".format(match))
            notify(_report_human_readable(config, parsed_certs))
        if parse_failures:
            notify("\nThe following renewal configuration files "
               "were invalid:")
            notify(_report_lines(parse_failures))

    disp = zope.component.getUtility(interfaces.IDisplay)
    disp.notification("\n".join(out), pause=False, wrap=False)

def _search_lineages(cli_config, func, initial_rv, *args):
    """Iterate func over unbroken lineages, allowing custom return conditions.

    Allows flexible customization of return values, including multiple
    return values and complex checks.

    :param `configuration.NamespaceConfig` cli_config: parsed command line arguments
    :param function func: function used while searching over lineages
    :param initial_rv: initial return value of the function (any type)

    :returns: Whatever was specified by `func` if a match is found.
    """
    configs_dir = cli_config.renewal_configs_dir
    # Verify the directory is there
    util.make_or_verify_dir(configs_dir, mode=0o755, uid=os.geteuid())

    rv = initial_rv
    for renewal_file in storage.renewal_conf_files(cli_config):
        try:
            candidate_lineage = storage.RenewableCert(renewal_file, cli_config)
        except (errors.CertStorageError, IOError):
            logger.debug("Renewal conf file %s is broken. Skipping.", renewal_file)
            logger.debug("Traceback was:\n%s", traceback.format_exc())
            continue
        rv = func(candidate_lineage, rv, *args)
    return rv