starred/certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-28 16:54:56 +02:00
Code Issues Packages Projects Releases Wiki Activity

IJSONSerializable Message, Signature, JWK

Browse Source
This commit is contained in:
Jakub Warmuz
2015-02-01 23:07:27 +00:00
parent 143b002d7e
commit a6addfa55a
24 changed files with 1051 additions and 386 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/api/acme/errors.rst
+5
View File
@@ -0,0 +1,5 @@
:mod:`letsencrypt.acme.errors`
------------------------------
.. automodule:: letsencrypt.acme.errors
:members:
docs/api/acme/interfaces.rst
+5
View File
@@ -0,0 +1,5 @@
:mod:`letsencrypt.acme.interfaces`
----------------------------------
.. automodule:: letsencrypt.acme.interfaces
:members:
docs/api/acme/jose.rst
+5
View File
@@ -0,0 +1,5 @@
:mod:`letsencrypt.acme.jose`
----------------------------
.. automodule:: letsencrypt.acme.jose
:members:
docs/api/acme/messages.rst
+5
View File
@@ -0,0 +1,5 @@
:mod:`letsencrypt.acme.messages`
--------------------------------
.. automodule:: letsencrypt.acme.messages
:members:
docs/api/acme/other.rst
+5
View File
@@ -0,0 +1,5 @@
:mod:`letsencrypt.acme.other`
-----------------------------
.. automodule:: letsencrypt.acme.other
:members:
docs/api/acme/util.rst
+5
View File
@@ -0,0 +1,5 @@
:mod:`letsencrypt.acme.util`
----------------------------
.. automodule:: letsencrypt.acme.util
:members:
docs/api/client/acme.rst
-5
View File
@@ -1,5 +0,0 @@
:mod:`letsencrypt.client.acme`
------------------------------
.. automodule:: letsencrypt.client.acme
:members:
letsencrypt/acme/errors.py
+13
View File
@@ -0,0 +1,13 @@
"""ACME errors."""
class Error(Exception):
"""Generic ACME error."""
class ValidationError(Error):
"""ACME message validation error."""
class UnrecognnizedMessageTypeError(ValidationError):
"""Unrecognized ACME message type error."""
class SchemaValidationError(ValidationError):
"""JSON schema ACME message validation error."""
letsencrypt/acme/interfaces.py
+22
View File
@@ -0,0 +1,22 @@
"""ACME interfaces."""
import zope.interface
# pylint: disable=no-self-argument,no-method-argument,no-init,inherit-non-class
class IJSONSerializable(zope.interface.Interface):
# pylint: disable=too-few-public-methods
"""JSON serializable object."""
def to_json():
"""Prepare JSON serializable object.
:returns: JSON object ready to be serialized. Note, however, that
this might return other
:class:`letsencrypt.acme.interfaces.IJSONSerializable`
objects, that haven't been serialized yet, which is fine as
long as :func:`letsencrypt.acme.util.dump_ijsonserializable`
is used.
:rtype: dict
"""
letsencrypt/acme/jose.py
+68
View File
@@ -0,0 +1,68 @@
"""JOSE."""
import binascii
import zope.interface
import Crypto.PublicKey.RSA
from letsencrypt.acme import interfaces
from letsencrypt.client import le_util
def _leading_zeros(arg):
if len(arg) % 2:
return "0" + arg
return arg
class JWK(object):
"""JSON Web Key.
.. todo:: Currently works for RSA keys only.
"""
zope.interface.implements(interfaces.IJSONSerializable)
def __init__(self, key):
self.key = key
def __eq__(self, other):
if isinstance(other, JWK):
return self.key == other.key
else:
raise TypeError(
'Unable to compare JWK object with: {0}'.format(other))
def same_public_key(self, other):
"""Does ``other`` have the same public key?"""
if isinstance(other, JWK):
return self.key.publickey() == other.key.publickey()
else:
raise TypeError(
'Unable to compare JWK object with: {0}'.format(other))
@classmethod
def _encode_param(cls, param):
"""Encode numeric key parameter."""
return le_util.jose_b64encode(binascii.unhexlify(
_leading_zeros(hex(param)[2:].rstrip("L"))))
@classmethod
def _decode_param(cls, param):
"""Decode numeric key parameter."""
return long(binascii.hexlify(le_util.jose_b64decode(param)), 16)
def to_json(self):
"""Serialize to JSON."""
return {
"kty": "RSA", # TODO
"n": self._encode_param(self.key.n),
"e": self._encode_param(self.key.e),
}
@classmethod
def from_json(cls, json_object):
"""Deserialize from JSON."""
assert "RSA" == json_object["kty"] # TODO
return cls(Crypto.PublicKey.RSA.construct(
(cls._decode_param(json_object["n"]),
cls._decode_param(json_object["e"]))))
letsencrypt/acme/jose_test.py
+61
View File
@@ -0,0 +1,61 @@
"""Tests for letsencrypt.acme.jose."""
import pkg_resources
import unittest
import Crypto.PublicKey.RSA
RSA256_KEY_PATH = pkg_resources.resource_string(
'letsencrypt.client.tests', 'testdata/rsa256_key.pem')
RSA256_KEY = Crypto.PublicKey.RSA.importKey(RSA256_KEY_PATH)
RSA512_KEY_PATH = pkg_resources.resource_string(
'letsencrypt.client.tests', 'testdata/rsa512_key.pem')
RSA512_KEY = Crypto.PublicKey.RSA.importKey(RSA512_KEY_PATH)
class JWKTest(unittest.TestCase):
def setUp(self):
from letsencrypt.acme.jose import JWK
self.jwk256 = JWK(RSA256_KEY)
self.jwk256json = {
'kty': 'RSA',
'e': 'AQAB',
'n': 'rHVztFHtH92ucFJD_N_HW9AsdRsUuHUBBBDlHwNlRd3fp5'
'80rv2-6QWE30cWgdmJS86ObRz6lUTor4R0T-3C5Q',
}
self.jwk512 = JWK(RSA512_KEY)
def test_equals(self):
self.assertEqual(self.jwk256, self.jwk256)
self.assertEqual(self.jwk512, self.jwk512)
def test_not_equals(self):
self.assertNotEqual(self.jwk256, self.jwk512)
self.assertNotEqual(self.jwk512, self.jwk256)
def test_equals_raises_type_error(self):
self.assertRaises(TypeError, self.jwk256.__eq__, 123)
def test_same_public_key(self):
from letsencrypt.acme.jose import JWK
self.assertTrue(self.jwk256.same_public_key(
JWK(Crypto.PublicKey.RSA.importKey(RSA256_KEY_PATH))))
def test_not_same_public_key(self):
self.assertFalse(self.jwk256.same_public_key(self.jwk512))
def test_same_public_key_raises_type_error(self):
self.assertRaises(TypeError, self.jwk256.same_public_key, 5)
def test_to_json(self):
self.assertEqual(self.jwk256.to_json(), self.jwk256json)
def test_from_json(self):
from letsencrypt.acme.jose import JWK
self.assertTrue(self.jwk256.same_public_key(
JWK.from_json(self.jwk256json)))
if __name__ == "__main__":
unittest.main()
letsencrypt/acme/messages.py
+501 -100
View File
@@ -3,8 +3,15 @@ import json
import pkg_resources
import jsonschema
import M2Crypto
import zope.interface
from letsencrypt.acme import errors
from letsencrypt.acme import interfaces
from letsencrypt.acme import jose
from letsencrypt.acme import other
from letsencrypt.acme import util
from letsencrypt.client import crypto_util
from letsencrypt.client import le_util
@@ -21,135 +28,529 @@ SCHEMATA = dict([
"error",
"revocation",
"revocationRequest",
"statusRequest"
"statusRequest",
]
])
def acme_object_validate(json_string, schemata=None):
"""Validate a JSON string against the ACME protocol using JSON Schema.
class Message(object):
"""ACME message.
:param str json_string: Well-formed input JSON string.
:param dict schemata: Mapping from type name to JSON Schema
definition. Useful for testing.
:returns: None if validation was successful.
:raises jsonschema.ValidationError: if validation was unsuccessful
:raises ValueError: if the object cannot even be parsed as valid JSON
Messages are considered immutable.
"""
schemata = SCHEMATA if schemata is None else schemata
json_object = json.loads(json_string)
if not isinstance(json_object, dict):
raise jsonschema.ValidationError("this is not a dictionary object")
if "type" not in json_object:
raise jsonschema.ValidationError("missing type field")
if json_object["type"] not in schemata:
raise jsonschema.ValidationError(
"unknown type %s" % json_object["type"])
jsonschema.validate(json_object, schemata[json_object["type"]])
zope.interface.implements(interfaces.IJSONSerializable)
acme_type = NotImplemented
"""ACME message "type" field. Subclasses must override."""
TYPES = {}
"""Message types registered for JSON deserialization"""
@classmethod
def register(cls, msg_cls):
"""Register class for JSON deserialization."""
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.
:returns: Serializable JSON object representing ACME message.
:meth:`validate` will almost certianly not work, due to reasons
explained in :class:`letsencrypt.acme.interfaces.IJSONSerializable`.
:rtype: dict
"""
json_object = self._fields_to_json()
json_object["type"] = self.acme_type
return json_object
def _fields_to_json(self):
"""Prepare ACME message fields for JSON serialiazation.
Subclasses must override this method.
:returns: Serializable JSON object containg all ACME message fields
apart from "type".
: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)
@classmethod
def validate(cls, json_object, schemata=None):
"""Is JSON object a valid ACME message?
:param str json_object: JSON object
:param dict schemata: Mapping from type name to JSON Schema
definition. Useful for testing.
:returns: ACME message class, subclassing :class:`Message`.
:raises letsencrypt.acme.errors.ValidationError: if validation
was unsuccessful
"""
schemata = SCHEMATA if schemata is None else schemata
if not isinstance(json_object, dict):
raise errors.ValidationError(
"{0} is not a dictionary object".format(json_object))
try:
msg_type = json_object["type"]
except KeyError:
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(json_object, 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`
"""
json_object = json.loads(json_string)
msg_cls = cls.validate(json_object, schemata)
# pylint: disable=protected-access
return msg_cls._valid_from_json(json_object)
@classmethod
def _valid_from_json(cls, json_object):
"""Deserialize from valid ACME message JSON object.
Subclasses must override.
:param json_object: Schema validated ACME message JSON object.
:type json_object: dict
:returns: Valid ACME message.
:rtype: subclass of :class:`Message`
"""
raise NotImplementedError
def pretty(json_string):
"""Return a pretty-printed version of any JSON string.
@Message.register # pylint: disable=too-few-public-methods
class Challenge(Message):
"""ACME "challenge" message."""
acme_type = "challenge"
Useful when printing out protocol messages for debugging purposes.
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
def _fields_to_json(self):
fields = {
"sessionID": self.session_id,
"nonce": le_util.jose_b64encode(self.nonce),
"challenges": self.challenges,
}
if self.combinations:
fields["combinations"] = self.combinations
return fields
@classmethod
def _valid_from_json(cls, json_object):
return cls(json_object["sessionID"],
le_util.jose_b64decode(json_object["nonce"]),
json_object["challenges"], json_object.get("combinations"))
@Message.register # pylint: disable=too-few-public-methods
class ChallengeRequest(Message):
"""ACME "challengeRequest" message.
:ivar str identifier: Domain name.
"""
return json.dumps(json.loads(json_string), indent=4)
acme_type = "challengeRequest"
def __init__(self, identifier):
self.identifier = identifier
def _fields_to_json(self):
return {
"identifier": self.identifier,
}
@classmethod
def _valid_from_json(cls, json_string):
return cls(json_string["identifier"])
def challenge_request(name):
"""Create ACME "challengeRequest message.
@Message.register # pylint: disable=too-few-public-methods
class Authorization(Message):
"""ACME "authorization" message."""
acme_type = "authorization"
:param str name: Domain name
def __init__(self, recovery_token=None, identifier=None, jwk=None):
self.recovery_token = recovery_token
self.identifier = identifier
self.jwk = jwk
:returns: ACME "challengeRequest" message.
:rtype: dict
def _fields_to_json(self):
fields = {}
if self.recovery_token is not None:
fields["recoveryToken"] = self.recovery_token
if self.identifier is not None:
fields["identifier"] = self.identifier
if self.jwk is not None:
fields["jwk"] = self.jwk
return fields
@classmethod
def _valid_from_json(cls, json_object):
jwk = json_object.get("jwk")
if jwk is not None:
jwk = jose.JWK.from_json(jwk)
return cls(json_object.get("recoveryToken"),
json_object.get("identifier"),
jwk)
@Message.register
class AuthorizationRequest(Message):
"""ACME "authorizationRequest" message.
:ivar str session_id: "sessionID" from the server challenge
:ivar str name: Hostname
:ivar str nonce: Nonce from the server challenge
:ivar list responses: List of completed challenges
:ivar contact: TODO
"""
return {
"type": "challengeRequest",
"identifier": name,
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
@classmethod
def create(cls, session_id, nonce, responses, name, key,
sig_nonce=None, contact=None):
"""Create signed "authorizationRequest".
:param key: Key used for signing.
:type key: :class:`Crypto.PublicKey.RSA`
:param str sig_nonce: Nonce used for signature. Useful for testing.
:returns: Signed "authorizationRequest" ACME message.
:rtype: :class:`AuthorizationRequest`
"""
# pylint: disable=too-many-arguments
signature = other.Signature.from_msg(name + nonce, key, sig_nonce)
return cls(session_id, nonce, responses, signature, contact)
def verify(self, name):
"""Verify signature.
:param str name: Hostname
:returns: True iff ``signature`` can be verified, False otherwise.
:rtype: bool
"""
# TODO: must also check that the public key encoded in the JWK object
# is the correct key for a given context.
return self.signature.verify(name + self.nonce)
def _fields_to_json(self):
fields = {
"sessionID": self.session_id,
"nonce": self.nonce,
"responses": self.responses,
"signature": self.signature,
}
if self.contact:
fields["contact"] = self.contact
return fields
@classmethod
def _valid_from_json(cls, json_object):
return cls(json_object["sessionID"], json_object["nonce"],
json_object["responses"],
other.Signature.from_json(json_object["signature"]),
json_object.get("contact"))
@Message.register # pylint: disable=too-few-public-methods
class Certificate(Message):
"""ACME "certificate" message.
:ivar certificate: TODO
:type certificate: :class:`M2Crypto.X509` TODO
"""
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
def _fields_to_json(self):
fields = {
"certificate": le_util.jose_b64encode(self.certificate.as_der())}
if self.chain is not None:
fields["chain"] = self.chain
if self.refresh is not None:
fields["refresh"] = self.refresh
return fields
@classmethod
def _valid_from_json(cls, json_object):
certificate = M2Crypto.X509.load_cert_der_string(
le_util.jose_b64decode(json_object["certificate"]))
return cls(certificate,
json_object.get("chain"),
json_object.get("refresh"))
@Message.register
class CertificateRequest(Message):
"""ACME "certificateRequest" message.
:ivar str csr: DER encoded CSR.
:ivar signature: Signature.
:type signature: :class:`letsencrypt.acme.other.Signature`
"""
acme_type = "certificateRequest"
def __init__(self, csr, signature):
self.csr = csr
self.signature = signature
@classmethod
def create(cls, csr, key, nonce=None):
"""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.
:returns: Signed "certificateRequest" ACME message.
:rtype: :class:`CertificateRequest`
"""
return cls(csr, other.Signature.from_msg(csr, key, nonce))
def verify(self):
"""Verify signature.
:returns: True iff ``signature`` can be verified, False otherwise.
:rtype: bool
"""
# TODO: must also check that the public key encoded in the JWK object
# is the correct key for a given context.
return self.signature.verify(self.csr)
def _fields_to_json(self):
return {
"csr": le_util.jose_b64encode(self.csr),
"signature": self.signature,
}
@classmethod
def _valid_from_json(cls, json_object):
return cls(le_util.jose_b64decode(json_object["csr"]),
other.Signature.from_json(json_object["signature"]))
@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
def _fields_to_json(self):
fields = {"token": self.token}
if self.interval is not None:
fields["interval"] = self.interval
if self.message is not None:
fields["message"] = self.message
return fields
@classmethod
def _valid_from_json(cls, json_object):
return cls(json_object["token"], json_object.get("interval"),
json_object.get("message"))
@Message.register # pylint: disable=too-few-public-methods
class Error(Message):
"""ACME "error" message."""
acme_type = "error"
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)",
}
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 authorization_request(req_id, name, server_nonce, responses, key,
nonce=None):
"""Create ACME "authorizationRequest" message.
def _fields_to_json(self):
fields = {"error": self.error}
if self.message is not None:
fields["message"] = self.message
if self.more_info is not None:
fields["moreInfo"] = self.more_info
return fields
:param str req_id: SessionID from the server challenge
:param str name: Hostname
:param str server_nonce: Nonce from the server challenge
:param list responses: List of completed challenges
:param str key: Key in string form. Accepted formats
@classmethod
def _valid_from_json(cls, json_object):
return cls(json_object["error"], json_object.get("message"),
json_object.get("more_info"))
@Message.register # pylint: disable=too-few-public-methods
class Revocation(Message):
"""ACME "revocation" message."""
acme_type = "revocation"
def _fields_to_json(self):
return {}
@classmethod
def _valid_from_json(cls, json_object):
return cls()
@Message.register
class RevocationRequest(Message):
"""ACME "revocationRequest" message.
:iver str certificate: DER encoded certificate.
:iver str key: Key in string form. Accepted formats
are the same as for `Crypto.PublicKey.RSA.importKey`.
:param str nonce: Nonce used for signature. Useful for testing.
:returns: ACME "authorizationRequest" message.
:rtype: dict
:ivar str nonce: Nonce used for signature. Useful for testing.
"""
return {
"type": "authorizationRequest",
"sessionID": req_id,
"nonce": server_nonce,
"responses": responses,
"signature": crypto_util.create_sig(
name + le_util.jose_b64decode(server_nonce), key, nonce),
}
acme_type = "revocationRequest"
def __init__(self, certificate, signature):
self.certificate = certificate
self.signature = signature
@classmethod
def create(cls, certificate, key, nonce=None):
"""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.
:returns: Signed "revocationRequest" ACME message.
:rtype: :class:`RevocationRequest`
"""
return cls(certificate,
other.Signature.from_msg(certificate, key, nonce))
def verify(self):
"""Verify signature.
:returns: True iff ``signature`` can be verified, False otherwise.
:rtype: bool
"""
# TODO: must also check that the public key encoded in the JWK object
# is the correct key for a given context.
return self.signature.verify(self.certificate)
def _fields_to_json(self):
return {
"certificate": le_util.jose_b64encode(self.certificate),
"signature": self.signature,
}
@classmethod
def _valid_from_json(cls, json_string):
return cls(le_util.jose_b64decode(json_string["certificate"]),
other.Signature.from_json(json_string["signature"]))
def certificate_request(csr_der, key, nonce=None):
"""Create ACME "certificateRequest" message.
@Message.register # pylint: disable=too-few-public-methods
class StatusRequest(Message):
"""ACME "statusRequest" message.
:param str csr_der: DER encoded CSR.
:param str key: Key in string form. Accepted formats
are the same as for `Crypto.PublicKey.RSA.importKey`.
:param str nonce: Nonce used for signature. Useful for testing.
:returns: ACME "certificateRequest" message.
:rtype: dict
:ivar unicode token: Token provided in ACME "defer" message.
"""
return {
"type": "certificateRequest",
"csr": le_util.jose_b64encode(csr_der),
"signature": crypto_util.create_sig(csr_der, key, nonce),
}
acme_type = "statusRequest"
def __init__(self, token):
self.token = token
def revocation_request(cert_der, key, nonce=None):
"""Create ACME "revocationRequest" message.
def _fields_to_json(self):
return {
"token": self.token,
}
:param str cert_der: DER encoded certificate.
:param str key: Key in string form. Accepted formats
are the same as for `Crypto.PublicKey.RSA.importKey`.
:param str nonce: Nonce used for signature. Useful for testing.
:returns: ACME "revocationRequest" message.
:rtype: dict
"""
return {
"type": "revocationRequest",
"certificate": le_util.jose_b64encode(cert_der),
"signature": crypto_util.create_sig(cert_der, key, nonce),
}
def status_request(token):
"""Create ACME "statusRequest" message.
:param unicode token: Token provided in ACME "defer" message.
:returns: ACME "statusRequest" message.
:rtype: dict
"""
return {
"type": "statusRequest",
"token": token,
}
@classmethod
def _valid_from_json(cls, json_string):
return cls(json_string["token"])
letsencrypt/acme/messages_test.py
+108 -93
View File
@@ -2,11 +2,17 @@
import pkg_resources
import unittest
import jsonschema
import Crypto.PublicKey.RSA
import mock
from letsencrypt.acme import errors
KEY = Crypto.PublicKey.RSA.importKey(pkg_resources.resource_string(
'letsencrypt.client.tests', 'testdata/rsa256_key.pem'))
class ACMEObjectValidateTest(unittest.TestCase):
"""Tests for letsencrypt.acme.messages.acme_object_validate."""
class MessageTest(unittest.TestCase):
"""Tests for letsencrypt.acme.messages.Message."""
def setUp(self):
self.schemata = {
@@ -19,68 +25,59 @@ class ACMEObjectValidateTest(unittest.TestCase):
},
}
def _call(self, json_string):
from letsencrypt.acme.messages import acme_object_validate
return acme_object_validate(json_string, self.schemata)
def _test_fails(self, json_string):
self.assertRaises(jsonschema.ValidationError, self._call, json_string)
def _validate(self, json_object):
from letsencrypt.acme.messages import Message
return Message.validate(json_object, self.schemata)
def test_non_dictionary_fails(self):
self._test_fails('[]')
def test_validate_non_dictionary_fails(self):
self.assertRaises(errors.ValidationError, self._validate, [])
def test_dict_without_type_fails(self):
self._test_fails('{}')
def test_validate_dict_without_type_fails(self):
self.assertRaises(errors.ValidationError, self._validate, {})
def test_unknown_type_fails(self):
self._test_fails('{"type": "bar"}')
def test_validate_unknown_type_fails(self):
self.assertRaises(errors.UnrecognnizedMessageTypeError,
self._validate, {"type": "bar"})
def test_valid_returns_none(self):
self.assertTrue(self._call('{"type": "foo"}') is None)
def test_validate_unregistered_type_fails(self):
self.assertRaises(errors.UnrecognnizedMessageTypeError,
self._validate, {"type": "foo"})
def test_invalid_fails(self):
self._test_fails('{"type": "foo", "price": "asd"}')
@mock.patch("letsencrypt.acme.messages.Message.TYPES")
def test_validate_invalid_fails(self, types):
types.__getitem__.side_effect = lambda x: {"foo": "bar"}[x]
self.assertRaises(errors.SchemaValidationError,
self._validate, {"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")
class PrettyTest(unittest.TestCase): # pylint: disable=too-few-public-methods
"""Tests for letsencrypt.acme.messages.pretty."""
@classmethod
def _call(cls, json_string):
from letsencrypt.acme.messages import pretty
return pretty(json_string)
class ChallengeRequestTest(unittest.TestCase):
# pylint: disable=too-few-public-methods
def test_it(self):
self.assertEqual(
self._call('{"foo": {"bar": "baz"}}'),
'{\n "foo": {\n "bar": "baz"\n }\n}')
from letsencrypt.acme.messages import ChallengeRequest
msg = ChallengeRequest('example.com')
class MessageFactoriesTest(unittest.TestCase):
"""Tests for ACME message factories from letsencrypt.acme.messages."""
def setUp(self):
self.privkey = pkg_resources.resource_string(
'letsencrypt.client.tests', 'testdata/rsa256_key.pem')
self.nonce = '\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9'
self.b64nonce = '7Nbyb1lI6xPVI3Hg3aKSqQ'
@classmethod
def _validate(cls, msg):
from letsencrypt.acme.messages import SCHEMATA
jsonschema.validate(msg, SCHEMATA[msg['type']])
def test_challenge_request(self):
from letsencrypt.acme.messages import challenge_request
msg = challenge_request('example.com')
self._validate(msg)
self.assertEqual(msg, {
'type': 'challengeRequest',
jmsg = msg._fields_to_json() # pylint: disable=protected-access
self.assertEqual(jmsg, {
'identifier': 'example.com',
})
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 authorization_request
from letsencrypt.acme.messages import AuthorizationRequest
responses = [
{
'type': 'simpleHttps',
@@ -92,66 +89,84 @@ class MessageFactoriesTest(unittest.TestCase):
'token': '23029d88d9e123e',
}
]
msg = authorization_request(
msg = AuthorizationRequest.create(
'aefoGaavieG9Wihuk2aufai3aeZ5EeW4',
'example.com',
'czpsrF0KMH6dgajig3TGHw',
responses,
self.privkey,
'example.com',
KEY,
self.nonce,
)
msg.verify('example.com')
self._validate(msg)
self.assertEqual(
msg.pop('signature')['sig'],
'VkpReso87ogwGul2MGck96TkYs4QoblIgNthgrm9O7EBGlzCRCnTHnx'
'bj6loqaC4f5bn1rgS927Gp1Kvbqnmqg'
)
self.assertEqual(msg, {
'type': 'authorizationRequest',
jmsg = msg._fields_to_json() # pylint: disable=protected-access
jmsg.pop('signature')
self.assertEqual(jmsg, {
'sessionID': 'aefoGaavieG9Wihuk2aufai3aeZ5EeW4',
'nonce': 'czpsrF0KMH6dgajig3TGHw',
'responses': responses,
})
def test_certificate_request(self):
from letsencrypt.acme.messages import certificate_request
msg = certificate_request(
'TODO: real DER CSR?', self.privkey, self.nonce)
self._validate(msg)
self.assertEqual(
msg.pop('signature')['sig'],
'HEQVN4MU1yDrArP2T7WZQ12XlHCn5DgTPgb5eWT5_vjRPppLSNe6uWE'
'x9SFwG9d9umqn49nZCSW7uskA2lcW6Q'
)
self.assertEqual(msg, {
'type': 'certificateRequest',
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?'
def test_it(self):
from letsencrypt.acme.messages import CertificateRequest
msg = CertificateRequest.create(self.csr, KEY, self.nonce)
self.assertTrue(msg.verify())
jmsg = msg._fields_to_json() # pylint: disable=protected-access
jmsg.pop('signature')
self.assertEqual(jmsg, {
'csr': 'VE9ETzogcmVhbCBERVIgQ1NSPw',
})
def test_revocation_request(self):
from letsencrypt.acme.messages import revocation_request
msg = revocation_request(
'TODO: real DER cert?', self.privkey, self.nonce)
self._validate(msg)
self.assertEqual(
msg.pop('signature')['sig'],
'ABXA1IsyTalTXIojxmGnIUGyZASmvqEvTQ98jJ5KFs2FTswLEmsoqFX'
'fU6l5_fous-tsbXOfLN-7PjfZ5XWPvg'
)
self.assertEqual(msg, {
'type': 'revocationRequest',
class RevocationRequestTest(unittest.TestCase):
def setUp(self):
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())
jmsg = msg._fields_to_json() # pylint: disable=protected-access
jmsg.pop('signature')
self.assertEqual(jmsg, {
'certificate': 'VE9ETzogcmVhbCBERVIgY2VydD8',
})
def test_status_request(self):
from letsencrypt.acme.messages import status_request
msg = status_request(u'O7-s9MNq1siZHlgrMzi9_A')
self._validate(msg)
self.assertEqual(msg, {
'type': 'statusRequest',
'token': u'O7-s9MNq1siZHlgrMzi9_A',
})
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.jmsg = {
'token': self.token,
}
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)
from letsencrypt.acme.messages import StatusRequest
# pylint: disable=protected-access
msg = StatusRequest._valid_from_json(self.jmsg)
self.assertEqual(msg.token, self.msg.token)
if __name__ == '__main__':
letsencrypt/acme/other.py
+102
View File
@@ -0,0 +1,102 @@
"""JSON objects in ACME protocol other than messages."""
import logging
from Crypto import Random
import Crypto.Hash.SHA256
import Crypto.Signature.PKCS1_v1_5
import zope.interface
from letsencrypt.acme import interfaces
from letsencrypt.acme import jose
from letsencrypt.client import CONFIG
from letsencrypt.client import le_util
class Signature(object):
"""ACME signature.
:ivar str alg: Signature algorithm.
:ivar str sig: Signature.
:ivar str nonce: Nonce.
:ivar jwk: JWK.
:type jwk: :class:`letsencrypt.acme.jose.JWK`
.. todo:: Currently works for RSA keys only.
"""
zope.interface.implements(interfaces.IJSONSerializable)
NONCE_LEN = CONFIG.NONCE_SIZE
def __init__(self, alg, sig, nonce, jwk):
self.alg = alg
self.sig = sig
self.nonce = nonce
self.jwk = jwk
@classmethod
def from_msg(cls, msg, key, nonce=None):
"""Create signature with nonce prepended to the message.
.. todo:: Change this over to M2Crypto... PKey
.. todo:: Protect against crypto unicode errors... is this sufficient?
Do I need to escape?
:param str msg: Message to be signed.
:param key: Key used for signing.
:type key: :class:`Crypto.PublicKey.RSA`
:param nonce: Nonce to be used. If None, nonce of
:const:`NONCE_LEN` size will be randomly generated.
:type nonce: str or None
"""
msg = str(msg) # TODO: ????
if nonce is None:
nonce = Random.get_random_bytes(cls.NONCE_LEN)
msg_with_nonce = nonce + msg
hashed = Crypto.Hash.SHA256.new(msg_with_nonce)
sig = Crypto.Signature.PKCS1_v1_5.new(key).sign(hashed)
logging.debug("%s signed as %s", msg_with_nonce, sig)
return cls("RS256", sig, nonce, jose.JWK(key))
def __eq__(self, other):
if isinstance(other, Signature):
return ((self.alg, self.sig, self.nonce, self.jwk) ==
(other.alg, other.sig, other.nonce, other.jwk))
else:
raise TypeError(
'Unable to compare Signature object with: {0}'.format(other))
def verify(self, msg):
"""Verify the signature.
:param str msg: Message that was used in signing.
"""
return self == self.from_msg(msg, self.jwk.key, self.nonce)
def to_json(self):
"""Seriliaze to JSON."""
return {
"alg": self.alg,
"sig": le_util.jose_b64encode(self.sig),
"nonce": le_util.jose_b64encode(self.nonce),
"jwk": self.jwk,
}
@classmethod
def from_json(cls, json_object):
"""Deserialize from JSON."""
return cls(json_object["alg"],
le_util.jose_b64decode(json_object["sig"]),
le_util.jose_b64decode(json_object["nonce"]),
jose.JWK.from_json(json_object["jwk"]))
letsencrypt/acme/other_test.py
+56
View File
@@ -0,0 +1,56 @@
"""Tests for letsencrypt.acme.sig."""
import pkg_resources
import unittest
import Crypto.PublicKey.RSA
from letsencrypt.acme import jose
RSA256_KEY = Crypto.PublicKey.RSA.importKey(pkg_resources.resource_string(
'letsencrypt.client.tests', 'testdata/rsa256_key.pem'))
RSA512_KEY = Crypto.PublicKey.RSA.importKey(pkg_resources.resource_string(
'letsencrypt.client.tests', 'testdata/rsa512_key.pem'))
class SigatureTest(unittest.TestCase):
"""Tests for letsencrypt.acme.sig.Signature."""
def setUp(self):
self.alg = 'RS256'
self.sig = ('IC\xd8*\xe7\x14\x9e\x19S\xb7\xcf\xec3\x12\xe2\x8a\x03'
'\x98u\xff\xf0\x94\xe2\xd7<\x8f\xa8\xed\xa4KN\xc3\xaa'
'\xb9X\xc3w\xaa\xc0_\xd0\x05$y>l#\x10<\x96\xd2\xcdr\xa3'
'\x1b\xa1\xf5!f\xef\xc64\xb6\x13')
self.nonce = '\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9'
self.jwk = jose.JWK(RSA256_KEY)
self.b64sig = ('SUPYKucUnhlTt8_sMxLiigOYdf_wlOLXPI-o7aRLTsOquVjDd6r'
'AX9AFJHk-bCMQPJbSzXKjG6H1IWbvxjS2Ew')
self.b64nonce = '7Nbyb1lI6xPVI3Hg3aKSqQ'
self.jsig = {
'nonce': self.b64nonce,
'alg': self.alg,
'jwk': self.jwk,
'sig': self.b64sig,
}
@classmethod
def _from_msg(cls, *args, **kwargs):
from letsencrypt.acme.other import Signature
return Signature.from_msg(*args, **kwargs)
def test_from_msg(self):
sig = self._from_msg('message', RSA256_KEY, self.nonce)
self.assertEqual(sig.alg, self.alg)
self.assertEqual(sig.sig, self.sig)
self.assertEqual(sig.nonce, self.nonce)
self.assertEqual(sig.jwk, self.jwk)
def test_from_random_nonce(self):
sig = self._from_msg('message', RSA256_KEY)
self.assertEqual(sig.alg, self.alg)
self.assertEqual(sig.jwk, self.jwk)
if __name__ == "__main__":
unittest.main()
letsencrypt/acme/util.py
+15
View File
@@ -0,0 +1,15 @@
"""ACME utilities."""
from letsencrypt.acme import interfaces
def dump_ijsonserializable(python_object):
"""Serialize IJSONSerializable to JSON.
This is meant to be passed to :func:`json.dumps` as ``default``
argument.
"""
if interfaces.IJSONSerializable.providedBy(python_object):
return python_object.to_json()
else:
raise TypeError(repr(python_object) + ' is not JSON serializable')
letsencrypt/client/auth_handler.py
+16 -12
View File
@@ -2,6 +2,8 @@
import logging
import sys
import Crypto.PublicKey.RSA
from letsencrypt import acme
from letsencrypt.client import CONFIG
@@ -52,7 +54,9 @@ class AuthHandler(object): # pylint: disable=too-many-instance-attributes
"""Add a challenge message to the AuthHandler.
:param str domain: domain for authorization
:param dict msg: ACME challenge message
:param msg: ACME "challenge" message
:type msg: :class:`letsencrypt.acme.message.Challenge`
:param authkey: authorized key for the challenge
:type authkey: :class:`letsencrypt.client.client.Client.Key`
@@ -63,7 +67,7 @@ class AuthHandler(object): # pylint: disable=too-many-instance-attributes
"Multiple ACMEChallengeMessages for the same domain "
"is not supported.")
self.domains.append(domain)
self.responses[domain] = ["null"] * len(msg["challenges"])
self.responses[domain] = ["null"] * len(msg.challenges)
self.msgs[domain] = msg
self.authkey[domain] = authkey
@@ -101,18 +105,18 @@ class AuthHandler(object): # pylint: disable=too-many-instance-attributes
:param str domain: domain that is requesting authorization
:returns: ACME "authorization" message.
:rtype: dict
:rtype: :class:`letsencrypt.acme.messages.Authorization`
"""
try:
auth = self.network.send_and_receive_expected(
acme.messages.authorization_request(
self.msgs[domain]["sessionID"],
domain,
self.msgs[domain]["nonce"],
acme.messages.AuthorizationRequest.create(
self.msgs[domain].session_id,
self.msgs[domain].nonce,
self.responses[domain],
self.authkey[domain].pem),
"authorization")
domain,
Crypto.PublicKey.RSA.importKey(self.authkey[domain].pem)),
acme.messages.Authorization)
logging.info("Received Authorization for %s", domain)
return auth
except errors.LetsEncryptClientError as err:
@@ -128,9 +132,9 @@ class AuthHandler(object): # pylint: disable=too-many-instance-attributes
logging.info("Performing the following challenges:")
for dom in self.domains:
self.paths[dom] = gen_challenge_path(
self.msgs[dom]["challenges"],
self.msgs[dom].challenges,
self._get_chall_pref(dom),
self.msgs[dom].get("combinations", None))
self.msgs[dom].combinations)
self.dv_c[dom], self.client_c[dom] = self._challenge_factory(
dom, self.paths[dom])
@@ -230,7 +234,7 @@ class AuthHandler(object): # pylint: disable=too-many-instance-attributes
recognized
"""
challenges = self.msgs[domain]["challenges"]
challenges = self.msgs[domain].challenges
dv_chall = []
client_chall = []
letsencrypt/client/client.py
+16 -13
View File
@@ -6,6 +6,7 @@ import os
import shutil
import sys
import Crypto.PublicKey.RSA
import M2Crypto
import zope.component
@@ -103,11 +104,11 @@ class Client(object):
csr = init_csr(self.authkey, domains)
# Retrieve certificate
certificate_dict = self.acme_certificate(csr.data)
certificate_msg = self.acme_certificate(csr.data)
# Save Certificate
cert_file, chain_file = self.save_certificate(
certificate_dict, cert_path, chain_path)
certificate_msg, cert_path, chain_path)
self.store_cert_key(cert_file, False)
@@ -117,11 +118,11 @@ class Client(object):
"""Handle ACME "challenge" phase.
:returns: ACME "challenge" message.
:rtype: dict
:rtype: :class:`letsencrypt.acme.messages.Challenge`
"""
return self.network.send_and_receive_expected(
acme.messages.challenge_request(domain), "challenge")
acme.messages.ChallengeRequest(domain), acme.messages.Challenge)
def acme_certificate(self, csr_der):
"""Handle ACME "certificate" phase.
@@ -129,19 +130,22 @@ class Client(object):
:param str csr_der: CSR in DER format.
:returns: ACME "certificate" message.
:rtype: dict
:rtype: :class:`letsencrypt.acme.message.Certificate`
"""
logging.info("Preparing and sending CSR...")
return self.network.send_and_receive_expected(
acme.messages.certificate_request(
csr_der, self.authkey.pem), "certificate")
acme.messages.CertificateRequest.create(
csr_der, Crypto.PublicKey.RSA.importKey(self.authkey.pem)),
acme.messages.Certificate)
def save_certificate(self, certificate_dict, cert_path, chain_path):
def save_certificate(self, certificate_msg, cert_path, chain_path):
# pylint: disable=no-self-use
"""Saves the certificate received from the ACME server.
:param dict certificate_dict: certificate message from server
:param certificate_msg: ACME "certificate" message from server.
:type certificate_msg: :class:`letsencrypt.acme.messages.Certificate`
:param str cert_path: Path to attempt to save the cert file
:param str chain_path: Path to attempt to save the chain file
@@ -153,15 +157,14 @@ class Client(object):
"""
cert_chain_abspath = None
cert_fd, cert_file = le_util.unique_file(cert_path, 0o644)
cert_fd.write(
crypto_util.b64_cert_to_pem(certificate_dict["certificate"]))
cert_fd.write(certificate_msg.certificate.as_pem())
cert_fd.close()
logging.info(
"Server issued certificate; certificate written to %s", cert_file)
if certificate_dict.get("chain", None):
if certificate_msg.chain:
chain_fd, chain_fn = le_util.unique_file(chain_path, 0o644)
for cert in certificate_dict.get("chain", []):
for cert in certificate_msg.chain:
chain_fd.write(crypto_util.b64_cert_to_pem(cert))
chain_fd.close()
letsencrypt/client/crypto_util.py
-58
View File
@@ -1,73 +1,15 @@
"""Let's Encrypt client crypto utility functions"""
import binascii
import logging
import time
from Crypto import Random
import Crypto.Hash.SHA256
import Crypto.PublicKey.RSA
import Crypto.Signature.PKCS1_v1_5
import M2Crypto
from letsencrypt.client import CONFIG
from letsencrypt.client import le_util
def create_sig(msg, key_str, nonce=None, nonce_len=CONFIG.NONCE_SIZE):
"""Create signature with nonce prepended to the message.
.. todo:: Change this over to M2Crypto... PKey
.. todo:: Protect against crypto unicode errors... is this sufficient?
Do I need to escape?
:param str key_str: Key in string form. Accepted formats
are the same as for `Crypto.PublicKey.RSA.importKey`.
:param str msg: Message to be signed
:param nonce: Nonce to be used. If None, nonce of `nonce_len` size
will be randomly generated.
:type nonce: str or None
:param int nonce_len: Size of the automatically generated nonce.
:returns: Signature.
:rtype: dict
"""
msg = str(msg)
key = Crypto.PublicKey.RSA.importKey(key_str)
nonce = Random.get_random_bytes(nonce_len) if nonce is None else nonce
msg_with_nonce = nonce + msg
hashed = Crypto.Hash.SHA256.new(msg_with_nonce)
signature = Crypto.Signature.PKCS1_v1_5.new(key).sign(hashed)
logging.debug("%s signed as %s", msg_with_nonce, signature)
n_bytes = binascii.unhexlify(_leading_zeros(hex(key.n)[2:].rstrip("L")))
e_bytes = binascii.unhexlify(_leading_zeros(hex(key.e)[2:].rstrip("L")))
return {
"nonce": le_util.jose_b64encode(nonce),
"alg": "RS256",
"jwk": {
"kty": "RSA",
"n": le_util.jose_b64encode(n_bytes),
"e": le_util.jose_b64encode(e_bytes),
},
"sig": le_util.jose_b64encode(signature),
}
def _leading_zeros(arg):
if len(arg) % 2:
return "0" + arg
return arg
def make_csr(key_str, domains):
"""Generate a CSR.
letsencrypt/client/network.py
+23 -39
View File
@@ -1,10 +1,8 @@
"""Network Module."""
import json
import logging
import sys
import time
import jsonschema
import requests
from letsencrypt import acme
@@ -32,10 +30,11 @@ class Network(object):
def send(self, msg):
"""Send ACME message to server.
:param dict msg: ACME message (JSON serializable).
:param msg: ACME message.
:type msg: :class:`letsencrypt.acme.messages.Message`
:returns: Server response message.
:rtype: dict
:rtype: :class:`letsencrypt.acme.messages.Message`
:raises TypeError: if `msg` is not JSON serializable
:raises jsonschema.ValidationError: if not valid ACME message
@@ -43,13 +42,10 @@ class Network(object):
or if response from server is not a valid ACME message.
"""
json_encoded = json.dumps(msg)
acme.messages.acme_object_validate(json_encoded)
try:
response = requests.post(
self.server_url,
data=json_encoded,
data=msg.json_dumps(),
headers={"Content-Type": "application/json"},
verify=True
)
@@ -57,67 +53,55 @@ class Network(object):
raise errors.LetsEncryptClientError(
'Sending ACME message to server has failed: %s' % error)
try:
acme.messages.acme_object_validate(response.content)
except ValueError:
raise errors.LetsEncryptClientError(
'Server did not send JSON serializable message')
except jsonschema.ValidationError as error:
raise errors.LetsEncryptClientError(
'Response from server is not a valid ACME message')
return response.json()
return acme.messages.Message.from_json(response.json())
def send_and_receive_expected(self, msg, expected):
"""Send ACME message to server and return expected message.
:param dict msg: ACME message (JSON serializable).
:param str expected: Name of the expected response ACME message type.
:param msg: ACME message.
:type msg: :class:`letsencrypt.acme.Message`
:returns: ACME response message of expected type.
:rtype: dict
:rtype: :class:`letsencrypt.acme.messages.Message`
:raises errors.LetsEncryptClientError: An exception is thrown
"""
response = self.send(msg)
try:
return self.is_expected_msg(response, expected)
except: # TODO: too generic exception
raise errors.LetsEncryptClientError(
'Expected message (%s) not received' % expected)
return self.is_expected_msg(response, expected)
def is_expected_msg(self, response, expected, delay=3, rounds=20):
"""Is response expected ACME message?
:param dict response: ACME response message from server.
:param str expected: Name of the expected response ACME message type.
:param response: ACME response message from server.
:type response: :class:`letsencrypt.acme.messages.Message`
:param expected: Expected response type.
:type expected: subclass of :class:`letsencrypt.acme.messages.Message`
:param int delay: Number of seconds to delay before next round
in case of ACME "defer" response message.
:param int rounds: Number of resend attempts in case of ACME "defer"
response message.
:returns: ACME response message from server.
:rtype: dict
:rtype: :class:`letsencrypt.acme.messages.Message`
:raises LetsEncryptClientError: if server sent ACME "error" message
"""
for _ in xrange(rounds):
if response["type"] == expected:
if isinstance(response, expected):
return response
elif response["type"] == "error":
logging.error(
"%s: %s - More Info: %s", response["error"],
response.get("message", ""), response.get("moreInfo", ""))
raise errors.LetsEncryptClientError(response["error"])
elif response["type"] == "defer":
elif isinstance(response, acme.messages.Error):
logging.error("%s", response)
raise errors.LetsEncryptClientError(response.error)
elif isinstance(response, acme.messages.Defer):
logging.info("Waiting for %d seconds...", delay)
time.sleep(delay)
response = self.send(
acme.messages.status_request(response["token"]))
acme.messages.StatusRequest(response.token))
else:
logging.fatal("Received unexpected message")
logging.fatal("Expected: %s", expected)
letsencrypt/client/revoker.py
+5 -2
View File
@@ -4,6 +4,7 @@ import logging
import os
import shutil
import Crypto.PublicKey.RSA
import M2Crypto
import zope.component
@@ -28,7 +29,7 @@ class Revoker(object):
:param dict cert: TODO
:returns: ACME "revocation" message.
:rtype: dict
:rtype: :class:`letsencrypt.acme.message.Revocation`
"""
cert_der = M2Crypto.X509.load_cert(cert["backup_cert_file"]).as_der()
@@ -36,7 +37,9 @@ class Revoker(object):
key = backup_key_file.read()
revocation = self.network.send_and_receive_expected(
acme.messages.revocation_request(cert_der, key), "revocation")
acme.messages.RevocationRequest.create(
cert_der, Crypto.PublicKey.RSA.importKey(key)),
acme.messages.Revocation)
zope.component.getUtility(interfaces.IDisplay).generic_notification(
"You have successfully revoked the certificate for "
letsencrypt/client/tests/acme_util.py
-16
View File
@@ -94,19 +94,3 @@ def gen_combos(challs):
combos.append([i, j])
return combos
def get_chall_msg(iden, nonce, challenges, combos=None):
"""Produce an ACME challenge message."""
chall_msg = {
"type": "challenge",
"sessionID": iden,
"nonce": nonce,
"challenges": challenges
}
if combos is None:
return chall_msg
chall_msg["combinations"] = combos
return chall_msg
letsencrypt/client/tests/auth_handler_test.py
+15 -11
View File
@@ -3,10 +3,13 @@ import unittest
import mock
from letsencrypt import acme
from letsencrypt.client import errors
from letsencrypt.client.tests import acme_util
TRANSLATE = {
"dvsni": "DvsniChall",
"simpleHttps": "SimpleHttpsChall",
@@ -38,7 +41,7 @@ class SatisfyChallengesTest(unittest.TestCase):
def test_name1_dvsni1(self):
dom = "0"
challenge = [acme_util.CHALLENGES["dvsni"]]
msg = acme_util.get_chall_msg(dom, "nonce0", challenge)
msg = acme.messages.Challenge(dom, "nonce0", challenge)
self.handler.add_chall_msg(dom, msg, "dummy_key")
self.handler._satisfy_challenges() # pylint: disable=protected-access
@@ -57,7 +60,7 @@ class SatisfyChallengesTest(unittest.TestCase):
for i in range(5):
self.handler.add_chall_msg(
str(i),
acme_util.get_chall_msg(str(i), "nonce%d" % i, challenge),
acme.messages.Challenge(str(i), "nonce%d" % i, challenge),
"dummy_key")
self.handler._satisfy_challenges() # pylint: disable=protected-access
@@ -84,7 +87,7 @@ class SatisfyChallengesTest(unittest.TestCase):
combos = acme_util.gen_combos(challenges)
self.handler.add_chall_msg(
dom,
acme_util.get_chall_msg("0", "nonce0", challenges, combos),
acme.messages.Challenge("0", "nonce0", challenges, combos),
"dummy_key")
path = gen_path(["simpleHttps"], challenges)
@@ -113,7 +116,7 @@ class SatisfyChallengesTest(unittest.TestCase):
combos = acme_util.gen_combos(challenges)
self.handler.add_chall_msg(
dom,
acme_util.get_chall_msg(dom, "nonce0", challenges, combos),
acme.messages.Challenge(dom, "nonce0", challenges, combos),
"dummy_key")
path = gen_path(["simpleHttps", "recoveryToken"], challenges)
@@ -143,7 +146,7 @@ class SatisfyChallengesTest(unittest.TestCase):
for i in range(5):
self.handler.add_chall_msg(
str(i),
acme_util.get_chall_msg(
acme.messages.Challenge(
str(i), "nonce%d" % i, challenges, combos),
"dummy_key")
@@ -193,7 +196,7 @@ class SatisfyChallengesTest(unittest.TestCase):
paths.append(gen_path(chosen_chall[i], challenge_list[i]))
self.handler.add_chall_msg(
dom,
acme_util.get_chall_msg(
acme.messages.Challenge(
dom, "nonce%d" % i, challenge_list[i]),
"dummy_key")
@@ -229,7 +232,8 @@ class SatisfyChallengesTest(unittest.TestCase):
self.assertEqual(
type(self.handler.client_c["4"][0].chall).__name__, "RecTokenChall")
def _get_exp_response(self, domain, path, challenges): # pylint: disable=no-self-use
def _get_exp_response(self, domain, path, challenges):
# pylint: disable=no-self-use
exp_resp = ["null"] * len(challenges)
for i in path:
exp_resp[i] = TRANSLATE[challenges[i]["type"]] + str(domain)
@@ -262,7 +266,7 @@ class GetAuthorizationsTest(unittest.TestCase):
for i in range(3):
self.handler.add_chall_msg(
str(i),
acme_util.get_chall_msg(str(i), "nonce%d" % i, challenge),
acme.messages.Challenge(str(i), "nonce%d" % i, challenge),
"dummy_key")
self.mock_sat_chall.side_effect = self._sat_solved_at_once
@@ -290,7 +294,7 @@ class GetAuthorizationsTest(unittest.TestCase):
challenges = acme_util.get_challenges()
self.handler.add_chall_msg(
"0",
acme_util.get_chall_msg("0", "nonce0", challenges),
acme.messages.Challenge("0", "nonce0", challenges),
"dummy_key")
# Don't do anything to satisfy challenges
@@ -305,7 +309,7 @@ class GetAuthorizationsTest(unittest.TestCase):
def _sat_failure(self):
dom = "0"
self.handler.paths[dom] = gen_path(
["dns", "recoveryToken"], self.handler.msgs[dom]["challenges"])
["dns", "recoveryToken"], self.handler.msgs[dom].challenges)
dv_c, c_c = self.handler._challenge_factory(
dom, self.handler.paths[dom])
self.handler.dv_c[dom], self.handler.client_c[dom] = dv_c, c_c
@@ -318,7 +322,7 @@ class GetAuthorizationsTest(unittest.TestCase):
dom = str(i)
self.handler.add_chall_msg(
dom,
acme_util.get_chall_msg(dom, "nonce%d" % i, challs[i]),
acme.messages.Challenge(dom, "nonce%d" % i, challs[i]),
"dummy_key")
self.mock_sat_chall.side_effect = self._sat_incremental
letsencrypt/client/tests/crypto_util_test.py
-37
View File
@@ -11,43 +11,6 @@ RSA256_KEY = pkg_resources.resource_string(__name__, 'testdata/rsa256_key.pem')
RSA512_KEY = pkg_resources.resource_string(__name__, 'testdata/rsa512_key.pem')
class CreateSigTest(unittest.TestCase):
"""Tests for letsencrypt.client.crypto_util.create_sig."""
def setUp(self):
self.nonce = '\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9'
self.b64nonce = '7Nbyb1lI6xPVI3Hg3aKSqQ'
self.signature = {
'nonce': self.b64nonce,
'alg': 'RS256',
'jwk': {
'kty': 'RSA',
'e': 'AQAB',
'n': 'rHVztFHtH92ucFJD_N_HW9AsdRsUuHUBBBDlHwNlRd3fp5'
'80rv2-6QWE30cWgdmJS86ObRz6lUTor4R0T-3C5Q',
},
'sig': 'SUPYKucUnhlTt8_sMxLiigOYdf_wlOLXPI-o7aRLTsOquVjDd6r'
'AX9AFJHk-bCMQPJbSzXKjG6H1IWbvxjS2Ew',
}
@classmethod
def _call(cls, *args, **kwargs):
from letsencrypt.client.crypto_util import create_sig
return create_sig(*args, **kwargs)
def test_it(self):
self.assertEqual(
self._call('message', RSA256_KEY, self.nonce), self.signature)
def test_random_nonce(self):
signature = self._call('message', RSA256_KEY)
signature.pop('sig')
signature.pop('nonce')
del self.signature['sig']
del self.signature['nonce']
self.assertEqual(signature, self.signature)
class ValidCSRTest(unittest.TestCase):
"""Tests for letsencrypt.client.crypto_util.valid_csr."""