mirror of
https://github.com/certbot/certbot.git
synced 2026-08-03 08:03:10 +02:00
Merge pull request #325 from kuba/network2
Restified messages2 and network2
This commit is contained in:
James Kasten
@@ -1,6 +1,8 @@
|
||||
:mod:`letsencrypt.acme`
|
||||
=======================
|
||||
|
||||
.. contents::
|
||||
|
||||
.. automodule:: letsencrypt.acme
|
||||
:members:
|
||||
|
||||
@@ -8,9 +10,18 @@
|
||||
Messages
|
||||
--------
|
||||
|
||||
v00
|
||||
~~~
|
||||
|
||||
.. automodule:: letsencrypt.acme.messages
|
||||
:members:
|
||||
|
||||
v02
|
||||
~~~
|
||||
|
||||
.. automodule:: letsencrypt.acme.messages2
|
||||
:members:
|
||||
|
||||
|
||||
Challenges
|
||||
----------
|
||||
@@ -21,10 +32,18 @@ Challenges
|
||||
|
||||
Other ACME objects
|
||||
------------------
|
||||
|
||||
.. automodule:: letsencrypt.acme.other
|
||||
:members:
|
||||
|
||||
|
||||
Fields
|
||||
------
|
||||
|
||||
.. automodule:: letsencrypt.acme.fields
|
||||
:members:
|
||||
|
||||
|
||||
Errors
|
||||
------
|
||||
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
:mod:`letsencrypt.client.network2`
|
||||
----------------------------------
|
||||
|
||||
.. automodule:: letsencrypt.client.network2
|
||||
:members:
|
||||
@@ -0,0 +1,42 @@
|
||||
import logging
|
||||
import os
|
||||
import pkg_resources
|
||||
|
||||
import M2Crypto
|
||||
|
||||
from letsencrypt.acme import messages2
|
||||
from letsencrypt.acme import jose
|
||||
|
||||
from letsencrypt.client import network2
|
||||
|
||||
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel(logging.DEBUG)
|
||||
|
||||
NEW_REG_URL = 'https://www.letsencrypt-demo.org/acme/new-reg'
|
||||
|
||||
key = jose.JWKRSA.load(pkg_resources.resource_string(
|
||||
'letsencrypt.acme.jose', os.path.join('testdata', 'rsa512_key.pem')))
|
||||
net = network2.Network(NEW_REG_URL, key)
|
||||
|
||||
regr = net.register(contact=(
|
||||
'mailto:cert-admin@example.com', 'tel:+12025551212'))
|
||||
logging.info('Auto-accepting TOS: %s', regr.terms_of_service)
|
||||
net.update_registration(regr.update(
|
||||
body=regr.body.update(agreement=regr.terms_of_service)))
|
||||
logging.debug(regr)
|
||||
|
||||
authzr = net.request_challenges(
|
||||
identifier=messages2.Identifier(
|
||||
typ=messages2.IDENTIFIER_FQDN, value='example1.com'),
|
||||
regr=regr)
|
||||
logging.debug(authzr)
|
||||
|
||||
authzr, authzr_response = net.poll(authzr)
|
||||
|
||||
csr = M2Crypto.X509.load_request_string(pkg_resources.resource_string(
|
||||
'letsencrypt.client.tests', os.path.join('testdata', 'csr.pem')))
|
||||
try:
|
||||
net.request_issuance(csr, (authzr,))
|
||||
except messages2.Error as error:
|
||||
print error.detail
|
||||
@@ -0,0 +1,25 @@
|
||||
"""ACME JSON fields."""
|
||||
import pyrfc3339
|
||||
|
||||
from letsencrypt.acme import jose
|
||||
|
||||
|
||||
class RFC3339Field(jose.Field):
|
||||
"""RFC3339 field encoder/decoder.
|
||||
|
||||
Handles decoding/encoding between RFC3339 strings and aware (not
|
||||
naive) `datetime.datetime` objects
|
||||
(e.g. ``datetime.datetime.now(pytz.utc)``).
|
||||
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def default_encoder(cls, value):
|
||||
return pyrfc3339.generate(value)
|
||||
|
||||
@classmethod
|
||||
def default_decoder(cls, value):
|
||||
try:
|
||||
return pyrfc3339.parse(value)
|
||||
except ValueError as error:
|
||||
raise jose.DeserializationError(error)
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Tests for letsencrypt.acme.fields."""
|
||||
import datetime
|
||||
import unittest
|
||||
|
||||
import pytz
|
||||
|
||||
from letsencrypt.acme import jose
|
||||
|
||||
|
||||
class RFC3339FieldTest(unittest.TestCase):
|
||||
"""Tests for letsencrypt.acme.fields.RFC3339Field."""
|
||||
|
||||
def setUp(self):
|
||||
self.decoded = datetime.datetime(2015, 3, 27, tzinfo=pytz.utc)
|
||||
self.encoded = '2015-03-27T00:00:00Z'
|
||||
|
||||
def test_default_encoder(self):
|
||||
from letsencrypt.acme.fields import RFC3339Field
|
||||
self.assertEqual(
|
||||
self.encoded, RFC3339Field.default_encoder(self.decoded))
|
||||
|
||||
def test_default_encoder_naive_fails(self):
|
||||
from letsencrypt.acme.fields import RFC3339Field
|
||||
self.assertRaises(
|
||||
ValueError, RFC3339Field.default_encoder, datetime.datetime.now())
|
||||
|
||||
def test_default_decoder(self):
|
||||
from letsencrypt.acme.fields import RFC3339Field
|
||||
self.assertEqual(
|
||||
self.decoded, RFC3339Field.default_decoder(self.encoded))
|
||||
|
||||
def test_default_decoder_raises_deserialization_error(self):
|
||||
from letsencrypt.acme.fields import RFC3339Field
|
||||
self.assertRaises(
|
||||
jose.DeserializationError, RFC3339Field.default_decoder, '')
|
||||
@@ -129,18 +129,24 @@ class JSONDeSerializable(object):
|
||||
:returns: Fully serialized object.
|
||||
|
||||
"""
|
||||
partial = self.to_json()
|
||||
try_serialize = (lambda x: x.fully_serialize()
|
||||
if isinstance(x, JSONDeSerializable) else x)
|
||||
if isinstance(partial, basestring): # strings are sequences
|
||||
return partial
|
||||
if isinstance(partial, collections.Sequence):
|
||||
return [try_serialize(elem) for elem in partial]
|
||||
elif isinstance(partial, collections.Mapping):
|
||||
return dict([(try_serialize(key), try_serialize(value))
|
||||
for key, value in partial.iteritems()])
|
||||
else:
|
||||
return partial
|
||||
def _serialize(obj):
|
||||
if isinstance(obj, JSONDeSerializable):
|
||||
return _serialize(obj.to_json())
|
||||
if isinstance(obj, basestring): # strings are sequence
|
||||
return obj
|
||||
elif isinstance(obj, list):
|
||||
return [_serialize(subobj) for subobj in obj]
|
||||
elif isinstance(obj, collections.Sequence):
|
||||
# default to tuple, otherwise Mapping could get
|
||||
# unhashable list
|
||||
return tuple(_serialize(subobj) for subobj in obj)
|
||||
elif isinstance(obj, collections.Mapping):
|
||||
return dict((_serialize(key), _serialize(value))
|
||||
for key, value in obj.iteritems())
|
||||
else:
|
||||
return obj
|
||||
|
||||
return _serialize(self)
|
||||
|
||||
@util.abstractclassmethod
|
||||
def from_json(cls, unused_jobj):
|
||||
|
||||
@@ -3,6 +3,7 @@ import unittest
|
||||
|
||||
|
||||
class JSONDeSerializableTest(unittest.TestCase):
|
||||
# pylint: disable=too-many-instance-attributes
|
||||
|
||||
def setUp(self):
|
||||
from letsencrypt.acme.jose.interfaces import JSONDeSerializable
|
||||
@@ -50,6 +51,8 @@ class JSONDeSerializableTest(unittest.TestCase):
|
||||
self.basic2 = Basic('foo2')
|
||||
self.seq = Sequence(self.basic1, self.basic2)
|
||||
self.mapping = Mapping(self.basic1, self.basic2)
|
||||
self.nested = Basic([[self.basic1]])
|
||||
self.tuple = Basic(('foo',))
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
self.Basic = Basic
|
||||
@@ -66,6 +69,12 @@ class JSONDeSerializableTest(unittest.TestCase):
|
||||
mock_value = object()
|
||||
self.assertTrue(self.Basic(mock_value).fully_serialize() is mock_value)
|
||||
|
||||
def test_fully_serialize_nested(self):
|
||||
self.assertEqual(self.nested.fully_serialize(), [['foo1']])
|
||||
|
||||
def test_fully_serialize(self):
|
||||
self.assertEqual(self.tuple.fully_serialize(), (('foo', )))
|
||||
|
||||
def test_from_json_not_implemented(self):
|
||||
from letsencrypt.acme.jose.interfaces import JSONDeSerializable
|
||||
self.assertRaises(TypeError, JSONDeSerializable.from_json, 'xxx')
|
||||
|
||||
@@ -77,6 +77,12 @@ class ImmutableMap(collections.Mapping, collections.Hashable):
|
||||
for slot in self.__slots__:
|
||||
object.__setattr__(self, slot, kwargs.pop(slot))
|
||||
|
||||
def update(self, **kwargs):
|
||||
"""Return updated map."""
|
||||
items = dict(self)
|
||||
items.update(kwargs)
|
||||
return type(self)(**items) # pylint: disable=star-args
|
||||
|
||||
def __getitem__(self, key):
|
||||
try:
|
||||
return getattr(self, key)
|
||||
|
||||
@@ -54,6 +54,10 @@ class ImmutableMapTest(unittest.TestCase):
|
||||
self.a2 = self.A(x=3, y=4)
|
||||
self.b = self.B(x=1, y=2)
|
||||
|
||||
def test_update(self):
|
||||
self.assertEqual(self.A(x=2, y=2), self.a1.update(x=2))
|
||||
self.assertEqual(self.a2, self.a1.update(x=3, y=4))
|
||||
|
||||
def test_get_missing_item_raises_key_error(self):
|
||||
self.assertRaises(KeyError, self.a1.__getitem__, 'z')
|
||||
|
||||
|
||||
@@ -0,0 +1,298 @@
|
||||
"""ACME protocol v02 messages."""
|
||||
from letsencrypt.acme import challenges
|
||||
from letsencrypt.acme import fields
|
||||
from letsencrypt.acme import jose
|
||||
|
||||
|
||||
class Error(jose.JSONObjectWithFields, Exception):
|
||||
"""ACME error.
|
||||
|
||||
https://tools.ietf.org/html/draft-ietf-appsawg-http-problem-00
|
||||
|
||||
"""
|
||||
|
||||
ERROR_TYPE_NAMESPACE = 'urn:acme:error:'
|
||||
ERROR_TYPE_DESCRIPTIONS = {
|
||||
'malformed': 'The request message was malformed',
|
||||
'unauthorized': 'The client lacks sufficient authorization',
|
||||
'serverInternal': 'The server experienced an internal error',
|
||||
'badCSR': 'The CSR is unacceptable (e.g., due to a short key)',
|
||||
}
|
||||
|
||||
# TODO: Boulder omits 'type' and 'instance', spec requires
|
||||
typ = jose.Field('type', omitempty=True)
|
||||
title = jose.Field('title', omitempty=True)
|
||||
detail = jose.Field('detail')
|
||||
instance = jose.Field('instance', omitempty=True)
|
||||
|
||||
@typ.encoder
|
||||
def typ(value): # pylint: disable=missing-docstring,no-self-argument
|
||||
return Error.ERROR_TYPE_NAMESPACE + value
|
||||
|
||||
@typ.decoder
|
||||
def typ(value): # pylint: disable=missing-docstring,no-self-argument
|
||||
# pylint thinks isinstance(value, Error), so startswith is not found
|
||||
# pylint: disable=no-member
|
||||
if not value.startswith(Error.ERROR_TYPE_NAMESPACE):
|
||||
raise jose.DeserializationError('Missing error type prefix')
|
||||
|
||||
without_prefix = value[len(Error.ERROR_TYPE_NAMESPACE):]
|
||||
if without_prefix not in Error.ERROR_TYPE_DESCRIPTIONS:
|
||||
raise jose.DeserializationError('Error type not recognized')
|
||||
|
||||
return without_prefix
|
||||
|
||||
@property
|
||||
def description(self):
|
||||
"""Hardcoded error description based on its type."""
|
||||
return self.ERROR_TYPE_DESCRIPTIONS[self.typ]
|
||||
|
||||
|
||||
class _Constant(jose.JSONDeSerializable):
|
||||
"""ACME constant."""
|
||||
__slots__ = ('name',)
|
||||
POSSIBLE_NAMES = NotImplemented
|
||||
|
||||
def __init__(self, name):
|
||||
self.POSSIBLE_NAMES[name] = self
|
||||
self.name = name
|
||||
|
||||
def to_json(self):
|
||||
return self.name
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, value):
|
||||
if value not in cls.POSSIBLE_NAMES:
|
||||
raise jose.DeserializationError(
|
||||
'{0} not recognized'.format(cls.__name__))
|
||||
return cls.POSSIBLE_NAMES[value]
|
||||
|
||||
def __repr__(self):
|
||||
return '{0}({1})'.format(self.__class__.__name__, self.name)
|
||||
|
||||
def __eq__(self, other):
|
||||
return isinstance(other, type(self)) and other.name == self.name
|
||||
|
||||
|
||||
class Status(_Constant):
|
||||
"""ACME "status" field."""
|
||||
POSSIBLE_NAMES = {}
|
||||
STATUS_UNKNOWN = Status('unknown')
|
||||
STATUS_PENDING = Status('pending')
|
||||
STATUS_PROCESSING = Status('processing')
|
||||
STATUS_VALID = Status('valid')
|
||||
STATUS_INVALID = Status('invalid')
|
||||
STATUS_REVOKED = Status('revoked')
|
||||
|
||||
|
||||
class IdentifierType(_Constant):
|
||||
"""ACME identifier type."""
|
||||
POSSIBLE_NAMES = {}
|
||||
IDENTIFIER_FQDN = IdentifierType('dns') # IdentifierDNS in Boulder
|
||||
|
||||
|
||||
class Identifier(jose.JSONObjectWithFields):
|
||||
"""ACME identifier.
|
||||
|
||||
:ivar letsencrypt.acme.messages2.IdentifierType typ:
|
||||
|
||||
"""
|
||||
typ = jose.Field('type', decoder=IdentifierType.from_json)
|
||||
value = jose.Field('value')
|
||||
|
||||
|
||||
class Resource(jose.ImmutableMap):
|
||||
"""ACME Resource.
|
||||
|
||||
:ivar letsencrypt.acme.messages2.ResourceBody body: Resource body.
|
||||
:ivar str uri: Location of the resource.
|
||||
|
||||
"""
|
||||
__slots__ = ('body', 'uri')
|
||||
|
||||
|
||||
class ResourceBody(jose.JSONObjectWithFields):
|
||||
"""ACME Resource Body."""
|
||||
|
||||
|
||||
class RegistrationResource(Resource):
|
||||
"""Registration Resource.
|
||||
|
||||
:ivar letsencrypt.acme.messages2.Registration body:
|
||||
:ivar str new_authzr_uri: URI found in the 'next' ``Link`` header
|
||||
:ivar str terms_of_service: URL for the CA TOS.
|
||||
|
||||
"""
|
||||
__slots__ = ('body', 'uri', 'new_authzr_uri', 'terms_of_service')
|
||||
|
||||
|
||||
class Registration(ResourceBody):
|
||||
"""Registration Resource Body.
|
||||
|
||||
:ivar letsencrypt.acme.jose.jwk.JWK key: Public key.
|
||||
:ivar tuple contact:
|
||||
|
||||
"""
|
||||
|
||||
# on new-reg key server ignores 'key' and populates it based on
|
||||
# JWS.signature.combined.jwk
|
||||
key = jose.Field('key', omitempty=True, decoder=jose.JWK.from_json)
|
||||
contact = jose.Field('contact', omitempty=True, default=())
|
||||
recovery_token = jose.Field('recoveryToken', omitempty=True)
|
||||
agreement = jose.Field('agreement', omitempty=True)
|
||||
|
||||
|
||||
class ChallengeResource(Resource, jose.JSONObjectWithFields):
|
||||
"""Challenge Resource.
|
||||
|
||||
:ivar letsencrypt.acme.messages2.ChallengeBody body:
|
||||
:ivar str authzr_uri: URI found in the 'up' ``Link`` header.
|
||||
|
||||
"""
|
||||
__slots__ = ('body', 'authzr_uri')
|
||||
|
||||
@property
|
||||
def uri(self): # pylint: disable=missing-docstring,no-self-argument
|
||||
# bug? 'method already defined line None'
|
||||
# pylint: disable=function-redefined
|
||||
return self.body.uri
|
||||
|
||||
|
||||
class ChallengeBody(ResourceBody):
|
||||
"""Challenge Resource Body.
|
||||
|
||||
.. todo::
|
||||
Confusingly, this has a similar name to `.challenges.Challenge`,
|
||||
as well as `.achallenges.AnnotatedChallenge` or
|
||||
`.achallenges.Indexed`... Once `messages2` and `network2` is
|
||||
integrated with the rest of the client, this class functionality
|
||||
will be merged with `.challenges.Challenge`. Meanwhile,
|
||||
separation allows the ``master`` to be still interoperable with
|
||||
Node.js server (protocol v00). For the time being use names such
|
||||
as ``challb`` to distinguish instances of this class from
|
||||
``achall`` or ``ichall``.
|
||||
|
||||
:ivar letsencrypt.acme.messages2.Status status:
|
||||
:ivar datetime.datetime validated:
|
||||
|
||||
"""
|
||||
|
||||
__slots__ = ('chall',)
|
||||
uri = jose.Field('uri')
|
||||
status = jose.Field('status', decoder=Status.from_json)
|
||||
validated = fields.RFC3339Field('validated', omitempty=True)
|
||||
|
||||
def to_json(self):
|
||||
jobj = super(ChallengeBody, self).to_json()
|
||||
jobj.update(self.chall.to_json())
|
||||
return jobj
|
||||
|
||||
@classmethod
|
||||
def fields_from_json(cls, jobj):
|
||||
jobj_fields = super(ChallengeBody, cls).fields_from_json(jobj)
|
||||
jobj_fields['chall'] = challenges.Challenge.from_json(jobj)
|
||||
return jobj_fields
|
||||
|
||||
|
||||
class AuthorizationResource(Resource):
|
||||
"""Authorization Resource.
|
||||
|
||||
:ivar letsencrypt.acme.messages2.Authorization body:
|
||||
:ivar str new_cert_uri: URI found in the 'next' ``Link`` header
|
||||
|
||||
"""
|
||||
__slots__ = ('body', 'uri', 'new_cert_uri')
|
||||
|
||||
|
||||
class Authorization(ResourceBody):
|
||||
"""Authorization Resource Body.
|
||||
|
||||
:ivar letsencrypt.acme.messages2.Identifier identifier:
|
||||
:ivar list challenges: `list` of `Challenge`
|
||||
:ivar tuple combinations: Challenge combinations (`tuple` of `tuple`
|
||||
of `int`, as opposed to `list` of `list` from the spec).
|
||||
:ivar letsencrypt.acme.jose.jwk.JWK key: Public key.
|
||||
:ivar tuple contact:
|
||||
:ivar letsencrypt.acme.messages2.Status status:
|
||||
:ivar datetime.datetime expires:
|
||||
|
||||
"""
|
||||
|
||||
identifier = jose.Field('identifier', decoder=Identifier.from_json)
|
||||
challenges = jose.Field('challenges', omitempty=True)
|
||||
combinations = jose.Field('combinations', omitempty=True)
|
||||
|
||||
# TODO: acme-spec #92, #98
|
||||
key = Registration._fields['key']
|
||||
contact = Registration._fields['contact']
|
||||
|
||||
status = jose.Field('status', omitempty=True, decoder=Status.from_json)
|
||||
# TODO: 'expires' is allowed for Authorization Resources in
|
||||
# general, but for Key Authorization '[t]he "expires" field MUST
|
||||
# be absent'... then acme-spec gives example with 'expires'
|
||||
# present... That's confusing!
|
||||
expires = fields.RFC3339Field('expires', omitempty=True)
|
||||
|
||||
@challenges.decoder
|
||||
def challenges(value): # pylint: disable=missing-docstring,no-self-argument
|
||||
return tuple(ChallengeBody.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)
|
||||
|
||||
|
||||
class CertificateRequest(jose.JSONObjectWithFields):
|
||||
"""ACME new-cert request.
|
||||
|
||||
:ivar letsencrypt.acme.jose.util.ComparableX509 csr:
|
||||
`M2Crypto.X509.Request` wrapped in `.ComparableX509`
|
||||
:ivar tuple authorizations: `tuple` of URIs (`str`)
|
||||
|
||||
"""
|
||||
csr = jose.Field('csr', decoder=jose.decode_csr, encoder=jose.encode_csr)
|
||||
authorizations = jose.Field('authorizations', decoder=tuple)
|
||||
|
||||
|
||||
class CertificateResource(Resource):
|
||||
"""Certificate Resource.
|
||||
|
||||
:ivar letsencrypt.acme.jose.util.ComparableX509 body:
|
||||
`M2Crypto.X509.X509` wrapped in `.ComparableX509`
|
||||
:ivar str cert_chain_uri: URI found in the 'up' ``Link`` header
|
||||
:ivar tuple authzrs: `tuple` of `AuthorizationResource`.
|
||||
|
||||
"""
|
||||
__slots__ = ('body', 'uri', 'cert_chain_uri', 'authzrs')
|
||||
|
||||
|
||||
class Revocation(jose.JSONObjectWithFields):
|
||||
"""Revocation message.
|
||||
|
||||
:ivar revoke: Either a `datetime.datetime` or `Revocation.NOW`.
|
||||
:ivar tuple authorizations: Same as `CertificateRequest.authorizations`
|
||||
|
||||
"""
|
||||
|
||||
NOW = 'now'
|
||||
"""A possible value for `revoke`, denoting that certificate should
|
||||
be revoked now."""
|
||||
|
||||
revoke = jose.Field('revoke')
|
||||
authorizations = CertificateRequest._fields['authorizations']
|
||||
|
||||
@revoke.decoder
|
||||
def revoke(value): # pylint: disable=missing-docstring,no-self-argument
|
||||
if value == Revocation.NOW:
|
||||
return value
|
||||
else:
|
||||
return fields.RFC3339Field.default_decoder(value)
|
||||
|
||||
@revoke.encoder
|
||||
def revoke(value): # pylint: disable=missing-docstring,no-self-argument
|
||||
if value == Revocation.NOW:
|
||||
return value
|
||||
else:
|
||||
return fields.RFC3339Field.default_encoder(value)
|
||||
@@ -0,0 +1,172 @@
|
||||
"""Tests for letsencrypt.acme.messages2."""
|
||||
import datetime
|
||||
import unittest
|
||||
|
||||
import mock
|
||||
import pytz
|
||||
|
||||
from letsencrypt.acme import challenges
|
||||
from letsencrypt.acme import jose
|
||||
|
||||
|
||||
class ErrorTest(unittest.TestCase):
|
||||
"""Tests for letsencrypt.acme.messages2.Error."""
|
||||
|
||||
def setUp(self):
|
||||
from letsencrypt.acme.messages2 import Error
|
||||
self.error = Error(detail='foo', typ='malformed')
|
||||
|
||||
def test_typ_prefix(self):
|
||||
self.assertEqual('malformed', self.error.typ)
|
||||
self.assertEqual(
|
||||
'urn:acme:error:malformed', self.error.to_json()['type'])
|
||||
self.assertEqual(
|
||||
'malformed', self.error.from_json(self.error.to_json()).typ)
|
||||
|
||||
def test_typ_decoder_missing_prefix(self):
|
||||
from letsencrypt.acme.messages2 import Error
|
||||
self.assertRaises(jose.DeserializationError, Error.from_json,
|
||||
{'detail': 'foo', 'type': 'malformed'})
|
||||
self.assertRaises(jose.DeserializationError, Error.from_json,
|
||||
{'detail': 'foo', 'type': 'not valid bare type'})
|
||||
|
||||
def test_typ_decoder_not_recognized(self):
|
||||
from letsencrypt.acme.messages2 import Error
|
||||
self.assertRaises(jose.DeserializationError, Error.from_json,
|
||||
{'detail': 'foo', 'type': 'urn:acme:error:baz'})
|
||||
|
||||
def test_description(self):
|
||||
self.assertEqual(
|
||||
'The request message was malformed', self.error.description)
|
||||
|
||||
|
||||
class ConstantTest(unittest.TestCase):
|
||||
"""Tests for letsencrypt.acme.messages2._Constant."""
|
||||
|
||||
def setUp(self):
|
||||
from letsencrypt.acme.messages2 import _Constant
|
||||
class MockConstant(_Constant): # pylint: disable=missing-docstring
|
||||
POSSIBLE_NAMES = {}
|
||||
|
||||
self.MockConstant = MockConstant # pylint: disable=invalid-name
|
||||
self.const_a = MockConstant('a')
|
||||
self.const_b = MockConstant('b')
|
||||
|
||||
def test_to_json(self):
|
||||
self.assertEqual('a', self.const_a.to_json())
|
||||
self.assertEqual('b', self.const_b.to_json())
|
||||
|
||||
def test_from_json(self):
|
||||
self.assertEqual(self.const_a, self.MockConstant.from_json('a'))
|
||||
self.assertRaises(
|
||||
jose.DeserializationError, self.MockConstant.from_json, 'c')
|
||||
|
||||
def test_repr(self):
|
||||
self.assertEqual('MockConstant(a)', repr(self.const_a))
|
||||
self.assertEqual('MockConstant(b)', repr(self.const_b))
|
||||
|
||||
|
||||
class ChallengeResourceTest(unittest.TestCase):
|
||||
"""Tests for letsencrypt.acme.messages2.ChallengeResource."""
|
||||
|
||||
def test_uri(self):
|
||||
from letsencrypt.acme.messages2 import ChallengeResource
|
||||
self.assertEqual('http://challb', ChallengeResource(body=mock.MagicMock(
|
||||
uri='http://challb'), authzr_uri='http://authz').uri)
|
||||
|
||||
|
||||
class ChallengeBodyTest(unittest.TestCase):
|
||||
"""Tests for letsencrypt.acme.messages2.ChallengeBody."""
|
||||
|
||||
def setUp(self):
|
||||
self.chall = challenges.DNS(token='foo')
|
||||
|
||||
from letsencrypt.acme.messages2 import ChallengeBody
|
||||
from letsencrypt.acme.messages2 import STATUS_VALID
|
||||
self.status = STATUS_VALID
|
||||
self.challb = ChallengeBody(
|
||||
uri='http://challb', chall=self.chall, status=self.status)
|
||||
|
||||
self.jobj_to = {
|
||||
'uri': 'http://challb',
|
||||
'status': self.status,
|
||||
'type': 'dns',
|
||||
'token': 'foo',
|
||||
}
|
||||
self.jobj_from = self.jobj_to.copy()
|
||||
self.jobj_from['status'] = 'valid'
|
||||
|
||||
def test_to_json(self):
|
||||
self.assertEqual(self.jobj_to, self.challb.to_json())
|
||||
|
||||
def test_fields_from_json(self):
|
||||
from letsencrypt.acme.messages2 import ChallengeBody
|
||||
self.assertEqual(self.challb, ChallengeBody.from_json(self.jobj_from))
|
||||
|
||||
|
||||
class AuthorizationTest(unittest.TestCase):
|
||||
"""Tests for letsencrypt.acme.messages2.Authorization."""
|
||||
|
||||
def setUp(self):
|
||||
from letsencrypt.acme.messages2 import ChallengeBody
|
||||
from letsencrypt.acme.messages2 import STATUS_VALID
|
||||
self.challbs = (
|
||||
ChallengeBody(
|
||||
uri='http://challb1', status=STATUS_VALID,
|
||||
chall=challenges.SimpleHTTPS(token='IlirfxKKXAsHtmzK29Pj8A')),
|
||||
ChallengeBody(uri='http://challb2', status=STATUS_VALID,
|
||||
chall=challenges.DNS(token='DGyRejmCefe7v4NfDGDKfA')),
|
||||
ChallengeBody(uri='http://challb3', status=STATUS_VALID,
|
||||
chall=challenges.RecoveryToken()),
|
||||
)
|
||||
combinations = ((0, 2), (1, 2))
|
||||
|
||||
from letsencrypt.acme.messages2 import Authorization
|
||||
from letsencrypt.acme.messages2 import Identifier
|
||||
from letsencrypt.acme.messages2 import IDENTIFIER_FQDN
|
||||
identifier = Identifier(typ=IDENTIFIER_FQDN, value='example.com')
|
||||
self.authz = Authorization(
|
||||
identifier=identifier, combinations=combinations,
|
||||
challenges=self.challbs)
|
||||
|
||||
self.jobj_from = {
|
||||
'identifier': identifier.fully_serialize(),
|
||||
'challenges': [challb.fully_serialize() for challb in self.challbs],
|
||||
'combinations': combinations,
|
||||
}
|
||||
|
||||
def test_from_json(self):
|
||||
from letsencrypt.acme.messages2 import Authorization
|
||||
Authorization.from_json(self.jobj_from)
|
||||
|
||||
def test_resolved_combinations(self):
|
||||
self.assertEqual(self.authz.resolved_combinations, (
|
||||
(self.challbs[0], self.challbs[2]),
|
||||
(self.challbs[1], self.challbs[2]),
|
||||
))
|
||||
|
||||
|
||||
class RevocationTest(unittest.TestCase):
|
||||
"""Tests for letsencrypt.acme.messages2.RevocationTest."""
|
||||
|
||||
def setUp(self):
|
||||
from letsencrypt.acme.messages2 import Revocation
|
||||
self.rev_now = Revocation(authorizations=(), revoke=Revocation.NOW)
|
||||
self.rev_date = Revocation(authorizations=(), revoke=datetime.datetime(
|
||||
2015, 3, 27, tzinfo=pytz.utc))
|
||||
self.jobj_now = {'authorizations': (), 'revoke': Revocation.NOW}
|
||||
self.jobj_date = {'authorizations': (),
|
||||
'revoke': '2015-03-27T00:00:00Z'}
|
||||
|
||||
def test_revoke_decoder(self):
|
||||
from letsencrypt.acme.messages2 import Revocation
|
||||
self.assertEqual(self.rev_now, Revocation.from_json(self.jobj_now))
|
||||
self.assertEqual(self.rev_date, Revocation.from_json(self.jobj_date))
|
||||
|
||||
def test_revoke_encoder(self):
|
||||
self.assertEqual(self.jobj_now, self.rev_now.to_json())
|
||||
self.assertEqual(self.jobj_date, self.rev_date.to_json())
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -5,6 +5,14 @@ class LetsEncryptClientError(Exception):
|
||||
"""Generic Let's Encrypt client error."""
|
||||
|
||||
|
||||
class NetworkError(LetsEncryptClientError):
|
||||
"""Network error."""
|
||||
|
||||
|
||||
class UnexpectedUpdate(NetworkError):
|
||||
"""Unexpected update."""
|
||||
|
||||
|
||||
class LetsEncryptReverterError(LetsEncryptClientError):
|
||||
"""Let's Encrypt Reverter error."""
|
||||
|
||||
|
||||
@@ -0,0 +1,506 @@
|
||||
"""Networking for ACME protocol v02."""
|
||||
import datetime
|
||||
import heapq
|
||||
import httplib
|
||||
import itertools
|
||||
import logging
|
||||
import time
|
||||
|
||||
import M2Crypto
|
||||
import requests
|
||||
import werkzeug
|
||||
|
||||
from letsencrypt.acme import jose
|
||||
from letsencrypt.acme import messages2
|
||||
|
||||
from letsencrypt.client import errors
|
||||
|
||||
|
||||
# https://urllib3.readthedocs.org/en/latest/security.html#insecureplatformwarning
|
||||
requests.packages.urllib3.contrib.pyopenssl.inject_into_urllib3()
|
||||
|
||||
|
||||
class Network(object):
|
||||
"""ACME networking.
|
||||
|
||||
.. todo::
|
||||
Clean up raised error types hierarchy, document, and handle (wrap)
|
||||
instances of `.DeserializationError` raised in `from_json()``.
|
||||
|
||||
:ivar str new_reg_uri: Location of new-reg
|
||||
:ivar key: `.JWK` (private)
|
||||
:ivar alg: `.JWASignature`
|
||||
|
||||
"""
|
||||
|
||||
DER_CONTENT_TYPE = 'application/pkix-cert'
|
||||
JSON_CONTENT_TYPE = 'application/json'
|
||||
JSON_ERROR_CONTENT_TYPE = 'application/problem+json'
|
||||
|
||||
def __init__(self, new_reg_uri, key, alg=jose.RS256):
|
||||
self.new_reg_uri = new_reg_uri
|
||||
self.key = key
|
||||
self.alg = alg
|
||||
|
||||
def _wrap_in_jws(self, obj):
|
||||
"""Wrap `JSONDeSerializable` object in JWS.
|
||||
|
||||
:rtype: `.JWS`
|
||||
|
||||
"""
|
||||
dumps = obj.json_dumps()
|
||||
logging.debug('Serialized JSON: %s', dumps)
|
||||
return jose.JWS.sign(
|
||||
payload=dumps, key=self.key, alg=self.alg).json_dumps()
|
||||
|
||||
@classmethod
|
||||
def _check_response(cls, response, content_type=None):
|
||||
"""Check response content and its type.
|
||||
|
||||
.. note::
|
||||
Checking is not strict: wrong server response ``Content-Type``
|
||||
HTTP header is ignored if response is an expected JSON object
|
||||
(c.f. Boulder #56).
|
||||
|
||||
:param str content_type: Expected Content-Type response header.
|
||||
If JSON is expected and not present in server response, this
|
||||
function will raise an error. Otherwise, wrong Content-Type
|
||||
is ignored, but logged.
|
||||
|
||||
:raises letsencrypt.messages2.Error: If server response body
|
||||
carries HTTP Problem (draft-ietf-appsawg-http-problem-00).
|
||||
:raises letsencrypt.errors.NetworkError: In case of other
|
||||
networking errors.
|
||||
|
||||
"""
|
||||
response_ct = response.headers.get('Content-Type')
|
||||
|
||||
try:
|
||||
# TODO: response.json() is called twice, once here, and
|
||||
# once in _get and _post clients
|
||||
jobj = response.json()
|
||||
except ValueError as error:
|
||||
jobj = None
|
||||
|
||||
if not response.ok:
|
||||
if jobj is not None:
|
||||
if response_ct != cls.JSON_ERROR_CONTENT_TYPE:
|
||||
logging.debug(
|
||||
'Ignoring wrong Content-Type (%r) for JSON Error',
|
||||
response_ct)
|
||||
|
||||
try:
|
||||
raise messages2.Error.from_json(jobj)
|
||||
except jose.DeserializationError as error:
|
||||
# Couldn't deserialize JSON object
|
||||
raise errors.NetworkError((response, error))
|
||||
else:
|
||||
# response is not JSON object
|
||||
raise errors.NetworkError(response)
|
||||
else:
|
||||
if jobj is not None and response_ct != cls.JSON_CONTENT_TYPE:
|
||||
logging.debug(
|
||||
'Ignoring wrong Content-Type (%r) for JSON decodable '
|
||||
'response', response_ct)
|
||||
|
||||
if content_type == cls.JSON_CONTENT_TYPE and jobj is None:
|
||||
raise errors.NetworkError(
|
||||
'Unexpected response Content-Type: {0}'.format(response_ct))
|
||||
|
||||
def _get(self, uri, content_type=JSON_CONTENT_TYPE, **kwargs):
|
||||
"""Send GET request.
|
||||
|
||||
:raises letsencrypt.client.errors.NetworkError:
|
||||
|
||||
:returns: HTTP Response
|
||||
:rtype: `requests.Response`
|
||||
|
||||
"""
|
||||
try:
|
||||
response = requests.get(uri, **kwargs)
|
||||
except requests.exceptions.RequestException as error:
|
||||
raise errors.NetworkError(error)
|
||||
self._check_response(response, content_type=content_type)
|
||||
return response
|
||||
|
||||
def _post(self, uri, data, content_type=JSON_CONTENT_TYPE, **kwargs):
|
||||
"""Send POST data.
|
||||
|
||||
:param str content_type: Expected ``Content-Type``, fails if not set.
|
||||
|
||||
:raises letsencrypt.acme.messages2.NetworkError:
|
||||
|
||||
:returns: HTTP Response
|
||||
:rtype: `requests.Response`
|
||||
|
||||
"""
|
||||
logging.debug('Sending POST data: %s', data)
|
||||
try:
|
||||
response = requests.post(uri, data=data, **kwargs)
|
||||
except requests.exceptions.RequestException as error:
|
||||
raise errors.NetworkError(error)
|
||||
logging.debug('Received response %s: %s', response, response.text)
|
||||
|
||||
self._check_response(response, content_type=content_type)
|
||||
return response
|
||||
|
||||
@classmethod
|
||||
def _regr_from_response(cls, response, uri=None, new_authzr_uri=None,
|
||||
terms_of_service=None):
|
||||
terms_of_service = (
|
||||
response.links['terms-of-service']['url']
|
||||
if 'terms-of-service' in response.links else terms_of_service)
|
||||
|
||||
if new_authzr_uri is None:
|
||||
try:
|
||||
new_authzr_uri = response.links['next']['url']
|
||||
except KeyError:
|
||||
raise errors.NetworkError('"next" link missing')
|
||||
|
||||
return messages2.RegistrationResource(
|
||||
body=messages2.Registration.from_json(response.json()),
|
||||
uri=response.headers.get('Location', uri),
|
||||
new_authzr_uri=new_authzr_uri,
|
||||
terms_of_service=terms_of_service)
|
||||
|
||||
def register(self, contact=messages2.Registration._fields[
|
||||
'contact'].default):
|
||||
"""Register.
|
||||
|
||||
:param contact: Contact list, as accpeted by `.RegistrationResource`
|
||||
:type contact: `tuple`
|
||||
|
||||
:returns: Registration Resource.
|
||||
:rtype: `.RegistrationResource`
|
||||
|
||||
:raises letsencrypt.client.errors.UnexpectedUpdate:
|
||||
|
||||
"""
|
||||
new_reg = messages2.Registration(contact=contact)
|
||||
|
||||
response = self._post(self.new_reg_uri, self._wrap_in_jws(new_reg))
|
||||
assert response.status_code == httplib.CREATED # TODO: handle errors
|
||||
|
||||
regr = self._regr_from_response(response)
|
||||
if regr.body.key != self.key.public() or regr.body.contact != contact:
|
||||
raise errors.UnexpectedUpdate(regr)
|
||||
|
||||
return regr
|
||||
|
||||
def update_registration(self, regr):
|
||||
"""Update registration.
|
||||
|
||||
:pram regr: Registration Resource.
|
||||
:type regr: `.RegistrationResource`
|
||||
|
||||
:returns: Updated Registration Resource.
|
||||
:rtype: `.RegistrationResource`
|
||||
|
||||
"""
|
||||
response = self._post(regr.uri, self._wrap_in_jws(regr.body))
|
||||
|
||||
# TODO: Boulder returns httplib.ACCEPTED
|
||||
#assert response.status_code == httplib.OK
|
||||
|
||||
# TODO: Boulder does not set Location or Link on update
|
||||
# (c.f. acme-spec #94)
|
||||
|
||||
updated_regr = self._regr_from_response(
|
||||
response, uri=regr.uri, new_authzr_uri=regr.new_authzr_uri,
|
||||
terms_of_service=regr.terms_of_service)
|
||||
if updated_regr != regr:
|
||||
# TODO: Boulder reregisters with new recoveryToken and new URI
|
||||
raise errors.UnexpectedUpdate(regr)
|
||||
return updated_regr
|
||||
|
||||
def _authzr_from_response(self, response, identifier,
|
||||
uri=None, new_cert_uri=None):
|
||||
if new_cert_uri is None:
|
||||
try:
|
||||
new_cert_uri = response.links['next']['url']
|
||||
except KeyError:
|
||||
raise errors.NetworkError('"next" link missing')
|
||||
|
||||
authzr = messages2.AuthorizationResource(
|
||||
body=messages2.Authorization.from_json(response.json()),
|
||||
uri=response.headers.get('Location', uri),
|
||||
new_cert_uri=new_cert_uri)
|
||||
if (authzr.body.key != self.key.public()
|
||||
or authzr.body.identifier != identifier):
|
||||
raise errors.UnexpectedUpdate(authzr)
|
||||
return authzr
|
||||
|
||||
def request_challenges(self, identifier, regr):
|
||||
"""Request challenges.
|
||||
|
||||
:param identifier: Identifier to be challenged.
|
||||
:type identifier: `.messages2.Identifier`
|
||||
|
||||
:param regr: Registration Resource.
|
||||
:type regr: `.RegistrationResource`
|
||||
|
||||
:returns: Authorization Resource.
|
||||
:rtype: `.AuthorizationResource`
|
||||
|
||||
"""
|
||||
new_authz = messages2.Authorization(identifier=identifier)
|
||||
response = self._post(regr.new_authzr_uri, self._wrap_in_jws(new_authz))
|
||||
assert response.status_code == httplib.CREATED # TODO: handle errors
|
||||
return self._authzr_from_response(response, identifier)
|
||||
|
||||
def request_domain_challenges(self, domain, regr):
|
||||
"""Request challenges for domain names.
|
||||
|
||||
This is simply a convenience function that wraps around
|
||||
`request_challenges`, but works with domain names instead of
|
||||
generic identifiers.
|
||||
|
||||
:param str domain: Domain name to be challenged.
|
||||
|
||||
"""
|
||||
return self.request_challenges(messages2.Identifier(
|
||||
typ=messages2.IDENTIFIER_FQDN, value=domain), regr)
|
||||
|
||||
def answer_challenge(self, challb, response):
|
||||
"""Answer challenge.
|
||||
|
||||
:param challb: Challenge Resource body.
|
||||
:type challb: `.ChallengeBody`
|
||||
|
||||
:param response: Corresponding Challenge response
|
||||
:type response: `.challenges.ChallengeResponse`
|
||||
|
||||
:returns: Challenge Resource with updated body.
|
||||
:rtype: `.ChallengeResource`
|
||||
|
||||
:raises errors.UnexpectedUpdate:
|
||||
|
||||
"""
|
||||
response = self._post(challb.uri, self._wrap_in_jws(response))
|
||||
try:
|
||||
authzr_uri = response.links['up']['url']
|
||||
except KeyError:
|
||||
raise errors.NetworkError('"up" Link header missing')
|
||||
challr = messages2.ChallengeResource(
|
||||
authzr_uri=authzr_uri,
|
||||
body=messages2.ChallengeBody.from_json(response.json()))
|
||||
# TODO: check that challr.uri == response.headers['Location']?
|
||||
if challr.uri != challb.uri:
|
||||
raise errors.UnexpectedUpdate(challr.uri)
|
||||
return challr
|
||||
|
||||
def answer_challenges(self, challbs, responses):
|
||||
"""Answer multiple challenges.
|
||||
|
||||
.. note:: This is a convenience function to make integration
|
||||
with old proto code easier and shall probably be removed
|
||||
once restification is over.
|
||||
|
||||
"""
|
||||
return [self.answer_challenge(challb, response)
|
||||
for challb, response in itertools.izip(challbs, responses)]
|
||||
|
||||
@classmethod
|
||||
def retry_after(cls, response, default):
|
||||
"""Compute next `poll` time based on response ``Retry-After`` header.
|
||||
|
||||
:param response: Response from `poll`.
|
||||
:type response: `requests.Response`
|
||||
|
||||
:param int default: Default value (in seconds), used when
|
||||
``Retry-After`` header is not present or invalid.
|
||||
|
||||
:returns: Time point when next `poll` should be performed.
|
||||
:rtype: `datetime.datetime`
|
||||
|
||||
"""
|
||||
retry_after = response.headers.get('Retry-After', str(default))
|
||||
try:
|
||||
seconds = int(retry_after)
|
||||
except ValueError:
|
||||
# pylint: disable=no-member
|
||||
decoded = werkzeug.parse_date(retry_after) # RFC1123
|
||||
if decoded is None:
|
||||
seconds = default
|
||||
else:
|
||||
return decoded
|
||||
|
||||
return datetime.datetime.now() + datetime.timedelta(seconds=seconds)
|
||||
|
||||
def poll(self, authzr):
|
||||
"""Poll Authorization Resource for status.
|
||||
|
||||
:param authzr: Authorization Resource
|
||||
:type authzr: `.AuthorizationResource`
|
||||
|
||||
:returns: Updated Authorization Resource and HTTP response.
|
||||
|
||||
:rtype: (`.AuthorizationResource`, `requests.Response`)
|
||||
|
||||
"""
|
||||
response = self._get(authzr.uri)
|
||||
updated_authzr = self._authzr_from_response(
|
||||
response, authzr.body.identifier, authzr.uri, authzr.new_cert_uri)
|
||||
# TODO: check and raise UnexpectedUpdate
|
||||
|
||||
return updated_authzr, response
|
||||
|
||||
def request_issuance(self, csr, authzrs):
|
||||
"""Request issuance.
|
||||
|
||||
:param csr: CSR
|
||||
:type csr: `M2Crypto.X509.Request` wrapped in `.ComparableX509`
|
||||
|
||||
:param authzrs: `list` of `.AuthorizationResource`
|
||||
|
||||
:returns: Issued certificate
|
||||
:rtype: `.messages2.CertificateResource`
|
||||
|
||||
"""
|
||||
assert authzrs, "Authorizations list is empty"
|
||||
|
||||
# TODO: assert len(authzrs) == number of SANs
|
||||
req = messages2.CertificateRequest(
|
||||
csr=csr, authorizations=tuple(authzr.uri for authzr in authzrs))
|
||||
|
||||
content_type = self.DER_CONTENT_TYPE # TODO: add 'cert_type 'argument
|
||||
response = self._post(
|
||||
authzrs[0].new_cert_uri, # TODO: acme-spec #90
|
||||
self._wrap_in_jws(req),
|
||||
content_type=content_type,
|
||||
headers={'Accept': content_type})
|
||||
|
||||
cert_chain_uri = response.links.get('up', {}).get('url')
|
||||
|
||||
try:
|
||||
uri = response.headers['Location']
|
||||
except KeyError:
|
||||
raise errors.NetworkError('"Location" Header missing')
|
||||
|
||||
return messages2.CertificateResource(
|
||||
uri=uri, authzrs=authzrs, cert_chain_uri=cert_chain_uri,
|
||||
body=jose.ComparableX509(
|
||||
M2Crypto.X509.load_cert_der_string(response.content)))
|
||||
|
||||
def poll_and_request_issuance(self, csr, authzrs, mintime=5):
|
||||
"""Poll and request issuance.
|
||||
|
||||
This function polls all provided Authorization Resource URIs
|
||||
until all challenges are valid, respecting ``Retry-After`` HTTP
|
||||
headers, and then calls `request_issuance`.
|
||||
|
||||
.. todo:: add `max_attempts` or `timeout`
|
||||
|
||||
:param csr: CSR.
|
||||
:type csr: `M2Crypto.X509.Request` wrapped in `.ComparableX509`
|
||||
|
||||
:param authzrs: `list` of `.AuthorizationResource`
|
||||
|
||||
:param int mintime: Minimum time before next attempt, used if
|
||||
``Retry-After`` is not present in the response.
|
||||
|
||||
:returns: ``(cert, updated_authzrs)`` `tuple` where ``cert`` is
|
||||
the issued certificate (`.messages2.CertificateResource.),
|
||||
and ``updated_authzrs`` is a `tuple` consisting of updated
|
||||
Authorization Resources (`.AuthorizationResource`) as
|
||||
present in the responses from server, and in the same order
|
||||
as the input ``authzrs``.
|
||||
:rtype: `tuple`
|
||||
|
||||
"""
|
||||
# priority queue with datetime (based od Retry-After) as key,
|
||||
# and original Authorization Resource as value
|
||||
waiting = [(datetime.datetime.now(), authzr) for authzr in authzrs]
|
||||
# mapping between original Authorization Resource and the most
|
||||
# recently updated one
|
||||
updated = dict((authzr, authzr) for authzr in authzrs)
|
||||
|
||||
while waiting:
|
||||
# find the smallest Retry-After, and sleep if necessary
|
||||
when, authzr = heapq.heappop(waiting)
|
||||
now = datetime.datetime.now()
|
||||
if when > now:
|
||||
seconds = (when - now).seconds
|
||||
logging.debug('Sleeping for %d seconds', seconds)
|
||||
time.sleep(seconds)
|
||||
|
||||
# Note that we poll with the latest updated Authorization
|
||||
# URI, which might have a different URI than initial one
|
||||
updated_authzr, response = self.poll(updated[authzr])
|
||||
updated[authzr] = updated_authzr
|
||||
|
||||
if updated_authzr.body.status != messages2.STATUS_VALID:
|
||||
# push back to the priority queue, with updated retry_after
|
||||
heapq.heappush(waiting, (self.retry_after(
|
||||
response, default=mintime), authzr))
|
||||
|
||||
updated_authzrs = tuple(updated[authzr] for authzr in authzrs)
|
||||
return self.request_issuance(csr, updated_authzrs), updated_authzrs
|
||||
|
||||
def _get_cert(self, uri):
|
||||
content_type = self.DER_CONTENT_TYPE # TODO: make it a param
|
||||
response = self._get(uri, headers={'Accept': content_type},
|
||||
content_type=content_type)
|
||||
return response, jose.ComparableX509(
|
||||
M2Crypto.X509.load_cert_der_string(response.content))
|
||||
|
||||
def check_cert(self, certr):
|
||||
"""Check for new cert.
|
||||
|
||||
:param certr: Certificate Resource
|
||||
:type certr: `.CertificateResource`
|
||||
|
||||
:returns: Updated Certificate Resource.
|
||||
:rtype: `.CertificateResource`
|
||||
|
||||
"""
|
||||
# TODO: acme-spec 5.1 table action should be renamed to
|
||||
# "refresh cert", and this method integrated with self.refresh
|
||||
response, cert = self._get_cert(certr.uri)
|
||||
if 'Location' not in response.headers:
|
||||
raise errors.NetworkError('Location header missing')
|
||||
if response.headers['Location'] != certr.uri:
|
||||
raise errors.UnexpectedUpdate(response.text)
|
||||
return certr.update(body=cert)
|
||||
|
||||
def refresh(self, certr):
|
||||
"""Refresh certificate.
|
||||
|
||||
:param certr: Certificate Resource
|
||||
:type certr: `.CertificateResource`
|
||||
|
||||
:returns: Updated Certificate Resource.
|
||||
:rtype: `.CertificateResource`
|
||||
|
||||
"""
|
||||
# TODO: If a client sends a refresh request and the server is
|
||||
# not willing to refresh the certificate, the server MUST
|
||||
# respond with status code 403 (Forbidden)
|
||||
return self.check_cert(certr)
|
||||
|
||||
def fetch_chain(self, certr):
|
||||
"""Fetch chain for certificate.
|
||||
|
||||
:param certr: Certificate Resource
|
||||
:type certr: `.CertificateResource`
|
||||
|
||||
:returns: Certificate chain, or `None` if no "up" Link was provided.
|
||||
:rtype: `M2Crypto.X509.X509` wrapped in `.ComparableX509`
|
||||
|
||||
"""
|
||||
if certr.cert_chain_uri is not None:
|
||||
return self._get_cert(certr.cert_chain_uri)
|
||||
|
||||
def revoke(self, certr, when=messages2.Revocation.NOW):
|
||||
"""Revoke certificate.
|
||||
|
||||
:param when: When should the revocation take place? Takes
|
||||
the same values as `.messages2.Revocation.revoke`.
|
||||
|
||||
"""
|
||||
rev = messages2.Revocation(revoke=when, authorizations=tuple(
|
||||
authzr.uri for authzr in certr.authzrs))
|
||||
response = self._post(certr.uri, self._wrap_in_jws(rev))
|
||||
if response.status_code != httplib.OK:
|
||||
raise errors.NetworkError(
|
||||
'Successful revocation must return HTTP OK status')
|
||||
@@ -0,0 +1,458 @@
|
||||
"""Tests for letsencrypt.client.network2."""
|
||||
import datetime
|
||||
import httplib
|
||||
import os
|
||||
import pkg_resources
|
||||
import unittest
|
||||
|
||||
import M2Crypto
|
||||
import mock
|
||||
import requests
|
||||
|
||||
from letsencrypt.client import errors
|
||||
|
||||
from letsencrypt.acme import challenges
|
||||
from letsencrypt.acme import jose
|
||||
from letsencrypt.acme import messages2
|
||||
|
||||
|
||||
CERT = jose.ComparableX509(M2Crypto.X509.load_cert_string(
|
||||
pkg_resources.resource_string(
|
||||
__name__, os.path.join('testdata/cert.pem'))))
|
||||
CERT2 = jose.ComparableX509(M2Crypto.X509.load_cert_string(
|
||||
pkg_resources.resource_string(
|
||||
__name__, os.path.join('testdata/cert-san.pem'))))
|
||||
CSR = jose.ComparableX509(M2Crypto.X509.load_request_string(
|
||||
pkg_resources.resource_string(
|
||||
__name__, os.path.join('testdata/csr.pem'))))
|
||||
KEY = jose.JWKRSA.load(pkg_resources.resource_string(
|
||||
__name__, os.path.join('testdata/rsa512_key.pem')))
|
||||
KEY2 = jose.JWKRSA.load(pkg_resources.resource_string(
|
||||
__name__, os.path.join('testdata/rsa256_key.pem')))
|
||||
|
||||
|
||||
class NetworkTest(unittest.TestCase):
|
||||
"""Tests for letsencrypt.client.network2.Network."""
|
||||
|
||||
# pylint: disable=too-many-instance-attributes,too-many-public-methods
|
||||
|
||||
def setUp(self):
|
||||
from letsencrypt.client.network2 import Network
|
||||
self.net = Network(
|
||||
new_reg_uri='https://www.letsencrypt-demo.org/acme/new-reg',
|
||||
key=KEY, alg=jose.RS256)
|
||||
self.response = mock.MagicMock(ok=True, status_code=httplib.OK)
|
||||
self.response.headers = {}
|
||||
self.response.links = {}
|
||||
|
||||
self.identifier = messages2.Identifier(
|
||||
typ=messages2.IDENTIFIER_FQDN, value='example.com')
|
||||
|
||||
# Registration
|
||||
self.contact = ('mailto:cert-admin@example.com', 'tel:+12025551212')
|
||||
reg = messages2.Registration(
|
||||
contact=self.contact, key=KEY.public(), recovery_token='t')
|
||||
self.regr = messages2.RegistrationResource(
|
||||
body=reg, uri='https://www.letsencrypt-demo.org/acme/reg/1',
|
||||
new_authzr_uri='https://www.letsencrypt-demo.org/acme/new-reg',
|
||||
terms_of_service='https://www.letsencrypt-demo.org/tos')
|
||||
|
||||
# Authorization
|
||||
authzr_uri = 'https://www.letsencrypt-demo.org/acme/authz/1'
|
||||
challb = messages2.ChallengeBody(
|
||||
uri=(authzr_uri + '/1'), status=messages2.STATUS_VALID,
|
||||
chall=challenges.DNS(token='foo'))
|
||||
self.challr = messages2.ChallengeResource(
|
||||
body=challb, authzr_uri=authzr_uri)
|
||||
self.authz = messages2.Authorization(
|
||||
identifier=messages2.Identifier(
|
||||
typ=messages2.IDENTIFIER_FQDN, value='example.com'),
|
||||
challenges=(challb,), combinations=None, key=KEY.public())
|
||||
self.authzr = messages2.AuthorizationResource(
|
||||
body=self.authz, uri=authzr_uri,
|
||||
new_cert_uri='https://www.letsencrypt-demo.org/acme/new-cert')
|
||||
|
||||
# Request issuance
|
||||
self.certr = messages2.CertificateResource(
|
||||
body=CERT, authzrs=(self.authzr,),
|
||||
uri='https://www.letsencrypt-demo.org/acme/cert/1',
|
||||
cert_chain_uri='https://www.letsencrypt-demo.org/ca')
|
||||
|
||||
def _mock_post_get(self):
|
||||
# pylint: disable=protected-access
|
||||
self.net._post = mock.MagicMock(return_value=self.response)
|
||||
self.net._get = mock.MagicMock(return_value=self.response)
|
||||
|
||||
def test_wrap_in_jws(self):
|
||||
class MockJSONDeSerializable(jose.JSONDeSerializable):
|
||||
# pylint: disable=missing-docstring
|
||||
def __init__(self, value):
|
||||
self.value = value
|
||||
def to_json(self):
|
||||
return self.value
|
||||
@classmethod
|
||||
def from_json(cls, value):
|
||||
return cls(value)
|
||||
# pylint: disable=protected-access
|
||||
jws = self.net._wrap_in_jws(MockJSONDeSerializable('foo'))
|
||||
self.assertEqual(jose.JWS.json_loads(jws).payload, '"foo"')
|
||||
|
||||
def test_check_response_not_ok_jobj_no_error(self):
|
||||
self.response.ok = False
|
||||
self.response.json.return_value = {}
|
||||
# pylint: disable=protected-access
|
||||
self.assertRaises(
|
||||
errors.NetworkError, self.net._check_response, self.response)
|
||||
|
||||
def test_check_response_not_ok_jobj_error(self):
|
||||
self.response.ok = False
|
||||
self.response.json.return_value = messages2.Error(detail='foo')
|
||||
# pylint: disable=protected-access
|
||||
self.assertRaises(
|
||||
messages2.Error, self.net._check_response, self.response)
|
||||
|
||||
def test_check_response_not_ok_no_jobj(self):
|
||||
self.response.ok = False
|
||||
self.response.json.side_effect = ValueError
|
||||
# pylint: disable=protected-access
|
||||
self.assertRaises(
|
||||
errors.NetworkError, self.net._check_response, self.response)
|
||||
|
||||
def test_check_response_ok_no_jobj_ct_required(self):
|
||||
self.response.json.side_effect = ValueError
|
||||
for response_ct in [self.net.JSON_CONTENT_TYPE, 'foo']:
|
||||
self.response.headers['Content-Type'] = response_ct
|
||||
# pylint: disable=protected-access
|
||||
self.assertRaises(
|
||||
errors.NetworkError, self.net._check_response, self.response,
|
||||
content_type=self.net.JSON_CONTENT_TYPE)
|
||||
|
||||
def test_check_response_ok_no_jobj_no_ct(self):
|
||||
self.response.json.side_effect = ValueError
|
||||
for response_ct in [self.net.JSON_CONTENT_TYPE, 'foo']:
|
||||
self.response.headers['Content-Type'] = response_ct
|
||||
# pylint: disable=protected-access
|
||||
self.net._check_response(self.response)
|
||||
|
||||
def test_check_response_jobj(self):
|
||||
self.response.json.return_value = {}
|
||||
for response_ct in [self.net.JSON_CONTENT_TYPE, 'foo']:
|
||||
self.response.headers['Content-Type'] = response_ct
|
||||
# pylint: disable=protected-access
|
||||
self.net._check_response(self.response)
|
||||
|
||||
@mock.patch('letsencrypt.client.network2.requests')
|
||||
def test_get_requests_error_passthrough(self, requests_mock):
|
||||
requests_mock.exceptions = requests.exceptions
|
||||
requests_mock.get.side_effect = requests.exceptions.RequestException
|
||||
# pylint: disable=protected-access
|
||||
self.assertRaises(errors.NetworkError, self.net._get, 'uri')
|
||||
|
||||
@mock.patch('letsencrypt.client.network2.requests')
|
||||
def test_get(self, requests_mock):
|
||||
# pylint: disable=protected-access
|
||||
self.net._check_response = mock.MagicMock()
|
||||
self.net._get('uri', content_type='ct')
|
||||
self.net._check_response.assert_called_once_with(
|
||||
requests_mock.get('uri'), content_type='ct')
|
||||
|
||||
@mock.patch('letsencrypt.client.network2.requests')
|
||||
def test_post_requests_error_passthrough(self, requests_mock):
|
||||
requests_mock.exceptions = requests.exceptions
|
||||
requests_mock.post.side_effect = requests.exceptions.RequestException
|
||||
# pylint: disable=protected-access
|
||||
self.assertRaises(errors.NetworkError, self.net._post, 'uri', 'data')
|
||||
|
||||
@mock.patch('letsencrypt.client.network2.requests')
|
||||
def test_post(self, requests_mock):
|
||||
# pylint: disable=protected-access
|
||||
self.net._check_response = mock.MagicMock()
|
||||
self.net._post('uri', 'data', content_type='ct')
|
||||
self.net._check_response.assert_called_once_with(
|
||||
requests_mock.post('uri', 'data'), content_type='ct')
|
||||
|
||||
def test_register(self):
|
||||
self.response.status_code = httplib.CREATED
|
||||
self.response.json.return_value = self.regr.body.fully_serialize()
|
||||
self.response.headers['Location'] = self.regr.uri
|
||||
self.response.links.update({
|
||||
'next': {'url': self.regr.new_authzr_uri},
|
||||
'terms-of-service': {'url': self.regr.terms_of_service},
|
||||
})
|
||||
|
||||
self._mock_post_get()
|
||||
self.assertEqual(self.regr, self.net.register(self.contact))
|
||||
# TODO: test POST call arguments
|
||||
|
||||
# TODO: split here and separate test
|
||||
reg_wrong_key = self.regr.body.update(key=KEY2.public())
|
||||
self.response.json.return_value = reg_wrong_key.fully_serialize()
|
||||
self.assertRaises(
|
||||
errors.UnexpectedUpdate, self.net.register, self.contact)
|
||||
|
||||
def test_register_missing_next(self):
|
||||
self.response.status_code = httplib.CREATED
|
||||
self._mock_post_get()
|
||||
self.assertRaises(
|
||||
errors.NetworkError, self.net.register, self.regr.body)
|
||||
|
||||
def test_update_registration(self):
|
||||
self.response.headers['Location'] = self.regr.uri
|
||||
self.response.json.return_value = self.regr.body.fully_serialize()
|
||||
self._mock_post_get()
|
||||
self.assertEqual(self.regr, self.net.update_registration(self.regr))
|
||||
|
||||
# TODO: split here and separate test
|
||||
self.response.json.return_value = self.regr.body.update(
|
||||
contact=()).fully_serialize()
|
||||
self.assertRaises(
|
||||
errors.UnexpectedUpdate, self.net.update_registration, self.regr)
|
||||
|
||||
def test_request_challenges(self):
|
||||
self.response.status_code = httplib.CREATED
|
||||
self.response.headers['Location'] = self.authzr.uri
|
||||
self.response.json.return_value = self.authz.fully_serialize()
|
||||
self.response.links = {
|
||||
'next': {'url': self.authzr.new_cert_uri},
|
||||
}
|
||||
|
||||
self._mock_post_get()
|
||||
self.net.request_challenges(self.identifier, self.regr)
|
||||
# TODO: test POST call arguments
|
||||
|
||||
# TODO: split here and separate test
|
||||
authz_wrong_key = self.authz.update(key=KEY2.public())
|
||||
self.response.json.return_value = authz_wrong_key.fully_serialize()
|
||||
self.assertRaises(
|
||||
errors.UnexpectedUpdate, self.net.request_challenges,
|
||||
self.identifier, self.regr)
|
||||
|
||||
def test_request_challenges_missing_next(self):
|
||||
self.response.status_code = httplib.CREATED
|
||||
self._mock_post_get()
|
||||
self.assertRaises(
|
||||
errors.NetworkError, self.net.request_challenges,
|
||||
self.identifier, self.regr)
|
||||
|
||||
def test_request_domain_challenges(self):
|
||||
self.net.request_challenges = mock.MagicMock()
|
||||
self.assertEqual(
|
||||
self.net.request_challenges(self.identifier),
|
||||
self.net.request_domain_challenges('example.com', self.regr))
|
||||
|
||||
def test_answer_challenge(self):
|
||||
self.response.links['up'] = {'url': self.challr.authzr_uri}
|
||||
self.response.json.return_value = self.challr.body.fully_serialize()
|
||||
|
||||
chall_response = challenges.DNSResponse()
|
||||
|
||||
self._mock_post_get()
|
||||
self.net.answer_challenge(self.challr.body, chall_response)
|
||||
|
||||
# TODO: split here and separate test
|
||||
self.assertRaises(errors.UnexpectedUpdate, self.net.answer_challenge,
|
||||
self.challr.body.update(uri='foo'), chall_response)
|
||||
|
||||
def test_answer_challenge_missing_next(self):
|
||||
self._mock_post_get()
|
||||
self.assertRaises(errors.NetworkError, self.net.answer_challenge,
|
||||
self.challr.body, challenges.DNSResponse())
|
||||
|
||||
def test_answer_challenges(self):
|
||||
self.net.answer_challenge = mock.MagicMock()
|
||||
self.assertEqual(
|
||||
[self.net.answer_challenge(
|
||||
self.challr.body, challenges.DNSResponse())],
|
||||
self.net.answer_challenges(
|
||||
[self.challr.body], [challenges.DNSResponse()]))
|
||||
|
||||
def test_retry_after_date(self):
|
||||
self.response.headers['Retry-After'] = 'Fri, 31 Dec 1999 23:59:59 GMT'
|
||||
self.assertEqual(
|
||||
datetime.datetime(1999, 12, 31, 23, 59, 59),
|
||||
self.net.retry_after(response=self.response, default=10))
|
||||
|
||||
@mock.patch('letsencrypt.client.network2.datetime')
|
||||
def test_retry_after_invalid(self, dt_mock):
|
||||
dt_mock.datetime.now.return_value = datetime.datetime(2015, 3, 27)
|
||||
dt_mock.timedelta = datetime.timedelta
|
||||
|
||||
self.response.headers['Retry-After'] = 'foooo'
|
||||
self.assertEqual(
|
||||
datetime.datetime(2015, 3, 27, 0, 0, 10),
|
||||
self.net.retry_after(response=self.response, default=10))
|
||||
|
||||
@mock.patch('letsencrypt.client.network2.datetime')
|
||||
def test_retry_after_seconds(self, dt_mock):
|
||||
dt_mock.datetime.now.return_value = datetime.datetime(2015, 3, 27)
|
||||
dt_mock.timedelta = datetime.timedelta
|
||||
|
||||
self.response.headers['Retry-After'] = '50'
|
||||
self.assertEqual(
|
||||
datetime.datetime(2015, 3, 27, 0, 0, 50),
|
||||
self.net.retry_after(response=self.response, default=10))
|
||||
|
||||
@mock.patch('letsencrypt.client.network2.datetime')
|
||||
def test_retry_after_missing(self, dt_mock):
|
||||
dt_mock.datetime.now.return_value = datetime.datetime(2015, 3, 27)
|
||||
dt_mock.timedelta = datetime.timedelta
|
||||
|
||||
self.assertEqual(
|
||||
datetime.datetime(2015, 3, 27, 0, 0, 10),
|
||||
self.net.retry_after(response=self.response, default=10))
|
||||
|
||||
def test_poll(self):
|
||||
self.response.json.return_value = self.authzr.body.fully_serialize()
|
||||
self._mock_post_get()
|
||||
self.assertEqual((self.authzr, self.response),
|
||||
self.net.poll(self.authzr))
|
||||
|
||||
def test_request_issuance(self):
|
||||
self.response.content = CERT.as_der()
|
||||
self.response.headers['Location'] = self.certr.uri
|
||||
self.response.links['up'] = {'url': self.certr.cert_chain_uri}
|
||||
self._mock_post_get()
|
||||
self.assertEqual(
|
||||
self.certr, self.net.request_issuance(CSR, (self.authzr,)))
|
||||
# TODO: check POST args
|
||||
|
||||
def test_request_issuance_missing_up(self):
|
||||
self.response.content = CERT.as_der()
|
||||
self.response.headers['Location'] = self.certr.uri
|
||||
self._mock_post_get()
|
||||
self.assertEqual(
|
||||
self.certr.update(cert_chain_uri=None),
|
||||
self.net.request_issuance(CSR, (self.authzr,)))
|
||||
|
||||
def test_request_issuance_missing_location(self):
|
||||
self._mock_post_get()
|
||||
self.assertRaises(
|
||||
errors.NetworkError, self.net.request_issuance,
|
||||
CSR, (self.authzr,))
|
||||
|
||||
@mock.patch('letsencrypt.client.network2.datetime')
|
||||
@mock.patch('letsencrypt.client.network2.time')
|
||||
def test_poll_and_request_issuance(self, time_mock, dt_mock):
|
||||
# clock.dt | pylint: disable=no-member
|
||||
clock = mock.MagicMock(dt=datetime.datetime(2015, 3, 27))
|
||||
|
||||
def sleep(seconds):
|
||||
"""increment clock"""
|
||||
clock.dt += datetime.timedelta(seconds=seconds)
|
||||
time_mock.sleep.side_effect = sleep
|
||||
|
||||
def now():
|
||||
"""return current clock value"""
|
||||
return clock.dt
|
||||
dt_mock.datetime.now.side_effect = now
|
||||
dt_mock.timedelta = datetime.timedelta
|
||||
|
||||
def poll(authzr): # pylint: disable=missing-docstring
|
||||
# record poll start time based on the current clock value
|
||||
authzr.times.append(clock.dt)
|
||||
|
||||
# suppose it takes 2 seconds for server to produce the
|
||||
# result, increment clock
|
||||
clock.dt += datetime.timedelta(seconds=2)
|
||||
|
||||
if not authzr.retries: # no more retries
|
||||
done = mock.MagicMock(uri=authzr.uri, times=authzr.times)
|
||||
done.body.status = messages2.STATUS_VALID
|
||||
return done, []
|
||||
|
||||
# response (2nd result tuple element) is reduced to only
|
||||
# Retry-After header contents represented as integer
|
||||
# seconds; authzr.retries is a list of Retry-After
|
||||
# headers, head(retries) is peeled of as a current
|
||||
# Retry-After header, and tail(retries) is persisted for
|
||||
# later poll() calls
|
||||
return (mock.MagicMock(retries=authzr.retries[1:],
|
||||
uri=authzr.uri + '.', times=authzr.times),
|
||||
authzr.retries[0])
|
||||
self.net.poll = mock.MagicMock(side_effect=poll)
|
||||
|
||||
mintime = 7
|
||||
|
||||
def retry_after(response, default): # pylint: disable=missing-docstring
|
||||
# check that poll_and_request_issuance correctly passes mintime
|
||||
self.assertEqual(default, mintime)
|
||||
return clock.dt + datetime.timedelta(seconds=response)
|
||||
self.net.retry_after = mock.MagicMock(side_effect=retry_after)
|
||||
|
||||
def request_issuance(csr, authzrs): # pylint: disable=missing-docstring
|
||||
return csr, authzrs
|
||||
self.net.request_issuance = mock.MagicMock(side_effect=request_issuance)
|
||||
|
||||
csr = mock.MagicMock()
|
||||
authzrs = (
|
||||
mock.MagicMock(uri='a', times=[], retries=(8, 20, 30)),
|
||||
mock.MagicMock(uri='b', times=[], retries=(5,)),
|
||||
)
|
||||
|
||||
cert, updated_authzrs = self.net.poll_and_request_issuance(
|
||||
csr, authzrs, mintime=mintime)
|
||||
self.assertTrue(cert[0] is csr)
|
||||
self.assertTrue(cert[1] is updated_authzrs)
|
||||
self.assertEqual(updated_authzrs[0].uri, 'a...')
|
||||
self.assertEqual(updated_authzrs[1].uri, 'b.')
|
||||
self.assertEqual(updated_authzrs[0].times, [
|
||||
datetime.datetime(2015, 3, 27),
|
||||
# a is scheduled for 10, but b is polling [9..11), so it
|
||||
# will be picked up as soon as b is finished, without
|
||||
# additional sleeping
|
||||
datetime.datetime(2015, 3, 27, 0, 0, 11),
|
||||
datetime.datetime(2015, 3, 27, 0, 0, 33),
|
||||
datetime.datetime(2015, 3, 27, 0, 1, 5),
|
||||
])
|
||||
self.assertEqual(updated_authzrs[1].times, [
|
||||
datetime.datetime(2015, 3, 27, 0, 0, 2),
|
||||
datetime.datetime(2015, 3, 27, 0, 0, 9),
|
||||
])
|
||||
self.assertEqual(clock.dt, datetime.datetime(2015, 3, 27, 0, 1, 7))
|
||||
|
||||
def test_check_cert(self):
|
||||
self.response.headers['Location'] = self.certr.uri
|
||||
self.response.content = CERT2.as_der()
|
||||
self._mock_post_get()
|
||||
self.assertEqual(
|
||||
self.certr.update(body=CERT2), self.net.check_cert(self.certr))
|
||||
|
||||
# TODO: split here and separate test
|
||||
self.response.headers['Location'] = 'foo'
|
||||
self.assertRaises(
|
||||
errors.UnexpectedUpdate, self.net.check_cert, self.certr)
|
||||
|
||||
def test_check_cert_missing_location(self):
|
||||
self.response.content = CERT2.as_der()
|
||||
self._mock_post_get()
|
||||
self.assertRaises(errors.NetworkError, self.net.check_cert, self.certr)
|
||||
|
||||
def test_refresh(self):
|
||||
self.net.check_cert = mock.MagicMock()
|
||||
self.assertEqual(
|
||||
self.net.check_cert(self.certr), self.net.refresh(self.certr))
|
||||
|
||||
def test_fetch_chain(self):
|
||||
# pylint: disable=protected-access
|
||||
self.net._get_cert = mock.MagicMock()
|
||||
self.assertEqual(self.net._get_cert(self.certr.cert_chain_uri),
|
||||
self.net.fetch_chain(self.certr))
|
||||
|
||||
def test_fetch_chain_no_up_link(self):
|
||||
self.assertTrue(self.net.fetch_chain(self.certr.update(
|
||||
cert_chain_uri=None)) is None)
|
||||
|
||||
def test_revoke(self):
|
||||
self._mock_post_get()
|
||||
self.net.revoke(self.certr, when=messages2.Revocation.NOW)
|
||||
# pylint: disable=protected-access
|
||||
self.net._post.assert_called_once_with(self.certr.uri, mock.ANY)
|
||||
|
||||
def test_revoke_bad_status_raises_error(self):
|
||||
self.response.status_code = httplib.METHOD_NOT_ALLOWED
|
||||
self._mock_post_get()
|
||||
self.assertRaises(errors.NetworkError, self.net.revoke, self.certr)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -36,9 +36,12 @@ install_requires = [
|
||||
'pyasn1', # urllib3 InsecurePlatformWarning (#304)
|
||||
'pycrypto',
|
||||
'PyOpenSSL',
|
||||
'pyrfc3339',
|
||||
'python-augeas',
|
||||
'python2-pythondialog',
|
||||
'pytz',
|
||||
'requests',
|
||||
'werkzeug',
|
||||
'zope.component',
|
||||
'zope.interface',
|
||||
# order of items in install_requires DOES matter and M2Crypto has
|
||||
|
||||
Reference in New Issue
Block a user