starred/certbot-certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-31 18:34:41 +02:00
Code Issues Packages Projects Releases Wiki Activity

cleanup challenge infrastructure

Browse Source
This commit is contained in:
James Kasten
2015-01-09 05:30:15 -08:00
parent 8eaeb1bc5e
commit 0bef6769ba
16 changed files with 837 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
letsencrypt/client/CONFIG.py
+9 -8
View File
@@ -30,9 +30,10 @@ IN_PROGRESS_DIR = os.path.join(BACKUP_DIR, "IN_PROGRESS/")
"""Directory used before a permanent checkpoint is finalized"""
CERT_KEY_BACKUP = os.path.join(WORK_DIR, "keys-certs/")
"""Directory where all certificates/keys are stored.
"""Directory where all certificates/keys are stored. Used for easy revocation"""
Used for easy revocation"""
REV_TOKENS_DIR = os.path.join(WORK_DIR, "revocation_tokens/")
"""Directory where all revocation tokens are saved."""
KEY_DIR = os.path.join(SERVER_ROOT, "ssl/")
"""Where all keys should be stored"""
@@ -56,15 +57,15 @@ CHAIN_PATH = CERT_DIR + "chain-letsencrypt.pem"
INVALID_EXT = ".acme.invalid"
"""Invalid Extension"""
# Challenge Information
#CHALLENGE_PREFERENCES = ["dvsni", "recoveryToken"]
"""Challenge Preferences Dict for currently supported challenges"""
EXCLUSIVE_CHALLENGES = [frozenset(["dvsni", "simpleHttps"])]
"""Mutually Exclusive Challenges - only solve 1"""
CONFIG_CHALLENGES = frozenset(["dvsni", "simpleHttps"])
"""These are challenges that must be solved by a Configurator object"""
AUTH_CHALLENGES = frozenset(["dvsni", "simpleHttps", "dns"])
"""These are challenges that must be solved by an Authenticator object"""
CLIENT_CHALLENGES = frozenset(
["recoveryToken", "recoveryContact", "proofOfPossession"])
"""These are challenges that are handled by client.py"""
# Challenge Constants
S_SIZE = 32
letsencrypt/client/apache/configurator.py
+5 -4
View File
@@ -811,7 +811,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
"""
enabled_dir = os.path.join(self.parser.root, "sites-enabled/")
for entry in os.listdir(enabled_dir):
if os.path.realpath(enabled_dir + entry) == avail_fp:
if os.path.realpath(os.path.join(enabled_dir, entry)) == avail_fp:
return True
return False
@@ -926,7 +926,8 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
###########################################################################
# Challenges Section
###########################################################################
def get_chall_pref(self): # pylint: disable=no-self-use
# pylint: disable=no-self-use, unused-argument
def get_chall_pref(self, domain):
"""Return list of challenge preferences."""
return ["dvsni"]
@@ -948,10 +949,10 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
"""
self.chall_out += len(chall_list)
responses = [None] * len(chall_list)
apache_dvsni = dvsni.ApacheDVSNI(self)
apache_dvsni = dvsni.ApacheDvsni(self)
for i, chall in enumerate(chall_list):
if isinstance(chall, challenge_util.DVSNI_Chall):
if isinstance(chall, challenge_util.DvsniChall):
apache_dvsni.add_chall(chall, i)
sni_response = apache_dvsni.perform()
letsencrypt/client/apache/dvsni.py
+5 -17
View File
@@ -7,18 +7,19 @@ from letsencrypt.client import CONFIG
from letsencrypt.client.apache import parser
class ApacheDVSNI(object):
class ApacheDvsni(object):
"""Class performs DVSNI challenges within the Apache configurator.
:ivar config: ApacheConfigurator object
:type config: :class:`letsencrypt.client.apache.configurator`
:ivar dvsni_chall: Data required for challenges.
where DVSNI_Chall tuples have the following fields
where DvsniChall tuples have the following fields
`domain` (`str`), `r_b64` (base64 `str`), `nonce` (hex `str`)
`key` (:class:`letsencrypt.client.client.Client.Key`)
:type dvsni_chall: `list` of
:class:`letsencrypt.client.challenge_util.DVSNI_Chall`
:class:`letsencrypt.client.challenge_util.DvsniChall`
"""
def __init__(self, config):
@@ -33,7 +34,7 @@ class ApacheDVSNI(object):
"""Add challenge to DVSNI object to perform at once.
:param chall: DVSNI challenge info
:type chall: :class:`letsencrypt.client.challenge_util.DVSNI_Chall`
:type chall: :class:`letsencrypt.client.challenge_util.DvsniChall`
:param int idx: index to challenge in a larger array
@@ -91,19 +92,6 @@ class ApacheDVSNI(object):
return responses
# def chall_complete(self, chall):
# """Used by Authenticator to notify the DVSNI challenge.
# :param chall: Challenge info
# :type chall: :class:`letsencrypt.client.client.Client.DVSNI_Chall`
# """
# self.completed += 1
# if self.completed < len(self.dvsni_chall):
# return False
# return True
# TODO: Variable names
def mod_config(self, ll_addrs):
"""Modifies Apache config files to include challenge vhosts.
letsencrypt/client/apache/parser.py
+1
View File
@@ -225,6 +225,7 @@ class ApacheParser(object):
:rtype: str
"""
# Checkout fnmatch.py in venv/local/lib/python2.7/fnmatch.py
regex = ""
for letter in clean_fn_match:
if letter == '.':
letsencrypt/client/challenge.py
+1
View File
@@ -105,6 +105,7 @@ def _find_dumb_path(challenges, preferences):
def is_preferred(offered_challenge_type, path):
"""Return whether or not the challenge is preferred in path."""
for _, challenge_type in path:
for mutually_exclusive in CONFIG.EXCLUSIVE_CHALLENGES:
# Second part is in case we eventually allow multiple names
letsencrypt/client/challenge_util.py
+10 -1
View File
@@ -8,8 +8,17 @@ from letsencrypt.client import CONFIG
from letsencrypt.client import crypto_util
from letsencrypt.client import le_util
# Authenticator Challenges
DvsniChall = collections.namedtuple("DvsniChall", "domain, r_b64, nonce, key")
SimpleHttpsChall = collections.namedtuple(
"SimpleHttpsChall", "domain, token, key")
DnsChall = collections.namedtuple("DnsChall", "domain, token, key")
DVSNI_Chall = collections.namedtuple("DVSNI_Chall", "domain, r_b64, nonce, key")
# Client Challenges
RecContactChall = collections.namedtuple(
"RecContactChall", "domain, a_url, s_url, contact")
RecTokenChall = collections.namedtuple("RecTokenChall", "domain")
PopChall = collections.namedtuple("PopChall", "domain, alg, nonce, hints")
# DVSNI Challenge functions
letsencrypt/client/client.py
+129 -42
View File
@@ -20,6 +20,7 @@ from letsencrypt.client import errors
from letsencrypt.client import interfaces
from letsencrypt.client import le_util
from letsencrypt.client import network
from letsencrypt.client import recovery_token
# it's weird to point to chocolate servers via raw IPv6 addresses, and
@@ -40,12 +41,15 @@ class Client(object):
:type authkey: :class:`letsencrypt.client.client.Client.Key`
:ivar auth: Object that supports the IAuthenticator interface.
`auth` is used specifically for CONFIG.AUTH_CHALLENGES
:type auth: :class:`letsencrypt.client.interfaces.IAuthenticator`
:ivar installer: Object supporting the IInstaller interface.
:type installer: :class:`letsencrypt.client.interfaces.IInstraller`
"""
zope.interface.implements(interfaces.IAuthenticator)
Key = collections.namedtuple("Key", "file pem")
CSR = collections.namedtuple("CSR", "file data form")
@@ -60,14 +64,7 @@ class Client(object):
self.auth = auth
self.installer = installer
# Client challenges and Authenticator challenges should be separate
# and really should not be conflicting along the same path.
# I have chosen to make client challenges preferred
# as the client challenges should be able to be completely handled
# by this module and does not require outside config changes.
# (which may be costly)
self.preferences = ["recoveryToken"]
self.preferences.extend(auth.get_chall_pref())
self.rec_token = recovery_token.RecoveryToken(server)
def obtain_certificate(self, csr,
cert_path=CONFIG.CERT_PATH,
@@ -104,12 +101,9 @@ class Client(object):
while i < len(responses):
# Get Authorization
if responses[i] is not None:
print "client chall_msgs:", challenge_msgs[i]
print "client responses:", responses[i]
print "client auth_c:", auth_c[i]
print "client client_c:", client_c[i]
self.acme_authorization(
challenge_msgs[i], auth_c[i], client_c[i], responses[i])
challenge_msgs[i], self.names[i],
auth_c[i], client_c[i], responses[i])
# Received authorization, remove challenge from list
# We have also cleaned up challenges... keep index
# in sync
@@ -146,13 +140,15 @@ class Client(object):
return self.network.send_and_receive_expected(
acme.challenge_request(domain), "challenge")
def acme_authorization(self, challenge_msg, auth_c, client_c, responses):
def acme_authorization(
self, challenge_msg, domain, auth_c, client_c, responses):
"""Handle ACME "authorization" phase.
:param dict challenge_msg: ACME "challenge" message.
:param chal_objs: TODO - this will be a new object...
:param responses: TODO
:param str domain: domain that is requesting authorization
:param list auth_c: auth challenges
:param list client_c: client challenges
:param list responses: Responses to all challenges in challenge_msg
:returns: ACME "authorization" message.
:rtype: dict
@@ -161,7 +157,7 @@ class Client(object):
try:
return self.network.send_and_receive_expected(
acme.authorization_request(
challenge_msg["sessionID"], self.names[0],
challenge_msg["sessionID"], domain,
challenge_msg["nonce"], responses, self.authkey.pem),
"authorization")
except errors.LetsEncryptClientError as err:
@@ -322,10 +318,20 @@ class Client(object):
auth_idx = []
client_idx = []
# Client challenges and Authenticator challenges should be separate
# and really should not be conflicting along the same path.
# I have chosen to make client challenges preferred
# as the client challenges should be able to be completely handled
# by this module and does not require outside config changes.
# (which may be costly)
for i, msg in enumerate(challenge_msgs):
prefs = self.get_chall_pref(self.names[i])
prefs.extend(self.auth.get_chall_pref(self.names[i]))
paths.append(challenge.gen_challenge_path(
msg["challenges"],
self.preferences,
prefs,
msg.get("combinations", [])))
logging.info("Performing the following challenges:")
@@ -340,20 +346,16 @@ class Client(object):
responses.append(["null"] * len(msg["challenges"]))
# Do client centric challenges here...
# Since this isn't implemented yet...
# Client challenge responses should be cached...
# The client should be able to solve all challenges the first time
assert not client_i
# Flatten list for authenticator
# Flatten list for client authenticator functions
client_resp = self.perform(
[chall for sublist in client_chall for chall in sublist])
self._assign_responses(client_resp, client_idx, responses)
# Flatten list for auth authenticator
auth_resp = self.auth.perform(
[chall for sublist in auth_chall for chall in sublist])
self._assign_responses(auth_resp, auth_idx, responses)
print 'auth_resp:', auth_resp
print 'auth_idx:', auth_idx
print 'auth_responses:', responses
for i in range(len(paths)):
# If challenges failed to complete... zero them out
if not self._path_satisfied(responses[i], paths[i]):
@@ -476,9 +478,17 @@ class Client(object):
:param list path: List of indices from `challenges`.
:returns: A pair of TODO
:returns: auth_chall, list of `collections.namedtuples`
auth_satisfies, list of indices, each associated auth_chall
satisfieswithin the challenge_msg
client_chall, list of `collections.namedtuples`
client_satisfies, list of indices each associated client_chall
satisfies within the challenge_msg
:rtype: tuple
:raises errors.LetsEncryptClientError: If Challenge type is not
recognized
"""
auth_chall = []
# Since a single invocation of SNI challenge can satisfy multiple
@@ -487,29 +497,106 @@ class Client(object):
client_chall = []
client_satisfies = []
domain = str(domain)
for index in path:
chall = challenges[index]
if chall["type"] == "dvsni":
logging.info(" DVSNI challenge for name %s.", domain)
# Authenticator Challenges
if chall["type"] in CONFIG.AUTH_CHALLENGES:
auth_chall.append(self._construct_auth_chall(chall, domain))
auth_satisfies.append(index)
auth_chall.append(challenge_util.DVSNI_Chall(
str(domain), str(chall["r"]),
str(chall["nonce"]), self.authkey))
elif chall["type"] == "recoveryToken":
logging.info(" Recovery Token Challenge for name: %s.", domain)
# Client Challenges
elif chall["type"] in CONFIG.CLIENT_CHALLENGES:
client_chall.append(self._construct_client_chall(chall, domain))
client_satisfies.append(index)
client_chall.append({
type: "recoveryToken",
})
else:
logging.fatal("Challenge not currently supported")
sys.exit(82)
raise errors.LetsEncryptClientError(
"Received unrecognized challenge of type: "
"%s" % chall["type"])
return auth_chall, auth_satisfies, client_chall, client_satisfies
def _construct_auth_chall(self, chall, domain):
"""Construct Auth Type Challenges.
:param dict chall: Single challenge
:returns: challenge_util named tuple Chall object
:rtype: `collections.namedtuple`
:raises errors.LetsEncryptClientError: If unimplemented challenge exists
"""
if chall["type"] == "dvsni":
logging.info(" DVSNI challenge for name %s.", domain)
return challenge_util.DvsniChall(
domain, str(chall["r"]), str(chall["nonce"]), self.authkey)
elif chall["type"] == "simpleHttps":
logging.info(" SimpleHTTPS challenge for name %s.", domain)
return challenge_util.SimpleHttpsChall(
domain, str(chall["token"]), self.authkey)
elif chall["type"] == "dns":
logging.info(" DNS challenge for name %s.", domain)
return challenge_util.DnsChall(
domain, str(chall["token"]), self.authkey)
else:
raise errors.LetsEncryptClientError(
"Unimplemented Auth Challenge: %s" % chall["type"])
def _construct_client_chall(self, chall, domain):
"""Construct Client Type Challenges.
:param dict chall: Single challenge
:returns: challenge_util named tuple Chall object
:rtype: `collections.namedtuple`
:raises errors.LetsEncryptClientError: If unimplemented challenge exists
"""
if chall["type"] == "recoveryToken":
logging.info(" Recovery Token Challenge for name: %s.", domain)
return challenge_util.RecTokenChall(domain)
elif chall["type"] == "recoveryContact":
logging.info(" Recovery Contact Challenge for name: %s.", domain)
return challenge_util.RecContactChall(
domain,
chall.get("activationURL", None),
chall.get("successURL", None),
chall.get("contact", None))
elif chall["type"] == "proofOfPossession":
logging.info(" Proof-of-Possession Challenge for name: "
"%s", domain)
return challenge_util.PopChall(
domain, chall["alg"], chall["nonce"], chall["hints"])
else:
raise errors.LetsEncryptClientError(
"Unimplemented Client Challenge: %s" % chall["type"])
# pylint: disable=unused-argument
def get_chall_pref(self, domain):
"""Return list of challenge preferences."""
return ["recoveryToken"]
def perform(self, chall_list):
"""Perform client specific challenges."""
responses = []
for chall in chall_list:
if isinstance(chall, challenge_util.RecTokenChall):
responses.append(self.rec_token.perform(chall))
else:
raise errors.LetsEncryptClientError("Unexpected Challenge")
return responses
def validate_key_csr(privkey, csr):
"""Validate CSR and key files.
letsencrypt/client/interfaces.py
+5 -3
View File
@@ -11,9 +11,11 @@ class IAuthenticator(zope.interface.Interface):
ability to perform challenges and attain a certificate.
"""
def get_chall_pref():
def get_chall_pref(domain):
"""Return list of challenge preferences.
:param str domain: Domain for which challenge preferences are sought.
:returns: list of strings with the most preferred challenges first.
:rtype: list
@@ -22,7 +24,7 @@ class IAuthenticator(zope.interface.Interface):
"""Perform the given challenge.
:param list chall_list: List of challenge types defined in client.py
:returns: List of responses
If the challenge cant be completed...
None - Authenticator can perform challenge, but can't at this time
letsencrypt/client/recovery_token.py
+82
View File
@@ -0,0 +1,82 @@
"""Recovery Token Identifier Validation Challenge."""
import errno
import os
import zope.component
# import zope.interface
from letsencrypt.client import CONFIG
from letsencrypt.client import le_util
from letsencrypt.client import interfaces
class RecoveryToken(object):
"""Recovery Token Identifier Validation Challenge.
Based on draft-barnes-acme, section 6.4.
"""
# zope.interface.implements(interfaces.IChallenge)
def __init__(self, server, direc=CONFIG.REV_TOKENS_DIR):
# super(RecoveryToken, self).__init__()
self.token_dir = os.path.join(direc, server)
def perform(self, chall):
"""Perform the Recovery Token Challenge.
:param chall: Recovery Token Challenge
:type chall: :class:`letsencrypt.client.challenge_util.RecTokenChall`
:returns: response
:rtype: dict
"""
token_fp = os.path.join(self.token_dir, chall.domain)
if os.path.isfile(token_fp):
with open(token_fp) as token_fd:
return self.generate_response(token_fd.read())
cancel, token = zope.component.getUtility(
interfaces.IDisplay).generic_input(
"%s - Input Recovery Token: " % chall.domain)
if cancel != 1:
return self.generate_response(token)
return None
def cleanup(self, chall):
"""Cleanup the saved recovery token if it exists.
:param chall: Recovery Token Challenge
:type chall: :class:`letsencrypt.client.challenge_util.RecTokenChall`
"""
try:
os.remove(os.path.join(self.token_dir, chall.domain))
except OSError as err:
if err.errno != errno.ENOENT:
raise
def generate_response(self, token): # pylint: disable=no-self-use
"""Generate json response."""
return {
"type": "recoveryToken",
"token": token,
}
def requires_human(self, domain):
"""Indicates whether or not domain can be auto solved."""
return not os.path.isfile(os.path.join(self.token_dir, domain))
def store_token(self, domain, token):
"""Store token for later automatic use.
:param str domain: domain associated with the token
:param str token: token from authorization
"""
le_util.make_or_verify_dir(self.token_dir, 0o700, os.geteuid())
with open(os.path.join(self.token_dir, domain), 'w') as token_fd:
token_fd.write(str(token))
letsencrypt/client/recovery_token_challenge.py
-37
View File
@@ -1,37 +0,0 @@
"""Recovery Token Identifier Validation Challenge.
.. note:: This challenge has not been implemented into the project yet
"""
import zope.component
import zope.interface
from letsencrypt.client import interfaces
class RecoveryToken(object):
"""Recovery Token Identifier Validation Challenge.
Based on draft-barnes-acme, section 6.4.
"""
zope.interface.implements(interfaces.IChallenge)
def __init__(self):
super(RecoveryToken, self).__init__()
self.token = ""
def perform(self, quiet=True):
cancel, self.token = zope.component.getUtility(
interfaces.IDisplay).generic_input(
"Please Input Recovery Token: ")
return cancel != 1
def cleanup(self):
pass
def generate_response(self):
return {
"type": "recoveryToken",
"token": self.token,
}
letsencrypt/client/tests/acme_util.py
+112
View File
@@ -0,0 +1,112 @@
"""Class helps construct valid ACME messages for testing."""
from letsencrypt.client import CONFIG
CHALLENGES = {
"simpleHttps":
{
"type": "simpleHttps",
"token": "evaGxfADs6pSRb2LAv9IZf17Dt3juxGJ+PCt92wr+oA"
},
"dvsni":
{
"type": "dvsni",
"r": "Tyq0La3slT7tqQ0wlOiXnCY2vyez7Zo5blgPJ1xt5xI",
"nonce": "a82d5ff8ef740d12881f6d3c2277ab2e"
},
"dns":
{
"type": "dns",
"token": "17817c66b60ce2e4012dfad92657527a"
},
"recoveryContact":
{
"type": "recoveryContact",
"activationURL": "https://example.ca/sendrecovery/a5bd99383fb0",
"successURL": "https://example.ca/confirmrecovery/bb1b9928932",
"contact": "c********n@example.com"
},
"recoveryTokent":
{
"type": "recoveryToken"
},
"proofOfPossession":
{
"type": "proofOfPossession",
"alg": "RS256",
"nonce": "eET5udtV7aoX8Xl8gYiZIA",
"hints": {
"jwk": {
"kty": "RSA",
"e": "AQAB",
"n": "KxITJ0rNlfDMAtfDr8eAw...fSSoehDFNZKQKzTZPtQ"
},
"certFingerprints": [
"93416768eb85e33adc4277f4c9acd63e7418fcfe",
"16d95b7b63f1972b980b14c20291f3c0d1855d95",
"48b46570d9fc6358108af43ad1649484def0debf"
],
"subjectKeyIdentifiers":
["d0083162dcc4c8a23ecb8aecbd86120e56fd24e5"],
"serialNumbers": [34234239832, 23993939911, 17],
"issuers": [
"C=US, O=SuperT LLC, CN=SuperTrustworthy Public CA",
"O=LessTrustworthy CA Inc, CN=LessTrustworthy But StillSecure"
],
"authorizedFor": ["www.example.com", "example.net"]
}
}
}
def get_auth_challenges():
"""Returns all auth challenges."""
return [chall for typ, chall in CHALLENGES.iteritems()
if typ in CONFIG.AUTH_CHALLENGES]
def get_client_challenges():
"""Returns all client challenges."""
return [chall for typ, chall in CHALLENGES.iteritems()
if typ in CONFIG.CLIENT_CHALLENGES]
def get_challenges():
"""Returns all challenges."""
return [chall for chall in CHALLENGES.itervalues()]
def gen_combos(challs):
"""Generate natural combinations for challs."""
dv_chall = []
renewal_chall = []
combos = []
for i, chall in enumerate(challs):
if chall["type"] in CONFIG.AUTH_CHALLENGES:
dv_chall.append(i)
else:
renewal_chall.append(i)
# Gen combos for 1 of each type
for i in range(len(dv_chall)):
for j in range(len(renewal_chall)):
combos.append([i, j])
return combos
def get_chall_msg(iden, nonce, challenges, combos=None):
"""Produce an ACME challenge message."""
chall_msg = {
"type": "challenge",
"sessionID": iden,
"nonce": nonce,
"challenges": challenges
}
if combos is None:
return chall_msg
chall_msg["combinations"] = combos
return chall_msg
letsencrypt/client/tests/apache_configurator_test.py
+11 -6
View File
@@ -162,8 +162,11 @@ class TwoVhost80Test(unittest.TestCase):
self.assertRaises(
errors.LetsEncryptConfiguratorError, self.config.get_version)
@mock.patch("letsencrypt.client.apache.dvsni")
def test_perform(self, mock_dvsni, mock_restart):
@mock.patch("letsencrypt.client.apache.configurator."
"dvsni.ApacheDvsni.perform")
@mock.patch("letsencrypt.client.apache.configurator."
"ApacheConfigurator.restart")
def test_perform(self, mock_restart, mock_dvsni_perform):
# Only tests functionality specific to configurator.perform
# Note: As more challenges are offered this will have to be expanded
rsa256_file = pkg_resources.resource_filename(
@@ -172,12 +175,12 @@ class TwoVhost80Test(unittest.TestCase):
__name__, 'testdata/rsa256_key.pem')
auth_key = client.Client.Key(rsa256_file, rsa256_pem)
chall1 = challenge_util.DVSNI_Chall(
chall1 = challenge_util.DvsniChall(
"encryption-example.demo",
"jIq_Xy1mXGN37tb4L6Xj_es58fW571ZNyXekdZzhh7Q",
"37bc5eb75d3e00a19b4f6355845e5a18",
auth_key)
chall2 = challenge_util.DVSNI_Chall(
chall2 = challenge_util.DvsniChall(
"letsencrypt.demo",
"uqnaPzxtrndteOqtrXb0Asl5gOJfWAnnx6QJyvcmlDU",
"59ed014cac95f77057b1d7a1b2c596ba",
@@ -188,11 +191,13 @@ class TwoVhost80Test(unittest.TestCase):
{"type": "dvsni", "s": "randomS2"}
]
mock_dvsni().perform.return_value = dvsni_ret_val
mock_dvsni_perform.return_value = dvsni_ret_val
responses = self.config.perform([chall1, chall2])
self.assertEqual(mock_dvsni.perform.call_count, 1)
self.assertEqual(mock_dvsni_perform.call_count, 1)
self.assertEqual(responses, dvsni_ret_val)
self.assertEqual(mock_restart.call_count, 1)
if __name__ == '__main__':
unittest.main()
letsencrypt/client/tests/apache_dvsni_test.py
+120
View File
@@ -0,0 +1,120 @@
"""Test for letsencrypt.client.apache.dvsni."""
import os
import pkg_resources
import unittest
import shutil
import mock
import zope.component
from letsencrypt.client import challenge_util
from letsencrypt.client import client
from letsencrypt.client import CONFIG
from letsencrypt.client import display
from letsencrypt.client.apache import obj
from letsencrypt.client.tests import config_util
class DvsniPerformTest(unittest.TestCase):
def setUp(self):
from letsencrypt.client.apache import dvsni
zope.component.provideUtility(display.NcursesDisplay())
self.temp_dir, self.config_dir, self.work_dir = config_util.dir_setup(
"debian_apache_2_4/two_vhost_80")
self.ssl_options = config_util.setup_apache_ssl_options(self.config_dir)
# Final slash is currently important
self.config_path = os.path.join(
self.temp_dir, "debian_apache_2_4/two_vhost_80/apache2/")
config = config_util.get_apache_configurator(
self.config_path, self.config_dir, self.work_dir, self.ssl_options)
self.sni = dvsni.ApacheDvsni(config)
rsa256_file = pkg_resources.resource_filename(
__name__, 'testdata/rsa256_key.pem')
rsa256_pem = pkg_resources.resource_string(
__name__, 'testdata/rsa256_key.pem')
auth_key = client.Client.Key(rsa256_file, rsa256_pem)
self.chall1 = challenge_util.DvsniChall(
"encryption-example.demo",
"jIq_Xy1mXGN37tb4L6Xj_es58fW571ZNyXekdZzhh7Q",
"37bc5eb75d3e00a19b4f6355845e5a18",
auth_key)
self.chall2 = challenge_util.DvsniChall(
"letsencrypt.demo",
"uqnaPzxtrndteOqtrXb0Asl5gOJfWAnnx6QJyvcmlDU",
"59ed014cac95f77057b1d7a1b2c596ba",
auth_key)
self.sni.add_chall(self.chall1)
self.sni.add_chall(self.chall2)
def tearDown(self):
shutil.rmtree(self.temp_dir)
shutil.rmtree(self.config_dir)
shutil.rmtree(self.work_dir)
@mock.patch("letsencrypt.client.apache.configurator."
"ApacheConfigurator.restart")
@mock.patch("letsencrypt.client.challenge_util.dvsni_gen_cert")
def test_perform(self, mock_dvsni_gen_cert, mock_restart):
mock_dvsni_gen_cert.side_effect = ["randomS1", "randomS2"]
responses = self.sni.perform()
self.assertEqual(mock_dvsni_gen_cert.call_count, 2)
calls = mock_dvsni_gen_cert.call_args_list
expected_call_list = [
(self.sni.get_cert_file(self.chall1.nonce), self.chall1.domain,
self.chall1.r_b64, self.chall1.nonce, self.chall1.key),
(self.sni.get_cert_file(self.chall2.nonce), self.chall2.domain,
self.chall2.r_b64, self.chall2.nonce, self.chall2.key)
]
for i in range(len(expected_call_list)):
for j in range(len(expected_call_list[0])):
self.assertEqual(calls[i][0][j], expected_call_list[i][j])
self.assertEqual(
len(self.sni.config.parser.find_dir(
"Include", self.sni.challenge_conf)),
1)
self.assertEqual(len(responses), 2)
self.assertEqual(responses[0]["s"], "randomS1")
self.assertEqual(responses[1]["s"], "randomS2")
def test_mod_config(self):
v_addr1 = [obj.Addr(("1.2.3.4", "443")), obj.Addr(("5.6.7.8", "443"))]
v_addr2 = [obj.Addr(("127.0.0.1", "443"))]
ll_addr = []
ll_addr.append(v_addr1)
ll_addr.append(v_addr2)
self.sni.mod_config(ll_addr)
self.sni.config.save()
self.sni.config.parser.find_dir("Include", self.sni.challenge_conf)
vh_match = self.sni.config.aug.match(
"/files" + self.sni.challenge_conf + "//VirtualHost")
vhs = []
for match in vh_match:
# pylint: disable=protected-access
vhs.append(self.sni.config._create_vhost(match))
self.assertEqual(len(vhs), 2)
for vhost in vhs:
if vhost.addrs == set(v_addr1):
self.assertEqual(
vhost.names,
set([str(self.chall1.nonce + CONFIG.INVALID_EXT)]))
else:
self.assertEqual(vhost.addrs, set(v_addr2))
self.assertEqual(
vhost.names,
set([str(self.chall2.nonce + CONFIG.INVALID_EXT)]))
letsencrypt/client/tests/client_test.py
+282
View File
@@ -0,0 +1,282 @@
"""Test client.py."""
import unittest
import mock
import pkg_resources
from letsencrypt.client.tests import acme_util
class VerifyIdentityTest(unittest.TestCase):
"""verify_identities test."""
def setUp(self):
from letsencrypt.client.client import Client
from letsencrypt.client import CONFIG
rsa256_file = pkg_resources.resource_filename(
__name__, 'testdata/rsa256_key.pem')
rsa256_pem = pkg_resources.resource_string(
__name__, 'testdata/rsa256_key.pem')
auth_key = Client.Key(rsa256_file, rsa256_pem)
self.mock_auth = mock.MagicMock(name='ApacheConfigurator')
self.mock_auth.get_chall_pref.return_value = ["dvsni"]
self.mock_auth.perform.side_effect = gen_auth_resp
self.client = Client(
CONFIG.ACME_SERVER, ["0", "1", "2", "3", "4"],
auth_key, self.mock_auth, None)
self.client.perform = mock.MagicMock(
name='perform', side_effect=gen_auth_resp)
def test_name1_dvsni1(self):
self.client.names = ["0"]
challenge = [acme_util.CHALLENGES["dvsni"]]
msgs = [acme_util.get_chall_msg("0", "nonce0", challenge)]
responses, auth_c, client_c = self.client.verify_identities(msgs)
self.assertEqual(len(responses), 1)
self.assertEqual(len(responses[0]), 1)
self.assertEqual("DvsniChall0", responses[0][0])
self.assertEqual(len(auth_c), 1)
self.assertEqual(len(client_c), 1)
self.assertEqual(len(auth_c[0]), 1)
self.assertEqual(len(client_c[0]), 0)
def test_name5_dvsni5(self):
challenge = [acme_util.CHALLENGES["dvsni"]]
msgs = []
for i in range(5):
msgs.append(
acme_util.get_chall_msg(str(i), "nonce%d" % i, challenge))
responses, auth_c, client_c = self.client.verify_identities(msgs)
self.assertEqual(len(responses), 5)
self.assertEqual(len(auth_c), 5)
self.assertEqual(len(client_c), 5)
# Each message contains 1 auth, 0 client
for i in range(5):
self.assertEqual(len(responses[i]), 1)
self.assertEqual(responses[i][0], "DvsniChall%d" % i)
self.assertEqual(len(auth_c[i]), 1)
self.assertEqual(len(client_c[i]), 0)
self.assertEqual(type(auth_c[i][0]).__name__, "DvsniChall")
@mock.patch("letsencrypt.client.client."
"challenge.gen_challenge_path")
def test_name1_auth(self, mock_chall_path):
self.client.names = ["0"]
challenges = acme_util.get_auth_challenges()
combos = acme_util.gen_combos(challenges)
msgs = [acme_util.get_chall_msg("0", "nonce0", challenges, combos)]
path = gen_path(["simpleHttps"], challenges)
mock_chall_path.return_value = path
responses, auth_c, client_c = self.client.verify_identities(msgs)
self.assertEqual(len(responses), 1)
self.assertEqual(len(responses[0]), len(challenges))
self.assertEqual(len(auth_c), 1)
self.assertEqual(len(client_c), 1)
self.assertEqual(
responses[0],
self._get_exp_response("0", path, challenges))
self.assertEqual(len(auth_c[0]), 1)
self.assertEqual(len(client_c[0]), 0)
self.assertEqual(type(auth_c[0][0]).__name__, "SimpleHttpsChall")
@mock.patch("letsencrypt.client.client."
"challenge.gen_challenge_path")
def test_name1_all(self, mock_chall_path):
self.client.names = ["0"]
challenges = acme_util.get_challenges()
combos = acme_util.gen_combos(challenges)
msgs = [acme_util.get_chall_msg("0", "nonce0", challenges, combos)]
path = gen_path(["simpleHttps", "recoveryToken"], challenges)
mock_chall_path.return_value = path
responses, auth_c, client_c = self.client.verify_identities(msgs)
self.assertEqual(len(responses), 1)
self.assertEqual(len(responses[0]), len(challenges))
self.assertEqual(len(auth_c), 1)
self.assertEqual(len(client_c), 1)
self.assertEqual(len(auth_c[0]), 1)
self.assertEqual(len(client_c[0]), 1)
self.assertEqual(
responses[0],
self._get_exp_response("0", path, challenges))
self.assertEqual(type(auth_c[0][0]).__name__, "SimpleHttpsChall")
self.assertEqual(type(client_c[0][0]).__name__, "RecTokenChall")
@mock.patch("letsencrypt.client.client."
"challenge.gen_challenge_path")
def test_name5_all(self, mock_chall_path):
challenges = acme_util.get_challenges()
combos = acme_util.gen_combos(challenges)
msgs = []
for i in range(5):
msgs.append(
acme_util.get_chall_msg(
str(i), "nonce%d" % i, challenges, combos))
path = gen_path(["dvsni", "recoveryContact"], challenges)
mock_chall_path.return_value = path
responses, auth_c, client_c = self.client.verify_identities(msgs)
self.assertEqual(len(responses), 5)
for i in range(5):
self.assertEqual(len(responses[i]), len(challenges))
self.assertEqual(len(auth_c), 5)
self.assertEqual(len(client_c), 5)
for i in range(5):
self.assertEqual(
responses[i], self._get_exp_response(i, path, challenges))
self.assertEqual(len(auth_c[0]), 1)
self.assertEqual(len(client_c[0]), 1)
self.assertEqual(type(auth_c[i][0]).__name__, "DvsniChall")
self.assertEqual(type(client_c[i][0]).__name__, "RecContactChall")
@mock.patch("letsencrypt.client.client."
"challenge.gen_challenge_path")
def test_name5_mix(self, mock_chall_path):
paths = []
msgs = []
chosen_chall = [["dns"],
["dvsni"],
["simpleHttps", "proofOfPossession"],
["simpleHttps"],
["dns", "recoveryToken"]]
challenge_list = [acme_util.get_auth_challenges(),
[acme_util.CHALLENGES["dvsni"]],
acme_util.get_challenges(),
acme_util.get_auth_challenges(),
acme_util.get_challenges()]
# Combos doesn't matter since I am overriding the gen_path function
for i in range(5):
paths.append(gen_path(chosen_chall[i], challenge_list[i]))
msgs.append(
acme_util.get_chall_msg(
str(i), "nonce%d" % i, challenge_list[i]))
mock_chall_path.side_effect = paths
responses, auth_c, client_c = self.client.verify_identities(msgs)
self.assertEqual(len(responses), 5)
self.assertEqual(len(auth_c), 5)
self.assertEqual(len(client_c), 5)
for i in range(5):
resp = self._get_exp_response(i, paths[i], challenge_list[i])
self.assertEqual(responses[i], resp)
self.assertEqual(len(auth_c[i]), 1)
self.assertEqual(len(client_c[i]), len(chosen_chall[i]) - 1)
self.assertEqual(type(auth_c[0][0]).__name__, "DnsChall")
self.assertEqual(type(auth_c[1][0]).__name__, "DvsniChall")
self.assertEqual(type(auth_c[2][0]).__name__, "SimpleHttpsChall")
self.assertEqual(type(auth_c[3][0]).__name__, "SimpleHttpsChall")
self.assertEqual(type(auth_c[4][0]).__name__, "DnsChall")
self.assertEqual(type(client_c[2][0]).__name__, "PopChall")
self.assertEqual(type(client_c[4][0]).__name__, "RecTokenChall")
def _get_exp_response(self, domain, path, challenges):
exp_resp = ["null"] * len(challenges)
for i in path:
exp_resp[i] = translate[challenges[i]["type"]] + str(domain)
return exp_resp
class ClientPerformTest(unittest.TestCase):
"""Test client perform function."""
def setUp(self):
from letsencrypt.client.client import Client
from letsencrypt.client import CONFIG
rsa256_file = pkg_resources.resource_filename(
__name__, 'testdata/rsa256_key.pem')
rsa256_pem = pkg_resources.resource_string(
__name__, 'testdata/rsa256_key.pem')
auth_key = Client.Key(rsa256_file, rsa256_pem)
self.client = Client(
CONFIG.ACME_SERVER, ["example.com"], auth_key, None, None)
self.client.rec_token.perform = mock.MagicMock(
name="rec_token_perform", side_effect=gen_client_resp)
def test_rec_token1(self):
from letsencrypt.client.challenge_util import RecTokenChall
token = RecTokenChall("0")
responses = self.client.perform([token])
self.assertEqual(responses, ["RecTokenChall0"])
def test_rec_token5(self):
from letsencrypt.client.challenge_util import RecTokenChall
tokens = []
for i in range(5):
tokens.append(RecTokenChall(str(i)))
responses = self.client.perform(tokens)
self.assertEqual(len(responses), 5)
for i in range(5):
self.assertEqual(responses[i], "RecTokenChall%d" % i)
def test_unexpected(self):
from letsencrypt.client.challenge_util import DvsniChall
from letsencrypt.client.errors import LetsEncryptClientError
unexpected = DvsniChall("0", "rb64", "123", "invalid_key")
self.assertRaises(
LetsEncryptClientError, self.client.perform, [unexpected])
translate = {"dvsni": "DvsniChall",
"simpleHttps": "SimpleHttpsChall",
"dns": "DnsChall",
"recoveryToken": "RecTokenChall",
"recoveryContact": "RecContactChall",
"proofOfPossession": "PopChall"}
def gen_auth_resp(chall_list):
return ["%s%s" % (type(chall).__name__, chall.domain)
for chall in chall_list]
def gen_client_resp(chall):
return "%s%s" % (type(chall).__name__, chall.domain)
def gen_path(str_list, challenges):
path = []
for i, chall in enumerate(challenges):
for str_chall in str_list:
if chall["type"] == str_chall:
path.append(i)
continue
return path
if __name__ == '__main__':
unittest.main()
letsencrypt/client/tests/recovery_token_test.py
+64
View File
@@ -0,0 +1,64 @@
"""Tests for recovery_token.py."""
import os
import unittest
import shutil
import tempfile
import mock
class RecoveryTokenTest(unittest.TestCase):
def setUp(self):
from letsencrypt.client.recovery_token import RecoveryToken
server = "demo_server"
self.base_dir = tempfile.mkdtemp("tokens")
self.token_dir = os.path.join(self.base_dir, server)
self.rec_token = RecoveryToken(server, self.base_dir)
def tearDown(self):
shutil.rmtree(self.base_dir)
def test_store_token(self):
self.rec_token.store_token("example.com", 111)
path = os.path.join(self.token_dir, "example.com")
self.assertTrue(os.path.isfile(path))
with open(path) as token_fd:
self.assertEqual(token_fd.read(), "111")
def test_requires_human(self):
self.rec_token.store_token("example2.com", 222)
self.assertFalse(self.rec_token.requires_human("example2.com"))
self.assertTrue(self.rec_token.requires_human("example3.com"))
def test_cleanup(self):
from letsencrypt.client.challenge_util import RecTokenChall
self.rec_token.store_token("example3.com", 333)
self.assertFalse(self.rec_token.requires_human("example3.com"))
self.rec_token.cleanup(RecTokenChall("example3.com"))
self.assertTrue(self.rec_token.requires_human("example3.com"))
# Shouldn't throw an error
self.rec_token.cleanup(RecTokenChall("example4.com"))
def test_perform_stored(self):
from letsencrypt.client.challenge_util import RecTokenChall
self.rec_token.store_token("example4.com", 444)
response = self.rec_token.perform(RecTokenChall("example4.com"))
self.assertEqual(response, {"type": "recoveryToken", "token": "444"})
@mock.patch("letsencrypt.client.recovery_token.zope.component.getUtility")
def test_perform_not_stored(self, mock_input):
from letsencrypt.client.challenge_util import RecTokenChall
mock_input().generic_input.side_effect = [(0, "555"), (1, "000")]
response = self.rec_token.perform(RecTokenChall("example5.com"))
self.assertEqual(response, {"type": "recoveryToken", "token": "555"})
response = self.rec_token.perform(RecTokenChall("example6.com"))
self.assertEqual(response, None)
if __name__ == '__main__':
unittest.main()
tox.ini
+1 -1
View File
@@ -14,7 +14,7 @@ commands =
[testenv:cover]
commands =
python setup.py dev
python setup.py nosetests --with-coverage --cover-min-percentage=47
python setup.py nosetests --with-coverage --cover-min-percentage=60
[testenv:lint]
commands =