diff --git a/acme/messages.py b/acme/messages.py deleted file mode 100644 index 6d46f894c..000000000 --- a/acme/messages.py +++ /dev/null @@ -1,367 +0,0 @@ -"""ACME protocol v00 messages. - -.. warning:: This module is an implementation of the draft `ACME - protocol version 00`_, and not the "RESTified" `ACME protocol version - 01`_ or later. It should work with `older Node.js implementation`_, - but will definitely not work with Boulder_. It is kept for reference - purposes only. - - -.. _`ACME protocol version 00`: - https://github.com/letsencrypt/acme-spec/blob/v00/draft-barnes-acme.md - -.. _`ACME protocol version 01`: - https://github.com/letsencrypt/acme-spec/blob/v01/draft-barnes-acme.md - -.. _Boulder: https://github.com/letsencrypt/boulder - -.. _`older Node.js implementation`: - https://github.com/letsencrypt/node-acme/commit/f42aa5b7fad4cd2fc289653c4ab14f18052367b3 - - -""" -import jsonschema - -from acme import challenges -from acme import errors -from acme import jose -from acme import other -from acme import util - - -class Message(jose.TypedJSONObjectWithFields): - # _fields_to_partial_json | pylint: disable=abstract-method - # pylint: disable=too-few-public-methods - """ACME message.""" - TYPES = {} - type_field_name = "type" - - schema = NotImplemented - """JSON schema the object is tested against in :meth:`from_json`. - - Subclasses must overrride it with a value that is acceptable by - :func:`jsonschema.validate`, most probably using - :func:`acme.util.load_schema`. - - """ - - @classmethod - def from_json(cls, jobj): - """Deserialize from (possibly invalid) JSON object. - - Note that the input ``jobj`` has not been sanitized in any way. - - :param jobj: JSON object. - - :raises acme.errors.SchemaValidationError: if the input - JSON object could not be validated against JSON schema specified - in :attr:`schema`. - :raises acme.jose.errors.DeserializationError: for any - other generic error in decoding. - - :returns: instance of the class - - """ - msg_cls = cls.get_type_cls(jobj) - - # TODO: is that schema testing still relevant? - try: - jsonschema.validate(jobj, msg_cls.schema) - except jsonschema.ValidationError as error: - raise errors.SchemaValidationError(error) - - return super(Message, cls).from_json(jobj) - - -@Message.register # pylint: disable=too-few-public-methods -class Challenge(Message): - """ACME "challenge" message. - - :ivar str nonce: Random data, **not** base64-encoded. - :ivar list challenges: List of - :class:`~acme.challenges.Challenge` objects. - - .. todo:: - 1. can challenges contain two challenges of the same type? - 2. can challenges contain duplicates? - 3. check "combinations" indices are in valid range - 4. turn "combinations" elements into sets? - 5. turn "combinations" into set? - - """ - typ = "challenge" - schema = util.load_schema(typ) - - session_id = jose.Field("sessionID") - nonce = jose.Field("nonce", encoder=jose.b64encode, - decoder=jose.decode_b64jose) - challenges = jose.Field("challenges") - combinations = jose.Field("combinations", omitempty=True, default=()) - - @challenges.decoder - def challenges(value): # pylint: disable=missing-docstring,no-self-argument - return tuple(challenges.Challenge.from_json(chall) for chall in value) - - @property - def resolved_combinations(self): - """Combinations with challenges instead of indices.""" - return tuple(tuple(self.challenges[idx] for idx in combo) - for combo in self.combinations) - - -@Message.register # pylint: disable=too-few-public-methods -class ChallengeRequest(Message): - """ACME "challengeRequest" message.""" - typ = "challengeRequest" - schema = util.load_schema(typ) - identifier = jose.Field("identifier") - - -@Message.register # pylint: disable=too-few-public-methods -class Authorization(Message): - """ACME "authorization" message. - - :ivar jwk: :class:`acme.jose.JWK` - - """ - typ = "authorization" - schema = util.load_schema(typ) - - recovery_token = jose.Field("recoveryToken", omitempty=True) - identifier = jose.Field("identifier", omitempty=True) - jwk = jose.Field("jwk", decoder=jose.JWK.from_json, omitempty=True) - - -@Message.register -class AuthorizationRequest(Message): - """ACME "authorizationRequest" message. - - :ivar str nonce: Random data from the corresponding - :attr:`Challenge.nonce`, **not** base64-encoded. - :ivar list responses: List of completed challenges ( - :class:`acme.challenges.ChallengeResponse`). - :ivar signature: Signature (:class:`acme.other.Signature`). - - """ - typ = "authorizationRequest" - schema = util.load_schema(typ) - - session_id = jose.Field("sessionID") - nonce = jose.Field("nonce", encoder=jose.b64encode, - decoder=jose.decode_b64jose) - responses = jose.Field("responses") - signature = jose.Field("signature", decoder=other.Signature.from_json) - contact = jose.Field("contact", omitempty=True, default=()) - - @responses.decoder - def responses(value): # pylint: disable=missing-docstring,no-self-argument - return tuple(challenges.ChallengeResponse.from_json(chall) - for chall in value) - - @classmethod - def create(cls, name, key, sig_nonce=None, **kwargs): - """Create signed "authorizationRequest". - - :param str name: Hostname - - :param key: Key used for signing. - :type key: :class:`Crypto.PublicKey.RSA` - - :param str sig_nonce: Nonce used for signature. Useful for testing. - :kwargs: Any other arguments accepted by the class constructor. - - :returns: Signed "authorizationRequest" ACME message. - :rtype: :class:`AuthorizationRequest` - - """ - # pylint: disable=too-many-arguments - signature = other.Signature.from_msg( - name + kwargs["nonce"], key, sig_nonce) - return cls( - signature=signature, contact=kwargs.pop("contact", ()), **kwargs) - - def verify(self, name): - """Verify signature. - - .. warning:: Caller must check that the public key encoded in the - :attr:`signature`'s :class:`acme.jose.JWK` object - is the correct key for a given context. - - :param str name: Hostname - - :returns: True iff ``signature`` can be verified, False otherwise. - :rtype: bool - - """ - # self.signature is not Field | pylint: disable=no-member - return self.signature.verify(name + self.nonce) - - -@Message.register # pylint: disable=too-few-public-methods -class Certificate(Message): - """ACME "certificate" message. - - :ivar certificate: The certificate (:class:`M2Crypto.X509.X509` - wrapped in :class:`acme.util.ComparableX509`). - - :ivar list chain: Chain of certificates (:class:`M2Crypto.X509.X509` - wrapped in :class:`acme.util.ComparableX509` ). - - """ - typ = "certificate" - schema = util.load_schema(typ) - - certificate = jose.Field("certificate", encoder=jose.encode_cert, - decoder=jose.decode_cert) - chain = jose.Field("chain", omitempty=True, default=()) - refresh = jose.Field("refresh", omitempty=True) - - @chain.decoder - def chain(value): # pylint: disable=missing-docstring,no-self-argument - return tuple(jose.decode_cert(cert) for cert in value) - - @chain.encoder - def chain(value): # pylint: disable=missing-docstring,no-self-argument - return tuple(jose.encode_cert(cert) for cert in value) - - -@Message.register -class CertificateRequest(Message): - """ACME "certificateRequest" message. - - :ivar csr: Certificate Signing Request (:class:`M2Crypto.X509.Request` - wrapped in :class:`acme.util.ComparableX509`. - :ivar signature: Signature (:class:`acme.other.Signature`). - - """ - typ = "certificateRequest" - schema = util.load_schema(typ) - - csr = jose.Field("csr", encoder=jose.encode_csr, - decoder=jose.decode_csr) - signature = jose.Field("signature", decoder=other.Signature.from_json) - - @classmethod - def create(cls, key, sig_nonce=None, **kwargs): - """Create signed "certificateRequest". - - :param key: Key used for signing. - :type key: :class:`Crypto.PublicKey.RSA` - - :param str sig_nonce: Nonce used for signature. Useful for testing. - :kwargs: Any other arguments accepted by the class constructor. - - :returns: Signed "certificateRequest" ACME message. - :rtype: :class:`CertificateRequest` - - """ - return cls(signature=other.Signature.from_msg( - kwargs["csr"].as_der(), key, sig_nonce), **kwargs) - - def verify(self): - """Verify signature. - - .. warning:: Caller must check that the public key encoded in the - :attr:`signature`'s :class:`acme.jose.JWK` object - is the correct key for a given context. - - :returns: True iff ``signature`` can be verified, False otherwise. - :rtype: bool - - """ - # self.signature is not Field | pylint: disable=no-member - return self.signature.verify(self.csr.as_der()) - - -@Message.register # pylint: disable=too-few-public-methods -class Defer(Message): - """ACME "defer" message.""" - typ = "defer" - schema = util.load_schema(typ) - - token = jose.Field("token") - interval = jose.Field("interval", omitempty=True) - message = jose.Field("message", omitempty=True) - - -@Message.register # pylint: disable=too-few-public-methods -class Error(Message): - """ACME "error" message.""" - typ = "error" - schema = util.load_schema(typ) - - error = jose.Field("error") - message = jose.Field("message", omitempty=True) - more_info = jose.Field("moreInfo", omitempty=True) - - MESSAGE_CODES = { - "malformed": "The request message was malformed", - "unauthorized": "The client lacks sufficient authorization", - "serverInternal": "The server experienced an internal error", - "notSupported": "The request type is not supported", - "unknown": "The server does not recognize an ID/token in the request", - "badCSR": "The CSR is unacceptable (e.g., due to a short key)", - } - - -@Message.register # pylint: disable=too-few-public-methods -class Revocation(Message): - """ACME "revocation" message.""" - typ = "revocation" - schema = util.load_schema(typ) - - -@Message.register -class RevocationRequest(Message): - """ACME "revocationRequest" message. - - :ivar certificate: Certificate (:class:`M2Crypto.X509.X509` - wrapped in :class:`acme.util.ComparableX509`). - :ivar signature: Signature (:class:`acme.other.Signature`). - - """ - typ = "revocationRequest" - schema = util.load_schema(typ) - - certificate = jose.Field("certificate", decoder=jose.decode_cert, - encoder=jose.encode_cert) - signature = jose.Field("signature", decoder=other.Signature.from_json) - - @classmethod - def create(cls, key, sig_nonce=None, **kwargs): - """Create signed "revocationRequest". - - :param key: Key used for signing. - :type key: :class:`Crypto.PublicKey.RSA` - - :param str sig_nonce: Nonce used for signature. Useful for testing. - :kwargs: Any other arguments accepted by the class constructor. - - :returns: Signed "revocationRequest" ACME message. - :rtype: :class:`RevocationRequest` - - """ - return cls(signature=other.Signature.from_msg( - kwargs["certificate"].as_der(), key, sig_nonce), **kwargs) - - def verify(self): - """Verify signature. - - .. warning:: Caller must check that the public key encoded in the - :attr:`signature`'s :class:`acme.jose.JWK` object - is the correct key for a given context. - - :returns: True iff ``signature`` can be verified, False otherwise. - :rtype: bool - - """ - # self.signature is not Field | pylint: disable=no-member - return self.signature.verify(self.certificate.as_der()) - - -@Message.register # pylint: disable=too-few-public-methods -class StatusRequest(Message): - """ACME "statusRequest" message.""" - typ = "statusRequest" - schema = util.load_schema(typ) - token = jose.Field("token") diff --git a/acme/messages_test.py b/acme/messages_test.py deleted file mode 100644 index baff2a21a..000000000 --- a/acme/messages_test.py +++ /dev/null @@ -1,480 +0,0 @@ -"""Tests for acme.messages.""" -import os -import pkg_resources -import unittest - -import Crypto.PublicKey.RSA -import M2Crypto - -from acme import challenges -from acme import errors -from acme import jose -from acme import other - - -KEY = jose.HashableRSAKey(Crypto.PublicKey.RSA.importKey( - pkg_resources.resource_string( - 'acme.jose', os.path.join('testdata', 'rsa512_key.pem')))) -CERT = jose.ComparableX509(M2Crypto.X509.load_cert( - pkg_resources.resource_filename( - 'letsencrypt.tests', os.path.join('testdata', 'cert.pem')))) -CSR = jose.ComparableX509(M2Crypto.X509.load_request( - pkg_resources.resource_filename( - 'letsencrypt.tests', os.path.join('testdata', 'csr.pem')))) -CSR2 = jose.ComparableX509(M2Crypto.X509.load_request( - pkg_resources.resource_filename( - 'acme.jose', os.path.join('testdata', 'csr2.pem')))) - - -class MessageTest(unittest.TestCase): - """Tests for acme.messages.Message.""" - - def setUp(self): - # pylint: disable=missing-docstring,too-few-public-methods - from acme.messages import Message - - class MockParentMessage(Message): - # pylint: disable=abstract-method - TYPES = {} - - @MockParentMessage.register - class MockMessage(MockParentMessage): - typ = 'test' - schema = { - 'type': 'object', - 'properties': { - 'price': {'type': 'number'}, - 'name': {'type': 'string'}, - }, - } - price = jose.Field('price') - name = jose.Field('name') - - self.parent_cls = MockParentMessage - self.msg = MockMessage(price=123, name='foo') - - def test_from_json_validates(self): - self.assertRaises(errors.SchemaValidationError, - self.parent_cls.from_json, - {'type': 'test', 'price': 'asd'}) - - -class ChallengeTest(unittest.TestCase): - - def setUp(self): - challs = ( - challenges.SimpleHTTP(token='IlirfxKKXAsHtmzK29Pj8A'), - challenges.DNS(token='DGyRejmCefe7v4NfDGDKfA'), - challenges.RecoveryToken(), - ) - combinations = ((0, 2), (1, 2)) - - from acme.messages import Challenge - self.msg = Challenge( - session_id='aefoGaavieG9Wihuk2aufai3aeZ5EeW4', - nonce='\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9', - challenges=challs, combinations=combinations) - - self.jmsg_to = { - 'type': 'challenge', - 'sessionID': 'aefoGaavieG9Wihuk2aufai3aeZ5EeW4', - 'nonce': '7Nbyb1lI6xPVI3Hg3aKSqQ', - 'challenges': challs, - 'combinations': combinations, - } - - self.jmsg_from = { - 'type': 'challenge', - 'sessionID': 'aefoGaavieG9Wihuk2aufai3aeZ5EeW4', - 'nonce': '7Nbyb1lI6xPVI3Hg3aKSqQ', - 'challenges': [chall.to_json() for chall in challs], - 'combinations': [[0, 2], [1, 2]], # TODO array tuples - } - - def test_resolved_combinations(self): - self.assertEqual(self.msg.resolved_combinations, ( - ( - challenges.SimpleHTTP(token='IlirfxKKXAsHtmzK29Pj8A'), - challenges.RecoveryToken() - ), - ( - challenges.DNS(token='DGyRejmCefe7v4NfDGDKfA'), - challenges.RecoveryToken(), - ) - )) - - def test_to_partial_json(self): - self.assertEqual(self.msg.to_partial_json(), self.jmsg_to) - - def test_from_json(self): - from acme.messages import Challenge - self.assertEqual(Challenge.from_json(self.jmsg_from), self.msg) - - def test_json_without_optionals(self): - del self.jmsg_from['combinations'] - del self.jmsg_to['combinations'] - - from acme.messages import Challenge - msg = Challenge.from_json(self.jmsg_from) - - self.assertEqual(msg.combinations, ()) - self.assertEqual(msg.to_partial_json(), self.jmsg_to) - - -class ChallengeRequestTest(unittest.TestCase): - - def setUp(self): - from acme.messages import ChallengeRequest - self.msg = ChallengeRequest(identifier='example.com') - - self.jmsg = { - 'type': 'challengeRequest', - 'identifier': 'example.com', - } - - def test_to_partial_json(self): - self.assertEqual(self.msg.to_partial_json(), self.jmsg) - - def test_from_json(self): - from acme.messages import ChallengeRequest - self.assertEqual(ChallengeRequest.from_json(self.jmsg), self.msg) - - -class AuthorizationTest(unittest.TestCase): - - def setUp(self): - jwk = jose.JWKRSA(key=KEY.publickey()) - - from 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_partial_json(self): - self.assertEqual(self.msg.to_partial_json(), self.jmsg) - - def test_from_json(self): - self.jmsg['jwk'] = self.jmsg['jwk'].to_partial_json() - - from acme.messages import Authorization - self.assertEqual(Authorization.from_json(self.jmsg), self.msg) - - def test_json_without_optionals(self): - del self.jmsg['recoveryToken'] - del self.jmsg['identifier'] - del self.jmsg['jwk'] - - from acme.messages import Authorization - msg = Authorization.from_json(self.jmsg) - - self.assertTrue(msg.recovery_token is None) - self.assertTrue(msg.identifier is None) - self.assertTrue(msg.jwk is None) - self.assertEqual(self.jmsg, msg.to_partial_json()) - - -class AuthorizationRequestTest(unittest.TestCase): - - def setUp(self): - self.responses = ( - challenges.SimpleHTTPResponse(path='Hf5GrX4Q7EBax9hc2jJnfw'), - None, # null - challenges.RecoveryTokenResponse(token='23029d88d9e123e'), - ) - self.contact = ("mailto:cert-admin@example.com", "tel:+12025551212") - signature = other.Signature( - alg=jose.RS256, jwk=jose.JWKRSA(key=KEY.publickey()), - sig='-v\xd8\xc2\xa3\xba0\xd6\x92\x16\xb5.\xbe\xa1[\x04\xbe' - '\x1b\xa1X\xd2)\x18\x94\x8f\xd7\xd0\xc0\xbbcI`W\xdf v' - '\xe4\xed\xe8\x03J\xe8\xc8