Gecko [ 167.86.127.34 ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
display	[ DIR ]	drwxr-xr-x
plugins	[ DIR ]	drwxr-xr-x
tests	[ DIR ]	drwxr-xr-x
__init__.py	114 B	-rw-r--r--
account.py	13.98 KB	-rw-r--r--
achallenges.py	1.59 KB	-rw-r--r--
auth_handler.py	20.92 KB	-rw-r--r--
cert_manager.py	15.1 KB	-rw-r--r--
cli.py	71.49 KB	-rw-r--r--
client.py	28.72 KB	-rw-r--r--
compat.py	6.91 KB	-rw-r--r--
configuration.py	5.66 KB	-rw-r--r--
constants.py	6.54 KB	-rw-r--r--
crypto_util.py	15.29 KB	-rw-r--r--
eff.py	3.07 KB	-rw-r--r--
error_handler.py	5.81 KB	-rw-r--r--
errors.py	2.59 KB	-rw-r--r--
hooks.py	8.44 KB	-rw-r--r--
interfaces.py	22.02 KB	-rw-r--r--
lock.py	3.56 KB	-rw-r--r--
log.py	12.39 KB	-rw-r--r--
main.py	48.47 KB	-rw-r--r--
notify.py	1.04 KB	-rw-r--r--
ocsp.py	4.1 KB	-rw-r--r--
renewal.py	20.91 KB	-rw-r--r--
reporter.py	3.46 KB	-rw-r--r--
reverter.py	23.32 KB	-rw-r--r--
ssl-dhparams.pem	424 B	-rw-r--r--
storage.py	44.91 KB	-rw-r--r--
updater.py	3.86 KB	-rw-r--r--
util.py	20.35 KB	-rw-r--r--

Code Editor : 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 compat
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=compat.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 configurations "
               "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=compat.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

${this.title}

Code Editor : cert_manager.py