add dns-account-01 tests

acme/acme/_internal/tests/challenges_test.py

@@ -4,6 +4,8 @@
 from unittest import mock
 import urllib.parse as urllib_parse
 
+from base64 import b32encode
+from hashlib import sha256
 import josepy as jose
 from josepy.jwk import JWKEC
 import OpenSSL
@@ -127,7 +129,7 @@ def setUp(self):
 
     def test_validation_domain_name(self):
         assert '_acme-challenge.www.example.com' == \
-                         self.msg.validation_domain_name('www.example.com')
+            self.msg.validation_domain_name('www.example.com')
 
     def test_validation(self):
         assert "rAa7iIg4K2y63fvUhCfy8dP1Xl7wEhmQq0oChTcE3Zk" == \
@@ -145,6 +147,82 @@ def test_from_json_hashable(self):
         hash(DNS01.from_json(self.jmsg))
 
 
+class DNSACCOUNT01ResponseTest(unittest.TestCase):
+
+    def setUp(self):
+        from acme.challenges import DNSACCOUNT01Response
+        self.msg = DNSACCOUNT01Response(key_authorization=u'foo')
+        self.jmsg = {
+            'resource': 'challenge',
+            'type': 'dns-account-01',
+            'keyAuthorization': u'foo',
+        }
+
+        from acme.challenges import DNSACCOUNT01
+        self.chall = DNSACCOUNT01(token=(b'x' * 16))
+        self.response = self.chall.response(KEY)
+
+    def test_to_partial_json(self):
+        assert {} == self.msg.to_partial_json()
+
+    def test_from_json(self):
+        from acme.challenges import DNSACCOUNT01Response
+        assert self.msg == DNSACCOUNT01Response.from_json(self.jmsg)
+
+    def test_from_json_hashable(self):
+        from acme.challenges import DNSACCOUNT01Response
+        hash(DNSACCOUNT01Response.from_json(self.jmsg))
+
+    def test_simple_verify_failure(self):
+        key2 = jose.JWKRSA.load(test_util.load_vector('rsa256_key.pem'))
+        public_key = key2.public_key()
+        verified = self.response.simple_verify(self.chall, "local", public_key)
+        assert not verified
+
+    def test_simple_verify_success(self):
+        public_key = KEY.public_key()
+        verified = self.response.simple_verify(self.chall, "local", public_key)
+        assert verified
+
+
+class DNSACCOUNT01Test(unittest.TestCase):
+
+    def setUp(self):
+        from acme.challenges import DNSACCOUNT01
+        self.msg = DNSACCOUNT01(token=jose.decode_b64jose(
+            'evaGxfADs6pSRb2LAv9IZf17Dt3juxGJ+PCt92wr+oA'))
+        self.jmsg = {
+            'type': 'dns-account-01',
+            'token': 'evaGxfADs6pSRb2LAv9IZf17Dt3juxGJ-PCt92wr-oA',
+        }
+
+        from acme.challenges import DNSACCOUNT01
+        self.accountURI = "https://example.com/acme/acct/1234"
+        # "_" || base32(SHA-256(Account Resource URL)[0:9])
+        self.accountLabel = '_' + b32encode(
+            sha256(self.accountURI.encode()).digest()[:10]
+        ).decode().lower()
+
+    def test_validation_domain_name(self):
+        assert self.accountLabel + '._acme-challenge.www.example.com' == \
+            self.msg.validation_domain_name(self.accountURI, 'www.example.com')
+
+    def test_validation(self):
+        assert "rAa7iIg4K2y63fvUhCfy8dP1Xl7wEhmQq0oChTcE3Zk" == \
+            self.msg.validation(KEY)
+
+    def test_to_partial_json(self):
+        assert self.jmsg == self.msg.to_partial_json()
+
+    def test_from_json(self):
+        from acme.challenges import DNSACCOUNT01
+        assert self.msg == DNSACCOUNT01.from_json(self.jmsg)
+
+    def test_from_json_hashable(self):
+        from acme.challenges import DNSACCOUNT01
+        hash(DNSACCOUNT01.from_json(self.jmsg))
+
+
 class HTTP01ResponseTest(unittest.TestCase):
 
     def setUp(self):
@@ -221,7 +299,8 @@ def test_simple_verify_timeout(self, mock_get):
         mock_get.assert_called_once_with(self.chall.uri("local"), verify=False,
                                          timeout=30)
         mock_get.reset_mock()
-        self.response.simple_verify(self.chall, "local", KEY.public_key(), timeout=1234)
+        self.response.simple_verify(
+            self.chall, "local", KEY.public_key(), timeout=1234)
         mock_get.assert_called_once_with(self.chall.uri("local"), verify=False,
                                          timeout=1234)
 
@@ -240,7 +319,7 @@ def setUp(self):
 
     def test_path(self):
         assert self.msg.path == '/.well-known/acme-challenge/' \
-                         'evaGxfADs6pSRb2LAv9IZf17Dt3juxGJ-PCt92wr-oA'
+            'evaGxfADs6pSRb2LAv9IZf17Dt3juxGJ-PCt92wr-oA'
 
     def test_uri(self):
         assert 'http://example.com/.well-known/acme-challenge/' \
@@ -303,7 +382,7 @@ def test_gen_verify_cert_gen_key(self):
 
     def test_verify_bad_cert(self):
         assert not self.response.verify_cert(self.domain,
-            test_util.load_cert('cert.pem'))
+                                             test_util.load_cert('cert.pem'))
 
     def test_verify_bad_domain(self):
         key1 = test_util.load_pyopenssl_private_key('rsa512_key.pem')
@@ -319,8 +398,8 @@ def test_simple_verify_bad_key_authorization(self):
     def test_simple_verify(self, mock_verify_cert):
         mock_verify_cert.return_value = mock.sentinel.verification
         assert mock.sentinel.verification == self.response.simple_verify(
-                self.chall, self.domain, KEY.public_key(),
-                cert=mock.sentinel.cert)
+            self.chall, self.domain, KEY.public_key(),
+            cert=mock.sentinel.cert)
         mock_verify_cert.assert_called_once_with(
             self.response, self.domain, mock.sentinel.cert)
 
@@ -406,7 +485,8 @@ def test_from_json_hashable(self):
         hash(DNS.from_json(self.jmsg))
 
     def test_gen_check_validation(self):
-        ec_key_secp384r1 = JWKEC(key=test_util.load_ecdsa_private_key('ec_secp384r1_key.pem'))
+        ec_key_secp384r1 = JWKEC(
+            key=test_util.load_ecdsa_private_key('ec_secp384r1_key.pem'))
         for key, alg in [(KEY, jose.RS256), (ec_key_secp384r1, jose.ES384)]:
             with self.subTest(key=key, alg=alg):
                 assert self.msg.check_validation(
@@ -442,10 +522,12 @@ def test_gen_response(self):
         assert response.validation == mock.sentinel.validation
 
     def test_validation_domain_name(self):
-        assert '_acme-challenge.le.wtf' == self.msg.validation_domain_name('le.wtf')
+        assert '_acme-challenge.le.wtf' == self.msg.validation_domain_name(
+            'le.wtf')
 
     def test_validation_domain_name_ecdsa(self):
-        ec_key_secp384r1 = JWKEC(key=test_util.load_ecdsa_private_key('ec_secp384r1_key.pem'))
+        ec_key_secp384r1 = JWKEC(
+            key=test_util.load_ecdsa_private_key('ec_secp384r1_key.pem'))
         assert self.msg.check_validation(
             self.msg.gen_validation(ec_key_secp384r1, alg=jose.ES384),
             ec_key_secp384r1.public_key()) is True
@@ -489,6 +571,7 @@ def test_check_validation(self):
 
 class JWSPayloadRFC8555Compliant(unittest.TestCase):
     """Test for RFC8555 compliance of JWS generated from resources/challenges"""
+
     def test_challenge_payload(self):
         from acme.challenges import HTTP01Response
 

acme/acme/challenges.py

@@ -1,6 +1,6 @@
 """ACME Identifier Validation Challenges."""
 import abc
-import base64
+from base64 import b32encode
 import codecs
 import functools
 import hashlib
@@ -345,7 +345,7 @@ def validation_domain_name(self, acctURI: str, name: str) -> str:
         :rtype: str
 
         """
-        acctLabel = base64.b32encode(hashlib.sha256(
+        acctLabel = b32encode(hashlib.sha256(
             acctURI.encode()).digest()[:10]).decode().lower()
         return f"_{acctLabel}.{self.LABEL}.{name}"