mirror of
https://github.com/certbot/certbot.git
synced 2026-07-30 18:14:21 +02:00
merge standalone, plus further development
This commit is contained in:
James Kasten
@@ -21,3 +21,9 @@
|
||||
|
||||
.. automodule:: letsencrypt.client.display.enhancements
|
||||
:members:
|
||||
|
||||
:mod:`letsencrypt.client.display.revocation`
|
||||
============================================
|
||||
|
||||
.. automodule:: letsencrypt.client.display.revocation
|
||||
:members:
|
||||
|
||||
@@ -250,7 +250,7 @@ def validate_key_csr(privkey, csr=None):
|
||||
:type privkey: :class:`letsencrypt.client.le_util.Key`
|
||||
|
||||
:param csr: CSR
|
||||
:type csr: :class:`letsencrypt.client.client.Client.CSR`
|
||||
:type csr: :class:`letsencrypt.client.le_util.CSR`
|
||||
|
||||
:raises LetsEncryptClientError: if validation fails
|
||||
|
||||
|
||||
@@ -45,8 +45,10 @@ class NamespaceConfig(object):
|
||||
@property
|
||||
def cert_key_backup(self): # pylint: disable=missing-docstring
|
||||
return os.path.join(
|
||||
self.namespace.work_dir, constants.CERT_KEY_BACKUP_DIR)
|
||||
self.namespace.work_dir, constants.CERT_KEY_BACKUP_DIR,
|
||||
self.namespace.server.partition(":")[0])
|
||||
|
||||
# TODO: This should probably include the server name
|
||||
@property
|
||||
def rec_token_dir(self): # pylint: disable=missing-docstring
|
||||
return os.path.join(self.namespace.work_dir, constants.REC_TOKEN_DIR)
|
||||
|
||||
@@ -45,6 +45,9 @@ APACHE_REWRITE_HTTPS_ARGS = [
|
||||
"""Apache rewrite rule arguments used for redirections to https vhost"""
|
||||
|
||||
|
||||
DVSNI_CHALLENGE_PORT = 443
|
||||
"""Port to perform DVSNI challenge."""
|
||||
|
||||
DVSNI_DOMAIN_SUFFIX = ".acme.invalid"
|
||||
"""Suffix appended to domains in DVSNI validation."""
|
||||
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Utilities for all Let"s Encrypt."""
|
||||
"""Utilities for all Let's Encrypt."""
|
||||
import base64
|
||||
import collections
|
||||
import errno
|
||||
|
||||
@@ -20,7 +20,7 @@ from letsencrypt.client.display import revocation
|
||||
class Revoker(object):
|
||||
"""A revocation class for LE.
|
||||
|
||||
..todo:: Add a method to specify your own certificate for revocation - CLI
|
||||
.. todo:: Add a method to specify your own certificate for revocation - CLI
|
||||
|
||||
:ivar network: Network object
|
||||
:type network: :class:`letsencrypt.client.network`
|
||||
@@ -32,20 +32,15 @@ class Revoker(object):
|
||||
:type config: :class:`~letsencrypt.client.interfaces.IConfig`
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, installer, config):
|
||||
self.network = network.Network(config.server)
|
||||
self.installer = installer
|
||||
self.config = config
|
||||
|
||||
# This will go through and make sure that nothing almost got revoked...
|
||||
# but didn't quite make it... also, guarantees no orphan cert/key files
|
||||
self.recovery_routine()
|
||||
le_util.make_or_verify_dir(config.cert_key_backup, 0o700)
|
||||
|
||||
# TODO: WTF do I do with these...
|
||||
# TODO: Find a better solution for this...
|
||||
self.list_path = os.path.join(config.cert_key_backup, "LIST")
|
||||
self.marked_path = os.path.join(config.cert_key_backup, "MARKED")
|
||||
|
||||
|
||||
def revoke_from_interface(self, cert):
|
||||
"""Handle ACME "revocation" phase.
|
||||
@@ -54,12 +49,9 @@ class Revoker(object):
|
||||
:type cert: :class:`letsencrypt.client.revoker.Cert`
|
||||
|
||||
"""
|
||||
self._mark_for_revocation(cert)
|
||||
|
||||
revoc = self.revoke(cert.backup_path, cert.backup_key_path)
|
||||
|
||||
self.remove_cert_key(cert)
|
||||
self._remove_mark()
|
||||
self.remove_cert_key([cert.idx, cert.backup_path, cert.backup_key_path])
|
||||
|
||||
if revoc is not None:
|
||||
revocation.success_revocation(cert)
|
||||
@@ -69,6 +61,22 @@ class Revoker(object):
|
||||
|
||||
self.display_menu()
|
||||
|
||||
def revoke_from_key(self, auth_key):
|
||||
marked = []
|
||||
with open(self.list_path, "r") as csvfile:
|
||||
csvreader = csv.reader(csvfile)
|
||||
for row in csvreader:
|
||||
# idx, cert, key
|
||||
# Add all keys that match to marked list
|
||||
# TODO: This doesn't account for padding in file that might
|
||||
# differ. This should only consider the key material.
|
||||
# Note: The key can be different than the pub key found in the
|
||||
# certificate.
|
||||
if auth_key.pem == open(row[2]).read():
|
||||
marked.append(row)
|
||||
|
||||
self.remove_certs_keys(marked)
|
||||
|
||||
def revoke(self, cert_path, key_path):
|
||||
"""Revoke the certificate with the ACME server.
|
||||
|
||||
@@ -89,32 +97,6 @@ class Revoker(object):
|
||||
return self.network.send_and_receive_expected(
|
||||
acme.revocation_request(cert_der, key), "revocation")
|
||||
|
||||
def recovery_routine(self):
|
||||
"""Intended to make sure files aren't orphaned."""
|
||||
if not os.path.isfile(self.marked_path):
|
||||
return
|
||||
with open(self.marked_path, "r") as marked_file:
|
||||
csvreader = csv.reader(marked_file)
|
||||
for row in csvreader:
|
||||
self.revoke(row[0], row[1])
|
||||
le_util.safely_remove(row[0])
|
||||
le_util.safely_remove(row[1])
|
||||
|
||||
self._remove_mark()
|
||||
|
||||
def _mark_for_revocation(self, cert): # pylint: disable=no-self-use
|
||||
"""Marks a cert for revocation."""
|
||||
if os.path.isfile(self.marked_path):
|
||||
raise errors.LetsEncryptRevokerError(
|
||||
"MARKED file was never cleaned.")
|
||||
with open(self.marked_path, "w") as marked_file:
|
||||
csvwriter = csv.writer(marked_file)
|
||||
csvwriter.writerow([cert.backup_path, cert.backup_key_path])
|
||||
|
||||
def _remove_mark(self): # pylint: disable=no-self-use
|
||||
"""Remove the marked file."""
|
||||
os.remove(self.marked_path)
|
||||
|
||||
def display_menu(self):
|
||||
"""List trusted Let's Encrypt certificates."""
|
||||
|
||||
@@ -136,7 +118,15 @@ class Revoker(object):
|
||||
|
||||
def _populate_saved_certs(self, csha1_vhlist):
|
||||
# pylint: disable=no-self-use
|
||||
"""Populate a list of all the saved certs."""
|
||||
"""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.
|
||||
|
||||
"""
|
||||
certs = []
|
||||
with open(self.list_path, "rb") as csvfile:
|
||||
csvreader = csv.reader(csvfile)
|
||||
@@ -183,23 +173,32 @@ class Revoker(object):
|
||||
|
||||
return csha1_vhlist
|
||||
|
||||
def remove_cert_key(self, cert): # pylint: disable=no-self-use
|
||||
def remove_certs_keys(self, del_list): # pylint: disable=no-self-use
|
||||
"""Remove certificate and key.
|
||||
|
||||
:param cert: cert object
|
||||
:type cert: :class:`letsencrypt.client.revoker.Cert`
|
||||
:param list del_list: each is a `list` in the form
|
||||
[idx, cert_path, key_path] all entries must be in the original
|
||||
LIST order
|
||||
|
||||
"""
|
||||
self._remove_cert_from_list(cert)
|
||||
# This must occur first, LIST is the official key
|
||||
self._remove_certs_from_list(del_list)
|
||||
|
||||
# Remove files
|
||||
os.remove(cert.backup_path)
|
||||
os.remove(cert.backup_key_path)
|
||||
for row in del_list:
|
||||
os.remove(row[1])
|
||||
os.remove(row[2])
|
||||
|
||||
def _remove_cert_from_list(self, cert): # pylint: disable=no-self-use
|
||||
"""Remove a certificate from the LIST file."""
|
||||
def _remove_certs_from_list(self, del_list): # pylint: disable=no-self-use
|
||||
"""Remove a certificate from the LIST file.
|
||||
|
||||
:param list del_list: each is a csv row, all items must be in the
|
||||
proper file order.
|
||||
|
||||
"""
|
||||
list_path2 = os.path.join(self.config.cert_key_backup, "LIST.tmp")
|
||||
|
||||
idx = 0
|
||||
with open(self.list_path, "rb") as orgfile:
|
||||
csvreader = csv.reader(orgfile)
|
||||
|
||||
@@ -207,10 +206,16 @@ class Revoker(object):
|
||||
csvwriter = csv.writer(newfile)
|
||||
|
||||
for row in csvreader:
|
||||
if not (row[0] == str(cert.idx) and
|
||||
row[1] == cert.orig.path and
|
||||
row[2] == cert.orig_key.path):
|
||||
if not (row[0] == str(del_list[idx][0]) and
|
||||
row[1] == del_list[idx][1] and
|
||||
row[2] == del_list[idx][2]):
|
||||
csvwriter.writerow(row)
|
||||
else:
|
||||
# Found one of the marked rows... on to the next
|
||||
idx += 1
|
||||
|
||||
if idx != len(del_list):
|
||||
errors.LetsEncryptRevokerError("Did not find all items in del_list")
|
||||
|
||||
shutil.copy2(list_path2, self.list_path)
|
||||
os.remove(list_path2)
|
||||
|
||||
+336
@@ -0,0 +1,336 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""An authenticator that doesn't rely on any existing server program.
|
||||
|
||||
This authenticator creates its own ephemeral TCP listener on the specified
|
||||
port in order to respond to incoming DVSNI challenges from the certificate
|
||||
authority."""
|
||||
|
||||
import os
|
||||
import signal
|
||||
import socket
|
||||
import sys
|
||||
import time
|
||||
|
||||
import Crypto.Random
|
||||
import OpenSSL.crypto
|
||||
import OpenSSL.SSL
|
||||
import zope.component
|
||||
import zope.interface
|
||||
|
||||
from letsencrypt.client import challenge_util
|
||||
from letsencrypt.client import constants
|
||||
from letsencrypt.client import interfaces
|
||||
|
||||
|
||||
class StandaloneAuthenticator(object):
|
||||
# pylint: disable=too-many-instance-attributes
|
||||
"""The StandaloneAuthenticator class itself.
|
||||
|
||||
This authenticator can be invoked by the Let's Encrypt client
|
||||
according to the IAuthenticator API interface. It creates a local
|
||||
TCP listener on a specified port and satisfies DVSNI challenges."""
|
||||
zope.interface.implements(interfaces.IAuthenticator)
|
||||
|
||||
def __init__(self):
|
||||
self.child_pid = None
|
||||
self.parent_pid = os.getpid()
|
||||
self.subproc_state = None
|
||||
self.tasks = {}
|
||||
self.sock = None
|
||||
self.connection = None
|
||||
self.private_key = None
|
||||
self.ssl_conn = None
|
||||
|
||||
def client_signal_handler(self, sig, unused_frame):
|
||||
"""Signal handler for the parent process.
|
||||
|
||||
This handler receives inter-process communication from the
|
||||
child process in the form of Unix signals.
|
||||
|
||||
:param int sig: Which signal the process received."""
|
||||
# subprocess → client READY : SIGIO
|
||||
# subprocess → client INUSE : SIGUSR1
|
||||
# subprocess → client CANTBIND: SIGUSR2
|
||||
if sig == signal.SIGIO:
|
||||
self.subproc_state = "ready"
|
||||
elif sig == signal.SIGUSR1:
|
||||
self.subproc_state = "inuse"
|
||||
elif sig == signal.SIGUSR2:
|
||||
self.subproc_state = "cantbind"
|
||||
else:
|
||||
# NOTREACHED
|
||||
raise ValueError("Unexpected signal in signal handler")
|
||||
|
||||
def subproc_signal_handler(self, sig, unused_frame):
|
||||
"""Signal handler for the child process.
|
||||
|
||||
This handler receives inter-process communication from the parent
|
||||
process in the form of Unix signals.
|
||||
|
||||
:param int sig: Which signal the process received."""
|
||||
# client → subprocess CLEANUP : SIGINT
|
||||
if sig == signal.SIGINT:
|
||||
try:
|
||||
self.ssl_conn.shutdown()
|
||||
self.ssl_conn.close()
|
||||
except BaseException:
|
||||
# There might not even be any currently active SSL connection.
|
||||
pass
|
||||
try:
|
||||
self.connection.close()
|
||||
except BaseException:
|
||||
# There might not even be any currently active connection.
|
||||
pass
|
||||
try:
|
||||
self.sock.close()
|
||||
except BaseException:
|
||||
# Various things can go wrong in the course of closing these
|
||||
# connections, but none of them can clearly be usefully
|
||||
# reported here and none of them should impede us from
|
||||
# exiting as gracefully as possible.
|
||||
pass
|
||||
os.kill(self.parent_pid, signal.SIGUSR1)
|
||||
sys.exit(0)
|
||||
|
||||
def sni_callback(self, connection):
|
||||
"""Used internally to respond to incoming SNI names.
|
||||
|
||||
This method will set a new OpenSSL context object for this
|
||||
connection when an incoming connection provides an SNI name
|
||||
(in order to serve the appropriate certificate, if any).
|
||||
|
||||
:param OpenSSL.Connection connection: The TLS connection object
|
||||
on which the SNI extension was received."""
|
||||
|
||||
sni_name = connection.get_servername()
|
||||
if sni_name in self.tasks:
|
||||
pem_cert = self.tasks[sni_name]
|
||||
else:
|
||||
# TODO: Should we really present a certificate if we get an
|
||||
# unexpected SNI name? Or should we just disconnect?
|
||||
pem_cert = self.tasks.values()[0]
|
||||
cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
|
||||
pem_cert)
|
||||
new_ctx = OpenSSL.SSL.Context(OpenSSL.SSL.TLSv1_METHOD)
|
||||
new_ctx.set_verify(OpenSSL.SSL.VERIFY_NONE, lambda: False)
|
||||
new_ctx.use_certificate(cert)
|
||||
new_ctx.use_privatekey(self.private_key)
|
||||
connection.set_context(new_ctx)
|
||||
|
||||
def do_parent_process(self, port, delay_amount=5):
|
||||
"""Perform the parent process side of the TCP listener task.
|
||||
|
||||
This should only be called by start_listener(). We will wait
|
||||
up to delay_amount seconds to hear from the child process via
|
||||
a signal.
|
||||
|
||||
:param int port: Which TCP port to bind.
|
||||
:param float delay_amount: How long in seconds to wait for the
|
||||
subprocess to notify us whether it succeeded.
|
||||
|
||||
:returns: True or False according to whether we were notified
|
||||
that the child process succeeded or failed in binding the port."""
|
||||
|
||||
signal.signal(signal.SIGIO, self.client_signal_handler)
|
||||
signal.signal(signal.SIGUSR1, self.client_signal_handler)
|
||||
signal.signal(signal.SIGUSR2, self.client_signal_handler)
|
||||
display = zope.component.getUtility(interfaces.IDisplay)
|
||||
start_time = time.time()
|
||||
while time.time() < start_time + delay_amount:
|
||||
if self.subproc_state == "ready":
|
||||
return True
|
||||
if self.subproc_state == "inuse":
|
||||
display.generic_notification(
|
||||
"Could not bind TCP port {0} because it is already in "
|
||||
"use it is already in use by another process on this "
|
||||
"system (such as a web server).".format(port))
|
||||
return False
|
||||
if self.subproc_state == "cantbind":
|
||||
display.generic_notification(
|
||||
"Could not bind TCP port {0} because you don't have "
|
||||
"the appropriate permissions (for example, you "
|
||||
"aren't running this program as "
|
||||
"root).".format(port))
|
||||
return False
|
||||
time.sleep(0.1)
|
||||
display.generic_notification(
|
||||
"Subprocess unexpectedly timed out while trying to bind TCP "
|
||||
"port {0}.".format(port))
|
||||
return False
|
||||
|
||||
def do_child_process(self, port, key):
|
||||
"""Perform the child process side of the TCP listener task.
|
||||
|
||||
This should only be called by start_listener().
|
||||
|
||||
Normally does not return; instead, the child process exits from
|
||||
within this function or from within the child process signal
|
||||
handler.
|
||||
|
||||
:param int port: Which TCP port to bind.
|
||||
:param le_util.Key key: The private key to use to respond to
|
||||
DVSNI challenge requests."""
|
||||
signal.signal(signal.SIGINT, self.subproc_signal_handler)
|
||||
self.sock = socket.socket()
|
||||
try:
|
||||
self.sock.bind(("0.0.0.0", port))
|
||||
except socket.error, error:
|
||||
if error.errno == socket.errno.EACCES:
|
||||
# Signal permissions denied to bind TCP port
|
||||
os.kill(self.parent_pid, signal.SIGUSR2)
|
||||
elif error.errno == socket.errno.EADDRINUSE:
|
||||
# Signal TCP port is already in use
|
||||
os.kill(self.parent_pid, signal.SIGUSR1)
|
||||
else:
|
||||
# XXX: How to handle unknown errors in binding?
|
||||
raise error
|
||||
sys.exit(1)
|
||||
# XXX: We could use poll mechanism to handle simultaneous
|
||||
# XXX: rather than sequential inbound TCP connections here
|
||||
self.sock.listen(1)
|
||||
# Signal that we've successfully bound TCP port
|
||||
os.kill(self.parent_pid, signal.SIGIO)
|
||||
self.private_key = OpenSSL.crypto.load_privatekey(
|
||||
OpenSSL.crypto.FILETYPE_PEM, key.pem)
|
||||
|
||||
while True:
|
||||
self.connection, _ = self.sock.accept()
|
||||
|
||||
# The code below uses the PyOpenSSL bindings to respond to
|
||||
# the client. This may expose us to bugs and vulnerabilities
|
||||
# in OpenSSL (and creates additional dependencies).
|
||||
ctx = OpenSSL.SSL.Context(OpenSSL.SSL.TLSv1_METHOD)
|
||||
ctx.set_verify(OpenSSL.SSL.VERIFY_NONE, lambda: False)
|
||||
pem_cert = self.tasks.values()[0]
|
||||
first_cert = OpenSSL.crypto.load_certificate(
|
||||
OpenSSL.crypto.FILETYPE_PEM, pem_cert)
|
||||
ctx.use_certificate(first_cert)
|
||||
ctx.use_privatekey(self.private_key)
|
||||
ctx.set_cipher_list("HIGH")
|
||||
ctx.set_tlsext_servername_callback(self.sni_callback)
|
||||
self.ssl_conn = OpenSSL.SSL.Connection(ctx, self.connection)
|
||||
self.ssl_conn.set_accept_state()
|
||||
self.ssl_conn.do_handshake()
|
||||
self.ssl_conn.shutdown()
|
||||
self.ssl_conn.close()
|
||||
|
||||
def start_listener(self, port, key):
|
||||
"""Create a child process which will start a TCP listener on the
|
||||
specified port to perform the specified DVSNI challenges.
|
||||
|
||||
:param int port: The TCP port to bind.
|
||||
:param le_util.Key key: The private key to use to respond to
|
||||
DVSNI challenge requests.
|
||||
:returns: True or False to indicate success or failure creating
|
||||
the subprocess.
|
||||
"""
|
||||
fork_result = os.fork()
|
||||
Crypto.Random.atfork()
|
||||
if fork_result:
|
||||
# PARENT process (still the Let's Encrypt client process)
|
||||
self.child_pid = fork_result
|
||||
# do_parent_process() can return True or False to indicate
|
||||
# reported success or failure creating the listener.
|
||||
return self.do_parent_process(port)
|
||||
else:
|
||||
# CHILD process (the TCP listener subprocess)
|
||||
self.child_pid = os.getpid()
|
||||
# do_child_process() is normally not expected to return but
|
||||
# should terminate via sys.exit().
|
||||
return self.do_child_process(port, key)
|
||||
|
||||
# IAuthenticator method implementations follow
|
||||
|
||||
def get_chall_pref(self, unused_domain):
|
||||
# pylint: disable=no-self-use
|
||||
"""IAuthenticator interface method get_chall_pref.
|
||||
|
||||
Return a list of challenge types that this authenticator
|
||||
can perform for this domain. In the case of the
|
||||
StandaloneAuthenticator, the only challenge type that can ever
|
||||
be performed is dvsni.
|
||||
|
||||
:returns: A list containing only 'dvsni'."""
|
||||
return ["dvsni"]
|
||||
|
||||
def perform(self, chall_list):
|
||||
"""IAuthenticator interface method perform.
|
||||
|
||||
Attempt to perform the
|
||||
specified challenges, returning the status of each. For the
|
||||
StandaloneAuthenticator, because there is no convenient way to add
|
||||
additional requests, this should only be invoked once; subsequent
|
||||
invocations are an error. To perform validations for multiple
|
||||
independent sets of domains, a separate StandaloneAuthenticator
|
||||
should be instantiated.
|
||||
|
||||
:param list chall_list: A list of the the challenge objects to
|
||||
be attempted by this authenticator.
|
||||
:returns: A list in the same order containing, in each position,
|
||||
the successfully configured challenge, False, or None."""
|
||||
if self.child_pid or self.tasks:
|
||||
# We should not be willing to continue with perform
|
||||
# if there were existing pending challenges.
|
||||
raise ValueError(".perform() was called with pending tasks!")
|
||||
results_if_success = []
|
||||
results_if_failure = []
|
||||
if not chall_list or not isinstance(chall_list, list):
|
||||
raise ValueError(".perform() was called without challenge list")
|
||||
for chall in chall_list:
|
||||
if isinstance(chall, challenge_util.DvsniChall):
|
||||
# We will attempt to do it
|
||||
name, r_b64 = chall.domain, chall.r_b64
|
||||
nonce, key = chall.nonce, chall.key
|
||||
cert, s_b64 = challenge_util.dvsni_gen_cert(
|
||||
name, r_b64, nonce, key)
|
||||
self.tasks[nonce + constants.DVSNI_DOMAIN_SUFFIX] = cert
|
||||
results_if_success.append({"type": "dvsni", "s": s_b64})
|
||||
results_if_failure.append(None)
|
||||
else:
|
||||
# We will not attempt to do this challenge because it
|
||||
# is not a type we can handle
|
||||
results_if_success.append(False)
|
||||
results_if_failure.append(False)
|
||||
if not self.tasks:
|
||||
raise ValueError("nothing for .perform() to do")
|
||||
# Try to do the authentication; note that this creates
|
||||
# the listener subprocess via os.fork()
|
||||
if self.start_listener(constants.DVSNI_CHALLENGE_PORT, key):
|
||||
return results_if_success
|
||||
else:
|
||||
# TODO: This should probably raise a DVAuthError exception
|
||||
# rather than returning a list of None objects.
|
||||
return results_if_failure
|
||||
|
||||
def cleanup(self, chall_list):
|
||||
"""IAuthenticator interface method cleanup.
|
||||
|
||||
Remove each of the specified challenges from the list of
|
||||
challenges that still need to be performed. (In the case of
|
||||
the StandaloneAuthenticator, if some challenges are removed
|
||||
from the list, the authenticator socket will still respond to
|
||||
those challenges.) Once all challenges have been removed from
|
||||
the list, the listener is deactivated and stops listening.
|
||||
|
||||
:param list chall_list: A list of the the challenge objects to
|
||||
be deactivated."""
|
||||
# Remove this from pending tasks list
|
||||
for chall in chall_list:
|
||||
assert isinstance(chall, challenge_util.DvsniChall)
|
||||
nonce = chall.nonce
|
||||
if nonce + constants.DVSNI_DOMAIN_SUFFIX in self.tasks:
|
||||
del self.tasks[nonce + constants.DVSNI_DOMAIN_SUFFIX]
|
||||
else:
|
||||
# Could not find the challenge to remove!
|
||||
raise ValueError("could not find the challenge to remove")
|
||||
if self.child_pid and not self.tasks:
|
||||
# There are no remaining challenges, so
|
||||
# try to shutdown self.child_pid cleanly.
|
||||
# TODO: ignore any signals from child during this process
|
||||
os.kill(self.child_pid, signal.SIGINT)
|
||||
time.sleep(1)
|
||||
# TODO: restore original signal handlers in parent process
|
||||
# by resetting their actions to SIG_DFL
|
||||
# print "TCP listener subprocess has been told to shut down"
|
||||
@@ -0,0 +1,6 @@
|
||||
import unittest
|
||||
import package
|
||||
|
||||
|
||||
class CertTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
@@ -0,0 +1,461 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
"""Tests for standalone_authenticator.py."""
|
||||
import mock
|
||||
import unittest
|
||||
|
||||
import os
|
||||
import pkg_resources
|
||||
import signal
|
||||
import socket
|
||||
|
||||
import OpenSSL.crypto
|
||||
import OpenSSL.SSL
|
||||
|
||||
from letsencrypt.client import challenge_util
|
||||
from letsencrypt.client import le_util
|
||||
|
||||
|
||||
# Classes based on to allow interrupting infinite loop under test
|
||||
# after one iteration, based on.
|
||||
# http://igorsobreira.com/2013/03/17/testing-infinite-loops.html
|
||||
|
||||
class SocketAcceptOnlyNTimes(object):
|
||||
# pylint: disable=too-few-public-methods
|
||||
"""
|
||||
Callable that will raise `CallableExhausted`
|
||||
exception after `limit` calls, modified to also return
|
||||
a tuple simulating the return values of a socket.accept()
|
||||
call
|
||||
"""
|
||||
def __init__(self, limit):
|
||||
self.limit = limit
|
||||
self.calls = 0
|
||||
|
||||
def __call__(self):
|
||||
self.calls += 1
|
||||
if self.calls > self.limit:
|
||||
raise CallableExhausted
|
||||
# Modified here for a single use as socket.accept()
|
||||
return (mock.MagicMock(), "ignored")
|
||||
|
||||
class CallableExhausted(Exception):
|
||||
# pylint: disable=too-few-public-methods
|
||||
"""Exception raised when a method is called more than the
|
||||
specified number of times."""
|
||||
|
||||
|
||||
class ChallPrefTest(unittest.TestCase):
|
||||
"""Tests for chall_pref() method."""
|
||||
def setUp(self):
|
||||
from letsencrypt.client.standalone_authenticator import \
|
||||
StandaloneAuthenticator
|
||||
self.authenticator = StandaloneAuthenticator()
|
||||
|
||||
def test_chall_pref(self):
|
||||
self.assertEqual(
|
||||
self.authenticator.get_chall_pref("example.com"), ["dvsni"])
|
||||
|
||||
|
||||
class SNICallbackTest(unittest.TestCase):
|
||||
"""Tests for sni_callback() method."""
|
||||
def setUp(self):
|
||||
from letsencrypt.client.standalone_authenticator import \
|
||||
StandaloneAuthenticator
|
||||
self.authenticator = StandaloneAuthenticator()
|
||||
name, r_b64 = "example.com", le_util.jose_b64encode("x" * 32)
|
||||
test_key = pkg_resources.resource_string(
|
||||
__name__, 'testdata/rsa256_key.pem')
|
||||
nonce, key = "abcdef", le_util.Key("foo", test_key)
|
||||
self.cert = challenge_util.dvsni_gen_cert(name, r_b64, nonce, key)[0]
|
||||
private_key = OpenSSL.crypto.load_privatekey(
|
||||
OpenSSL.crypto.FILETYPE_PEM, key.pem)
|
||||
self.authenticator.private_key = private_key
|
||||
self.authenticator.tasks = {"abcdef.acme.invalid": self.cert}
|
||||
self.authenticator.child_pid = 12345
|
||||
|
||||
def test_real_servername(self):
|
||||
connection = mock.MagicMock()
|
||||
connection.get_servername.return_value = "abcdef.acme.invalid"
|
||||
self.authenticator.sni_callback(connection)
|
||||
self.assertEqual(connection.set_context.call_count, 1)
|
||||
called_ctx = connection.set_context.call_args[0][0]
|
||||
self.assertTrue(isinstance(called_ctx, OpenSSL.SSL.Context))
|
||||
|
||||
def test_fake_servername(self):
|
||||
"""Test behavior of SNI callback when an unexpected name is received.
|
||||
|
||||
(Currently the expected behavior in this case is to return the
|
||||
"first" certificate with which the listener was configured,
|
||||
although they are stored in an unordered data structure so
|
||||
this might not be the one that was first in the challenge list
|
||||
passed to the perform method. In the future, this might result
|
||||
in dropping the connection instead.)"""
|
||||
connection = mock.MagicMock()
|
||||
connection.get_servername.return_value = "example.com"
|
||||
self.authenticator.sni_callback(connection)
|
||||
self.assertEqual(connection.set_context.call_count, 1)
|
||||
called_ctx = connection.set_context.call_args[0][0]
|
||||
self.assertTrue(isinstance(called_ctx, OpenSSL.SSL.Context))
|
||||
|
||||
class ClientSignalHandlerTest(unittest.TestCase):
|
||||
"""Tests for client_signal_handler() method."""
|
||||
def setUp(self):
|
||||
from letsencrypt.client.standalone_authenticator import \
|
||||
StandaloneAuthenticator
|
||||
self.authenticator = StandaloneAuthenticator()
|
||||
self.authenticator.tasks = {"foononce.acme.invalid": "stuff"}
|
||||
self.authenticator.child_pid = 12345
|
||||
|
||||
def test_client_signal_handler(self):
|
||||
self.assertTrue(self.authenticator.subproc_state is None)
|
||||
self.authenticator.client_signal_handler(signal.SIGIO, None)
|
||||
self.assertEqual(self.authenticator.subproc_state, "ready")
|
||||
|
||||
self.authenticator.client_signal_handler(signal.SIGUSR1, None)
|
||||
self.assertEqual(self.authenticator.subproc_state, "inuse")
|
||||
|
||||
self.authenticator.client_signal_handler(signal.SIGUSR2, None)
|
||||
self.assertEqual(self.authenticator.subproc_state, "cantbind")
|
||||
|
||||
# Testing the unreached path for a signal other than these
|
||||
# specified (which can't occur in normal use because this
|
||||
# function is only set as a signal handler for the above three
|
||||
# signals).
|
||||
self.assertRaises(
|
||||
ValueError, self.authenticator.client_signal_handler,
|
||||
signal.SIGPIPE, None)
|
||||
|
||||
|
||||
class SubprocSignalHandlerTest(unittest.TestCase):
|
||||
"""Tests for subproc_signal_handler() method."""
|
||||
def setUp(self):
|
||||
from letsencrypt.client.standalone_authenticator import \
|
||||
StandaloneAuthenticator
|
||||
self.authenticator = StandaloneAuthenticator()
|
||||
self.authenticator.tasks = {"foononce.acme.invalid": "stuff"}
|
||||
self.authenticator.child_pid = 12345
|
||||
self.authenticator.parent_pid = 23456
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.os.kill")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.sys.exit")
|
||||
def test_subproc_signal_handler(self, mock_exit, mock_kill):
|
||||
self.authenticator.ssl_conn = mock.MagicMock()
|
||||
self.authenticator.connection = mock.MagicMock()
|
||||
self.authenticator.sock = mock.MagicMock()
|
||||
self.authenticator.subproc_signal_handler(signal.SIGINT, None)
|
||||
self.assertEquals(self.authenticator.ssl_conn.shutdown.call_count, 1)
|
||||
self.assertEquals(self.authenticator.ssl_conn.close.call_count, 1)
|
||||
self.assertEquals(self.authenticator.connection.close.call_count, 1)
|
||||
self.assertEquals(self.authenticator.sock.close.call_count, 1)
|
||||
mock_kill.assert_called_once_with(
|
||||
self.authenticator.parent_pid, signal.SIGUSR1)
|
||||
mock_exit.assert_called_once_with(0)
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.os.kill")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.sys.exit")
|
||||
def test_subproc_signal_handler_trouble(self, mock_exit, mock_kill):
|
||||
"""Test attempting to shut down a non-existent connection.
|
||||
|
||||
(This could occur because none was established or active at the
|
||||
time the signal handler tried to perform the cleanup)."""
|
||||
self.authenticator.ssl_conn = mock.MagicMock()
|
||||
self.authenticator.connection = mock.MagicMock()
|
||||
self.authenticator.sock = mock.MagicMock()
|
||||
# AttributeError simulates the case where one of these properties
|
||||
# is None because no connection exists. We raise it for
|
||||
# ssl_conn.close() instead of ssl_conn.shutdown() for better code
|
||||
# coverage.
|
||||
self.authenticator.ssl_conn.close.side_effect = AttributeError("!")
|
||||
self.authenticator.connection.close.side_effect = AttributeError("!")
|
||||
self.authenticator.sock.close.side_effect = AttributeError("!")
|
||||
self.authenticator.subproc_signal_handler(signal.SIGINT, None)
|
||||
self.assertEquals(self.authenticator.ssl_conn.shutdown.call_count, 1)
|
||||
self.assertEquals(self.authenticator.ssl_conn.close.call_count, 1)
|
||||
self.assertEquals(self.authenticator.connection.close.call_count, 1)
|
||||
self.assertEquals(self.authenticator.sock.close.call_count, 1)
|
||||
mock_kill.assert_called_once_with(
|
||||
self.authenticator.parent_pid, signal.SIGUSR1)
|
||||
mock_exit.assert_called_once_with(0)
|
||||
|
||||
|
||||
class PerformTest(unittest.TestCase):
|
||||
"""Tests for perform() method."""
|
||||
def setUp(self):
|
||||
from letsencrypt.client.standalone_authenticator import \
|
||||
StandaloneAuthenticator
|
||||
self.authenticator = StandaloneAuthenticator()
|
||||
|
||||
def test_can_perform(self):
|
||||
"""What happens if start_listener() returns True."""
|
||||
test_key = pkg_resources.resource_string(
|
||||
__name__, 'testdata/rsa256_key.pem')
|
||||
key = le_util.Key("something", test_key)
|
||||
chall1 = challenge_util.DvsniChall(
|
||||
"foo.example.com", "whee", "foononce", key)
|
||||
chall2 = challenge_util.DvsniChall(
|
||||
"bar.example.com", "whee", "barnonce", key)
|
||||
bad_chall = ("This", "Represents", "A Non-DVSNI", "Challenge")
|
||||
self.authenticator.start_listener = mock.Mock()
|
||||
self.authenticator.start_listener.return_value = True
|
||||
result = self.authenticator.perform([chall1, chall2, bad_chall])
|
||||
self.assertEqual(len(self.authenticator.tasks), 2)
|
||||
self.assertTrue(
|
||||
self.authenticator.tasks.has_key("foononce.acme.invalid"))
|
||||
self.assertTrue(
|
||||
self.authenticator.tasks.has_key("barnonce.acme.invalid"))
|
||||
self.assertTrue(isinstance(result, list))
|
||||
self.assertEqual(len(result), 3)
|
||||
self.assertTrue(isinstance(result[0], dict))
|
||||
self.assertTrue(isinstance(result[1], dict))
|
||||
self.assertFalse(result[2])
|
||||
self.assertTrue(result[0].has_key("s"))
|
||||
self.assertTrue(result[1].has_key("s"))
|
||||
self.authenticator.start_listener.assert_called_once_with(443, key)
|
||||
|
||||
def test_cannot_perform(self):
|
||||
"""What happens if start_listener() returns False."""
|
||||
test_key = pkg_resources.resource_string(
|
||||
__name__, 'testdata/rsa256_key.pem')
|
||||
key = le_util.Key("something", test_key)
|
||||
chall1 = challenge_util.DvsniChall(
|
||||
"foo.example.com", "whee", "foononce", key)
|
||||
chall2 = challenge_util.DvsniChall(
|
||||
"bar.example.com", "whee", "barnonce", key)
|
||||
bad_chall = ("This", "Represents", "A Non-DVSNI", "Challenge")
|
||||
self.authenticator.start_listener = mock.Mock()
|
||||
self.authenticator.start_listener.return_value = False
|
||||
result = self.authenticator.perform([chall1, chall2, bad_chall])
|
||||
self.assertEqual(len(self.authenticator.tasks), 2)
|
||||
self.assertTrue(
|
||||
self.authenticator.tasks.has_key("foononce.acme.invalid"))
|
||||
self.assertTrue(
|
||||
self.authenticator.tasks.has_key("barnonce.acme.invalid"))
|
||||
self.assertTrue(isinstance(result, list))
|
||||
self.assertEqual(len(result), 3)
|
||||
self.assertEqual(result, [None, None, False])
|
||||
self.authenticator.start_listener.assert_called_once_with(443, key)
|
||||
|
||||
def test_perform_with_pending_tasks(self):
|
||||
self.authenticator.tasks = {"foononce.acme.invalid": "cert_data"}
|
||||
extra_challenge = challenge_util.DvsniChall("a", "b", "c", "d")
|
||||
self.assertRaises(
|
||||
ValueError, self.authenticator.perform, [extra_challenge])
|
||||
|
||||
def test_perform_without_challenge_list(self):
|
||||
extra_challenge = challenge_util.DvsniChall("a", "b", "c", "d")
|
||||
# This is wrong because a challenge must be specified.
|
||||
self.assertRaises(ValueError, self.authenticator.perform, [])
|
||||
# This is wrong because it must be a list, not a bare challenge.
|
||||
self.assertRaises(
|
||||
ValueError, self.authenticator.perform, extra_challenge)
|
||||
# This is wrong because the list must contain at least one challenge.
|
||||
self.assertRaises(
|
||||
ValueError, self.authenticator.perform, range(20))
|
||||
|
||||
|
||||
class StartListenerTest(unittest.TestCase):
|
||||
"""Tests for start_listener() method."""
|
||||
def setUp(self):
|
||||
from letsencrypt.client.standalone_authenticator import \
|
||||
StandaloneAuthenticator
|
||||
self.authenticator = StandaloneAuthenticator()
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator."
|
||||
"Crypto.Random.atfork")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.os.fork")
|
||||
def test_start_listener_fork_parent(self, mock_fork, mock_atfork):
|
||||
self.authenticator.do_parent_process = mock.Mock()
|
||||
self.authenticator.do_parent_process.return_value = True
|
||||
mock_fork.return_value = 22222
|
||||
result = self.authenticator.start_listener(1717, "key")
|
||||
# start_listener is expected to return the True or False return
|
||||
# value from do_parent_process.
|
||||
self.assertTrue(result)
|
||||
self.assertEqual(self.authenticator.child_pid, 22222)
|
||||
self.authenticator.do_parent_process.assert_called_once_with(1717)
|
||||
mock_atfork.assert_called_once_with()
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator."
|
||||
"Crypto.Random.atfork")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.os.fork")
|
||||
def test_start_listener_fork_child(self, mock_fork, mock_atfork):
|
||||
self.authenticator.do_parent_process = mock.Mock()
|
||||
self.authenticator.do_child_process = mock.Mock()
|
||||
mock_fork.return_value = 0
|
||||
self.authenticator.start_listener(1717, "key")
|
||||
self.assertEqual(self.authenticator.child_pid, os.getpid())
|
||||
self.authenticator.do_child_process.assert_called_once_with(
|
||||
1717, "key")
|
||||
mock_atfork.assert_called_once_with()
|
||||
|
||||
class DoParentProcessTest(unittest.TestCase):
|
||||
"""Tests for do_parent_process() method."""
|
||||
def setUp(self):
|
||||
from letsencrypt.client.standalone_authenticator import \
|
||||
StandaloneAuthenticator
|
||||
self.authenticator = StandaloneAuthenticator()
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.signal.signal")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator."
|
||||
"zope.component.getUtility")
|
||||
def test_do_parent_process_ok(self, mock_get_utility, mock_signal):
|
||||
self.authenticator.subproc_state = "ready"
|
||||
result = self.authenticator.do_parent_process(1717)
|
||||
self.assertTrue(result)
|
||||
self.assertEqual(mock_get_utility.call_count, 1)
|
||||
self.assertEqual(mock_signal.call_count, 3)
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.signal.signal")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator."
|
||||
"zope.component.getUtility")
|
||||
def test_do_parent_process_inuse(self, mock_get_utility, mock_signal):
|
||||
self.authenticator.subproc_state = "inuse"
|
||||
result = self.authenticator.do_parent_process(1717)
|
||||
self.assertFalse(result)
|
||||
self.assertEqual(mock_get_utility.call_count, 1)
|
||||
self.assertEqual(mock_signal.call_count, 3)
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.signal.signal")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator."
|
||||
"zope.component.getUtility")
|
||||
def test_do_parent_process_cantbind(self, mock_get_utility, mock_signal):
|
||||
self.authenticator.subproc_state = "cantbind"
|
||||
result = self.authenticator.do_parent_process(1717)
|
||||
self.assertFalse(result)
|
||||
self.assertEqual(mock_get_utility.call_count, 1)
|
||||
self.assertEqual(mock_signal.call_count, 3)
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.signal.signal")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator."
|
||||
"zope.component.getUtility")
|
||||
def test_do_parent_process_timeout(self, mock_get_utility, mock_signal):
|
||||
# Normally times out in 5 seconds and returns False. We can
|
||||
# now set delay_amount to a lower value so that it times out
|
||||
# faster than it would under normal use.
|
||||
result = self.authenticator.do_parent_process(1717, delay_amount=1)
|
||||
self.assertFalse(result)
|
||||
self.assertEqual(mock_get_utility.call_count, 1)
|
||||
self.assertEqual(mock_signal.call_count, 3)
|
||||
|
||||
|
||||
class DoChildProcessTest(unittest.TestCase):
|
||||
"""Tests for do_child_process() method."""
|
||||
def setUp(self):
|
||||
from letsencrypt.client.standalone_authenticator import \
|
||||
StandaloneAuthenticator
|
||||
self.authenticator = StandaloneAuthenticator()
|
||||
name, r_b64 = "example.com", le_util.jose_b64encode("x" * 32)
|
||||
test_key = pkg_resources.resource_string(
|
||||
__name__, 'testdata/rsa256_key.pem')
|
||||
nonce, key = "abcdef", le_util.Key("foo", test_key)
|
||||
self.key = key
|
||||
self.cert = challenge_util.dvsni_gen_cert(name, r_b64, nonce, key)[0]
|
||||
private_key = OpenSSL.crypto.load_privatekey(
|
||||
OpenSSL.crypto.FILETYPE_PEM, key.pem)
|
||||
self.authenticator.private_key = private_key
|
||||
self.authenticator.tasks = {"abcdef.acme.invalid": self.cert}
|
||||
self.authenticator.parent_pid = 12345
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.socket.socket")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.os.kill")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.sys.exit")
|
||||
def test_do_child_process_cantbind1(
|
||||
self, mock_exit, mock_kill, mock_socket):
|
||||
mock_exit.side_effect = IndentationError("subprocess would exit here")
|
||||
eaccess = socket.error(socket.errno.EACCES, "Permission denied")
|
||||
sample_socket = mock.MagicMock()
|
||||
sample_socket.bind.side_effect = eaccess
|
||||
mock_socket.return_value = sample_socket
|
||||
# Using the IndentationError as an error that cannot easily be
|
||||
# generated at runtime, to indicate the behavior of sys.exit has
|
||||
# taken effect without actually causing the test process to exit.
|
||||
# (Just replacing it with a no-op causes logic errors because the
|
||||
# do_child_process code assumes that calling sys.exit() will
|
||||
# cause subsequent code not to be executed.)
|
||||
self.assertRaises(
|
||||
IndentationError, self.authenticator.do_child_process, 1717,
|
||||
self.key)
|
||||
mock_exit.assert_called_once_with(1)
|
||||
mock_kill.assert_called_once_with(12345, signal.SIGUSR2)
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.socket.socket")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.os.kill")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.sys.exit")
|
||||
def test_do_child_process_cantbind2(self, mock_exit, mock_kill,
|
||||
mock_socket):
|
||||
mock_exit.side_effect = IndentationError("subprocess would exit here")
|
||||
eaccess = socket.error(socket.errno.EADDRINUSE, "Port already in use")
|
||||
sample_socket = mock.MagicMock()
|
||||
sample_socket.bind.side_effect = eaccess
|
||||
mock_socket.return_value = sample_socket
|
||||
self.assertRaises(
|
||||
IndentationError, self.authenticator.do_child_process, 1717,
|
||||
self.key)
|
||||
mock_exit.assert_called_once_with(1)
|
||||
mock_kill.assert_called_once_with(12345, signal.SIGUSR1)
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.socket.socket")
|
||||
def test_do_child_process_cantbind3(self, mock_socket):
|
||||
"""Test case where attempt to bind socket results in an unhandled
|
||||
socket error. (The expected behavior is arguably wrong because it
|
||||
will crash the program; the reason for the expected behavior is
|
||||
that we don't have a way to report arbitrary socket errors.)"""
|
||||
eio = socket.error(socket.errno.EIO, "Imaginary unhandled error")
|
||||
sample_socket = mock.MagicMock()
|
||||
sample_socket.bind.side_effect = eio
|
||||
mock_socket.return_value = sample_socket
|
||||
self.assertRaises(
|
||||
socket.error, self.authenticator.do_child_process, 1717, self.key)
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator."
|
||||
"OpenSSL.SSL.Connection")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.socket.socket")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.os.kill")
|
||||
def test_do_child_process_success(self, mock_kill, mock_socket,
|
||||
mock_connection):
|
||||
sample_socket = mock.MagicMock()
|
||||
sample_socket.accept.side_effect = SocketAcceptOnlyNTimes(2)
|
||||
mock_socket.return_value = sample_socket
|
||||
mock_connection.return_value = mock.MagicMock()
|
||||
self.assertRaises(
|
||||
CallableExhausted, self.authenticator.do_child_process, 1717,
|
||||
self.key)
|
||||
mock_socket.assert_called_once_with()
|
||||
sample_socket.bind.assert_called_once_with(("0.0.0.0", 1717))
|
||||
sample_socket.listen.assert_called_once_with(1)
|
||||
self.assertEqual(sample_socket.accept.call_count, 3)
|
||||
mock_kill.assert_called_once_with(12345, signal.SIGIO)
|
||||
# TODO: We could have some tests about the fact that the listener
|
||||
# asks OpenSSL to negotiate a TLS connection (and correctly
|
||||
# sets the SNI callback function).
|
||||
|
||||
|
||||
class CleanupTest(unittest.TestCase):
|
||||
"""Tests for cleanup() method."""
|
||||
def setUp(self):
|
||||
from letsencrypt.client.standalone_authenticator import \
|
||||
StandaloneAuthenticator
|
||||
self.authenticator = StandaloneAuthenticator()
|
||||
self.authenticator.tasks = {"foononce.acme.invalid": "stuff"}
|
||||
self.authenticator.child_pid = 12345
|
||||
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.os.kill")
|
||||
@mock.patch("letsencrypt.client.standalone_authenticator.time.sleep")
|
||||
def test_cleanup(self, mock_sleep, mock_kill):
|
||||
mock_sleep.return_value = None
|
||||
mock_kill.return_value = None
|
||||
chall = challenge_util.DvsniChall(
|
||||
"foo.example.com", "whee", "foononce", "key")
|
||||
self.authenticator.cleanup([chall])
|
||||
mock_kill.assert_called_once_with(12345, signal.SIGINT)
|
||||
mock_sleep.assert_called_once_with(1)
|
||||
|
||||
def test_bad_cleanup(self):
|
||||
chall = challenge_util.DvsniChall(
|
||||
"bad.example.com", "whee", "badnonce", "key")
|
||||
self.assertRaises(ValueError, self.authenticator.cleanup, [chall])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -1,7 +1,7 @@
|
||||
#!/usr/bin/env python
|
||||
"""Parse command line and call the appropriate functions.
|
||||
|
||||
..todo:: Sanity check all input. Be sure to avoid shell code etc...
|
||||
.. todo:: Sanity check all input. Be sure to avoid shell code etc...
|
||||
|
||||
"""
|
||||
import argparse
|
||||
|
||||
@@ -5,3 +5,4 @@ python-augeas==0.5.0
|
||||
requests==2.4.3
|
||||
argparse==1.2.2
|
||||
mock==1.0.1
|
||||
PyOpenSSL==0.13
|
||||
|
||||
Reference in New Issue
Block a user