diff --git a/acme/acme/challenges.py b/acme/acme/challenges.py index 1e456d325..f48c20443 100644 --- a/acme/acme/challenges.py +++ b/acme/acme/challenges.py @@ -1,5 +1,8 @@ """ACME Identifier Validation Challenges.""" import abc +import base64 +import dns.resolver +import dns.exception import functools import hashlib import logging @@ -215,6 +218,82 @@ class KeyAuthorizationChallenge(_TokenDVChallenge): self.validation(account_key, *args, **kwargs)) +@ChallengeResponse.register +class DNS01Response(KeyAuthorizationChallengeResponse): + """ACME "dns-01" challenge response.""" + typ = "dns-01" + + def simple_verify(self, chall, domain, account_public_key): + """Simple verify. + + :param challenges.DNS01 chall: Corresponding challenge. + :param unicode domain: Domain name being verified. + :param account_public_key: Public key for the key pair + being authorized. If ``None`` key verification is not + performed! + :param JWK account_public_key: + + :returns: ``True`` iff validation is successful, ``False`` + otherwise. + :rtype: bool + + """ + if not self.verify(chall, account_public_key): + logger.debug("Verification of key authorization in response failed") + return False + + validation_name = chall.validation_domain_name(domain) + validation = chall.validation(account_public_key) + logger.debug("Verifying %s at %s...", chall.typ, validation_name) + txt_records = [] + try: + dns_response = dns.resolver.query(validation_name, 'TXT') + for rdata in dns_response: + for txt_record in rdata.strings: + txt_records.append(txt_record) + except dns.exception.DNSException as error: + logger.error("Unable to resolve %s: %s", validation_name, error) + return False + + for txt_record in txt_records: + if txt_record == validation: + return True + + logger.debug("Key authorization from response (%r) doesn't match any " + "DNS response in %r", self.key_authorization, txt_records) + return False + +@Challenge.register # pylint: disable=too-many-ancestors +class DNS01(KeyAuthorizationChallenge): + """ACME "dns-01" challenge.""" + + response_cls = DNS01Response + typ = response_cls.typ + + LABEL = "_acme-challenge" + """Label clients prepend to the domain name being validated.""" + + def validation(self, account_key, **unused_kwargs): + """Generate validation. + + :param JWK account_key: + :rtype: unicode + + """ + key_authorization = self.key_authorization(account_key) + # FIXME Once boulder response according to the spec this needs to be fixed + # return base64.b64encode(hashlib.sha256(key_authorization).digest()) + return hashlib.sha256(key_authorization).hexdigest() + + def validation_domain_name(self, name): + """Domain name for TXT validation record. + + :param unicode name: Domain name being validated. + + """ + return "{0}.{1}".format(self.LABEL, name) + + @ChallengeResponse.register class HTTP01Response(KeyAuthorizationChallengeResponse): """ACME http-01 challenge response.""" diff --git a/acme/acme/challenges_test.py b/acme/acme/challenges_test.py index a4e78ebe9..a587d3f61 100644 --- a/acme/acme/challenges_test.py +++ b/acme/acme/challenges_test.py @@ -1,6 +1,7 @@ """Tests for acme.challenges.""" import unittest +import dns.rrset import mock import OpenSSL import requests @@ -76,6 +77,98 @@ class KeyAuthorizationChallengeResponseTest(unittest.TestCase): key_authorization='.foo.oKGqedy-b-acd5eoybm2f-NVFxvyOoET5CNy3xnv8WY') self.assertFalse(response.verify(self.chall, KEY.public_key())) +class DNS01ResponseTest(unittest.TestCase): + # pylint: disable=too-many-instance-attributes + + def setUp(self): + from acme.challenges import DNS01Response + self.msg = DNS01Response(key_authorization=u'foo') + self.jmsg = { + 'resource': 'challenge', + 'type': 'dns-01', + 'keyAuthorization': u'foo', + } + + from acme.challenges import DNS01 + self.chall = DNS01(token=(b'x' * 16)) + self.response = self.chall.response(KEY) + + # This takes advantage of the fact that an answer object mostly behaves like + # an RRset + def create_txt_response(self, name, txt_record): + return dns.rrset.from_text(name, 60, "IN", "TXT", txt_record) + + def test_to_partial_json(self): + self.assertEqual(self.jmsg, self.msg.to_partial_json()) + + def test_from_json(self): + from acme.challenges import DNS01Response + self.assertEqual( + self.msg, DNS01Response.from_json(self.jmsg)) + + def test_from_json_hashable(self): + from acme.challenges import DNS01Response + hash(DNS01Response.from_json(self.jmsg)) + + def test_simple_verify_bad_key_authorization(self): + key2 = jose.JWKRSA.load(test_util.load_vector('rsa256_key.pem')) + self.response.simple_verify(self.chall, "local", key2.public_key()) + + @mock.patch("acme.challenges.dns.resolver.query") + def test_simple_verify_good_validation(self, mock_dns): + mock_dns.return_value = self.create_txt_response( + self.chall.validation_domain_name("local"), + self.chall.validation(KEY.public_key())) + self.assertTrue(self.response.simple_verify( + self.chall, "local", KEY.public_key())) + mock_dns.assert_called_once_with( + self.chall.validation_domain_name("local"), "TXT") + + @mock.patch("acme.challenges.dns.resolver.query") + def test_simple_verify_bad_validation(self, mock_dns): + mock_dns.return_value = self.create_txt_response( + self.chall.validation_domain_name("local"), "!") + self.assertFalse(self.response.simple_verify( + self.chall, "local", KEY.public_key())) + + @mock.patch("acme.challenges.dns.resolver.query") + def test_simple_verify_connection_error(self, mock_dns): + mock_dns.side_effect = dns.exception.DNSException + self.assertFalse(self.response.simple_verify( + self.chall, "local", KEY.public_key())) + +class DNS01Test(unittest.TestCase): + + def setUp(self): + from acme.challenges import DNS01 + self.msg = DNS01( + token=jose.decode_b64jose( + 'evaGxfADs6pSRb2LAv9IZf17Dt3juxGJ+PCt92wr+oA')) + self.jmsg = { + 'type': 'dns-01', + 'token': 'evaGxfADs6pSRb2LAv9IZf17Dt3juxGJ-PCt92wr-oA', + } + + def test_validation_domain_name(self): + self.assertEqual('_acme-challenge.www.example.com', + self.msg.validation_domain_name('www.example.com')) + + def test_validation(self): + self.assertEqual( + "ac06bb8888382b6cbaddfbd48427f2f1d3f55e5ef0121990ab4a02853704dd99", + self.msg.validation(KEY)) + + def test_to_partial_json(self): + self.assertEqual(self.jmsg, self.msg.to_partial_json()) + + def test_from_json(self): + from acme.challenges import DNS01 + self.assertEqual(self.msg, DNS01.from_json(self.jmsg)) + + def test_from_json_hashable(self): + from acme.challenges import DNS01 + hash(DNS01.from_json(self.jmsg)) + class HTTP01ResponseTest(unittest.TestCase): # pylint: disable=too-many-instance-attributes diff --git a/acme/setup.py b/acme/setup.py index ba2c88394..dd2bce5d9 100644 --- a/acme/setup.py +++ b/acme/setup.py @@ -12,6 +12,7 @@ install_requires = [ 'cryptography>=0.8', # Connection.set_tlsext_host_name (>=0.13), X509Req.get_extensions (>=0.15) 'PyOpenSSL>=0.15', + 'dnspython', 'pyrfc3339', 'pytz', 'requests',