upstream/certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-06-09 00:32:12 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Merge remote-tracking branch 'upstream/master'

Browse source
This commit is contained in:
Brandon Kreisel 2015-09-27 14:58:34 -04:00
commit 756cfb75ef
32 changed files with 502 additions and 1339 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
acme/setup.py
View file

@ -16,6 +16,7 @@ install_requires = [
'pyrfc3339',
'pytz',
'requests',
'setuptools', # pkg_resources
'six',
'werkzeug',
]

1
examples/dev-cli.ini
View file

@ -9,6 +9,7 @@ domains = example.com
text = True
agree-eula = True
agree-tos = True
debug = True
# Unfortunately, it's not possible to specify "verbose" multiple times
# (correspondingly to -vvvvvv)

1
letsencrypt-apache/setup.py
View file

@ -7,6 +7,7 @@ install_requires = [
'letsencrypt',
'mock<1.1.0', # py26
'python-augeas',
'setuptools', # pkg_resources
'zope.component',
'zope.interface',
]

1
letsencrypt-nginx/setup.py
View file

@ -8,6 +8,7 @@ install_requires = [
'mock<1.1.0', # py26
'PyOpenSSL',
'pyparsing>=1.5.5', # Python3 support; perhaps unnecessary?
'setuptools', # pkg_resources
'zope.interface',
]

22
letsencrypt/auth_handler.py
View file

@ -11,6 +11,7 @@ from acme import messages
from letsencrypt import achallenges
from letsencrypt import constants
from letsencrypt import errors
from letsencrypt import error_handler
from letsencrypt import interfaces
@ -106,17 +107,16 @@ class AuthHandler(object):
"""Get Responses for challenges from authenticators."""
cont_resp = []
dv_resp = []
try:
if self.cont_c:
cont_resp = self.cont_auth.perform(self.cont_c)
if self.dv_c:
dv_resp = self.dv_auth.perform(self.dv_c)
# This will catch both specific types of errors.
except errors.AuthorizationError:
logger.critical("Failure in setting up challenges.")
logger.info("Attempting to clean up outstanding challenges...")
self._cleanup_challenges()
raise
with error_handler.ErrorHandler(self._cleanup_challenges):
try:
if self.cont_c:
cont_resp = self.cont_auth.perform(self.cont_c)
if self.dv_c:
dv_resp = self.dv_auth.perform(self.dv_c)
except errors.AuthorizationError:
logger.critical("Failure in setting up challenges.")
logger.info("Attempting to clean up outstanding challenges...")
raise
assert len(cont_resp) == len(self.cont_c)
assert len(dv_resp) == len(self.dv_c)

205
letsencrypt/cli.py
View file

@ -24,6 +24,7 @@ from acme import jose
import letsencrypt
from letsencrypt import account
from letsencrypt import colored_logging
from letsencrypt import configuration
from letsencrypt import constants
from letsencrypt import client
@ -172,6 +173,7 @@ def _find_duplicative_certs(domains, config, renew_config):
identical_names_cert, subset_names_cert = None, None
configs_dir = renew_config.renewal_configs_dir
# Verify the directory is there
le_util.make_or_verify_dir(configs_dir, mode=0o755, uid=os.geteuid())
cli_config = configuration.RenewerConfiguration(config)
@ -198,8 +200,106 @@ def _find_duplicative_certs(domains, config, renew_config):
return identical_names_cert, subset_names_cert
def _treat_as_renewal(config, domains):
"""Determine whether or not the call should be treated as a renewal.
:returns: RenewableCert or None if renewal shouldn't occur.
:rtype: :class:`.storage.RenewableCert`
:raises .Error: If the user would like to rerun the client again.
"""
renewal = False
# Considering the possibility that the requested certificate is
# related to an existing certificate. (config.duplicate, which
# is set with --duplicate, skips all of this logic and forces any
# kind of certificate to be obtained with renewal = False.)
if not config.duplicate:
ident_names_cert, subset_names_cert = _find_duplicative_certs(
domains, config, configuration.RenewerConfiguration(config))
# I am not sure whether that correctly reads the systemwide
# configuration file.
question = None
if ident_names_cert is not None:
question = (
"You have an existing certificate that contains exactly the "
"same domains you requested (ref: {0}){br}{br}Do you want to "
"renew and replace this certificate with a newly-issued one?"
).format(ident_names_cert.configfile.filename, br=os.linesep)
elif subset_names_cert is not None:
question = (
"You have an existing certificate that contains a portion of "
"the domains you requested (ref: {0}){br}{br}It contains these "
"names: {1}{br}{br}You requested these names for the new "
"certificate: {2}.{br}{br}Do you want to replace this existing "
"certificate with the new certificate?"
).format(subset_names_cert.configfile.filename,
", ".join(subset_names_cert.names()),
", ".join(domains),
br=os.linesep)
if question is None:
# We aren't in a duplicative-names situation at all, so we don't
# have to tell or ask the user anything about this.
pass
elif config.renew_by_default or zope.component.getUtility(
interfaces.IDisplay).yesno(question, "Replace", "Cancel"):
renewal = True
else:
reporter_util = zope.component.getUtility(interfaces.IReporter)
reporter_util.add_message(
"To obtain a new certificate that {0} an existing certificate "
"in its domain-name coverage, you must use the --duplicate "
"option.{br}{br}For example:{br}{br}{1} --duplicate {2}".format(
"duplicates" if ident_names_cert is not None else
"overlaps with",
sys.argv[0], " ".join(sys.argv[1:]),
br=os.linesep
),
reporter_util.HIGH_PRIORITY)
raise errors.Error(
"User did not use proper CLI and would like "
"to reinvoke the client.")
if renewal:
return ident_names_cert if ident_names_cert is not None else subset_names_cert
return None
def _auth_from_domains(le_client, config, domains, plugins):
"""Authenticate and enroll certificate."""
# Note: This can raise errors... caught above us though.
lineage = _treat_as_renewal(config, domains)
if lineage is not None:
# TODO: schoen wishes to reuse key - discussion
# https://github.com/letsencrypt/letsencrypt/pull/777/files#r40498574
new_certr, new_chain, new_key, _ = le_client.obtain_certificate(domains)
# TODO: Check whether it worked! <- or make sure errors are thrown (jdk)
lineage.save_successor(
lineage.latest_common_version(), OpenSSL.crypto.dump_certificate(
OpenSSL.crypto.FILETYPE_PEM, new_certr.body),
new_key.pem, crypto_util.dump_pyopenssl_chain(new_chain))
lineage.update_all_links_to(lineage.latest_common_version())
# TODO: Check return value of save_successor
# TODO: Also update lineage renewal config with any relevant
# configuration values from this attempt? <- Absolutely (jdkasten)
else:
# TREAT AS NEW REQUEST
lineage = le_client.obtain_and_enroll_certificate(domains, plugins)
if not lineage:
raise errors.Error("Certificate could not be obtained")
return lineage
# TODO: Make run as close to auth + install as possible
# Possible difficulties: args.csr was hacked into auth
def run(args, config, plugins): # pylint: disable=too-many-branches,too-many-locals
"""Obtain a certificate and install."""
# Begin authenticator and installer setup
if args.configurator is not None and (args.installer is not None or
args.authenticator is not None):
return ("Either --configurator or --authenticator/--installer"
@ -218,92 +318,28 @@ def run(args, config, plugins): # pylint: disable=too-many-branches,too-many-lo
if installer is None or authenticator is None:
return "Configurator could not be determined"
# End authenticator and installer setup
domains = _find_domains(args, installer)
treat_as_renewal = False
# Considering the possibility that the requested certificate is
# related to an existing certificate. (config.duplicate, which
# is set with --duplicate, skips all of this logic and forces any
# kind of certificate to be obtained with treat_as_renewal = False.)
if not config.duplicate:
identical_names_cert, subset_names_cert = _find_duplicative_certs(
domains, config, configuration.RenewerConfiguration(config))
# I am not sure whether that correctly reads the systemwide
# configuration file.
question = None
if identical_names_cert is not None:
question = (
"You have an existing certificate that contains exactly the "
"same domains you requested (ref: {0})\n\nDo you want to "
"renew and replace this certificate with a newly-issued one?"
).format(identical_names_cert.configfile.filename)
elif subset_names_cert is not None:
question = (
"You have an existing certificate that contains a portion of "
"the domains you requested (ref: {0})\n\nIt contains these "
"names: {1}\n\nYou requested these names for the new "
"certificate: {2}.\n\nDo you want to replace this existing "
"certificate with the new certificate?"
).format(subset_names_cert.configfile.filename,
", ".join(subset_names_cert.names()),
", ".join(domains))
if question is None:
# We aren't in a duplicative-names situation at all, so we don't
# have to tell or ask the user anything about this.
pass
elif zope.component.getUtility(interfaces.IDisplay).yesno(
question, "Replace", "Cancel"):
treat_as_renewal = True
else:
reporter_util = zope.component.getUtility(interfaces.IReporter)
reporter_util.add_message(
"To obtain a new certificate that {0} an existing certificate "
"in its domain-name coverage, you must use the --duplicate "
"option.\n\nFor example:\n\n{1} --duplicate {2}".format(
"duplicates" if identical_names_cert is not None else
"overlaps with", sys.argv[0], " ".join(sys.argv[1:])),
reporter_util.HIGH_PRIORITY)
return 1
# Attempting to obtain the certificate
# TODO: Handle errors from _init_le_client?
le_client = _init_le_client(args, config, authenticator, installer)
if treat_as_renewal:
lineage = identical_names_cert if identical_names_cert is not None else subset_names_cert
# TODO: Use existing privkey instead of generating a new one
new_certr, new_chain, new_key, _ = le_client.obtain_certificate(domains)
# TODO: Check whether it worked!
lineage.save_successor(
lineage.latest_common_version(), OpenSSL.crypto.dump_certificate(
OpenSSL.crypto.FILETYPE_PEM, new_certr.body),
new_key.pem, crypto_util.dump_pyopenssl_chain(new_chain))
lineage.update_all_links_to(lineage.latest_common_version())
# TODO: Check return value of save_successor
# TODO: Also update lineage renewal config with any relevant
# configuration values from this attempt?
le_client.deploy_certificate(
domains, lineage.privkey, lineage.cert, lineage.chain)
display_ops.success_renewal(domains)
else:
# TREAT AS NEW REQUEST
lineage = le_client.obtain_and_enroll_certificate(
domains, authenticator, installer, plugins)
if not lineage:
return "Certificate could not be obtained"
# TODO: This treats the key as changed even when it wasn't
# TODO: We also need to pass the fullchain (for Nginx)
le_client.deploy_certificate(
domains, lineage.privkey, lineage.cert, lineage.chain)
le_client.enhance_config(domains, args.redirect)
lineage = _auth_from_domains(le_client, config, domains, plugins)
# TODO: We also need to pass the fullchain (for Nginx)
le_client.deploy_certificate(
domains, lineage.privkey, lineage.cert, lineage.chain)
le_client.enhance_config(domains, args.redirect)
if len(lineage.available_versions("cert")) == 1:
display_ops.success_installation(domains)
else:
display_ops.success_renewal(domains)
def auth(args, config, plugins):
"""Authenticate & obtain cert, but do not install it."""
# XXX: Update for renewer / RenewableCert
if args.domains is not None and args.csr is not None:
# TODO: --csr could have a priority, when --domains is
@ -323,6 +359,7 @@ def auth(args, config, plugins):
# TODO: Handle errors from _init_le_client?
le_client = _init_le_client(args, config, authenticator, installer)
# This is a special case; cert and chain are simply saved
if args.csr is not None:
certr, chain = le_client.obtain_certificate_from_csr(le_util.CSR(
file=args.csr[0], data=args.csr[1], form="der"))
@ -330,9 +367,7 @@ def auth(args, config, plugins):
certr, chain, args.cert_path, args.chain_path)
else:
domains = _find_domains(args, installer)
if not le_client.obtain_and_enroll_certificate(
domains, authenticator, installer, plugins):
return "Certificate could not be obtained"
_auth_from_domains(le_client, config, domains, plugins)
def install(args, config, plugins):
@ -483,7 +518,7 @@ class HelpfulArgumentParser(object):
help2 = self.prescan_for_flag("--help", self.help_topics)
assert max(True, "a") == "a", "Gravity changed direction"
help_arg = max(help1, help2)
if help_arg:
if help_arg == True:
# just --help with no topic; avoid argparse altogether
print USAGE
sys.exit(0)
@ -620,8 +655,9 @@ def create_parser(plugins, args):
version="%(prog)s {0}".format(letsencrypt.__version__),
help="show program's version number and exit")
helpful.add(
"automation", "--no-confirm", dest="no_confirm", action="store_true",
help="Turn off confirmation screens, currently used for --revoke")
"automation", "--renew-by-default", action="store_true",
help="Select renewal by default when domains are a superset of a "
"a previously attained cert")
helpful.add(
"automation", "--agree-eula", dest="eula", action="store_true",
help="Agree to the Let's Encrypt Developer Preview EULA")
@ -678,7 +714,7 @@ def create_parser(plugins, args):
# For now unfortunately this constant just needs to match the code below;
# there isn't an elegant way to autogenerate it in time.
VERBS = ["run", "auth", "install", "revoke", "rollback", "config_changes",
"plugins"]
"plugins", "--help"]
def _create_subparsers(helpful):
@ -786,7 +822,7 @@ def _setup_logging(args):
level = -args.verbose_count * 10
fmt = "%(asctime)s:%(levelname)s:%(name)s:%(message)s"
if args.text_mode:
handler = logging.StreamHandler()
handler = colored_logging.StreamHandler()
handler.setFormatter(logging.Formatter(fmt))
else:
handler = log.DialogHandler()
@ -831,7 +867,8 @@ def _handle_exception(exc_type, exc_value, trace, args):
"""
logger.debug(
"Exiting abnormally:\n%s",
"Exiting abnormally:%s%s",
os.linesep,
"".join(traceback.format_exception(exc_type, exc_value, trace)))
if issubclass(exc_type, Exception) and (args is None or not args.debug):

88
letsencrypt/client.py
View file

@ -18,10 +18,10 @@ from letsencrypt import constants
from letsencrypt import continuity_auth
from letsencrypt import crypto_util
from letsencrypt import errors
from letsencrypt import error_handler
from letsencrypt import interfaces
from letsencrypt import le_util
from letsencrypt import reverter
from letsencrypt import revoker
from letsencrypt import storage
from letsencrypt.display import ops as display_ops
@ -111,6 +111,8 @@ class Client(object):
:ivar .AuthHandler auth_handler: Authorizations handler that will
dispatch DV and Continuity challenges to appropriate
authenticators (providing `.IAuthenticator` interface).
:ivar .IAuthenticator dv_auth: Prepared (`.IAuthenticator.prepare`)
authenticator that can solve the `.constants.DV_CHALLENGES`.
:ivar .IInstaller installer: Installer.
:ivar acme.client.Client acme: Optional ACME client API handle.
You might already have one from `register`.
@ -118,14 +120,10 @@ class Client(object):
"""
def __init__(self, config, account_, dv_auth, installer, acme=None):
"""Initialize a client.
:param .IAuthenticator dv_auth: Prepared (`.IAuthenticator.prepare`)
authenticator that can solve the `.constants.DV_CHALLENGES`.
"""
"""Initialize a client."""
self.config = config
self.account = account_
self.dv_auth = dv_auth
self.installer = installer
# Initialize ACME if account is provided
@ -211,12 +209,11 @@ class Client(object):
# Create CSR from names
key = crypto_util.init_save_key(
self.config.rsa_key_size, self.config.key_dir)
csr = crypto_util.init_save_csr(key, domains, self.config.cert_dir)
csr = crypto_util.init_save_csr(key, domains, self.config.csr_dir)
return self._obtain_certificate(domains, csr) + (key, csr)
def obtain_and_enroll_certificate(
self, domains, authenticator, installer, plugins):
def obtain_and_enroll_certificate(self, domains, plugins):
"""Obtain and enroll certificate.
Get a new certificate for the specified domains using the specified
@ -224,12 +221,6 @@ class Client(object):
containing it.
:param list domains: Domains to request.
:param authenticator: The authenticator to use.
:type authenticator: :class:`letsencrypt.interfaces.IAuthenticator`
:param installer: The installer to use.
:type installer: :class:`letsencrypt.interfaces.IInstaller`
:param plugins: A PluginsFactory object.
:returns: A new :class:`letsencrypt.storage.RenewableCert` instance
@ -241,9 +232,10 @@ class Client(object):
# TODO: remove this dirty hack
self.config.namespace.authenticator = plugins.find_init(
authenticator).name
if installer is not None:
self.config.namespace.installer = plugins.find_init(installer).name
self.dv_auth).name
if self.installer is not None:
self.config.namespace.installer = plugins.find_init(
self.installer).name
# XXX: We clearly need a more general and correct way of getting
# options into the configobj for the RenewableCert instance.
@ -364,16 +356,17 @@ class Client(object):
chain_path = None if chain_path is None else os.path.abspath(chain_path)
for dom in domains:
# TODO: Provide a fullchain reference for installers like
# nginx that want it
self.installer.deploy_cert(
dom, os.path.abspath(cert_path),
os.path.abspath(privkey_path), chain_path)
with error_handler.ErrorHandler(self.installer.recovery_routine):
for dom in domains:
# TODO: Provide a fullchain reference for installers like
# nginx that want it
self.installer.deploy_cert(
dom, os.path.abspath(cert_path),
os.path.abspath(privkey_path), chain_path)
self.installer.save("Deployed Let's Encrypt Certificate")
# sites may have been enabled / final cleanup
self.installer.restart()
self.installer.save("Deployed Let's Encrypt Certificate")
# sites may have been enabled / final cleanup
self.installer.restart()
def enhance_config(self, domains, redirect=None):
"""Enhance the configuration.
@ -399,6 +392,8 @@ class Client(object):
if redirect is None:
redirect = enhancements.ask("redirect")
# When support for more enhancements are added, the call to the
# plugin's `enhance` function should be wrapped by an ErrorHandler
if redirect:
self.redirect_to_ssl(domains)
@ -409,14 +404,16 @@ class Client(object):
:type vhost: :class:`letsencrypt.interfaces.IInstaller`
"""
for dom in domains:
try:
self.installer.enhance(dom, "redirect")
except errors.PluginError:
logger.warn("Unable to perform redirect for %s", dom)
with error_handler.ErrorHandler(self.installer.recovery_routine):
for dom in domains:
try:
self.installer.enhance(dom, "redirect")
except errors.PluginError:
logger.warn("Unable to perform redirect for %s", dom)
raise
self.installer.save("Add Redirects")
self.installer.restart()
self.installer.save("Add Redirects")
self.installer.restart()
def validate_key_csr(privkey, csr=None):
@ -485,27 +482,6 @@ def rollback(default_installer, checkpoints, config, plugins):
installer.restart()
def revoke(default_installer, config, plugins, no_confirm, cert, authkey):
"""Revoke certificates.
:param config: Configuration.
:type config: :class:`letsencrypt.interfaces.IConfig`
"""
installer = display_ops.pick_installer(
config, default_installer, plugins, question="Which installer "
"should be used for certificate revocation?")
revoc = revoker.Revoker(installer, config, no_confirm)
# Cert is most selective, so it is chosen first.
if cert is not None:
revoc.revoke_from_cert(cert[0])
elif authkey is not None:
revoc.revoke_from_key(le_util.Key(authkey[0], authkey[1]))
else:
revoc.revoke_from_menu()
def view_config_changes(config):
"""View checkpoints and associated configuration changes.

40
letsencrypt/colored_logging.py Normal file
View file

@ -0,0 +1,40 @@
"""A formatter and StreamHandler for colorizing logging output."""
import logging
import sys
from letsencrypt import le_util
class StreamHandler(logging.StreamHandler):
"""Sends colored logging output to a stream.
If the specified stream is not a tty, the class works like the
standard logging.StreamHandler. Default red_level is logging.WARNING.
:ivar bool colored: True if output should be colored
:ivar bool red_level: The level at which to output
"""
def __init__(self, stream=None):
super(StreamHandler, self).__init__(stream)
self.colored = (sys.stderr.isatty() if stream is None else
stream.isatty())
self.red_level = logging.WARNING
def format(self, record):
"""Formats the string representation of record.
:param logging.LogRecord record: Record to be formatted
:returns: Formatted, string representation of record
:rtype: str
"""
output = super(StreamHandler, self).format(record)
if self.colored and record.levelno >= self.red_level:
return ''.join((le_util.ANSI_SGR_RED,
output,
le_util.ANSI_SGR_RESET))
else:
return output

12
letsencrypt/configuration.py
View file

@ -18,8 +18,7 @@ class NamespaceConfig(object):
paths defined in :py:mod:`letsencrypt.constants`:
- `accounts_dir`
- `cert_dir`
- `cert_key_backup`
- `csr_dir`
- `in_progress_dir`
- `key_dir`
- `renewer_config_file`
@ -54,13 +53,8 @@ class NamespaceConfig(object):
return os.path.join(self.namespace.work_dir, constants.BACKUP_DIR)
@property
def cert_dir(self): # pylint: disable=missing-docstring
return os.path.join(self.namespace.config_dir, constants.CERT_DIR)
@property
def cert_key_backup(self): # pylint: disable=missing-docstring
return os.path.join(self.namespace.work_dir,
constants.CERT_KEY_BACKUP_DIR, self.server_path)
def csr_dir(self): # pylint: disable=missing-docstring
return os.path.join(self.namespace.config_dir, constants.CSR_DIR)
@property
def in_progress_dir(self): # pylint: disable=missing-docstring

8
letsencrypt/constants.py
View file

@ -68,12 +68,8 @@ ACCOUNTS_DIR = "accounts"
BACKUP_DIR = "backups"
"""Directory (relative to `IConfig.work_dir`) where backups are kept."""
CERT_DIR = "certs"
"""See `.IConfig.cert_dir`."""
CERT_KEY_BACKUP_DIR = "keys-certs"
"""Directory where all certificates and keys are stored (relative to
`IConfig.work_dir`). Used for easy revocation."""
CSR_DIR = "csr"
"""See `.IConfig.csr_dir`."""
IN_PROGRESS_DIR = "IN_PROGRESS"
"""Directory used before a permanent checkpoint is finalized (relative to

77
letsencrypt/display/revocation.py
View file

@ -1,77 +0,0 @@
"""Revocation UI class."""
import os
import zope.component
from letsencrypt import interfaces
from letsencrypt.display import util as display_util
# Define a helper function to avoid verbose code
util = zope.component.getUtility # pylint: disable=invalid-name
def display_certs(certs):
"""Display the certificates in a menu for revocation.
:param list certs: each is a :class:`letsencrypt.revoker.Cert`
:returns: tuple of the form (code, selection) where
code is a display exit code
selection is the user's int selection
:rtype: tuple
"""
list_choices = [
"%s | %s | %s" % (
str(cert.get_cn().ljust(display_util.WIDTH - 39)),
cert.get_not_before().strftime("%m-%d-%y"),
"Installed" if cert.installed and cert.installed != ["Unknown"]
else "") for cert in certs
]
code, tag = util(interfaces.IDisplay).menu(
"Which certificates would you like to revoke?",
list_choices, help_label="More Info", ok_label="Revoke",
cancel_label="Exit")
return code, tag
def confirm_revocation(cert):
"""Confirm revocation screen.
:param cert: certificate object
:type cert: :class:
:returns: True if user would like to revoke, False otherwise
:rtype: bool
"""
return util(interfaces.IDisplay).yesno(
"Are you sure you would like to revoke the following "
"certificate:{0}{cert}This action cannot be reversed!".format(
os.linesep, cert=cert.pretty_print()))
def more_info_cert(cert):
"""Displays more info about the cert.
:param dict cert: cert dict used throughout revoker.py
"""
util(interfaces.IDisplay).notification(
"Certificate Information:{0}{1}".format(
os.linesep, cert.pretty_print()),
height=display_util.HEIGHT)
def success_revocation(cert):
"""Display a success message.
:param cert: cert that was revoked
:type cert: :class:`letsencrypt.revoker.Cert`
"""
util(interfaces.IDisplay).notification(
"You have successfully revoked the certificate for "
"%s" % cert.get_cn())

98
letsencrypt/error_handler.py Normal file
View file

@ -0,0 +1,98 @@
"""Registers functions to be called if an exception or signal occurs."""
import logging
import os
import signal
import traceback
logger = logging.getLogger(__name__)
# _SIGNALS stores the signals that will be handled by the ErrorHandler. These
# signals were chosen as their default handler terminates the process and could
# potentially occur from inside Python. Signals such as SIGILL were not
# included as they could be a sign of something devious and we should terminate
# immediately.
_SIGNALS = ([signal.SIGTERM] if os.name == "nt" else
[signal.SIGTERM, signal.SIGHUP, signal.SIGQUIT,
signal.SIGXCPU, signal.SIGXFSZ, signal.SIGPWR])
class ErrorHandler(object):
"""Registers functions to be called if an exception or signal occurs.
This class allows you to register functions that will be called when
an exception or signal is encountered. The class works best as a
context manager. For example:
with ErrorHandler(cleanup_func):
do_something()
If an exception is raised out of do_something, cleanup_func will be
called. The exception is not caught by the ErrorHandler. Similarly,
if a signal is encountered, cleanup_func is called followed by the
previously registered signal handler.
Every registered function is attempted to be run to completion
exactly once. If a registered function raises an exception, it is
logged and the next function is called. If a (different) handled
signal occurs while calling a registered function, it is attempted
to be called again by the next signal handler.
"""
def __init__(self, func=None):
self.funcs = []
self.prev_handlers = {}
if func is not None:
self.register(func)
def __enter__(self):
self.set_signal_handlers()
def __exit__(self, exec_type, exec_value, trace):
if exec_value is not None:
logger.debug("Encountered exception:\n%s", "".join(
traceback.format_exception(exec_type, exec_value, trace)))
self.call_registered()
self.reset_signal_handlers()
def register(self, func):
"""Registers func to be called if an error occurs."""
self.funcs.append(func)
def call_registered(self):
"""Calls all registered functions"""
logger.debug("Calling registered functions")
while self.funcs:
try:
self.funcs[-1]()
except Exception as error: # pylint: disable=broad-except
logger.error("Encountered exception during recovery")
logger.exception(error)
self.funcs.pop()
def set_signal_handlers(self):
"""Sets signal handlers for signals in _SIGNALS."""
for signum in _SIGNALS:
prev_handler = signal.getsignal(signum)
# If prev_handler is None, the handler was set outside of Python
if prev_handler is not None:
self.prev_handlers[signum] = prev_handler
signal.signal(signum, self._signal_handler)
def reset_signal_handlers(self):
"""Resets signal handlers for signals in _SIGNALS."""
for signum in self.prev_handlers:
signal.signal(signum, self.prev_handlers[signum])
self.prev_handlers.clear()
def _signal_handler(self, signum, unused_frame):
"""Calls registered functions and the previous signal handler.
:param int signum: number of current signal
"""
logger.debug("Singal %s encountered", signum)
self.call_registered()
signal.signal(signum, self.prev_handlers[signum])
os.kill(os.getpid(), signum)

18
letsencrypt/interfaces.py
View file

@ -205,12 +205,9 @@ class IConfig(zope.interface.Interface):
accounts_dir = zope.interface.Attribute(
"Directory where all account information is stored.")
backup_dir = zope.interface.Attribute("Configuration backups directory.")
cert_dir = zope.interface.Attribute(
csr_dir = zope.interface.Attribute(
"Directory where newly generated Certificate Signing Requests "
"(CSRs) and certificates not enrolled in the renewer are saved.")
cert_key_backup = zope.interface.Attribute(
"Directory where all certificates and keys are stored. "
"Used for easy revocation.")
"(CSRs) are saved.")
in_progress_dir = zope.interface.Attribute(
"Directory used before a permanent checkpoint is finalized.")
key_dir = zope.interface.Attribute("Keys storage.")
@ -321,6 +318,17 @@ class IInstaller(IPlugin):
"""
def recovery_routine():
"""Revert configuration to most recent finalized checkpoint.
Remove all changes (temporary and permanent) that have not been
finalized. This is useful to protect against crashes and other
execution interruptions.
:raises .errors.PluginError: If unable to recover the configuration
"""
def view_config_changes():
"""Display all of the LE config changes.

9
letsencrypt/le_util.py
View file

@ -18,6 +18,15 @@ Key = collections.namedtuple("Key", "file pem")
CSR = collections.namedtuple("CSR", "file data form")
# ANSI SGR escape codes
# Formats text as bold or with increased intensity
ANSI_SGR_BOLD = '\033[1m'
# Colors text red
ANSI_SGR_RED = "\033[31m"
# Resets output format
ANSI_SGR_RESET = "\033[0m"
def run_script(params):
"""Run the script with the given params.

4
letsencrypt/plugins/manual.py
View file

@ -23,7 +23,7 @@ from letsencrypt.plugins import common
logger = logging.getLogger(__name__)
class ManualAuthenticator(common.Plugin):
class Authenticator(common.Plugin):
"""Manual Authenticator.
.. todo:: Support for `~.challenges.DVSNI`.
@ -87,7 +87,7 @@ s.serve_forever()" """
"""
def __init__(self, *args, **kwargs):
super(ManualAuthenticator, self).__init__(*args, **kwargs)
super(Authenticator, self).__init__(*args, **kwargs)
self.template = (self.HTTP_TEMPLATE if self.config.no_simple_http_tls
else self.HTTPS_TEMPLATE)
self._root = (tempfile.mkdtemp() if self.conf("test-mode")

10
letsencrypt/plugins/manual_test.py
View file

@ -17,22 +17,22 @@ from letsencrypt.tests import test_util
KEY = jose.JWKRSA.load(test_util.load_vector("rsa512_key.pem"))
class ManualAuthenticatorTest(unittest.TestCase):
"""Tests for letsencrypt.plugins.manual.ManualAuthenticator."""
class AuthenticatorTest(unittest.TestCase):
"""Tests for letsencrypt.plugins.manual.Authenticator."""
def setUp(self):
from letsencrypt.plugins.manual import ManualAuthenticator
from letsencrypt.plugins.manual import Authenticator
self.config = mock.MagicMock(
no_simple_http_tls=True, simple_http_port=4430,
manual_test_mode=False)
self.auth = ManualAuthenticator(config=self.config, name="manual")
self.auth = Authenticator(config=self.config, name="manual")
self.achalls = [achallenges.SimpleHTTP(
challb=acme_util.SIMPLE_HTTP_P, domain="foo.com", account_key=KEY)]
config_test_mode = mock.MagicMock(
no_simple_http_tls=True, simple_http_port=4430,
manual_test_mode=True)
self.auth_test_mode = ManualAuthenticator(
self.auth_test_mode = Authenticator(
config=config_test_mode, name="manual")
def test_more_info(self):

3
letsencrypt/plugins/null.py
View file

@ -47,6 +47,9 @@ class Installer(common.Plugin):
def rollback_checkpoints(self, rollback=1):
pass # pragma: no cover
def recovery_routine(self):
pass # pragma: no cover
def view_config_changes(self):
pass # pragma: no cover

9
letsencrypt/reporter.py
View file

@ -9,6 +9,7 @@ import textwrap
import zope.interface
from letsencrypt import interfaces
from letsencrypt import le_util
logger = logging.getLogger(__name__)
@ -30,8 +31,6 @@ class Reporter(object):
LOW_PRIORITY = 2
"""Low priority constant. See `add_message`."""
_RESET = '\033[0m'
_BOLD = '\033[1m'
_msg_type = collections.namedtuple('ReporterMsg', 'priority text on_crash')
def __init__(self):
@ -76,7 +75,7 @@ class Reporter(object):
no_exception = sys.exc_info()[0] is None
bold_on = sys.stdout.isatty()
if bold_on:
print self._BOLD
print le_util.ANSI_SGR_BOLD
print 'IMPORTANT NOTES:'
first_wrapper = textwrap.TextWrapper(
initial_indent=' - ', subsequent_indent=(' ' * 3))
@ -87,7 +86,7 @@ class Reporter(object):
msg = self.messages.get()
if no_exception or msg.on_crash:
if bold_on and msg.priority > self.HIGH_PRIORITY:
sys.stdout.write(self._RESET)
sys.stdout.write(le_util.ANSI_SGR_RESET)
bold_on = False
lines = msg.text.splitlines()
print first_wrapper.fill(lines[0])
@ -95,4 +94,4 @@ class Reporter(object):
print "\n".join(
next_wrapper.fill(line) for line in lines[1:])
if bold_on:
sys.stdout.write(self._RESET)
sys.stdout.write(le_util.ANSI_SGR_RESET)

560
letsencrypt/revoker.py
View file

@ -1,560 +0,0 @@
"""Revoker module to enable LE revocations.
The backend of this module would fit a database quite nicely, but in order to
minimize dependencies and maintain transparency, the class currently implements
its own storage system. The number of certs that will likely be stored on any
given client might not warrant requiring a database.
"""
import collections
import csv
import logging
import os
import shutil
import tempfile
import OpenSSL
from acme import client as acme_client
from acme import crypto_util as acme_crypto_util
from acme.jose import util as jose_util
from letsencrypt import crypto_util
from letsencrypt import errors
from letsencrypt import le_util
from letsencrypt.display import util as display_util
from letsencrypt.display import revocation
logger = logging.getLogger(__name__)
class Revoker(object):
"""A revocation class for LE.
.. todo:: Add a method to specify your own certificate for revocation - CLI
:ivar .acme.client.Client acme: ACME client
:ivar installer: Installer object
:type installer: :class:`~letsencrypt.interfaces.IInstaller`
:ivar config: Configuration.
:type config: :class:`~letsencrypt.interfaces.IConfig`
:ivar bool no_confirm: Whether or not to ask for confirmation for revocation
"""
def __init__(self, installer, config, no_confirm=False):
# XXX
self.acme = acme_client.Client(directory=None, key=None, alg=None)
self.installer = installer
self.config = config
self.no_confirm = no_confirm
le_util.make_or_verify_dir(config.cert_key_backup, 0o700, os.geteuid(),
self.config.strict_permissions)
# TODO: Find a better solution for this...
self.list_path = os.path.join(config.cert_key_backup, "LIST")
# Make sure that the file is available for use for rest of class
open(self.list_path, "a").close()
def revoke_from_key(self, authkey):
"""Revoke all certificates under an authorized key.
:param authkey: Authorized key used in previous transactions
:type authkey: :class:`letsencrypt.le_util.Key`
"""
certs = []
try:
clean_pem = OpenSSL.crypto.dump_privatekey(
OpenSSL.crypto.FILETYPE_PEM, OpenSSL.crypto.load_privatekey(
OpenSSL.crypto.FILETYPE_PEM, authkey.pem))
except OpenSSL.crypto.Error as error:
logger.debug(error, exc_info=True)
raise errors.RevokerError(
"Invalid key file specified to revoke_from_key")
with open(self.list_path, "rb") as csvfile:
csvreader = csv.reader(csvfile)
for row in csvreader:
# idx, cert, key
# Add all keys that match to marked list
# Note: The key can be different than the pub key found in the
# certificate.
_, b_k = self._row_to_backup(row)
try:
test_pem = OpenSSL.crypto.dump_privatekey(
OpenSSL.crypto.FILETYPE_PEM, OpenSSL.crypto.load_privatekey(
OpenSSL.crypto.FILETYPE_PEM, open(b_k).read()))
except OpenSSL.crypto.Error as error:
logger.debug(error, exc_info=True)
# This should never happen given the assumptions of the
# module. If it does, it is probably best to delete the
# the offending key/cert. For now... just raise an exception
raise errors.RevokerError("%s - backup file is corrupted.")
if clean_pem == test_pem:
certs.append(
Cert.fromrow(row, self.config.cert_key_backup))
if certs:
self._safe_revoke(certs)
else:
logger.info("No certificates using the authorized key were found.")
def revoke_from_cert(self, cert_path):
"""Revoke a certificate by specifying a file path.
.. todo:: Add the ability to revoke the certificate even if the cert
is not stored locally. A path to the auth key will need to be
attained from the user.
:param str cert_path: path to ACME certificate in pem form
"""
# Locate the correct certificate (do not rely on filename)
cert_to_revoke = Cert(cert_path)
with open(self.list_path, "rb") as csvfile:
csvreader = csv.reader(csvfile)
for row in csvreader:
cert = Cert.fromrow(row, self.config.cert_key_backup)
if cert.get_der() == cert_to_revoke.get_der():
self._safe_revoke([cert])
return
logger.info("Associated ACME certificate was not found.")
def revoke_from_menu(self):
"""List trusted Let's Encrypt certificates."""
csha1_vhlist = self._get_installed_locations()
certs = self._populate_saved_certs(csha1_vhlist)
while True:
if certs:
code, selection = revocation.display_certs(certs)
if code == display_util.OK:
revoked_certs = self._safe_revoke([certs[selection]])
# Since we are currently only revoking one cert at a time...
if revoked_certs:
del certs[selection]
elif code == display_util.HELP:
revocation.more_info_cert(certs[selection])
else:
return
else:
logger.info(
"There are not any trusted Let's Encrypt "
"certificates for this server.")
return
def _populate_saved_certs(self, csha1_vhlist):
# pylint: disable=no-self-use
"""Populate a list of all the saved certs.
It is important to read from the file rather than the directory.
We assume that the LIST file is the master record and depending on
program crashes, this may differ from what is actually in the directory.
Namely, additional certs/keys may exist. There should never be any
certs/keys in the LIST that don't exist in the directory however.
:param dict csha1_vhlist: map from cert sha1 fingerprints to a list
of it's installed location paths.
"""
certs = []
with open(self.list_path, "rb") as csvfile:
csvreader = csv.reader(csvfile)
# idx, orig_cert, orig_key
for row in csvreader:
cert = Cert.fromrow(row, self.config.cert_key_backup)
# If we were able to find the cert installed... update status
cert.installed = csha1_vhlist.get(cert.get_fingerprint(), [])
certs.append(cert)
return certs
def _get_installed_locations(self):
"""Get installed locations of certificates.
:returns: map from cert sha1 fingerprint to :class:`list` of vhosts
where the certificate is installed.
"""
csha1_vhlist = {}
if self.installer is None:
return csha1_vhlist
for (cert_path, _, path) in self.installer.get_all_certs_keys():
try:
with open(cert_path) as cert_file:
cert_data = cert_file.read()
except IOError:
continue
try:
cert_obj, _ = crypto_util.pyopenssl_load_certificate(cert_data)
except errors.Error:
continue
cert_sha1 = cert_obj.digest("sha1")
if cert_sha1 in csha1_vhlist:
csha1_vhlist[cert_sha1].append(path)
else:
csha1_vhlist[cert_sha1] = [path]
return csha1_vhlist
def _safe_revoke(self, certs):
"""Confirm and revoke certificates.
:param certs: certs intended to be revoked
:type certs: :class:`list` of :class:`letsencrypt.revoker.Cert`
:returns: certs successfully revoked
:rtype: :class:`list` of :class:`letsencrypt.revoker.Cert`
"""
success_list = []
try:
for cert in certs:
if self.no_confirm or revocation.confirm_revocation(cert):
try:
self._acme_revoke(cert)
except errors.Error:
# TODO: Improve error handling when networking is set...
logger.error(
"Unable to revoke cert:%s%s", os.linesep, str(cert))
success_list.append(cert)
revocation.success_revocation(cert)
finally:
if success_list:
self._remove_certs_keys(success_list)
return success_list
def _acme_revoke(self, cert):
"""Revoke the certificate with the ACME server.
:param cert: certificate to revoke
:type cert: :class:`letsencrypt.revoker.Cert`
:returns: TODO
"""
# XXX | pylint: disable=unused-variable
# pylint: disable=protected-access
certificate = jose_util.ComparableX509(cert._cert)
try:
with open(cert.backup_key_path, "rU") as backup_key_file:
key = OpenSSL.crypto.load_privatekey(
OpenSSL.crypto.FILETYPE_PEM, backup_key_file.read())
# If the key file doesn't exist... or is corrupted
except OpenSSL.crypto.Error as error:
logger.debug(error, exc_info=True)
raise errors.RevokerError(
"Corrupted backup key file: %s" % cert.backup_key_path)
return self.acme.revoke(cert=None) # XXX
def _remove_certs_keys(self, cert_list): # pylint: disable=no-self-use
"""Remove certificate and key.
:param list cert_list: Must contain certs, each is of type
:class:`letsencrypt.revoker.Cert`
"""
# This must occur first, LIST is the official key
self._remove_certs_from_list(cert_list)
# Remove files
for cert in cert_list:
os.remove(cert.backup_path)
os.remove(cert.backup_key_path)
def _remove_certs_from_list(self, cert_list): # pylint: disable=no-self-use
"""Remove a certificate from the LIST file.
:param list cert_list: Must contain valid certs, each is of type
:class:`letsencrypt.revoker.Cert`
"""
newfile_handle, list_path2 = tempfile.mkstemp(".tmp", "LIST")
idx = 0
with open(self.list_path, "rb") as orgfile:
csvreader = csv.reader(orgfile)
with os.fdopen(newfile_handle, "wb") as newfile:
csvwriter = csv.writer(newfile)
for row in csvreader:
if idx >= len(cert_list) or row != cert_list[idx].get_row():
csvwriter.writerow(row)
else:
idx += 1
# This should never happen...
if idx != len(cert_list):
raise errors.RevokerError(
"Did not find all cert_list items to remove from LIST")
shutil.copy2(list_path2, self.list_path)
os.remove(list_path2)
def _row_to_backup(self, row):
"""Convenience function
:param list row: csv file row 'idx', 'cert_path', 'key_path'
:returns: tuple of the form ('backup_cert_path', 'backup_key_path')
:rtype: tuple
"""
return (self._get_backup(self.config.cert_key_backup, row[0], row[1]),
self._get_backup(self.config.cert_key_backup, row[0], row[2]))
@classmethod
def store_cert_key(cls, cert_path, key_path, config):
"""Store certificate key. (Used to allow quick revocation)
:param str cert_path: Path to a certificate file.
:param str key_path: Path to authorized key for certificate
:ivar config: Configuration.
:type config: :class:`~letsencrypt.interfaces.IConfig`
"""
list_path = os.path.join(config.cert_key_backup, "LIST")
le_util.make_or_verify_dir(config.cert_key_backup, 0o700, os.geteuid(),
config.strict_permissions)
cls._catalog_files(
config.cert_key_backup, cert_path, key_path, list_path)
@classmethod
def _catalog_files(cls, backup_dir, cert_path, key_path, list_path):
idx = 0
if os.path.isfile(list_path):
with open(list_path, "r+b") as csvfile:
csvreader = csv.reader(csvfile)
# Find the highest index in the file
for row in csvreader:
idx = int(row[0]) + 1
csvwriter = csv.writer(csvfile)
# You must move the files before appending the row
cls._copy_files(backup_dir, idx, cert_path, key_path)
csvwriter.writerow([str(idx), cert_path, key_path])
else:
with open(list_path, "wb") as csvfile:
csvwriter = csv.writer(csvfile)
# You must move the files before appending the row
cls._copy_files(backup_dir, idx, cert_path, key_path)
csvwriter.writerow([str(idx), cert_path, key_path])
@classmethod
def _copy_files(cls, backup_dir, idx, cert_path, key_path):
"""Copies the files into the backup dir appropriately."""
shutil.copy2(cert_path, cls._get_backup(backup_dir, idx, cert_path))
shutil.copy2(key_path, cls._get_backup(backup_dir, idx, key_path))
@classmethod
def _get_backup(cls, backup_dir, idx, orig_path):
"""Returns the path to the backup."""
return os.path.join(
backup_dir, "{name}_{idx}".format(
name=os.path.basename(orig_path), idx=str(idx)))
class Cert(object):
"""Cert object used for Revocation convenience.
:ivar _cert: Certificate
:type _cert: :class:`OpenSSL.crypto.X509`
:ivar int idx: convenience index used for listing
:ivar orig: (`str` path - original certificate, `str` status)
:type orig: :class:`PathStatus`
:ivar orig_key: (`str` path - original auth key, `str` status)
:type orig_key: :class:`PathStatus`
:ivar str backup_path: backup filepath of the certificate
:ivar str backup_key_path: backup filepath of the authorized key
:ivar list installed: `list` of `str` describing all locations the cert
is installed
"""
PathStatus = collections.namedtuple("PathStatus", "path status")
"""Convenience container to hold path and status info"""
DELETED_MSG = "This file has been moved or deleted"
CHANGED_MSG = "This file has changed"
def __init__(self, cert_path):
"""Cert initialization
:param str cert_filepath: Name of file containing certificate in
PEM format.
"""
try:
with open(cert_path) as cert_file:
cert_data = cert_file.read()
except IOError:
raise errors.RevokerError(
"Error loading certificate: %s" % cert_path)
try:
self._cert = OpenSSL.crypto.load_certificate(
OpenSSL.crypto.FILETYPE_PEM, cert_data)
except OpenSSL.crypto.Error:
raise errors.RevokerError(
"Error loading certificate: %s" % cert_path)
self.idx = -1
self.orig = None
self.orig_key = None
self.backup_path = ""
self.backup_key_path = ""
self.installed = ["Unknown"]
@classmethod
def fromrow(cls, row, backup_dir):
# pylint: disable=protected-access
"""Initialize Cert from a csv row."""
idx = int(row[0])
backup = Revoker._get_backup(backup_dir, idx, row[1])
backup_key = Revoker._get_backup(backup_dir, idx, row[2])
obj = cls(backup)
obj.add_meta(idx, row[1], row[2], backup, backup_key)
return obj
def get_row(self):
"""Returns a list in CSV format. If meta data is available."""
if self.orig is not None and self.orig_key is not None:
return [str(self.idx), self.orig.path, self.orig_key.path]
return None
def add_meta(self, idx, orig, orig_key, backup, backup_key):
"""Add meta data to cert
:param int idx: convenience index for revoker
:param tuple orig: (`str` original certificate filepath, `str` status)
:param tuple orig_key: (`str` original auth key path, `str` status)
:param str backup: backup certificate filepath
:param str backup_key: backup key filepath
"""
status = ""
key_status = ""
# Verify original cert path
if not os.path.isfile(orig):
status = Cert.DELETED_MSG
else:
with open(orig) as orig_file:
orig_data = orig_file.read()
o_cert = OpenSSL.crypto.load_certificate(
OpenSSL.crypto.FILETYPE_PEM, orig_data)
if self.get_fingerprint() != o_cert.digest("sha1"):
status = Cert.CHANGED_MSG
# Verify original key path
if not os.path.isfile(orig_key):
key_status = Cert.DELETED_MSG
else:
with open(orig_key, "r") as fd:
key_pem = fd.read()
with open(backup_key, "r") as fd:
backup_key_pem = fd.read()
if key_pem != backup_key_pem:
key_status = Cert.CHANGED_MSG
self.idx = idx
self.orig = Cert.PathStatus(orig, status)
self.orig_key = Cert.PathStatus(orig_key, key_status)
self.backup_path = backup
self.backup_key_path = backup_key
def get_cn(self):
"""Get common name."""
return self._cert.get_subject().CN
def get_fingerprint(self):
"""Get SHA1 fingerprint."""
return self._cert.digest("sha1")
def get_not_before(self):
"""Get not_valid_before field."""
return crypto_util.asn1_generalizedtime_to_dt(
self._cert.get_notBefore())
def get_not_after(self):
"""Get not_valid_after field."""
return crypto_util.asn1_generalizedtime_to_dt(
self._cert.get_notAfter())
def get_der(self):
"""Get certificate in der format."""
return OpenSSL.crypto.dump_certificate(
OpenSSL.crypto.FILETYPE_ASN1, self._cert)
def get_pub_key(self):
"""Get public key size.
.. todo:: Support for ECC
"""
return "RSA {0}".format(self._cert.get_pubkey().bits)
def get_san(self):
"""Get subject alternative name if available."""
# pylint: disable=protected-access
return ", ".join(acme_crypto_util._pyopenssl_cert_or_req_san(self._cert))
def __str__(self):
text = [
"Subject: %s" % crypto_util.pyopenssl_x509_name_as_text(
self._cert.get_subject()),
"SAN: %s" % self.get_san(),
"Issuer: %s" % crypto_util.pyopenssl_x509_name_as_text(
self._cert.get_issuer()),
"Public Key: %s" % self.get_pub_key(),
"Not Before: %s" % str(self.get_not_before()),
"Not After: %s" % str(self.get_not_after()),
"Serial Number: %s" % self._cert.get_serial_number(),
"SHA1: %s%s" % (self.get_fingerprint(), os.linesep),
"Installed: %s" % ", ".join(self.installed),
]
if self.orig is not None:
if self.orig.status == "":
text.append("Path: %s" % self.orig.path)
else:
text.append("Orig Path: %s (%s)" % self.orig)
if self.orig_key is not None:
if self.orig_key.status == "":
text.append("Auth Key Path: %s" % self.orig_key.path)
else:
text.append("Orig Auth Key Path: %s (%s)" % self.orig_key)
text.append("")
return os.linesep.join(text)
def pretty_print(self):
"""Nicely frames a cert str"""
frame = "-" * (display_util.WIDTH - 4) + os.linesep
return "{frame}{cert}{frame}".format(frame=frame, cert=str(self))

5
letsencrypt/tests/cli_test.py
View file

@ -176,7 +176,8 @@ class DuplicativeCertsTest(renewer_test.BaseRenewableCertTest):
def tearDown(self):
shutil.rmtree(self.tempdir)
def test_find_duplicative_names(self):
@mock.patch("letsencrypt.le_util.make_or_verify_dir")
def test_find_duplicative_names(self, unused_makedir):
from letsencrypt.cli import _find_duplicative_certs
test_cert = test_util.load_vector("cert-san.pem")
with open(self.test_rc.cert, "w") as f:
@ -206,5 +207,5 @@ class DuplicativeCertsTest(renewer_test.BaseRenewableCertTest):
self.assertEqual(result, (None, None))
if __name__ == '__main__':
if __name__ == "__main__":
unittest.main() # pragma: no cover

35
letsencrypt/tests/client_test.py
View file

@ -113,7 +113,7 @@ class ClientTest(unittest.TestCase):
mock_crypto_util.init_save_key.assert_called_once_with(
self.config.rsa_key_size, self.config.key_dir)
mock_crypto_util.init_save_csr.assert_called_once_with(
mock.sentinel.key, domains, self.config.cert_dir)
mock.sentinel.key, domains, self.config.csr_dir)
self._check_obtain_certificate()
@mock.patch("letsencrypt.client.zope.component.getUtility")
@ -178,6 +178,39 @@ class ClientTest(unittest.TestCase):
shutil.rmtree(tmp_path)
def test_deploy_certificate(self):
self.assertRaises(errors.Error, self.client.deploy_certificate,
["foo.bar"], "key", "cert", "chain")
installer = mock.MagicMock()
self.client.installer = installer
self.client.deploy_certificate(["foo.bar"], "key", "cert", "chain")
installer.deploy_cert.assert_called_once_with(
"foo.bar", os.path.abspath("cert"),
os.path.abspath("key"), os.path.abspath("chain"))
self.assertEqual(installer.save.call_count, 1)
installer.restart.assert_called_once_with()
@mock.patch("letsencrypt.client.enhancements")
def test_enhance_config(self, mock_enhancements):
self.assertRaises(errors.Error,
self.client.enhance_config, ["foo.bar"])
mock_enhancements.ask.return_value = True
installer = mock.MagicMock()
self.client.installer = installer
self.client.enhance_config(["foo.bar"])
installer.enhance.assert_called_once_with("foo.bar", "redirect")
self.assertEqual(installer.save.call_count, 1)
installer.restart.assert_called_once_with()
installer.enhance.side_effect = errors.PluginError
self.assertRaises(errors.PluginError,
self.client.enhance_config, ["foo.bar"], True)
installer.recovery_routine.assert_called_once_with()
class RollbackTest(unittest.TestCase):
"""Tests for letsencrypt.client.rollback."""

40
letsencrypt/tests/colored_logging_test.py Normal file
View file

@ -0,0 +1,40 @@
"""Tests for letsencrypt.colored_logging."""
import logging
import StringIO
import unittest
from letsencrypt import le_util
class StreamHandlerTest(unittest.TestCase):
"""Tests for letsencrypt.colored_logging."""
def setUp(self):
from letsencrypt import colored_logging
self.stream = StringIO.StringIO()
self.stream.isatty = lambda: True
self.handler = colored_logging.StreamHandler(self.stream)
self.logger = logging.getLogger()
self.logger.setLevel(logging.DEBUG)
self.logger.addHandler(self.handler)
def test_format(self):
msg = 'I did a thing'
self.logger.debug(msg)
self.assertEqual(self.stream.getvalue(), '{0}\n'.format(msg))
def test_format_and_red_level(self):
msg = 'I did another thing'
self.handler.red_level = logging.DEBUG
self.logger.debug(msg)
self.assertEqual(self.stream.getvalue(),
'{0}{1}{2}\n'.format(le_util.ANSI_SGR_RED,
msg,
le_util.ANSI_SGR_RESET))
if __name__ == "__main__":
unittest.main() # pragma: no cover

12
letsencrypt/tests/configuration_test.py
View file

@ -32,8 +32,8 @@ class NamespaceConfigTest(unittest.TestCase):
def test_dynamic_dirs(self, constants):
constants.ACCOUNTS_DIR = 'acc'
constants.BACKUP_DIR = 'backups'
constants.CERT_KEY_BACKUP_DIR = 'c/'
constants.CERT_DIR = 'certs'
constants.CSR_DIR = 'csr'
constants.IN_PROGRESS_DIR = '../p'
constants.KEY_DIR = 'keys'
constants.TEMP_CHECKPOINT_DIR = 't'
@ -41,9 +41,7 @@ class NamespaceConfigTest(unittest.TestCase):
self.assertEqual(
self.config.accounts_dir, '/tmp/config/acc/acme-server.org:443/new')
self.assertEqual(self.config.backup_dir, '/tmp/foo/backups')
self.assertEqual(self.config.cert_dir, '/tmp/config/certs')
self.assertEqual(
self.config.cert_key_backup, '/tmp/foo/c/acme-server.org:443/new')
self.assertEqual(self.config.csr_dir, '/tmp/config/csr')
self.assertEqual(self.config.in_progress_dir, '/tmp/foo/../p')
self.assertEqual(self.config.key_dir, '/tmp/config/keys')
self.assertEqual(self.config.temp_checkpoint_dir, '/tmp/foo/t')
@ -59,9 +57,9 @@ class RenewerConfigurationTest(unittest.TestCase):
@mock.patch('letsencrypt.configuration.constants')
def test_dynamic_dirs(self, constants):
constants.ARCHIVE_DIR = "a"
constants.ARCHIVE_DIR = 'a'
constants.LIVE_DIR = 'l'
constants.RENEWAL_CONFIGS_DIR = "renewal_configs"
constants.RENEWAL_CONFIGS_DIR = 'renewal_configs'
constants.RENEWER_CONFIG_FILENAME = 'r.conf'
self.assertEqual(self.config.archive_dir, '/tmp/config/a')

6
letsencrypt/tests/crypto_util_test.py
View file

@ -6,7 +6,9 @@ import unittest
import OpenSSL
import mock
import zope.component
from letsencrypt import interfaces
from letsencrypt.tests import test_util
@ -20,6 +22,8 @@ class InitSaveKeyTest(unittest.TestCase):
"""Tests for letsencrypt.crypto_util.init_save_key."""
def setUp(self):
logging.disable(logging.CRITICAL)
zope.component.provideUtility(
mock.Mock(strict_permissions=True), interfaces.IConfig)
self.key_dir = tempfile.mkdtemp('key_dir')
def tearDown(self):
@ -48,6 +52,8 @@ class InitSaveCSRTest(unittest.TestCase):
"""Tests for letsencrypt.crypto_util.init_save_csr."""
def setUp(self):
zope.component.provideUtility(
mock.Mock(strict_permissions=True), interfaces.IConfig)
self.csr_dir = tempfile.mkdtemp('csr_dir')
def tearDown(self):

97
letsencrypt/tests/display/revocation_test.py
View file

@ -1,97 +0,0 @@
"""Test :mod:`letsencrypt.display.revocation`."""
import sys
import unittest
import mock
import zope.component
from letsencrypt.display import util as display_util
from letsencrypt.tests import test_util
class DisplayCertsTest(unittest.TestCase):
def setUp(self):
from letsencrypt.revoker import Cert
self.cert0 = Cert(test_util.vector_path("cert.pem"))
self.cert1 = Cert(test_util.vector_path("cert-san.pem"))
self.certs = [self.cert0, self.cert1]
zope.component.provideUtility(display_util.FileDisplay(sys.stdout))
@classmethod
def _call(cls, certs):
from letsencrypt.display.revocation import display_certs
return display_certs(certs)
@mock.patch("letsencrypt.display.revocation.util")
def test_revocation(self, mock_util):
mock_util().menu.return_value = (display_util.OK, 0)
code, choice = self._call(self.certs)
self.assertEqual(display_util.OK, code)
self.assertEqual(self.certs[choice], self.cert0)
@mock.patch("letsencrypt.display.revocation.util")
def test_cancel(self, mock_util):
mock_util().menu.return_value = (display_util.CANCEL, -1)
code, _ = self._call(self.certs)
self.assertEqual(display_util.CANCEL, code)
class MoreInfoCertTest(unittest.TestCase):
# pylint: disable=too-few-public-methods
@classmethod
def _call(cls, cert):
from letsencrypt.display.revocation import more_info_cert
more_info_cert(cert)
@mock.patch("letsencrypt.display.revocation.util")
def test_more_info(self, mock_util):
self._call(mock.MagicMock())
self.assertEqual(mock_util().notification.call_count, 1)
class SuccessRevocationTest(unittest.TestCase):
def setUp(self):
from letsencrypt.revoker import Cert
self.cert = Cert(test_util.vector_path("cert.pem"))
@classmethod
def _call(cls, cert):
from letsencrypt.display.revocation import success_revocation
success_revocation(cert)
# Pretty trivial test... something is displayed...
@mock.patch("letsencrypt.display.revocation.util")
def test_success_revocation(self, mock_util):
self._call(self.cert)
self.assertEqual(mock_util().notification.call_count, 1)
class ConfirmRevocationTest(unittest.TestCase):
def setUp(self):
from letsencrypt.revoker import Cert
self.cert = Cert(test_util.vector_path("cert.pem"))
@classmethod
def _call(cls, cert):
from letsencrypt.display.revocation import confirm_revocation
return confirm_revocation(cert)
@mock.patch("letsencrypt.display.revocation.util")
def test_confirm_revocation(self, mock_util):
mock_util().yesno.return_value = True
self.assertTrue(self._call(self.cert))
mock_util().yesno.return_value = False
self.assertFalse(self._call(self.cert))
if __name__ == "__main__":
unittest.main() # pragma: no cover

55
letsencrypt/tests/error_handler_test.py Normal file
View file

@ -0,0 +1,55 @@
"""Tests for letsencrypt.error_handler."""
import signal
import unittest
import mock
class ErrorHandlerTest(unittest.TestCase):
"""Tests for letsencrypt.error_handler."""
def setUp(self):
from letsencrypt import error_handler
self.init_func = mock.MagicMock()
self.handler = error_handler.ErrorHandler(self.init_func)
# pylint: disable=protected-access
self.signals = error_handler._SIGNALS
def test_context_manager(self):
try:
with self.handler:
raise ValueError
except ValueError:
pass
self.init_func.assert_called_once_with()
@mock.patch('letsencrypt.error_handler.os')
@mock.patch('letsencrypt.error_handler.signal')
def test_signal_handler(self, mock_signal, mock_os):
# pylint: disable=protected-access
mock_signal.getsignal.return_value = signal.SIG_DFL
self.handler.set_signal_handlers()
signal_handler = self.handler._signal_handler
for signum in self.signals:
mock_signal.signal.assert_any_call(signum, signal_handler)
signum = self.signals[0]
signal_handler(signum, None)
self.init_func.assert_called_once_with()
mock_os.kill.assert_called_once_with(mock_os.getpid(), signum)
self.handler.reset_signal_handlers()
for signum in self.signals:
mock_signal.signal.assert_any_call(signum, signal.SIG_DFL)
def test_bad_recovery(self):
bad_func = mock.MagicMock(side_effect=[ValueError])
self.handler.register(bad_func)
self.handler.call_registered()
self.init_func.assert_called_once_with()
bad_func.assert_called_once_with()
if __name__ == "__main__":
unittest.main() # pragma: no cover

5
letsencrypt/tests/renewer_test.py
View file

@ -33,7 +33,12 @@ def fill_with_sample_data(rc_object):
class BaseRenewableCertTest(unittest.TestCase):
"""Base class for setting up Renewable Cert tests.
.. note:: It may be required to write out self.config for
your test. Check :class:`.cli_test.DuplicateCertTest` for an example.
"""
def setUp(self):
from letsencrypt import storage
self.tempdir = tempfile.mkdtemp()

409
letsencrypt/tests/revoker_test.py
View file

@ -1,409 +0,0 @@
"""Test letsencrypt.revoker."""
import csv
import os
import shutil
import tempfile
import unittest
import mock
import OpenSSL
from letsencrypt import errors
from letsencrypt import le_util
from letsencrypt.display import util as display_util
from letsencrypt.tests import test_util
KEY = OpenSSL.crypto.load_privatekey(
OpenSSL.crypto.FILETYPE_PEM, test_util.load_vector("rsa512_key.pem"))
class RevokerBase(unittest.TestCase): # pylint: disable=too-few-public-methods
"""Base Class for Revoker Tests."""
def setUp(self):
self.paths, self.certs, self.key_path = create_revoker_certs()
self.backup_dir = tempfile.mkdtemp("cert_backup")
self.mock_config = mock.MagicMock(cert_key_backup=self.backup_dir)
self.list_path = os.path.join(self.backup_dir, "LIST")
def _store_certs(self):
# pylint: disable=protected-access
from letsencrypt.revoker import Revoker
Revoker.store_cert_key(self.paths[0], self.key_path, self.mock_config)
Revoker.store_cert_key(self.paths[1], self.key_path, self.mock_config)
# Set metadata
for i in xrange(2):
self.certs[i].add_meta(
i, self.paths[i], self.key_path,
Revoker._get_backup(self.backup_dir, i, self.paths[i]),
Revoker._get_backup(self.backup_dir, i, self.key_path))
def _get_rows(self):
with open(self.list_path, "rb") as csvfile:
return [row for row in csv.reader(csvfile)]
def _write_rows(self, rows):
with open(self.list_path, "wb") as csvfile:
csvwriter = csv.writer(csvfile)
for row in rows:
csvwriter.writerow(row)
class RevokerTest(RevokerBase):
def setUp(self):
from letsencrypt.revoker import Revoker
super(RevokerTest, self).setUp()
with open(self.key_path) as key_file:
self.key = le_util.Key(self.key_path, key_file.read())
self._store_certs()
self.revoker = Revoker(
installer=mock.MagicMock(), config=self.mock_config)
def tearDown(self):
shutil.rmtree(self.backup_dir)
@mock.patch("acme.client.Client.revoke")
@mock.patch("letsencrypt.revoker.revocation")
def test_revoke_by_key_all(self, mock_display, mock_acme):
mock_display().confirm_revocation.return_value = True
self.revoker.revoke_from_key(self.key)
self.assertEqual(self._get_rows(), [])
# Check to make sure backups were eliminated
for i in xrange(2):
self.assertFalse(self._backups_exist(self.certs[i].get_row()))
self.assertEqual(mock_acme.call_count, 2)
@mock.patch("letsencrypt.revoker.OpenSSL.crypto.load_privatekey")
def test_revoke_by_invalid_keys(self, mock_load_privatekey):
mock_load_privatekey.side_effect = OpenSSL.crypto.Error
self.assertRaises(
errors.RevokerError, self.revoker.revoke_from_key, self.key)
mock_load_privatekey.side_effect = [KEY, OpenSSL.crypto.Error]
self.assertRaises(
errors.RevokerError, self.revoker.revoke_from_key, self.key)
@mock.patch("acme.client.Client.revoke")
@mock.patch("letsencrypt.revoker.revocation")
def test_revoke_by_wrong_key(self, mock_display, mock_acme):
mock_display().confirm_revocation.return_value = True
key_path = test_util.vector_path("rsa256_key.pem")
wrong_key = le_util.Key(key_path, open(key_path).read())
self.revoker.revoke_from_key(wrong_key)
# Nothing was removed
self.assertEqual(len(self._get_rows()), 2)
# No revocation went through
self.assertEqual(mock_acme.call_count, 0)
@mock.patch("acme.client.Client.revoke")
@mock.patch("letsencrypt.revoker.revocation")
def test_revoke_by_cert(self, mock_display, mock_acme):
mock_display().confirm_revocation.return_value = True
self.revoker.revoke_from_cert(self.paths[1])
row0 = self.certs[0].get_row()
row1 = self.certs[1].get_row()
self.assertEqual(self._get_rows(), [row0])
self.assertTrue(self._backups_exist(row0))
self.assertFalse(self._backups_exist(row1))
self.assertEqual(mock_acme.call_count, 1)
@mock.patch("acme.client.Client.revoke")
@mock.patch("letsencrypt.revoker.revocation")
def test_revoke_by_cert_not_found(self, mock_display, mock_acme):
mock_display().confirm_revocation.return_value = True
self.revoker.revoke_from_cert(self.paths[0])
self.revoker.revoke_from_cert(self.paths[0])
row0 = self.certs[0].get_row()
row1 = self.certs[1].get_row()
# Same check as last time... just reversed.
self.assertEqual(self._get_rows(), [row1])
self.assertTrue(self._backups_exist(row1))
self.assertFalse(self._backups_exist(row0))
self.assertEqual(mock_acme.call_count, 1)
@mock.patch("acme.client.Client.revoke")
@mock.patch("letsencrypt.revoker.revocation")
def test_revoke_by_menu(self, mock_display, mock_acme):
mock_display().confirm_revocation.return_value = True
mock_display.display_certs.side_effect = [
(display_util.HELP, 0),
(display_util.OK, 0),
(display_util.CANCEL, -1),
]
self.revoker.revoke_from_menu()
row0 = self.certs[0].get_row()
row1 = self.certs[1].get_row()
self.assertEqual(self._get_rows(), [row1])
self.assertFalse(self._backups_exist(row0))
self.assertTrue(self._backups_exist(row1))
self.assertEqual(mock_acme.call_count, 1)
self.assertEqual(mock_display.more_info_cert.call_count, 1)
@mock.patch("letsencrypt.revoker.logger")
@mock.patch("acme.client.Client.revoke")
@mock.patch("letsencrypt.revoker.revocation")
def test_revoke_by_menu_delete_all(self, mock_display, mock_acme, mock_log):
mock_display().confirm_revocation.return_value = True
mock_display.display_certs.return_value = (display_util.OK, 0)
self.revoker.revoke_from_menu()
self.assertEqual(self._get_rows(), [])
# Everything should be deleted...
for i in xrange(2):
self.assertFalse(self._backups_exist(self.certs[i].get_row()))
self.assertEqual(mock_acme.call_count, 2)
# Info is called when there aren't any certs left...
self.assertTrue(mock_log.info.called)
@mock.patch("letsencrypt.revoker.revocation")
@mock.patch("letsencrypt.revoker.Revoker._acme_revoke")
@mock.patch("letsencrypt.revoker.logger")
def test_safe_revoke_acme_fail(self, mock_log, mock_revoke, mock_display):
# pylint: disable=protected-access
mock_revoke.side_effect = errors.Error
mock_display().confirm_revocation.return_value = True
self.revoker._safe_revoke(self.certs)
self.assertTrue(mock_log.error.called)
@mock.patch("letsencrypt.revoker.OpenSSL.crypto.load_privatekey")
def test_acme_revoke_failure(self, mock_load_privatekey):
# pylint: disable=protected-access
mock_load_privatekey.side_effect = OpenSSL.crypto.Error
self.assertRaises(
errors.Error, self.revoker._acme_revoke, self.certs[0])
def test_remove_certs_from_list_bad_certs(self):
# pylint: disable=protected-access
from letsencrypt.revoker import Cert
new_cert = Cert(self.paths[0])
# This isn't stored in the db
new_cert.idx = 10
new_cert.backup_path = self.paths[0]
new_cert.backup_key_path = self.key_path
new_cert.orig = Cert.PathStatus("false path", "not here")
new_cert.orig_key = Cert.PathStatus("false path", "not here")
self.assertRaises(errors.RevokerError,
self.revoker._remove_certs_from_list, [new_cert])
def _backups_exist(self, row):
# pylint: disable=protected-access
cert_path, key_path = self.revoker._row_to_backup(row)
return os.path.isfile(cert_path) and os.path.isfile(key_path)
class RevokerInstallerTest(RevokerBase):
def setUp(self):
super(RevokerInstallerTest, self).setUp()
self.installs = [
["installation/path0a", "installation/path0b"],
["installation/path1"],
]
self.certs_keys = [
(self.paths[0], self.key_path, self.installs[0][0]),
(self.paths[0], self.key_path, self.installs[0][1]),
(self.paths[1], self.key_path, self.installs[1][0]),
]
self._store_certs()
def _get_revoker(self, installer):
from letsencrypt.revoker import Revoker
return Revoker(installer, self.mock_config)
def test_no_installer_get_installed_locations(self):
# pylint: disable=protected-access
revoker = self._get_revoker(None)
self.assertEqual(revoker._get_installed_locations(), {})
def test_get_installed_locations(self):
# pylint: disable=protected-access
mock_installer = mock.MagicMock()
mock_installer.get_all_certs_keys.return_value = self.certs_keys
revoker = self._get_revoker(mock_installer)
sha_vh = revoker._get_installed_locations()
self.assertEqual(len(sha_vh), 2)
for i, cert in enumerate(self.certs):
self.assertTrue(cert.get_fingerprint() in sha_vh)
self.assertEqual(
sha_vh[cert.get_fingerprint()], self.installs[i])
@mock.patch("letsencrypt.revoker.OpenSSL.crypto.load_certificate")
def test_get_installed_load_failure(self, mock_load_certificate):
mock_installer = mock.MagicMock()
mock_installer.get_all_certs_keys.return_value = self.certs_keys
mock_load_certificate.side_effect = OpenSSL.crypto.Error
revoker = self._get_revoker(mock_installer)
# pylint: disable=protected-access
self.assertEqual(revoker._get_installed_locations(), {})
def test_get_installed_load_failure_open(self):
tmp = tempfile.mkdtemp()
mock_installer = mock.MagicMock()
mock_installer.get_all_certs_keys.return_value = [(
os.path.join(tmp, 'missing'), None, None)]
revoker = self._get_revoker(mock_installer)
# pylint: disable=protected-access
self.assertEqual(revoker._get_installed_locations(), {})
os.rmdir(tmp)
class RevokerClassMethodsTest(RevokerBase):
def setUp(self):
super(RevokerClassMethodsTest, self).setUp()
self.mock_config = mock.MagicMock(cert_key_backup=self.backup_dir)
def tearDown(self):
shutil.rmtree(self.backup_dir)
def _call(self, cert_path, key_path):
from letsencrypt.revoker import Revoker
Revoker.store_cert_key(cert_path, key_path, self.mock_config)
def test_store_two(self):
from letsencrypt.revoker import Revoker
self._call(self.paths[0], self.key_path)
self._call(self.paths[1], self.key_path)
self.assertTrue(os.path.isfile(self.list_path))
rows = self._get_rows()
for i, row in enumerate(rows):
# pylint: disable=protected-access
self.assertTrue(os.path.isfile(
Revoker._get_backup(self.backup_dir, i, self.paths[i])))
self.assertTrue(os.path.isfile(
Revoker._get_backup(self.backup_dir, i, self.key_path)))
self.assertEqual([str(i), self.paths[i], self.key_path], row)
self.assertEqual(len(rows), 2)
def test_store_one_mixed(self):
from letsencrypt.revoker import Revoker
self._write_rows(
[["5", "blank", "blank"], ["18", "dc", "dc"], ["21", "b", "b"]])
self._call(self.paths[0], self.key_path)
self.assertEqual(
self._get_rows()[3], ["22", self.paths[0], self.key_path])
# pylint: disable=protected-access
self.assertTrue(os.path.isfile(
Revoker._get_backup(self.backup_dir, 22, self.paths[0])))
self.assertTrue(os.path.isfile(
Revoker._get_backup(self.backup_dir, 22, self.key_path)))
class CertTest(unittest.TestCase):
def setUp(self):
self.paths, self.certs, self.key_path = create_revoker_certs()
def test_failed_load(self):
from letsencrypt.revoker import Cert
self.assertRaises(errors.RevokerError, Cert, self.key_path)
def test_failed_load_open(self):
tmp = tempfile.mkdtemp()
from letsencrypt.revoker import Cert
self.assertRaises(
errors.RevokerError, Cert, os.path.join(tmp, 'missing'))
os.rmdir(tmp)
def test_no_row(self):
self.assertEqual(self.certs[0].get_row(), None)
def test_meta_moved_files(self):
from letsencrypt.revoker import Cert
fake_path = "/not/a/real/path/r72d3t6"
self.certs[0].add_meta(
0, fake_path, fake_path, self.paths[0], self.key_path)
self.assertEqual(self.certs[0].orig.status, Cert.DELETED_MSG)
self.assertEqual(self.certs[0].orig_key.status, Cert.DELETED_MSG)
def test_meta_changed_files(self):
from letsencrypt.revoker import Cert
self.certs[0].add_meta(
0, self.paths[1], self.paths[1], self.paths[0], self.key_path)
self.assertEqual(self.certs[0].orig.status, Cert.CHANGED_MSG)
self.assertEqual(self.certs[0].orig_key.status, Cert.CHANGED_MSG)
def test_meta_no_status(self):
self.certs[0].add_meta(
0, self.paths[0], self.key_path, self.paths[0], self.key_path)
self.assertEqual(self.certs[0].orig.status, "")
self.assertEqual(self.certs[0].orig_key.status, "")
def test_print_meta(self):
"""Just make sure there aren't any major errors."""
self.certs[0].add_meta(
0, self.paths[0], self.key_path, self.paths[0], self.key_path)
# Changed path and deleted file
self.certs[1].add_meta(
1, self.paths[0], "/not/a/path", self.paths[1], self.key_path)
self.assertTrue(self.certs[0].pretty_print())
self.assertTrue(self.certs[1].pretty_print())
def test_print_no_meta(self):
self.assertTrue(self.certs[0].pretty_print())
self.assertTrue(self.certs[1].pretty_print())
def create_revoker_certs():
"""Create a few revoker.Cert objects."""
cert0_path = test_util.vector_path("cert.pem")
cert1_path = test_util.vector_path("cert-san.pem")
key_path = test_util.vector_path("rsa512_key.pem")
from letsencrypt.revoker import Cert
cert0 = Cert(cert0_path)
cert1 = Cert(cert1_path)
return [cert0_path, cert1_path], [cert0, cert1], key_path
if __name__ == "__main__":
unittest.main() # pragma: no cover

4
letshelp-letsencrypt/setup.py
View file

@ -4,7 +4,9 @@ from setuptools import setup
from setuptools import find_packages
install_requires = []
install_requires = [
'setuptools', # pkg_resources
]
if sys.version_info < (2, 7):
install_requires.append("mock<1.1.0")
else:

3
setup.py
View file

@ -42,6 +42,7 @@ install_requires = [
'python2-pythondialog>=3.2.2rc1', # Debian squeeze support, cf. #280
'pytz',
'requests',
'setuptools', # pkg_resources
'zope.component',
'zope.interface',
]
@ -116,7 +117,7 @@ setup(
'letsencrypt-renewer = letsencrypt.renewer:main',
],
'letsencrypt.plugins': [
'manual = letsencrypt.plugins.manual:ManualAuthenticator',
'manual = letsencrypt.plugins.manual:Authenticator',
# TODO: null should probably not be presented to the user
'null = letsencrypt.plugins.null:Installer',
'standalone = letsencrypt.plugins.standalone.authenticator'

1
tests/integration/_common.sh
View file

@ -23,6 +23,7 @@ letsencrypt_test () {
--agree-eula \
--agree-tos \
--email "" \
--renew-by-default \
--debug \
-vvvvvvv \
"$@"

2
tox.cover.sh
View file

@ -16,7 +16,7 @@ fi
cover () {
if [ "$1" = "letsencrypt" ]; then
min=97
min=96
elif [ "$1" = "acme" ]; then
min=100
elif [ "$1" = "letsencrypt_apache" ]; then