mirror of
https://github.com/certbot/certbot.git
synced 2026-07-28 08:05:31 +02:00
Remove old messages and network
This commit is contained in:
Jakub Warmuz
@@ -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")
|
||||
@@ -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<?\xc8W\x94\x94cj(\xe7\xaa$'
|
||||
'\x92\xe9\x96\x11\xc2\xefx\x0bR',
|
||||
nonce='\xab?\x08o\xe6\x81$\x9f\xa1\xc9\x025\x1c\x1b\xa5+')
|
||||
|
||||
from 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.contact,
|
||||
)
|
||||
|
||||
self.jmsg_to = {
|
||||
'type': 'authorizationRequest',
|
||||
'sessionID': 'aefoGaavieG9Wihuk2aufai3aeZ5EeW4',
|
||||
'nonce': '7Nbyb1lI6xPVI3Hg3aKSqQ',
|
||||
'responses': self.responses,
|
||||
'signature': signature,
|
||||
'contact': self.contact,
|
||||
}
|
||||
self.jmsg_from = {
|
||||
'type': 'authorizationRequest',
|
||||
'sessionID': 'aefoGaavieG9Wihuk2aufai3aeZ5EeW4',
|
||||
'nonce': '7Nbyb1lI6xPVI3Hg3aKSqQ',
|
||||
'responses': [None if response is None else response.to_json()
|
||||
for response in self.responses],
|
||||
'signature': signature.to_json(),
|
||||
# TODO: schema validation doesn't recognize tuples as
|
||||
# arrays :(
|
||||
'contact': list(self.contact),
|
||||
}
|
||||
|
||||
def test_create(self):
|
||||
from 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+',
|
||||
contact=self.contact))
|
||||
|
||||
def test_verify(self):
|
||||
self.assertTrue(self.msg.verify('example.com'))
|
||||
|
||||
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 AuthorizationRequest
|
||||
self.assertEqual(
|
||||
self.msg, AuthorizationRequest.from_json(self.jmsg_from))
|
||||
|
||||
def test_json_without_optionals(self):
|
||||
del self.jmsg_from['contact']
|
||||
del self.jmsg_to['contact']
|
||||
|
||||
from acme.messages import AuthorizationRequest
|
||||
msg = AuthorizationRequest.from_json(self.jmsg_from)
|
||||
|
||||
self.assertEqual(msg.contact, ())
|
||||
self.assertEqual(self.jmsg_to, msg.to_partial_json())
|
||||
|
||||
|
||||
class CertificateTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
refresh = 'https://example.com/refresh/Dr8eAwTVQfSS/'
|
||||
|
||||
from acme.messages import Certificate
|
||||
self.msg = Certificate(
|
||||
certificate=CERT, chain=(CERT,), refresh=refresh)
|
||||
|
||||
self.jmsg_to = {
|
||||
'type': 'certificate',
|
||||
'certificate': jose.b64encode(CERT.as_der()),
|
||||
'chain': (jose.b64encode(CERT.as_der()),),
|
||||
'refresh': refresh,
|
||||
}
|
||||
self.jmsg_from = self.jmsg_to.copy()
|
||||
# TODO: schema validation array tuples
|
||||
self.jmsg_from['chain'] = list(self.jmsg_from['chain'])
|
||||
|
||||
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 Certificate
|
||||
self.assertEqual(Certificate.from_json(self.jmsg_from), self.msg)
|
||||
|
||||
def test_json_without_optionals(self):
|
||||
del self.jmsg_from['chain']
|
||||
del self.jmsg_from['refresh']
|
||||
del self.jmsg_to['chain']
|
||||
del self.jmsg_to['refresh']
|
||||
|
||||
from acme.messages import Certificate
|
||||
msg = Certificate.from_json(self.jmsg_from)
|
||||
|
||||
self.assertEqual(msg.chain, ())
|
||||
self.assertTrue(msg.refresh is None)
|
||||
self.assertEqual(self.jmsg_to, msg.to_partial_json())
|
||||
|
||||
|
||||
class CertificateRequestTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
signature = other.Signature(
|
||||
alg=jose.RS256, jwk=jose.JWKRSA(key=KEY.publickey()),
|
||||
sig='\x15\xed\x84\xaa:\xf2DO\x0e9 \xbcg\xf8\xc0\xcf\x87\x9a'
|
||||
'\x95\xeb\xffT[\x84[\xec\x85\x7f\x8eK\xe9\xc2\x12\xc8Q'
|
||||
'\xafo\xc6h\x07\xba\xa6\xdf\xd1\xa7"$\xba=Z\x13n\x14\x0b'
|
||||
'k\xfe\xee\xb4\xe4\xc8\x05\x9a\x08\xa7',
|
||||
nonce='\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9')
|
||||
|
||||
from acme.messages import CertificateRequest
|
||||
self.msg = CertificateRequest(csr=CSR, signature=signature)
|
||||
|
||||
self.jmsg_to = {
|
||||
'type': 'certificateRequest',
|
||||
'csr': jose.b64encode(CSR.as_der()),
|
||||
'signature': signature,
|
||||
}
|
||||
self.jmsg_from = self.jmsg_to.copy()
|
||||
self.jmsg_from['signature'] = self.jmsg_from['signature'].to_json()
|
||||
|
||||
def test_create(self):
|
||||
from acme.messages import CertificateRequest
|
||||
self.assertEqual(self.msg, CertificateRequest.create(
|
||||
csr=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_partial_json(self):
|
||||
self.assertEqual(self.msg.to_partial_json(), self.jmsg_to)
|
||||
|
||||
def test_from_json(self):
|
||||
from acme.messages import CertificateRequest
|
||||
self.assertEqual(self.msg, CertificateRequest.from_json(self.jmsg_from))
|
||||
|
||||
|
||||
class DeferTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
from acme.messages import Defer
|
||||
self.msg = Defer(
|
||||
token='O7-s9MNq1siZHlgrMzi9_A', interval=60,
|
||||
message='Warming up the HSM')
|
||||
|
||||
self.jmsg = {
|
||||
'type': 'defer',
|
||||
'token': 'O7-s9MNq1siZHlgrMzi9_A',
|
||||
'interval': 60,
|
||||
'message': 'Warming up the HSM',
|
||||
}
|
||||
|
||||
def test_to_partial_json(self):
|
||||
self.assertEqual(self.msg.to_partial_json(), self.jmsg)
|
||||
|
||||
def test_from_json(self):
|
||||
from acme.messages import Defer
|
||||
self.assertEqual(Defer.from_json(self.jmsg), self.msg)
|
||||
|
||||
def test_json_without_optionals(self):
|
||||
del self.jmsg['interval']
|
||||
del self.jmsg['message']
|
||||
|
||||
from acme.messages import Defer
|
||||
msg = Defer.from_json(self.jmsg)
|
||||
|
||||
self.assertTrue(msg.interval is None)
|
||||
self.assertTrue(msg.message is None)
|
||||
self.assertEqual(self.jmsg, msg.to_partial_json())
|
||||
|
||||
|
||||
class ErrorTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
from acme.messages import Error
|
||||
self.msg = Error(
|
||||
error='badCSR', message='RSA keys must be at least 2048 bits long',
|
||||
more_info='https://ca.example.com/documentation/csr-requirements')
|
||||
|
||||
self.jmsg = {
|
||||
'type': 'error',
|
||||
'error': 'badCSR',
|
||||
'message':'RSA keys must be at least 2048 bits long',
|
||||
'moreInfo': 'https://ca.example.com/documentation/csr-requirements',
|
||||
}
|
||||
|
||||
def test_to_partial_json(self):
|
||||
self.assertEqual(self.msg.to_partial_json(), self.jmsg)
|
||||
|
||||
def test_from_json(self):
|
||||
from acme.messages import Error
|
||||
self.assertEqual(Error.from_json(self.jmsg), self.msg)
|
||||
|
||||
def test_json_without_optionals(self):
|
||||
del self.jmsg['message']
|
||||
del self.jmsg['moreInfo']
|
||||
|
||||
from acme.messages import Error
|
||||
msg = Error.from_json(self.jmsg)
|
||||
|
||||
self.assertTrue(msg.message is None)
|
||||
self.assertTrue(msg.more_info is None)
|
||||
self.assertEqual(self.jmsg, msg.to_partial_json())
|
||||
|
||||
|
||||
class RevocationTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
from acme.messages import Revocation
|
||||
self.msg = Revocation()
|
||||
self.jmsg = {'type': 'revocation'}
|
||||
|
||||
def test_to_partial_json(self):
|
||||
self.assertEqual(self.msg.to_partial_json(), self.jmsg)
|
||||
|
||||
def test_from_json(self):
|
||||
from acme.messages import Revocation
|
||||
self.assertEqual(Revocation.from_json(self.jmsg), self.msg)
|
||||
|
||||
|
||||
class RevocationRequestTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.sig_nonce = '\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9'
|
||||
|
||||
signature = other.Signature(
|
||||
alg=jose.RS256, jwk=jose.JWKRSA(key=KEY.publickey()),
|
||||
sig='eJ\xfe\x12"U\x87\x8b\xbf/ ,\xdeP\xb2\xdc1\xb00\xe5\x1dB'
|
||||
'\xfch<\xc6\x9eH@!\x1c\x16\xb2\x0b_\xc4\xddP\x89\xc8\xce?'
|
||||
'\x16g\x069I\xb9\xb3\x91\xb9\x0e$3\x9f\x87\x8e\x82\xca\xc5'
|
||||
's\xd9\xd0\xe7',
|
||||
nonce=self.sig_nonce)
|
||||
|
||||
from acme.messages import RevocationRequest
|
||||
self.msg = RevocationRequest(certificate=CERT, signature=signature)
|
||||
|
||||
self.jmsg_to = {
|
||||
'type': 'revocationRequest',
|
||||
'certificate': jose.b64encode(CERT.as_der()),
|
||||
'signature': signature,
|
||||
}
|
||||
self.jmsg_from = self.jmsg_to.copy()
|
||||
self.jmsg_from['signature'] = self.jmsg_from['signature'].to_json()
|
||||
|
||||
def test_create(self):
|
||||
from acme.messages import RevocationRequest
|
||||
self.assertEqual(self.msg, RevocationRequest.create(
|
||||
certificate=CERT, key=KEY, sig_nonce=self.sig_nonce))
|
||||
|
||||
def test_verify(self):
|
||||
self.assertTrue(self.msg.verify())
|
||||
|
||||
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 RevocationRequest
|
||||
self.assertEqual(self.msg, RevocationRequest.from_json(self.jmsg_from))
|
||||
|
||||
|
||||
class StatusRequestTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
from acme.messages import StatusRequest
|
||||
self.msg = StatusRequest(token=u'O7-s9MNq1siZHlgrMzi9_A')
|
||||
self.jmsg = {
|
||||
'type': 'statusRequest',
|
||||
'token': u'O7-s9MNq1siZHlgrMzi9_A',
|
||||
}
|
||||
|
||||
def test_to_partial_json(self):
|
||||
self.assertEqual(self.msg.to_partial_json(), self.jmsg)
|
||||
|
||||
def test_from_json(self):
|
||||
from acme.messages import StatusRequest
|
||||
self.assertEqual(StatusRequest.from_json(self.jmsg), self.msg)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main() # pragma: no cover
|
||||
@@ -1,121 +0,0 @@
|
||||
"""Network Module."""
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
|
||||
import requests
|
||||
|
||||
from acme import jose
|
||||
from acme import messages
|
||||
|
||||
from letsencrypt import errors
|
||||
|
||||
|
||||
# https://urllib3.readthedocs.org/en/latest/security.html#insecureplatformwarning
|
||||
requests.packages.urllib3.contrib.pyopenssl.inject_into_urllib3()
|
||||
|
||||
logging.getLogger("requests").setLevel(logging.WARNING)
|
||||
|
||||
|
||||
class Network(object):
|
||||
"""Class for communicating with ACME servers.
|
||||
|
||||
:ivar str server_url: Full URL of the ACME service
|
||||
|
||||
"""
|
||||
def __init__(self, server):
|
||||
"""Initialize Network instance.
|
||||
|
||||
:param str server: ACME (CA) server[:port]
|
||||
|
||||
"""
|
||||
self.server_url = "https://%s/acme/" % server
|
||||
|
||||
def send(self, msg):
|
||||
"""Send ACME message to server.
|
||||
|
||||
:param msg: ACME message.
|
||||
:type msg: :class:`acme.messages.Message`
|
||||
|
||||
:returns: Server response message.
|
||||
:rtype: :class:`acme.messages.Message`
|
||||
|
||||
:raises acme.errors.ValidationError: if `msg` is not
|
||||
valid serializable ACME JSON message.
|
||||
:raises errors.LetsEncryptClientError: in case of connection error
|
||||
or if response from server is not a valid ACME message.
|
||||
|
||||
"""
|
||||
try:
|
||||
response = requests.post(
|
||||
self.server_url,
|
||||
data=msg.json_dumps(),
|
||||
headers={"Content-Type": "application/json"},
|
||||
verify=True
|
||||
)
|
||||
except requests.exceptions.RequestException as error:
|
||||
raise errors.LetsEncryptClientError(
|
||||
'Sending ACME message to server has failed: %s' % error)
|
||||
|
||||
json_string = response.json()
|
||||
try:
|
||||
return messages.Message.from_json(json_string)
|
||||
except jose.DeserializationError as error:
|
||||
logging.error(json_string)
|
||||
raise # TODO
|
||||
|
||||
def send_and_receive_expected(self, msg, expected):
|
||||
"""Send ACME message to server and return expected message.
|
||||
|
||||
:param msg: ACME message.
|
||||
:type msg: :class:`acme.Message`
|
||||
|
||||
:returns: ACME response message of expected type.
|
||||
:rtype: :class:`acme.messages.Message`
|
||||
|
||||
:raises errors.LetsEncryptClientError: An exception is thrown
|
||||
|
||||
"""
|
||||
response = self.send(msg)
|
||||
return self.is_expected_msg(response, expected)
|
||||
|
||||
|
||||
def is_expected_msg(self, response, expected, delay=3, rounds=20):
|
||||
"""Is response expected ACME message?
|
||||
|
||||
:param response: ACME response message from server.
|
||||
:type response: :class:`acme.messages.Message`
|
||||
|
||||
:param expected: Expected response type.
|
||||
:type expected: subclass of :class:`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: :class:`acme.messages.Message`
|
||||
|
||||
:raises LetsEncryptClientError: if server sent ACME "error" message
|
||||
|
||||
"""
|
||||
for _ in xrange(rounds):
|
||||
if isinstance(response, expected):
|
||||
return response
|
||||
elif isinstance(response, messages.Error):
|
||||
logging.error("%s", response)
|
||||
raise errors.LetsEncryptClientError(response.error)
|
||||
elif isinstance(response, messages.Defer):
|
||||
logging.info("Waiting for %d seconds...", delay)
|
||||
time.sleep(delay)
|
||||
response = self.send(
|
||||
messages.StatusRequest(token=response.token))
|
||||
else:
|
||||
logging.fatal("Received unexpected message")
|
||||
logging.fatal("Expected: %s", expected)
|
||||
logging.fatal("Received: %s", response)
|
||||
sys.exit(33)
|
||||
|
||||
logging.error(
|
||||
"Server has deferred past the max of %d seconds", rounds * delay)
|
||||
Reference in New Issue
Block a user