mirror of
https://github.com/certbot/certbot.git
synced 2026-07-28 00:35:50 +02:00
Use new framework for ACME messages
This commit is contained in:
Jakub Warmuz
+89
-185
@@ -1,8 +1,4 @@
|
||||
"""ACME protocol messages."""
|
||||
import json
|
||||
import pkg_resources
|
||||
|
||||
import jsonschema
|
||||
import M2Crypto
|
||||
import zope.interface
|
||||
|
||||
@@ -13,25 +9,7 @@ from letsencrypt.acme import other
|
||||
from letsencrypt.acme import util
|
||||
|
||||
|
||||
SCHEMATA = dict([
|
||||
(schema, json.load(open(pkg_resources.resource_filename(
|
||||
__name__, "schemata/%s.json" % schema)))) for schema in [
|
||||
"authorization",
|
||||
"authorizationRequest",
|
||||
"certificate",
|
||||
"certificateRequest",
|
||||
"challenge",
|
||||
"challengeRequest",
|
||||
"defer",
|
||||
"error",
|
||||
"revocation",
|
||||
"revocationRequest",
|
||||
"statusRequest",
|
||||
]
|
||||
])
|
||||
|
||||
|
||||
class Message(object):
|
||||
class Message(util.JSONDeSerializable, util.ImmutableMap):
|
||||
"""ACME message.
|
||||
|
||||
Messages are considered immutable.
|
||||
@@ -51,17 +29,6 @@ class Message(object):
|
||||
cls.TYPES[msg_cls.acme_type] = msg_cls
|
||||
return msg_cls
|
||||
|
||||
@classmethod
|
||||
def schema(cls, schemata=None):
|
||||
"""Get JSON schema for this ACME message.
|
||||
|
||||
:param dict schemata: Mapping from type name to JSON Schema
|
||||
definition. Useful for testing.
|
||||
|
||||
"""
|
||||
schemata = SCHEMATA if schemata is None else schemata
|
||||
return schemata[cls.acme_type]
|
||||
|
||||
def to_json(self):
|
||||
"""Get JSON serializable object.
|
||||
|
||||
@@ -85,34 +52,23 @@ class Message(object):
|
||||
:rtype: dict
|
||||
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def json_dumps(self):
|
||||
"""Dump to JSON using proper serializer.
|
||||
|
||||
:returns: JSON serialized string.
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
return json.dumps(self, default=util.dump_ijsonserializable)
|
||||
raise NotImplementedError()
|
||||
|
||||
@classmethod
|
||||
def validate(cls, jobj, schemata=None):
|
||||
"""Is JSON object a valid ACME message?
|
||||
def from_json(cls, jobj, validate=True):
|
||||
"""Deserialize validated ACME message from JSON string.
|
||||
|
||||
:param str jobj: JSON object
|
||||
|
||||
:param dict schemata: Mapping from type name to JSON Schema
|
||||
definition. Useful for testing.
|
||||
|
||||
:returns: ACME message class, subclassing :class:`Message`.
|
||||
:param str jobj: JSON object.
|
||||
:param bool validate: Validate against schema before deserializing.
|
||||
Useful if :class:`JWK` is part of already validated json object.
|
||||
|
||||
:raises letsencrypt.acme.errors.ValidationError: if validation
|
||||
was unsuccessful
|
||||
|
||||
"""
|
||||
schemata = SCHEMATA if schemata is None else schemata
|
||||
:returns: Valid ACME message.
|
||||
:rtype: subclass of :class:`Message`
|
||||
|
||||
"""
|
||||
if not isinstance(jobj, dict):
|
||||
raise errors.ValidationError(
|
||||
"{0} is not a dictionary object".format(jobj))
|
||||
@@ -122,64 +78,22 @@ class Message(object):
|
||||
raise errors.ValidationError("missing type field")
|
||||
|
||||
try:
|
||||
schema = schemata[msg_type] # pylint: disable=redefined-outer-name
|
||||
msg_cls = cls.TYPES[msg_type]
|
||||
except KeyError:
|
||||
raise errors.UnrecognnizedMessageTypeError(msg_type)
|
||||
|
||||
try:
|
||||
jsonschema.validate(jobj, schema)
|
||||
except jsonschema.ValidationError as error:
|
||||
raise errors.SchemaValidationError(error)
|
||||
|
||||
return msg_cls
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, json_string, schemata=None):
|
||||
"""Deserialize validated ACME message from JSON string.
|
||||
|
||||
:param str json_string: JSON serialize string.
|
||||
:param dict schemata: Mapping from type name to JSON Schema
|
||||
definition. Useful for testing.
|
||||
|
||||
:raises letsencrypt.acme.errors.ValidationError: if validation
|
||||
was unsuccessful
|
||||
|
||||
:returns: Valid ACME message.
|
||||
:rtype: subclass of :class:`Message`
|
||||
|
||||
"""
|
||||
jobj = json.loads(json_string)
|
||||
msg_cls = cls.validate(jobj, schemata)
|
||||
if validate:
|
||||
msg_cls.validate_json(jobj)
|
||||
# pylint: disable=protected-access
|
||||
return msg_cls._valid_from_json(jobj)
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, jobj):
|
||||
"""Deserialize from valid ACME message JSON object.
|
||||
|
||||
Subclasses must override.
|
||||
|
||||
:param jobj: Schema validated ACME message JSON object.
|
||||
:type jobj: dict
|
||||
|
||||
:returns: Valid ACME message.
|
||||
:rtype: subclass of :class:`Message`
|
||||
|
||||
"""
|
||||
raise NotImplementedError
|
||||
return msg_cls._from_valid_json(jobj)
|
||||
|
||||
|
||||
@Message.register # pylint: disable=too-few-public-methods
|
||||
class Challenge(Message):
|
||||
"""ACME "challenge" message."""
|
||||
acme_type = "challenge"
|
||||
|
||||
def __init__(self, session_id, nonce, challenges, combinations=None):
|
||||
self.session_id = session_id
|
||||
self.nonce = nonce
|
||||
self.challenges = challenges
|
||||
self.combinations = [] if combinations is None else combinations
|
||||
schema = util.load_schema(acme_type)
|
||||
__slots__ = ("session_id", "nonce", "challenges", "combinations")
|
||||
|
||||
def _fields_to_json(self):
|
||||
fields = {
|
||||
@@ -192,9 +106,11 @@ class Challenge(Message):
|
||||
return fields
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, jobj):
|
||||
return cls(jobj["sessionID"], jose.b64decode(jobj["nonce"]),
|
||||
jobj["challenges"], jobj.get("combinations"))
|
||||
def _from_valid_json(cls, jobj):
|
||||
return cls(session_id=jobj["sessionID"],
|
||||
nonce=jose.b64decode(jobj["nonce"]),
|
||||
challenges=jobj["challenges"],
|
||||
combinations=jobj.get("combinations", []))
|
||||
|
||||
|
||||
@Message.register # pylint: disable=too-few-public-methods
|
||||
@@ -205,9 +121,8 @@ class ChallengeRequest(Message):
|
||||
|
||||
"""
|
||||
acme_type = "challengeRequest"
|
||||
|
||||
def __init__(self, identifier):
|
||||
self.identifier = identifier
|
||||
schema = util.load_schema(acme_type)
|
||||
__slots__ = ("identifier",)
|
||||
|
||||
def _fields_to_json(self):
|
||||
return {
|
||||
@@ -215,19 +130,16 @@ class ChallengeRequest(Message):
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, json_string):
|
||||
return cls(json_string["identifier"])
|
||||
def _from_valid_json(cls, jobj):
|
||||
return cls(identifier=jobj["identifier"])
|
||||
|
||||
|
||||
@Message.register # pylint: disable=too-few-public-methods
|
||||
class Authorization(Message):
|
||||
"""ACME "authorization" message."""
|
||||
acme_type = "authorization"
|
||||
|
||||
def __init__(self, recovery_token=None, identifier=None, jwk=None):
|
||||
self.recovery_token = recovery_token
|
||||
self.identifier = identifier
|
||||
self.jwk = jwk
|
||||
schema = util.load_schema(acme_type)
|
||||
__slots__ = ("recovery_token", "identifier", "jwk")
|
||||
|
||||
def _fields_to_json(self):
|
||||
fields = {}
|
||||
@@ -240,11 +152,12 @@ class Authorization(Message):
|
||||
return fields
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, jobj):
|
||||
def _from_valid_json(cls, jobj):
|
||||
jwk = jobj.get("jwk")
|
||||
if jwk is not None:
|
||||
jwk = jose.JWK.from_json(jwk)
|
||||
return cls(jobj.get("recoveryToken"), jobj.get("identifier"), jwk)
|
||||
jwk = jose.JWK.from_json(jwk, validate=False)
|
||||
return cls(recovery_token=jobj.get("recoveryToken"),
|
||||
identifier=jobj.get("identifier"), jwk=jwk)
|
||||
|
||||
|
||||
@Message.register
|
||||
@@ -259,19 +172,15 @@ class AuthorizationRequest(Message):
|
||||
|
||||
"""
|
||||
acme_type = "authorizationRequest"
|
||||
|
||||
def __init__(self, session_id, nonce, responses, signature, contact=None):
|
||||
self.session_id = session_id
|
||||
self.nonce = nonce
|
||||
self.responses = responses
|
||||
self.signature = signature
|
||||
self.contact = [] if contact is None else contact
|
||||
schema = util.load_schema(acme_type)
|
||||
__slots__ = ("session_id", "nonce", "responses", "signature", "contact")
|
||||
|
||||
@classmethod
|
||||
def create(cls, session_id, nonce, responses, name, key,
|
||||
sig_nonce=None, contact=None):
|
||||
def create(cls, name, key, sig_nonce=None, **kwargs):
|
||||
"""Create signed "authorizationRequest".
|
||||
|
||||
:param str name: TODO
|
||||
|
||||
:param key: Key used for signing.
|
||||
:type key: :class:`Crypto.PublicKey.RSA`
|
||||
|
||||
@@ -282,8 +191,10 @@ class AuthorizationRequest(Message):
|
||||
|
||||
"""
|
||||
# pylint: disable=too-many-arguments
|
||||
signature = other.Signature.from_msg(name + nonce, key, sig_nonce)
|
||||
return cls(session_id, nonce, responses, signature, contact)
|
||||
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.
|
||||
@@ -310,11 +221,13 @@ class AuthorizationRequest(Message):
|
||||
return fields
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, jobj):
|
||||
return cls(jobj["sessionID"], jose.b64decode(jobj["nonce"]),
|
||||
jobj["responses"],
|
||||
other.Signature.from_json(jobj["signature"]),
|
||||
jobj.get("contact"))
|
||||
def _from_valid_json(cls, jobj):
|
||||
return cls(session_id=jobj["sessionID"],
|
||||
nonce=jose.b64decode(jobj["nonce"]),
|
||||
responses=jobj["responses"],
|
||||
signature=other.Signature.from_json(
|
||||
jobj["signature"], validate=False),
|
||||
contact=jobj.get("contact", []))
|
||||
|
||||
|
||||
@Message.register # pylint: disable=too-few-public-methods
|
||||
@@ -326,11 +239,8 @@ class Certificate(Message):
|
||||
|
||||
"""
|
||||
acme_type = "certificate"
|
||||
|
||||
def __init__(self, certificate, chain=None, refresh=None):
|
||||
self.certificate = certificate
|
||||
self.chain = [] if chain is None else chain
|
||||
self.refresh = refresh
|
||||
schema = util.load_schema(acme_type)
|
||||
__slots__ = ("certificate", "chain", "refresh")
|
||||
|
||||
def _fields_to_json(self):
|
||||
fields = {
|
||||
@@ -342,10 +252,11 @@ class Certificate(Message):
|
||||
return fields
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, jobj):
|
||||
def _from_valid_json(cls, jobj):
|
||||
certificate = M2Crypto.X509.load_cert_der_string(
|
||||
jose.b64decode(jobj["certificate"]))
|
||||
return cls(certificate, jobj.get("chain"), jobj.get("refresh"))
|
||||
return cls(certificate=certificate, chain=jobj.get("chain", []),
|
||||
refresh=jobj.get("refresh"))
|
||||
|
||||
|
||||
@Message.register
|
||||
@@ -358,25 +269,24 @@ class CertificateRequest(Message):
|
||||
|
||||
"""
|
||||
acme_type = "certificateRequest"
|
||||
|
||||
def __init__(self, csr, signature):
|
||||
self.csr = csr
|
||||
self.signature = signature
|
||||
schema = util.load_schema(acme_type)
|
||||
__slots__ = ("csr", "signature")
|
||||
|
||||
@classmethod
|
||||
def create(cls, csr, key, nonce=None):
|
||||
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 nonce: Nonce used for signature. Useful for testing.
|
||||
:param str sig_nonce: Nonce used for signature. Useful for testing.
|
||||
|
||||
:returns: Signed "certificateRequest" ACME message.
|
||||
:rtype: :class:`CertificateRequest`
|
||||
|
||||
"""
|
||||
return cls(csr, other.Signature.from_msg(csr, key, nonce))
|
||||
return cls(signature=other.Signature.from_msg(
|
||||
kwargs["csr"], key, sig_nonce), **kwargs)
|
||||
|
||||
def verify(self):
|
||||
"""Verify signature.
|
||||
@@ -396,20 +306,18 @@ class CertificateRequest(Message):
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, jobj):
|
||||
return cls(jose.b64decode(jobj["csr"]),
|
||||
other.Signature.from_json(jobj["signature"]))
|
||||
def _from_valid_json(cls, jobj):
|
||||
return cls(csr=jose.b64decode(jobj["csr"]),
|
||||
signature=other.Signature.from_json(
|
||||
jobj["signature"], validate=False))
|
||||
|
||||
|
||||
@Message.register # pylint: disable=too-few-public-methods
|
||||
class Defer(Message):
|
||||
"""ACME "defer" message."""
|
||||
acme_type = "defer"
|
||||
|
||||
def __init__(self, token, interval=None, message=None):
|
||||
self.token = token
|
||||
self.interval = interval # TODO: int
|
||||
self.message = message
|
||||
schema = util.load_schema(acme_type)
|
||||
__slots__ = ("token", "interval", "message")
|
||||
|
||||
def _fields_to_json(self):
|
||||
fields = {"token": self.token}
|
||||
@@ -420,14 +328,17 @@ class Defer(Message):
|
||||
return fields
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, jobj):
|
||||
return cls(jobj["token"], jobj.get("interval"), jobj.get("message"))
|
||||
def _from_valid_json(cls, jobj):
|
||||
return cls(token=jobj["token"], interval=jobj.get("interval"),
|
||||
message=jobj.get("message"))
|
||||
|
||||
|
||||
@Message.register # pylint: disable=too-few-public-methods
|
||||
class Error(Message):
|
||||
"""ACME "error" message."""
|
||||
acme_type = "error"
|
||||
schema = util.load_schema(acme_type)
|
||||
__slots__ = ("error", "message", "more_info")
|
||||
|
||||
CODES = {
|
||||
"malformed": "The request message was malformed",
|
||||
@@ -438,12 +349,6 @@ class Error(Message):
|
||||
"badCSR": "The CSR is unacceptable (e.g., due to a short key)",
|
||||
}
|
||||
|
||||
def __init__(self, error, message=None, more_info=None):
|
||||
assert error in self.CODES # TODO: already checked by schema validation
|
||||
self.error = error
|
||||
self.message = message
|
||||
self.more_info = more_info
|
||||
|
||||
def _fields_to_json(self):
|
||||
fields = {"error": self.error}
|
||||
if self.message is not None:
|
||||
@@ -453,20 +358,23 @@ class Error(Message):
|
||||
return fields
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, jobj):
|
||||
return cls(jobj["error"], jobj.get("message"), jobj.get("more_info"))
|
||||
def _from_valid_json(cls, jobj):
|
||||
return cls(error=jobj["error"], message=jobj.get("message"),
|
||||
more_info=jobj.get("more_info"))
|
||||
|
||||
|
||||
@Message.register # pylint: disable=too-few-public-methods
|
||||
class Revocation(Message):
|
||||
"""ACME "revocation" message."""
|
||||
acme_type = "revocation"
|
||||
schema = util.load_schema(acme_type)
|
||||
__slots__ = ()
|
||||
|
||||
def _fields_to_json(self):
|
||||
return {}
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, jobj):
|
||||
def _from_valid_json(cls, jobj):
|
||||
return cls()
|
||||
|
||||
|
||||
@@ -481,26 +389,24 @@ class RevocationRequest(Message):
|
||||
|
||||
"""
|
||||
acme_type = "revocationRequest"
|
||||
|
||||
def __init__(self, certificate, signature):
|
||||
self.certificate = certificate
|
||||
self.signature = signature
|
||||
schema = util.load_schema(acme_type)
|
||||
__slots__ = ("certificate", "signature")
|
||||
|
||||
@classmethod
|
||||
def create(cls, certificate, key, nonce=None):
|
||||
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 nonce: Nonce used for signature. Useful for testing.
|
||||
:param str sig_nonce: Nonce used for signature. Useful for testing.
|
||||
|
||||
:returns: Signed "revocationRequest" ACME message.
|
||||
:rtype: :class:`RevocationRequest`
|
||||
|
||||
"""
|
||||
return cls(certificate,
|
||||
other.Signature.from_msg(certificate, key, nonce))
|
||||
return cls(signature=other.Signature.from_msg(
|
||||
kwargs["certificate"], key, sig_nonce), **kwargs)
|
||||
|
||||
def verify(self):
|
||||
"""Verify signature.
|
||||
@@ -520,9 +426,10 @@ class RevocationRequest(Message):
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, json_string):
|
||||
return cls(jose.b64decode(json_string["certificate"]),
|
||||
other.Signature.from_json(json_string["signature"]))
|
||||
def _from_valid_json(cls, jobj):
|
||||
return cls(certificate=jose.b64decode(jobj["certificate"]),
|
||||
signature=other.Signature.from_json(
|
||||
jobj["signature"], validate=False))
|
||||
|
||||
|
||||
@Message.register # pylint: disable=too-few-public-methods
|
||||
@@ -533,15 +440,12 @@ class StatusRequest(Message):
|
||||
|
||||
"""
|
||||
acme_type = "statusRequest"
|
||||
|
||||
def __init__(self, token):
|
||||
self.token = token
|
||||
schema = util.load_schema(acme_type)
|
||||
__slots__ = ("token",)
|
||||
|
||||
def _fields_to_json(self):
|
||||
return {
|
||||
"token": self.token,
|
||||
}
|
||||
return {"token": self.token}
|
||||
|
||||
@classmethod
|
||||
def _valid_from_json(cls, json_string):
|
||||
return cls(json_string["token"])
|
||||
def _from_valid_json(cls, jobj):
|
||||
return cls(token=jobj["token"])
|
||||
|
||||
@@ -6,6 +6,9 @@ import Crypto.PublicKey.RSA
|
||||
import mock
|
||||
|
||||
from letsencrypt.acme import errors
|
||||
from letsencrypt.acme import jose
|
||||
from letsencrypt.acme import other
|
||||
|
||||
|
||||
KEY = Crypto.PublicKey.RSA.importKey(pkg_resources.resource_string(
|
||||
'letsencrypt.client.tests', 'testdata/rsa256_key.pem'))
|
||||
@@ -15,158 +18,228 @@ class MessageTest(unittest.TestCase):
|
||||
"""Tests for letsencrypt.acme.messages.Message."""
|
||||
|
||||
def setUp(self):
|
||||
self.schemata = {
|
||||
'foo': {
|
||||
# pylint: disable=missing-docstring,too-few-public-methods
|
||||
from letsencrypt.acme.messages import Message
|
||||
class TestMessage(Message):
|
||||
acme_type = 'test'
|
||||
schema = {
|
||||
'type': 'object',
|
||||
'properties': {
|
||||
'price': {'type': 'number'},
|
||||
'name': {'type': 'string'},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _from_valid_json(cls, jobj):
|
||||
return jobj
|
||||
|
||||
def _validate(self, jobj):
|
||||
def _fields_to_json(self):
|
||||
pass
|
||||
|
||||
self.msg_cls = TestMessage
|
||||
|
||||
@classmethod
|
||||
def _from_json(cls, jobj, validate=True):
|
||||
from letsencrypt.acme.messages import Message
|
||||
return Message.validate(jobj, self.schemata)
|
||||
return Message.from_json(jobj, validate)
|
||||
|
||||
def test_validate_non_dictionary_fails(self):
|
||||
self.assertRaises(errors.ValidationError, self._validate, [])
|
||||
def test_from_json_non_dict_fails(self):
|
||||
self.assertRaises(errors.ValidationError, self._from_json, [])
|
||||
|
||||
def test_validate_dict_without_type_fails(self):
|
||||
self.assertRaises(errors.ValidationError, self._validate, {})
|
||||
def test_from_json_dict_no_type_fails(self):
|
||||
self.assertRaises(errors.ValidationError, self._from_json, {})
|
||||
|
||||
def test_validate_unknown_type_fails(self):
|
||||
def test_from_json_unknown_type_fails(self):
|
||||
self.assertRaises(errors.UnrecognnizedMessageTypeError,
|
||||
self._validate, {'type': 'bar'})
|
||||
|
||||
def test_validate_unregistered_type_fails(self):
|
||||
self.assertRaises(errors.UnrecognnizedMessageTypeError,
|
||||
self._validate, {'type': 'foo'})
|
||||
self._from_json, {'type': 'bar'})
|
||||
|
||||
@mock.patch('letsencrypt.acme.messages.Message.TYPES')
|
||||
def test_validate_invalid_fails(self, types):
|
||||
types.__getitem__.side_effect = lambda x: {'foo': 'bar'}[x]
|
||||
def test_from_json_validate_errors(self, types):
|
||||
types.__getitem__.side_effect = lambda x: {'foo': self.msg_cls}[x]
|
||||
self.assertRaises(errors.SchemaValidationError,
|
||||
self._validate, {'type': 'foo', 'price': 'asd'})
|
||||
self._from_json, {'type': 'foo', 'price': 'asd'})
|
||||
|
||||
@mock.patch('letsencrypt.acme.messages.Message.TYPES')
|
||||
def test_validate_valid_returns_cls(self, types):
|
||||
types.__getitem__.side_effect = lambda x: {'foo': 'bar'}[x]
|
||||
self.assertEqual(self._validate({'type': 'foo'}), 'bar')
|
||||
def test_from_json_valid_returns_cls(self, types):
|
||||
types.__getitem__.side_effect = lambda x: {'foo': self.msg_cls}[x]
|
||||
self.assertEqual(self._from_json({'type': 'foo'}, validate=False),
|
||||
{'type': 'foo'})
|
||||
|
||||
|
||||
class ChallengeRequestTest(unittest.TestCase):
|
||||
# pylint: disable=too-few-public-methods
|
||||
|
||||
def test_it(self):
|
||||
def setUp(self):
|
||||
from letsencrypt.acme.messages import ChallengeRequest
|
||||
msg = ChallengeRequest('example.com')
|
||||
self.msg = ChallengeRequest(identifier='example.com')
|
||||
|
||||
jmsg = msg._fields_to_json() # pylint: disable=protected-access
|
||||
self.assertEqual(jmsg, {
|
||||
self.jmsg = {
|
||||
'type': 'challengeRequest',
|
||||
'identifier': 'example.com',
|
||||
})
|
||||
}
|
||||
|
||||
def test_to_json(self):
|
||||
self.assertEqual(self.msg.to_json(), self.jmsg)
|
||||
|
||||
def test_from_json(self):
|
||||
from letsencrypt.acme.messages import ChallengeRequest
|
||||
self.assertEqual(ChallengeRequest.from_json(self.jmsg), self.msg)
|
||||
|
||||
|
||||
class AuthorizationRequestTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.nonce = '\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9'
|
||||
self.b64nonce = '7Nbyb1lI6xPVI3Hg3aKSqQ'
|
||||
self.csr = 'TODO: real DER CSR?'
|
||||
|
||||
def test_authorization_request(self):
|
||||
from letsencrypt.acme.messages import AuthorizationRequest
|
||||
responses = [
|
||||
{
|
||||
'type': 'simpleHttps',
|
||||
'path': 'Hf5GrX4Q7EBax9hc2jJnfw',
|
||||
},
|
||||
self.responses = [
|
||||
{'type': 'simpleHttps', 'path': 'Hf5GrX4Q7EBax9hc2jJnfw'},
|
||||
None, # null
|
||||
{
|
||||
'type': 'recoveryToken',
|
||||
'token': '23029d88d9e123e',
|
||||
}
|
||||
{'type': 'recoveryToken', 'token': '23029d88d9e123e'},
|
||||
]
|
||||
msg = AuthorizationRequest.create(
|
||||
'aefoGaavieG9Wihuk2aufai3aeZ5EeW4',
|
||||
'czpsrF0KMH6dgajig3TGHw',
|
||||
responses,
|
||||
'example.com',
|
||||
KEY,
|
||||
self.nonce,
|
||||
)
|
||||
msg.verify('example.com')
|
||||
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'
|
||||
'\x1b\xa1X\xd2)\x18\x94\x8f\xd7\xd0\xc0\xbbcI`W\xdf v'
|
||||
'\xe4\xed\xe8\x03J\xe8\xc8<?\xc8W\x94\x94cj(\xe7\xaa$'
|
||||
'\x92\xe9\x96\x11\xc2\xefx\x0bR',
|
||||
nonce='\xab?\x08o\xe6\x81$\x9f\xa1\xc9\x025\x1c\x1b\xa5+')
|
||||
|
||||
jmsg = msg._fields_to_json() # pylint: disable=protected-access
|
||||
jmsg.pop('signature')
|
||||
self.assertEqual(jmsg, {
|
||||
from letsencrypt.acme.messages import AuthorizationRequest
|
||||
self.msg = AuthorizationRequest(
|
||||
session_id='aefoGaavieG9Wihuk2aufai3aeZ5EeW4',
|
||||
nonce='\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9',
|
||||
responses=self.responses,
|
||||
signature=signature,
|
||||
contact=[],
|
||||
)
|
||||
|
||||
self.jmsg = {
|
||||
'type': 'authorizationRequest',
|
||||
'sessionID': 'aefoGaavieG9Wihuk2aufai3aeZ5EeW4',
|
||||
'nonce': 'Y3pwc3JGMEtNSDZkZ2FqaWczVEdIdw',
|
||||
'responses': responses,
|
||||
})
|
||||
'nonce': '7Nbyb1lI6xPVI3Hg3aKSqQ',
|
||||
'responses': self.responses,
|
||||
'signature': signature,
|
||||
}
|
||||
|
||||
def test_create(self):
|
||||
from letsencrypt.acme.messages import AuthorizationRequest
|
||||
self.assertEqual(self.msg, AuthorizationRequest.create(
|
||||
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+'))
|
||||
|
||||
def test_verify(self):
|
||||
self.assertTrue(self.msg.verify('example.com'))
|
||||
|
||||
def test_to_json(self):
|
||||
self.assertEqual(self.msg.to_json(), self.jmsg)
|
||||
|
||||
def test_from_json(self):
|
||||
from letsencrypt.acme.messages import AuthorizationRequest
|
||||
self.jmsg['signature'] = self.jmsg['signature'].to_json()
|
||||
self.jmsg['signature']['jwk'] = self.jmsg['signature']['jwk'].to_json()
|
||||
self.assertEqual(self.msg, AuthorizationRequest.from_json(self.jmsg))
|
||||
|
||||
|
||||
class CertificateRequestTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.nonce = '\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9'
|
||||
self.b64nonce = '7Nbyb1lI6xPVI3Hg3aKSqQ'
|
||||
self.csr = 'TODO: real DER CSR?'
|
||||
signature = other.Signature(
|
||||
alg='RS256', jwk=jose.JWK(key=KEY.publickey()),
|
||||
sig='\x1cD\x157\x83\x14\xd7 \xeb\x02\xb3\xf6O\xb5\x99C]\x97'
|
||||
'\x94p\xa7\xe48\x13>\x06\xf9yd\xf9\xfe\xf8\xd1>\x9aKH'
|
||||
'\xd7\xba\xb9a1\xf5!p\x1b\xd7}\xbaj\xa7\xe3\xd9\xd9\t%'
|
||||
'\xbb\xba\xc9\x00\xdaW\x16\xe9',
|
||||
nonce='\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9')
|
||||
|
||||
def test_it(self):
|
||||
from letsencrypt.acme.messages import CertificateRequest
|
||||
msg = CertificateRequest.create(self.csr, KEY, self.nonce)
|
||||
self.assertTrue(msg.verify())
|
||||
self.msg = CertificateRequest(csr=self.csr, signature=signature)
|
||||
|
||||
jmsg = msg._fields_to_json() # pylint: disable=protected-access
|
||||
jmsg.pop('signature')
|
||||
self.assertEqual(jmsg, {
|
||||
self.jmsg = {
|
||||
'type': 'certificateRequest',
|
||||
'csr': 'VE9ETzogcmVhbCBERVIgQ1NSPw',
|
||||
})
|
||||
'signature': signature,
|
||||
}
|
||||
|
||||
def test_create(self):
|
||||
from letsencrypt.acme.messages import CertificateRequest
|
||||
self.assertEqual(self.msg, CertificateRequest.create(
|
||||
csr=self.csr, key=KEY,
|
||||
sig_nonce='\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9'))
|
||||
|
||||
def test_verify(self):
|
||||
self.assertTrue(self.msg.verify())
|
||||
|
||||
def test_to_json(self):
|
||||
self.assertEqual(self.msg.to_json(), self.jmsg)
|
||||
|
||||
def test_from_json(self):
|
||||
from letsencrypt.acme.messages import CertificateRequest
|
||||
self.jmsg['signature'] = self.jmsg['signature'].to_json()
|
||||
self.jmsg['signature']['jwk'] = self.jmsg['signature']['jwk'].to_json()
|
||||
self.assertEqual(self.msg, CertificateRequest.from_json(self.jmsg))
|
||||
|
||||
|
||||
class RevocationRequestTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.sig_nonce = '\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9'
|
||||
|
||||
self.nonce = '\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9'
|
||||
self.b64nonce = '7Nbyb1lI6xPVI3Hg3aKSqQ'
|
||||
self.certificate = 'TODO: real DER cert?'
|
||||
|
||||
def test_it(self):
|
||||
from letsencrypt.acme.messages import RevocationRequest
|
||||
msg = RevocationRequest.create(self.certificate, KEY, self.nonce)
|
||||
self.assertTrue(msg.verify())
|
||||
signature = other.Signature(
|
||||
alg='RS256', jwk=jose.JWK(key=KEY.publickey()),
|
||||
sig='\x00\x15\xc0\xd4\x8b2M\xa9S\\\x8a#\xc6a\xa7!A\xb2d\x04'
|
||||
'\xa6\xbe\xa1/M\x0f|\x8c\x9eJ\x16\xcd\x85N\xcc\x0b\x12k('
|
||||
'\xa8U\xdfS\xa9y\xfd\xfa.\xb3\xeblms\x9f,\xdf\xbb>7\xd9'
|
||||
'\xe5u\x8f\xbe',
|
||||
nonce=self.sig_nonce)
|
||||
|
||||
jmsg = msg._fields_to_json() # pylint: disable=protected-access
|
||||
jmsg.pop('signature')
|
||||
self.assertEqual(jmsg, {
|
||||
from letsencrypt.acme.messages import RevocationRequest
|
||||
self.msg = RevocationRequest(
|
||||
certificate=self.certificate, signature=signature)
|
||||
|
||||
self.jmsg = {
|
||||
'type': 'revocationRequest',
|
||||
'certificate': 'VE9ETzogcmVhbCBERVIgY2VydD8',
|
||||
})
|
||||
'signature': signature,
|
||||
}
|
||||
|
||||
def test_create(self):
|
||||
from letsencrypt.acme.messages import RevocationRequest
|
||||
RevocationRequest.create(
|
||||
certificate=self.certificate, key=KEY, sig_nonce=self.sig_nonce)
|
||||
|
||||
def test_verify(self):
|
||||
self.assertTrue(self.msg.verify())
|
||||
|
||||
def test_to_json(self):
|
||||
self.assertEqual(self.msg.to_json(), self.jmsg)
|
||||
|
||||
def test_from_json(self):
|
||||
from letsencrypt.acme.messages import RevocationRequest
|
||||
self.jmsg['signature'] = self.jmsg['signature'].to_json()
|
||||
self.jmsg['signature']['jwk'] = self.jmsg['signature']['jwk'].to_json()
|
||||
self.assertEqual(self.msg, RevocationRequest.from_json(self.jmsg))
|
||||
|
||||
|
||||
class StatusRequestTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
from letsencrypt.acme.messages import StatusRequest
|
||||
self.token = u'O7-s9MNq1siZHlgrMzi9_A'
|
||||
self.msg = StatusRequest(self.token)
|
||||
self.msg = StatusRequest(token=u'O7-s9MNq1siZHlgrMzi9_A')
|
||||
self.jmsg = {
|
||||
'token': self.token,
|
||||
'type': 'statusRequest',
|
||||
'token': u'O7-s9MNq1siZHlgrMzi9_A',
|
||||
}
|
||||
|
||||
def test_attributes(self):
|
||||
self.assertEqual(self.msg.token, self.token)
|
||||
|
||||
def test_json(self):
|
||||
jmsg = self.msg._fields_to_json() # pylint: disable=protected-access
|
||||
self.assertEqual(jmsg, self.jmsg)
|
||||
def test_to_json(self):
|
||||
self.assertEqual(self.msg.to_json(), self.jmsg)
|
||||
|
||||
def test_from_json(self):
|
||||
from letsencrypt.acme.messages import StatusRequest
|
||||
# pylint: disable=protected-access
|
||||
msg = StatusRequest._valid_from_json(self.jmsg)
|
||||
self.assertEqual(msg.token, self.msg.token)
|
||||
self.assertEqual(StatusRequest.from_json(self.jmsg), self.msg)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
@@ -111,11 +111,12 @@ class AuthHandler(object): # pylint: disable=too-many-instance-attributes
|
||||
try:
|
||||
auth = self.network.send_and_receive_expected(
|
||||
acme.messages.AuthorizationRequest.create(
|
||||
self.msgs[domain].session_id,
|
||||
self.msgs[domain].nonce,
|
||||
self.responses[domain],
|
||||
domain,
|
||||
Crypto.PublicKey.RSA.importKey(self.authkey[domain].pem)),
|
||||
session_id=self.msgs[domain].session_id,
|
||||
nonce=self.msgs[domain].nonce,
|
||||
responses=self.responses[domain],
|
||||
name=domain,
|
||||
key=Crypto.PublicKey.RSA.importKey(
|
||||
self.authkey[domain].pem)),
|
||||
acme.messages.Authorization)
|
||||
logging.info("Received Authorization for %s", domain)
|
||||
return auth
|
||||
|
||||
@@ -122,7 +122,8 @@ class Client(object):
|
||||
|
||||
"""
|
||||
return self.network.send_and_receive_expected(
|
||||
acme.messages.ChallengeRequest(domain), acme.messages.Challenge)
|
||||
acme.messages.ChallengeRequest(identifier=domain),
|
||||
acme.messages.Challenge)
|
||||
|
||||
def acme_certificate(self, csr_der):
|
||||
"""Handle ACME "certificate" phase.
|
||||
@@ -136,7 +137,8 @@ class Client(object):
|
||||
logging.info("Preparing and sending CSR...")
|
||||
return self.network.send_and_receive_expected(
|
||||
acme.messages.CertificateRequest.create(
|
||||
csr_der, Crypto.PublicKey.RSA.importKey(self.authkey.pem)),
|
||||
csr=csr_der, key=Crypto.PublicKey.RSA.importKey(
|
||||
self.authkey.pem)),
|
||||
acme.messages.Certificate)
|
||||
|
||||
def save_certificate(self, certificate_msg, cert_path, chain_path):
|
||||
|
||||
@@ -53,7 +53,7 @@ class Network(object):
|
||||
raise errors.LetsEncryptClientError(
|
||||
'Sending ACME message to server has failed: %s' % error)
|
||||
|
||||
return acme.messages.Message.from_json(response.json())
|
||||
return acme.messages.Message.from_json(response.json(), validate=True)
|
||||
|
||||
def send_and_receive_expected(self, msg, expected):
|
||||
"""Send ACME message to server and return expected message.
|
||||
@@ -101,7 +101,7 @@ class Network(object):
|
||||
logging.info("Waiting for %d seconds...", delay)
|
||||
time.sleep(delay)
|
||||
response = self.send(
|
||||
acme.messages.StatusRequest(response.token))
|
||||
acme.messages.StatusRequest(token=response.token))
|
||||
else:
|
||||
logging.fatal("Received unexpected message")
|
||||
logging.fatal("Expected: %s", expected)
|
||||
|
||||
@@ -38,7 +38,7 @@ class Revoker(object):
|
||||
|
||||
revocation = self.network.send_and_receive_expected(
|
||||
acme.messages.RevocationRequest.create(
|
||||
cert_der, Crypto.PublicKey.RSA.importKey(key)),
|
||||
certificate=cert_der, key=Crypto.PublicKey.RSA.importKey(key)),
|
||||
acme.messages.Revocation)
|
||||
|
||||
zope.component.getUtility(interfaces.IDisplay).generic_notification(
|
||||
|
||||
@@ -41,7 +41,8 @@ class SatisfyChallengesTest(unittest.TestCase):
|
||||
def test_name1_dvsni1(self):
|
||||
dom = "0"
|
||||
challenge = [acme_util.CHALLENGES["dvsni"]]
|
||||
msg = acme.messages.Challenge(dom, "nonce0", challenge)
|
||||
msg = acme.messages.Challenge(session_id=dom, nonce="nonce0",
|
||||
challenges=challenge, combinations=[])
|
||||
self.handler.add_chall_msg(dom, msg, "dummy_key")
|
||||
|
||||
self.handler._satisfy_challenges() # pylint: disable=protected-access
|
||||
@@ -60,7 +61,8 @@ class SatisfyChallengesTest(unittest.TestCase):
|
||||
for i in range(5):
|
||||
self.handler.add_chall_msg(
|
||||
str(i),
|
||||
acme.messages.Challenge(str(i), "nonce%d" % i, challenge),
|
||||
acme.messages.Challenge(session_id=str(i), nonce="nonce%d" % i,
|
||||
challenges=challenge, combinations=[]),
|
||||
"dummy_key")
|
||||
|
||||
self.handler._satisfy_challenges() # pylint: disable=protected-access
|
||||
@@ -87,7 +89,8 @@ class SatisfyChallengesTest(unittest.TestCase):
|
||||
combos = acme_util.gen_combos(challenges)
|
||||
self.handler.add_chall_msg(
|
||||
dom,
|
||||
acme.messages.Challenge("0", "nonce0", challenges, combos),
|
||||
acme.messages.Challenge(session_id="0", nonce="nonce0",
|
||||
challenges=challenges, combinations=combos),
|
||||
"dummy_key")
|
||||
|
||||
path = gen_path(["simpleHttps"], challenges)
|
||||
@@ -116,7 +119,8 @@ class SatisfyChallengesTest(unittest.TestCase):
|
||||
combos = acme_util.gen_combos(challenges)
|
||||
self.handler.add_chall_msg(
|
||||
dom,
|
||||
acme.messages.Challenge(dom, "nonce0", challenges, combos),
|
||||
acme.messages.Challenge(session_id=dom, nonce="nonce0",
|
||||
challenges=challenges, combinations=combos),
|
||||
"dummy_key")
|
||||
|
||||
path = gen_path(["simpleHttps", "recoveryToken"], challenges)
|
||||
@@ -147,7 +151,8 @@ class SatisfyChallengesTest(unittest.TestCase):
|
||||
self.handler.add_chall_msg(
|
||||
str(i),
|
||||
acme.messages.Challenge(
|
||||
str(i), "nonce%d" % i, challenges, combos),
|
||||
session_id=str(i), nonce="nonce%d" % i,
|
||||
challenges=challenges, combinations=combos),
|
||||
"dummy_key")
|
||||
|
||||
path = gen_path(["dvsni", "recoveryContact"], challenges)
|
||||
@@ -197,7 +202,8 @@ class SatisfyChallengesTest(unittest.TestCase):
|
||||
self.handler.add_chall_msg(
|
||||
dom,
|
||||
acme.messages.Challenge(
|
||||
dom, "nonce%d" % i, challenge_list[i]),
|
||||
session_id=dom, nonce="nonce%d" % i,
|
||||
challenges=challenge_list[i], combinations=[]),
|
||||
"dummy_key")
|
||||
|
||||
mock_chall_path.side_effect = paths
|
||||
@@ -266,7 +272,8 @@ class GetAuthorizationsTest(unittest.TestCase):
|
||||
for i in range(3):
|
||||
self.handler.add_chall_msg(
|
||||
str(i),
|
||||
acme.messages.Challenge(str(i), "nonce%d" % i, challenge),
|
||||
acme.messages.Challenge(session_id=str(i), nonce="nonce%d" % i,
|
||||
challenges=challenge, combinations=[]),
|
||||
"dummy_key")
|
||||
|
||||
self.mock_sat_chall.side_effect = self._sat_solved_at_once
|
||||
@@ -294,7 +301,8 @@ class GetAuthorizationsTest(unittest.TestCase):
|
||||
challenges = acme_util.get_challenges()
|
||||
self.handler.add_chall_msg(
|
||||
"0",
|
||||
acme.messages.Challenge("0", "nonce0", challenges),
|
||||
acme.messages.Challenge(session_id="0", nonce="nonce0",
|
||||
challenges=challenges, combinations=[]),
|
||||
"dummy_key")
|
||||
|
||||
# Don't do anything to satisfy challenges
|
||||
@@ -322,7 +330,8 @@ class GetAuthorizationsTest(unittest.TestCase):
|
||||
dom = str(i)
|
||||
self.handler.add_chall_msg(
|
||||
dom,
|
||||
acme.messages.Challenge(dom, "nonce%d" % i, challs[i]),
|
||||
acme.messages.Challenge(session_id=dom, nonce="nonce%d" % i,
|
||||
challenges=challs[i], combinations=[]),
|
||||
"dummy_key")
|
||||
|
||||
self.mock_sat_chall.side_effect = self._sat_incremental
|
||||
|
||||
Reference in New Issue
Block a user