mirror of
https://github.com/certbot/certbot.git
synced 2026-07-28 00:35:50 +02:00
tests for new code
This commit is contained in:
James Kasten
@@ -61,7 +61,8 @@ class AuthHandler(object):
|
||||
"""
|
||||
if domain in self.domains:
|
||||
raise errors.LetsEncryptAuthHandlerError(
|
||||
"Multiple Challenges for the same domain is not supported.")
|
||||
"Multiple ACMEChallengeMessages for the same domain "
|
||||
"is not supported.")
|
||||
self.domains.append(domain)
|
||||
self.responses[domain] = ["null"] * len(msg["challenges"])
|
||||
self.msgs[domain] = msg
|
||||
@@ -121,12 +122,6 @@ class AuthHandler(object):
|
||||
finally:
|
||||
self._cleanup_challenges(domain)
|
||||
|
||||
def _path_satisfied(self, dom):
|
||||
"""Returns whether a path has been completely satisfied."""
|
||||
return all(
|
||||
None != self.responses[dom][i] and "null" != self.responses[dom][i]
|
||||
for i in self.paths[dom])
|
||||
|
||||
def _satisfy_challenges(self):
|
||||
"""Attempt to satisfy all saved challenge messages."""
|
||||
logging.info("Performing the following challenges:")
|
||||
@@ -151,6 +146,8 @@ class AuthHandler(object):
|
||||
client_resp = self.client_auth.perform(flat_client)
|
||||
dv_resp = self.dv_auth.perform(flat_auth)
|
||||
|
||||
logging.info("Ready for verification...")
|
||||
|
||||
# Assemble Responses
|
||||
self._assign_responses(client_resp, self.client_c)
|
||||
self._assign_responses(dv_resp, self.dv_c)
|
||||
@@ -163,9 +160,16 @@ class AuthHandler(object):
|
||||
self.responses[dom][ichall.index] = flat_list[flat_index]
|
||||
flat_index += 1
|
||||
|
||||
def _path_satisfied(self, dom):
|
||||
"""Returns whether a path has been completely satisfied."""
|
||||
return all(
|
||||
None != self.responses[dom][i] and "null" != self.responses[dom][i]
|
||||
for i in self.paths[dom])
|
||||
|
||||
def _get_chall_pref(self, domain):
|
||||
"""Return list of challenge preferences."""
|
||||
chall_prefs = self.client_auth.get_chall_pref(domain)
|
||||
chall_prefs = []
|
||||
chall_prefs.extend(self.client_auth.get_chall_pref(domain))
|
||||
chall_prefs.extend(self.dv_auth.get_chall_pref(domain))
|
||||
return chall_prefs
|
||||
|
||||
@@ -389,6 +393,10 @@ def _find_dumb_path(challenges, preferences):
|
||||
server message to be fulfilled by the client in order to prove
|
||||
possession of the identifier.
|
||||
|
||||
:param list preferences: A list of preferences representing the
|
||||
challenge type found within the ACME spec. Each challenge type
|
||||
can only be listed once.
|
||||
|
||||
:returns: List of indices from `challenges`.
|
||||
:rtype: list
|
||||
|
||||
@@ -396,6 +404,7 @@ def _find_dumb_path(challenges, preferences):
|
||||
# Add logic for a crappy server
|
||||
# Choose a DV
|
||||
path = []
|
||||
assert(len(preferences) == len(set(preferences)))
|
||||
for pref_c in preferences:
|
||||
for i, offered_challenge in enumerate(challenges):
|
||||
if (pref_c == offered_challenge["type"] and
|
||||
|
||||
@@ -229,104 +229,6 @@ class Client(object):
|
||||
# # TODO enable OCSP Stapling
|
||||
# continue
|
||||
|
||||
def verify_identities(self, challenge_msgs):
|
||||
"""Verify identities.
|
||||
|
||||
This is greatly complicated by the fact that the Authenticator can
|
||||
oftentimes solve many challenges at once. The strategy is to give
|
||||
the authenticator all of the appropriate challenges at once to
|
||||
speed up the process. This creates indexing issues as the challenges
|
||||
can come from many different messages and are not in an exact order
|
||||
because of the optimal path decision. All of this complicated indexing
|
||||
will be completely hidden from the authenticator and all the
|
||||
authenticator must do is return a list of responses in the same order
|
||||
the challenges were given.
|
||||
|
||||
:param list challenge_msgs: List of ACME "challenge" messages.
|
||||
|
||||
:returns: TODO
|
||||
:rtype: TODO
|
||||
|
||||
"""
|
||||
# Every msg's responses are a list within this list
|
||||
responses = []
|
||||
# Every msg's desired path
|
||||
paths = []
|
||||
|
||||
auth_chall = []
|
||||
client_chall = []
|
||||
|
||||
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"],
|
||||
prefs,
|
||||
msg.get("combinations", [])))
|
||||
|
||||
logging.info("Performing the following challenges:")
|
||||
|
||||
auth_c, auth_i, client_c, client_i = self.challenge_factory(
|
||||
self.names[i], msg["challenges"], paths[-1])
|
||||
|
||||
auth_chall.append(auth_c)
|
||||
auth_idx.append(auth_i)
|
||||
client_chall.append(client_c)
|
||||
client_idx.append(client_i)
|
||||
|
||||
responses.append(["null"] * len(msg["challenges"]))
|
||||
|
||||
# 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)
|
||||
|
||||
for i in range(len(paths)):
|
||||
# If challenges failed to complete... zero them out
|
||||
if not self._path_satisfied(responses[i], paths[i]):
|
||||
responses[i] = None
|
||||
auth_chall[i] = None
|
||||
client_chall[i] = None
|
||||
|
||||
logging.info(
|
||||
"Configured Apache for challenges; waiting for verification...")
|
||||
|
||||
return responses, auth_chall, client_chall
|
||||
|
||||
# pylint: disable=no-self-use
|
||||
def _assign_responses(self, flat_resp, idx_list, responses):
|
||||
"""Assign chall_response to appropriate places in response list.
|
||||
|
||||
:param resp: responses from a challenge
|
||||
:type resp: list of dicts
|
||||
|
||||
:param list idx_list: respective challenges flat_resp satisfies
|
||||
:param list responses: master list of responses
|
||||
|
||||
"""
|
||||
flat_index = 0
|
||||
# Every authorization_request message
|
||||
for msg_num in range(len(responses)):
|
||||
for idx in idx_list[msg_num]:
|
||||
responses[msg_num][idx] = flat_resp[flat_index]
|
||||
flat_index += 1
|
||||
|
||||
def store_cert_key(self, cert_file, encrypt=False):
|
||||
"""Store certificate key.
|
||||
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import zope.interface
|
||||
|
||||
from letsencrypt.client import challenge_util
|
||||
from letsencrypt.client import errors
|
||||
from letsencrypt.client import interfaces
|
||||
from letsencrypt.client import recovery_token
|
||||
@@ -27,7 +28,7 @@ class ClientAuthenticator(object):
|
||||
return ["recoveryToken"]
|
||||
|
||||
def perform(self, chall_list):
|
||||
"""Perform client specific challenges."""
|
||||
"""Perform client specific challenges for IAuthenticator"""
|
||||
responses = []
|
||||
for chall in chall_list:
|
||||
if isinstance(chall, challenge_util.RecTokenChall):
|
||||
@@ -37,6 +38,7 @@ class ClientAuthenticator(object):
|
||||
return responses
|
||||
|
||||
def cleanup(self, chall_list):
|
||||
"""Cleanup call for IAuthenticator."""
|
||||
for chall in chall_list:
|
||||
if isinstance(chall, challenge_util.RecTokenChall):
|
||||
self.rec_token.cleanup(chall)
|
||||
|
||||
@@ -59,10 +59,10 @@ CHALLENGES = {
|
||||
}
|
||||
|
||||
|
||||
def get_auth_challenges():
|
||||
def get_dv_challenges():
|
||||
"""Returns all auth challenges."""
|
||||
return [chall for typ, chall in CHALLENGES.iteritems()
|
||||
if typ in CONFIG.AUTH_CHALLENGES]
|
||||
if typ in CONFIG.DV_CHALLENGES]
|
||||
|
||||
|
||||
def get_client_challenges():
|
||||
@@ -83,7 +83,7 @@ def gen_combos(challs):
|
||||
combos = []
|
||||
|
||||
for i, chall in enumerate(challs):
|
||||
if chall["type"] in CONFIG.AUTH_CHALLENGES:
|
||||
if chall["type"] in CONFIG.DV_CHALLENGES:
|
||||
dv_chall.append(i)
|
||||
else:
|
||||
renewal_chall.append(i)
|
||||
|
||||
@@ -0,0 +1,444 @@
|
||||
"""Test auth_handler.py."""
|
||||
import unittest
|
||||
import mock
|
||||
import pkg_resources
|
||||
|
||||
from letsencrypt.client.tests import acme_util
|
||||
|
||||
|
||||
# pylint: disable=protected-access
|
||||
class SatisfyChallengesTest(unittest.TestCase):
|
||||
"""verify_identities test."""
|
||||
def setUp(self):
|
||||
from letsencrypt.client.auth_handler import AuthHandler
|
||||
|
||||
self.mock_dv_auth = mock.MagicMock(name='ApacheConfigurator')
|
||||
self.mock_client_auth = mock.MagicMock(name='ClientAuthenticator')
|
||||
|
||||
self.mock_dv_auth.get_chall_pref.return_value = ["dvsni"]
|
||||
self.mock_client_auth.get_chall_pref.return_value = ["recoveryToken"]
|
||||
|
||||
self.mock_client_auth.perform.side_effect = gen_auth_resp
|
||||
self.mock_dv_auth.perform.side_effect = gen_auth_resp
|
||||
|
||||
self.handler = AuthHandler(
|
||||
self.mock_dv_auth, self.mock_client_auth, None)
|
||||
|
||||
def test_name1_dvsni1(self):
|
||||
dom = "0"
|
||||
challenge = [acme_util.CHALLENGES["dvsni"]]
|
||||
msg = acme_util.get_chall_msg(dom, "nonce0", challenge)
|
||||
self.handler.add_chall_msg(dom, msg, "dummy_key")
|
||||
|
||||
self.handler._satisfy_challenges()
|
||||
|
||||
self.assertEqual(len(self.handler.responses), 1)
|
||||
self.assertEqual(len(self.handler.responses[dom]), 1)
|
||||
|
||||
self.assertEqual("DvsniChall0", self.handler.responses[dom][0])
|
||||
self.assertEqual(len(self.handler.dv_c), 1)
|
||||
self.assertEqual(len(self.handler.client_c), 1)
|
||||
self.assertEqual(len(self.handler.dv_c[dom]), 1)
|
||||
self.assertEqual(len(self.handler.client_c[dom]), 0)
|
||||
|
||||
def test_name5_dvsni5(self):
|
||||
challenge = [acme_util.CHALLENGES["dvsni"]]
|
||||
for i in range(5):
|
||||
self.handler.add_chall_msg(
|
||||
str(i),
|
||||
acme_util.get_chall_msg(str(i), "nonce%d" % i, challenge),
|
||||
"dummy_key")
|
||||
|
||||
self.handler._satisfy_challenges()
|
||||
|
||||
self.assertEqual(len(self.handler.responses), 5)
|
||||
self.assertEqual(len(self.handler.dv_c), 5)
|
||||
self.assertEqual(len(self.handler.client_c), 5)
|
||||
# Each message contains 1 auth, 0 client
|
||||
|
||||
for i in range(5):
|
||||
dom = str(i)
|
||||
self.assertEqual(len(self.handler.responses[dom]), 1)
|
||||
self.assertEqual(self.handler.responses[dom][0], "DvsniChall%d" % i)
|
||||
self.assertEqual(len(self.handler.dv_c[dom]), 1)
|
||||
self.assertEqual(len(self.handler.client_c[dom]), 0)
|
||||
self.assertEqual(
|
||||
type(self.handler.dv_c[dom][0].chall).__name__, "DvsniChall")
|
||||
|
||||
@mock.patch("letsencrypt.client.auth_handler.gen_challenge_path")
|
||||
def test_name1_auth(self, mock_chall_path):
|
||||
dom = "0"
|
||||
|
||||
challenges = acme_util.get_dv_challenges()
|
||||
combos = acme_util.gen_combos(challenges)
|
||||
self.handler.add_chall_msg(
|
||||
dom,
|
||||
acme_util.get_chall_msg("0", "nonce0", challenges, combos),
|
||||
"dummy_key")
|
||||
|
||||
path = gen_path(["simpleHttps"], challenges)
|
||||
mock_chall_path.return_value = path
|
||||
self.handler._satisfy_challenges()
|
||||
|
||||
self.assertEqual(len(self.handler.responses), 1)
|
||||
self.assertEqual(len(self.handler.responses[dom]), len(challenges))
|
||||
self.assertEqual(len(self.handler.dv_c), 1)
|
||||
self.assertEqual(len(self.handler.client_c), 1)
|
||||
|
||||
self.assertEqual(
|
||||
self.handler.responses[dom],
|
||||
self._get_exp_response(dom, path, challenges))
|
||||
|
||||
self.assertEqual(len(self.handler.dv_c[dom]), 1)
|
||||
self.assertEqual(len(self.handler.client_c[dom]), 0)
|
||||
self.assertEqual(
|
||||
type(self.handler.dv_c[dom][0].chall).__name__, "SimpleHttpsChall")
|
||||
|
||||
@mock.patch("letsencrypt.client.auth_handler.gen_challenge_path")
|
||||
def test_name1_all(self, mock_chall_path):
|
||||
dom = "0"
|
||||
|
||||
challenges = acme_util.get_challenges()
|
||||
combos = acme_util.gen_combos(challenges)
|
||||
self.handler.add_chall_msg(
|
||||
dom,
|
||||
acme_util.get_chall_msg(dom, "nonce0", challenges, combos),
|
||||
"dummy_key")
|
||||
|
||||
path = gen_path(["simpleHttps", "recoveryToken"], challenges)
|
||||
mock_chall_path.return_value = path
|
||||
|
||||
self.handler._satisfy_challenges()
|
||||
|
||||
self.assertEqual(len(self.handler.responses), 1)
|
||||
self.assertEqual(len(self.handler.responses[dom]), len(challenges))
|
||||
self.assertEqual(len(self.handler.dv_c), 1)
|
||||
self.assertEqual(len(self.handler.client_c), 1)
|
||||
self.assertEqual(len(self.handler.dv_c[dom]), 1)
|
||||
self.assertEqual(len(self.handler.client_c[dom]), 1)
|
||||
|
||||
self.assertEqual(
|
||||
self.handler.responses[dom],
|
||||
self._get_exp_response(dom, path, challenges))
|
||||
self.assertEqual(
|
||||
type(self.handler.dv_c[dom][0].chall).__name__, "SimpleHttpsChall")
|
||||
self.assertEqual(
|
||||
type(self.handler.client_c[dom][0].chall).__name__, "RecTokenChall")
|
||||
|
||||
@mock.patch("letsencrypt.client.auth_handler.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):
|
||||
self.handler.add_chall_msg(
|
||||
str(i),
|
||||
acme_util.get_chall_msg(
|
||||
str(i), "nonce%d" % i, challenges, combos),
|
||||
"dummy_key")
|
||||
|
||||
path = gen_path(["dvsni", "recoveryContact"], challenges)
|
||||
mock_chall_path.return_value = path
|
||||
|
||||
self.handler._satisfy_challenges()
|
||||
|
||||
self.assertEqual(len(self.handler.responses), 5)
|
||||
for i in range(5):
|
||||
self.assertEqual(
|
||||
len(self.handler.responses[str(i)]), len(challenges))
|
||||
self.assertEqual(len(self.handler.dv_c), 5)
|
||||
self.assertEqual(len(self.handler.client_c), 5)
|
||||
|
||||
for i in range(5):
|
||||
dom = str(i)
|
||||
self.assertEqual(
|
||||
self.handler.responses[dom],
|
||||
self._get_exp_response(dom, path, challenges))
|
||||
self.assertEqual(len(self.handler.dv_c[dom]), 1)
|
||||
self.assertEqual(len(self.handler.client_c[dom]), 1)
|
||||
|
||||
self.assertEqual(
|
||||
type(self.handler.dv_c[dom][0].chall).__name__, "DvsniChall")
|
||||
self.assertEqual(
|
||||
type(self.handler.client_c[dom][0].chall).__name__,
|
||||
"RecContactChall")
|
||||
|
||||
@mock.patch("letsencrypt.client.auth_handler.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_dv_challenges(),
|
||||
[acme_util.CHALLENGES["dvsni"]],
|
||||
acme_util.get_challenges(),
|
||||
acme_util.get_dv_challenges(),
|
||||
acme_util.get_challenges()]
|
||||
|
||||
# Combos doesn't matter since I am overriding the gen_path function
|
||||
for i in range(5):
|
||||
dom = str(i)
|
||||
paths.append(gen_path(chosen_chall[i], challenge_list[i]))
|
||||
self.handler.add_chall_msg(
|
||||
dom,
|
||||
acme_util.get_chall_msg(
|
||||
dom, "nonce%d" % i, challenge_list[i]),
|
||||
"dummy_key")
|
||||
|
||||
mock_chall_path.side_effect = paths
|
||||
|
||||
self.handler._satisfy_challenges()
|
||||
|
||||
self.assertEqual(len(self.handler.responses), 5)
|
||||
self.assertEqual(len(self.handler.dv_c), 5)
|
||||
self.assertEqual(len(self.handler.client_c), 5)
|
||||
|
||||
for i in range(5):
|
||||
dom = str(i)
|
||||
resp = self._get_exp_response(i, paths[i], challenge_list[i])
|
||||
self.assertEqual(self.handler.responses[dom], resp)
|
||||
self.assertEqual(len(self.handler.dv_c[dom]), 1)
|
||||
self.assertEqual(len(self.handler.client_c[dom]), len(chosen_chall[i]) - 1)
|
||||
|
||||
self.assertEqual(
|
||||
type(self.handler.dv_c["0"][0].chall).__name__, "DnsChall")
|
||||
self.assertEqual(
|
||||
type(self.handler.dv_c["1"][0].chall).__name__, "DvsniChall")
|
||||
self.assertEqual(
|
||||
type(self.handler.dv_c["2"][0].chall).__name__, "SimpleHttpsChall")
|
||||
self.assertEqual(
|
||||
type(self.handler.dv_c["3"][0].chall).__name__, "SimpleHttpsChall")
|
||||
self.assertEqual(
|
||||
type(self.handler.dv_c["4"][0].chall).__name__, "DnsChall")
|
||||
|
||||
self.assertEqual(
|
||||
type(self.handler.client_c["2"][0].chall).__name__, "PopChall")
|
||||
self.assertEqual(
|
||||
type(self.handler.client_c["4"][0].chall).__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
|
||||
|
||||
def printout_handler(self):
|
||||
print "***** Test Printout *****"
|
||||
for dom in self.handler.domains:
|
||||
print "Domain:", dom
|
||||
print "***Challenge Messages***"
|
||||
print self.handler.msgs[dom]
|
||||
print "**responses**"
|
||||
print self.handler.responses[dom]
|
||||
print "**path**"
|
||||
print self.handler.paths[dom]
|
||||
print "**dv_c**"
|
||||
for item in self.handler.dv_c[dom]:
|
||||
print item
|
||||
print "**client_c**"
|
||||
for item in self.handler.client_c[dom]:
|
||||
print item
|
||||
|
||||
|
||||
# pylint: diable=protected-access
|
||||
class GetAuthorizationsTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
from letsencrypt.client.auth_handler import AuthHandler
|
||||
|
||||
self.mock_dv_auth = mock.MagicMock(name='ApacheConfigurator')
|
||||
self.mock_client_auth = mock.MagicMock(name='ClientAuthenticator')
|
||||
|
||||
self.mock_sat_chall = mock.MagicMock(name="_satisfy_challenges")
|
||||
self.mock_acme_auth = mock.MagicMock(name="acme_authorization")
|
||||
|
||||
self.iteration = 0
|
||||
|
||||
self.handler = AuthHandler(
|
||||
self.mock_dv_auth, self.mock_client_auth, None)
|
||||
|
||||
self.handler._satisfy_challenges = self.mock_sat_chall
|
||||
self.handler.acme_authorization = self.mock_acme_auth
|
||||
|
||||
def test_solved3_at_once(self):
|
||||
# Set 3 DVSNI challenges
|
||||
challenge = [acme_util.CHALLENGES["dvsni"]]
|
||||
for i in range(3):
|
||||
self.handler.add_chall_msg(
|
||||
str(i),
|
||||
acme_util.get_chall_msg(str(i), "nonce%d" % i, challenge),
|
||||
"dummy_key")
|
||||
|
||||
self.mock_sat_chall.side_effect = self._sat_solved_at_once
|
||||
self.handler.get_authorizations()
|
||||
|
||||
self.assertEqual(self.mock_sat_chall.call_count, 1)
|
||||
self.assertEqual(self.mock_acme_auth.call_count, 3)
|
||||
|
||||
exp_call_list = [mock.call("0"), mock.call("1"), mock.call("2")]
|
||||
self.assertEqual(
|
||||
self.mock_acme_auth.call_args_list, exp_call_list)
|
||||
self._test_finished()
|
||||
|
||||
def _sat_solved_at_once(self):
|
||||
for i in range(3):
|
||||
dom = str(i)
|
||||
self.handler.responses[dom] = ["DvsniChall%d" % i]
|
||||
self.handler.paths[dom] = [0]
|
||||
# Assignment was > 80 char...
|
||||
dv_c, c_c = self.handler._challenge_factory(dom, [0])
|
||||
|
||||
self.handler.dv_c[dom], self.handler.client_c[dom] = dv_c, c_c
|
||||
|
||||
def test_progress_failure(self):
|
||||
from letsencrypt.client.errors import LetsEncryptAuthHandlerError
|
||||
challenges = acme_util.get_challenges()
|
||||
self.handler.add_chall_msg(
|
||||
"0",
|
||||
acme_util.get_chall_msg("0", "nonce0", challenges),
|
||||
"dummy_key")
|
||||
|
||||
# Don't do anything to satisfy challenges
|
||||
self.mock_sat_chall.side_effect = self._sat_failure
|
||||
|
||||
self.assertRaises(
|
||||
LetsEncryptAuthHandlerError, self.handler.get_authorizations)
|
||||
|
||||
# Check to make sure program didn't loop
|
||||
self.assertEqual(self.mock_sat_chall.call_count, 1)
|
||||
|
||||
def _sat_failure(self):
|
||||
dom = "0"
|
||||
self.handler.paths[dom] = gen_path(
|
||||
["dns", "recoveryToken"], self.handler.msgs[dom]["challenges"])
|
||||
dv_c, c_c = self.handler._challenge_factory(
|
||||
dom, self.handler.paths[dom])
|
||||
self.handler.dv_c[dom], self.handler.client_c[dom] = dv_c, c_c
|
||||
|
||||
def test_incremental_progress(self):
|
||||
challs = []
|
||||
challs.append(acme_util.get_challenges())
|
||||
challs.append(acme_util.get_dv_challenges())
|
||||
for i in range(2):
|
||||
dom = str(i)
|
||||
self.handler.add_chall_msg(
|
||||
dom,
|
||||
acme_util.get_chall_msg(dom, "nonce%d" % i, challs[i]),
|
||||
"dummy_key")
|
||||
|
||||
self.mock_sat_chall.side_effect = self._sat_incremental
|
||||
|
||||
self.handler.get_authorizations()
|
||||
|
||||
self._test_finished()
|
||||
self.assertEqual(self.mock_acme_auth.call_args_list,
|
||||
[mock.call("1"), mock.call("0")])
|
||||
|
||||
def _sat_incremental(self):
|
||||
from letsencrypt.client.errors import LetsEncryptAuthHandlerError
|
||||
|
||||
# Exact responses don't matter, just path/response match
|
||||
if self.iteration == 0:
|
||||
# Only solve one of "0" required challs
|
||||
self.handler.responses["0"][1] = "onecomplete"
|
||||
self.handler.responses["0"][3] = None
|
||||
self.handler.responses["1"] = ["null", "null", "goodresp"]
|
||||
self.handler.paths["0"] = [1, 3]
|
||||
self.handler.paths["1"] = [2]
|
||||
# This is probably overkill... but set it anyway
|
||||
dv_c, c_c = self.handler._challenge_factory("0", [1, 3])
|
||||
self.handler.dv_c["0"], self.handler.client_c["0"] = dv_c, c_c
|
||||
dv_c, c_c = self.handler._challenge_factory("1", [2])
|
||||
self.handler.dv_c["1"], self.handler.client_c["1"] = dv_c, c_c
|
||||
|
||||
self.iteration += 1
|
||||
|
||||
elif self.iteration == 1:
|
||||
# Quick check to make sure it was actually completed.
|
||||
self.assertEqual(
|
||||
self.mock_acme_auth.call_args_list, [mock.call("1")])
|
||||
self.handler.responses["0"][1] = "now_finish"
|
||||
self.handler.responses["0"][3] = "finally!"
|
||||
|
||||
else:
|
||||
raise LetsEncryptAuthHandlerError(
|
||||
"Failed incremental test: too many invocations")
|
||||
|
||||
def _test_finished(self):
|
||||
self.assertFalse(self.handler.msgs)
|
||||
self.assertFalse(self.handler.dv_c)
|
||||
self.assertFalse(self.handler.responses)
|
||||
self.assertFalse(self.handler.paths)
|
||||
self.assertFalse(self.handler.domains)
|
||||
|
||||
# pylint: disable=protected-access
|
||||
class PathSatisfiedTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
from letsencrypt.client.auth_handler import AuthHandler
|
||||
self.handler = AuthHandler(None, None, None)
|
||||
|
||||
def test_satisfied_true(self):
|
||||
dom = ["0", "1", "2", "3", "4"]
|
||||
self.handler.paths[dom[0]] = [1, 2]
|
||||
self.handler.responses[dom[0]] = ["null", "sat", "sat2", "null"]
|
||||
|
||||
self.handler.paths[dom[1]] = [0]
|
||||
self.handler.responses[dom[1]] = ["sat", None, None, "null"]
|
||||
|
||||
self.handler.paths[dom[2]] = [0]
|
||||
self.handler.responses[dom[2]] = ["sat"]
|
||||
|
||||
self.handler.paths[dom[3]] = []
|
||||
self.handler.responses[dom[3]] = []
|
||||
|
||||
self.handler.paths[dom[4]] = []
|
||||
self.handler.responses[dom[4]] = ["respond... sure"]
|
||||
|
||||
for i in range(5):
|
||||
self.assertTrue(self.handler._path_satisfied(dom[i]))
|
||||
|
||||
def test_not_satisfied(self):
|
||||
dom = ["0", "1", "2", "3", "4"]
|
||||
self.handler.paths[dom[0]] = [1, 2]
|
||||
self.handler.responses[dom[0]] = ["sat1", "null", "sat2", "null"]
|
||||
|
||||
self.handler.paths[dom[1]] = [0]
|
||||
self.handler.responses[dom[1]] = [None, "null", "null", "null"]
|
||||
|
||||
self.handler.paths[dom[2]] = [0]
|
||||
self.handler.responses[dom[2]] = [None]
|
||||
|
||||
self.handler.paths[dom[3]] = [0]
|
||||
self.handler.responses[dom[3]] = ["null"]
|
||||
|
||||
for i in range(4):
|
||||
self.assertFalse(self.handler._path_satisfied(dom[i]))
|
||||
|
||||
|
||||
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_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()
|
||||
@@ -0,0 +1,80 @@
|
||||
import unittest
|
||||
|
||||
import mock
|
||||
|
||||
|
||||
class PerformTest(unittest.TestCase):
|
||||
"""Test client perform function."""
|
||||
def setUp(self):
|
||||
from letsencrypt.client.client_authenticator import ClientAuthenticator
|
||||
|
||||
self.auth = ClientAuthenticator("demo_server.org")
|
||||
self.auth.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.auth.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.auth.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 LetsEncryptClientAuthError
|
||||
|
||||
unexpected = DvsniChall("0", "rb64", "123", "invalid_key")
|
||||
|
||||
self.assertRaises(
|
||||
LetsEncryptClientAuthError, self.auth.perform, [unexpected])
|
||||
|
||||
|
||||
class CleanupTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
from letsencrypt.client.client_authenticator import ClientAuthenticator
|
||||
|
||||
self.auth = ClientAuthenticator("demo_server.org")
|
||||
self.mock_cleanup = mock.MagicMock(name="rec_token_cleanup")
|
||||
self.auth.rec_token.cleanup = self.mock_cleanup
|
||||
|
||||
def test_rec_token2(self):
|
||||
from letsencrypt.client.challenge_util import RecTokenChall
|
||||
token1 = RecTokenChall("0")
|
||||
token2 = RecTokenChall("1")
|
||||
|
||||
self.auth.cleanup([token1, token2])
|
||||
|
||||
self.assertEqual(self.mock_cleanup.call_args_list,
|
||||
[mock.call(token1), mock.call(token2)])
|
||||
|
||||
def test_unexpected(self):
|
||||
from letsencrypt.client.challenge_util import DvsniChall
|
||||
from letsencrypt.client.challenge_util import RecTokenChall
|
||||
from letsencrypt.client.errors import LetsEncryptClientAuthError
|
||||
|
||||
token = RecTokenChall("0")
|
||||
unexpected = DvsniChall("0", "rb64", "123", "dummy_key")
|
||||
|
||||
self.assertRaises(
|
||||
LetsEncryptClientAuthError, self.auth.cleanup, [token, unexpected])
|
||||
|
||||
|
||||
def gen_client_resp(chall):
|
||||
return "%s%s" % (type(chall).__name__, chall.domain)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -1,282 +0,0 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user