diff --git a/letsencrypt/acme/messages.py b/letsencrypt/acme/messages.py index df7e2c0cc..e8bc86eaa 100644 --- a/letsencrypt/acme/messages.py +++ b/letsencrypt/acme/messages.py @@ -243,19 +243,34 @@ class Certificate(Message): __slots__ = ("certificate", "chain", "refresh") def _fields_to_json(self): - fields = { - "certificate": jose.b64encode(self.certificate.as_der())} + fields = {"certificate": self._encode_cert(self.certificate)} if self.chain is not None: - fields["chain"] = self.chain + fields["chain"] = [self._encode_cert(cert) for cert in self.chain] if self.refresh is not None: fields["refresh"] = self.refresh return fields + def __eq__(self, other): + # pylint: disable=redefined-outer-name + # M2Crypto.X509 does not implement __eq__, do it manually + return isinstance(other, Certificate) and self.certificate.as_der( + ) == other.certificate.as_der() and [ + cert.as_der() for cert in self.chain] == [ + cert.as_der() for cert in other.chain] + + @classmethod + def _decode_cert(cls, b64der): + return M2Crypto.X509.load_cert_der_string(jose.b64decode(b64der)) + + @classmethod + def _encode_cert(cls, cert): + return jose.b64encode(cert.as_der()) + @classmethod def _from_valid_json(cls, jobj): - certificate = M2Crypto.X509.load_cert_der_string( - jose.b64decode(jobj["certificate"])) - return cls(certificate=certificate, chain=jobj.get("chain", []), + return cls(certificate=cls._decode_cert(jobj["certificate"]), + chain=[cls._decode_cert(cert) for cert in + jobj.get("chain", [])], refresh=jobj.get("refresh")) @@ -360,7 +375,7 @@ class Error(Message): @classmethod def _from_valid_json(cls, jobj): return cls(error=jobj["error"], message=jobj.get("message"), - more_info=jobj.get("more_info")) + more_info=jobj.get("moreInfo")) @Message.register # pylint: disable=too-few-public-methods diff --git a/letsencrypt/acme/messages_test.py b/letsencrypt/acme/messages_test.py index 9637fe852..bd3984ef8 100644 --- a/letsencrypt/acme/messages_test.py +++ b/letsencrypt/acme/messages_test.py @@ -3,6 +3,7 @@ import pkg_resources import unittest import Crypto.PublicKey.RSA +import M2Crypto.X509 import mock from letsencrypt.acme import errors @@ -12,6 +13,8 @@ from letsencrypt.acme import other KEY = Crypto.PublicKey.RSA.importKey(pkg_resources.resource_string( 'letsencrypt.client.tests', 'testdata/rsa256_key.pem')) +CERT = M2Crypto.X509.load_cert_string(pkg_resources.resource_string( + 'letsencrypt.client.tests', 'testdata/cert.pem')) class MessageTest(unittest.TestCase): @@ -39,6 +42,11 @@ class MessageTest(unittest.TestCase): self.msg_cls = TestMessage + def test_fields_to_json_not_implemented(self): + from letsencrypt.acme.messages import Message + # pylint: disable=protected-access + self.assertRaises(NotImplementedError, Message()._fields_to_json) + @classmethod def _from_json(cls, jobj, validate=True): from letsencrypt.acme.messages import Message @@ -67,6 +75,38 @@ class MessageTest(unittest.TestCase): {'type': 'foo'}) +class ChallengeTest(unittest.TestCase): + + def setUp(self): + challenges = [ + {'type': 'simpleHttps', 'token': 'IlirfxKKXAsHtmzK29Pj8A'}, + {'type': 'dns', 'token': 'DGyRejmCefe7v4NfDGDKfA'}, + {'type': 'recoveryToken'}, + ] + combinations = [[0, 2], [1, 2]] + + from letsencrypt.acme.messages import Challenge + self.msg = Challenge( + session_id='aefoGaavieG9Wihuk2aufai3aeZ5EeW4', + nonce='\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9', + challenges=challenges, combinations=combinations) + + self.jmsg = { + 'type': 'challenge', + 'sessionID': 'aefoGaavieG9Wihuk2aufai3aeZ5EeW4', + 'nonce': '7Nbyb1lI6xPVI3Hg3aKSqQ', + 'challenges': challenges, + 'combinations': combinations, + } + + def test_to_json(self): + self.assertEqual(self.msg.to_json(), self.jmsg) + + def test_from_json(self): + from letsencrypt.acme.messages import Challenge + self.assertEqual(Challenge.from_json(self.jmsg), self.msg) + + class ChallengeRequestTest(unittest.TestCase): def setUp(self): @@ -86,6 +126,31 @@ class ChallengeRequestTest(unittest.TestCase): self.assertEqual(ChallengeRequest.from_json(self.jmsg), self.msg) +class AuthorizationTest(unittest.TestCase): + + def setUp(self): + jwk = jose.JWK(key=KEY.publickey()) + + from letsencrypt.acme.messages import Authorization + self.msg = Authorization(recovery_token='tok', jwk=jwk, + identifier='example.com') + + self.jmsg = { + 'type': 'authorization', + 'recoveryToken': 'tok', + 'identifier': 'example.com', + 'jwk': jwk, + } + + def test_to_json(self): + self.assertEqual(self.msg.to_json(), self.jmsg) + + def test_from_json(self): + from letsencrypt.acme.messages import Authorization + self.jmsg['jwk'] = self.jmsg['jwk'].to_json() + self.assertEqual(Authorization.from_json(self.jmsg), self.msg) + + class AuthorizationRequestTest(unittest.TestCase): def setUp(self): @@ -94,6 +159,7 @@ class AuthorizationRequestTest(unittest.TestCase): None, # null {'type': 'recoveryToken', 'token': '23029d88d9e123e'}, ] + self.contact = ["mailto:cert-admin@example.com", "tel:+12025551212"] signature = other.Signature( alg='RS256', jwk=jose.JWK(key=KEY.publickey()), sig='-v\xd8\xc2\xa3\xba0\xd6\x92\x16\xb5.\xbe\xa1[\x04\xbe' @@ -108,7 +174,7 @@ class AuthorizationRequestTest(unittest.TestCase): nonce='\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9', responses=self.responses, signature=signature, - contact=[], + contact=self.contact, ) self.jmsg = { @@ -117,6 +183,7 @@ class AuthorizationRequestTest(unittest.TestCase): 'nonce': '7Nbyb1lI6xPVI3Hg3aKSqQ', 'responses': self.responses, 'signature': signature, + 'contact': self.contact, } def test_create(self): @@ -125,7 +192,8 @@ class AuthorizationRequestTest(unittest.TestCase): name='example.com', key=KEY, responses=self.responses, nonce='\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9', session_id='aefoGaavieG9Wihuk2aufai3aeZ5EeW4', - sig_nonce='\xab?\x08o\xe6\x81$\x9f\xa1\xc9\x025\x1c\x1b\xa5+')) + sig_nonce='\xab?\x08o\xe6\x81$\x9f\xa1\xc9\x025\x1c\x1b\xa5+', + contact=self.contact)) def test_verify(self): self.assertTrue(self.msg.verify('example.com')) @@ -140,6 +208,30 @@ class AuthorizationRequestTest(unittest.TestCase): self.assertEqual(self.msg, AuthorizationRequest.from_json(self.jmsg)) +class CertificateTest(unittest.TestCase): + + def setUp(self): + refresh = 'https://example.com/refresh/Dr8eAwTVQfSS/' + + from letsencrypt.acme.messages import Certificate + self.msg = Certificate( + certificate=CERT, chain=[CERT], refresh=refresh) + + self.jmsg = { + 'type': 'certificate', + 'certificate': jose.b64encode(CERT.as_der()), + 'chain': [jose.b64encode(CERT.as_der())], + 'refresh': refresh, + } + + def test_to_json(self): + self.assertEqual(self.msg.to_json(), self.jmsg) + + def test_from_json(self): + from letsencrypt.acme.messages import Certificate + self.assertEqual(Certificate.from_json(self.jmsg), self.msg) + + class CertificateRequestTest(unittest.TestCase): def setUp(self): @@ -180,6 +272,70 @@ class CertificateRequestTest(unittest.TestCase): self.assertEqual(self.msg, CertificateRequest.from_json(self.jmsg)) +class DeferTest(unittest.TestCase): + + def setUp(self): + from letsencrypt.acme.messages import Defer + self.msg = Defer( + token='O7-s9MNq1siZHlgrMzi9_A', interval=60, + message='Warming up the HSM') + + self.jmsg = { + 'type': 'defer', + 'token': 'O7-s9MNq1siZHlgrMzi9_A', + 'interval': 60, + 'message': 'Warming up the HSM', + } + + def test_to_json(self): + self.assertEqual(self.msg.to_json(), self.jmsg) + + def test_from_json(self): + from letsencrypt.acme.messages import Defer + self.assertEqual(Defer.from_json(self.jmsg), self.msg) + + +class ErrorTest(unittest.TestCase): + + def setUp(self): + from letsencrypt.acme.messages import Error + self.msg = Error( + error='badCSR', message='RSA keys must be at least 2048 bits long', + more_info='https://ca.example.com/documentation/csr-requirements') + + self.jmsg = { + 'type': 'error', + 'error': 'badCSR', + 'message':'RSA keys must be at least 2048 bits long', + 'moreInfo': 'https://ca.example.com/documentation/csr-requirements', + } + + def test_to_json(self): + self.assertEqual(self.msg.to_json(), self.jmsg) + + def test_from_json(self): + from letsencrypt.acme.messages import Error + self.assertEqual(Error.from_json(self.jmsg), self.msg) + + +class RevocationTest(unittest.TestCase): + + def setUp(self): + from letsencrypt.acme.messages import Revocation + self.msg = Revocation() + + self.jmsg = { + 'type': 'revocation', + } + + def test_to_json(self): + self.assertEqual(self.msg.to_json(), self.jmsg) + + def test_from_json(self): + from letsencrypt.acme.messages import Error + self.assertEqual(Error.from_json(self.jmsg), self.msg) + + class RevocationRequestTest(unittest.TestCase): def setUp(self): diff --git a/tox.ini b/tox.ini index 4049c78a0..e429923b2 100644 --- a/tox.ini +++ b/tox.ini @@ -14,7 +14,7 @@ commands = basepython = python2.7 commands = python setup.py dev - python setup.py nosetests --with-coverage --cover-min-percentage=66 + python setup.py nosetests --with-coverage --cover-min-percentage=71 [testenv:lint] # recent versions of pylint do not support Python 2.6 (#97, #187)