Merge remote-tracking branch 'letsencrypt/master'

This commit is contained in:
Amjad Mashaal 2016-03-16 22:17:52 +02:00
commit d42070c856
9 changed files with 37 additions and 512 deletions

View file

@ -90,6 +90,11 @@ IRC Channel: #letsencrypt on `Freenode`_
Community: https://community.letsencrypt.org
ACME spec: http://ietf-wg-acme.github.io/acme/
ACME working area in github: https://github.com/ietf-wg-acme/acme
Mailing list: `client-dev`_ (to subscribe without a Google account, send an
email to client-dev+subscribe@letsencrypt.org)

View file

@ -13,7 +13,6 @@ from acme import errors
from acme import crypto_util
from acme import fields
from acme import jose
from acme import other
logger = logging.getLogger(__name__)
@ -36,14 +35,6 @@ class Challenge(jose.TypedJSONObjectWithFields):
return UnrecognizedChallenge.from_json(jobj)
class ContinuityChallenge(Challenge): # pylint: disable=abstract-method
"""Client validation challenges."""
class DVChallenge(Challenge): # pylint: disable=abstract-method
"""Domain validation challenges."""
class ChallengeResponse(jose.TypedJSONObjectWithFields):
# _fields_to_partial_json | pylint: disable=abstract-method
"""ACME challenge response."""
@ -78,8 +69,8 @@ class UnrecognizedChallenge(Challenge):
return cls(jobj)
class _TokenDVChallenge(DVChallenge):
"""DV Challenge with token.
class _TokenChallenge(Challenge):
"""Challenge with token.
:ivar bytes token:
@ -149,7 +140,7 @@ class KeyAuthorizationChallengeResponse(ChallengeResponse):
return True
class KeyAuthorizationChallenge(_TokenDVChallenge):
class KeyAuthorizationChallenge(_TokenChallenge):
# pylint: disable=abstract-class-little-used,too-many-ancestors
"""Challenge based on Key Authorization.
@ -460,108 +451,8 @@ class TLSSNI01(KeyAuthorizationChallenge):
return self.response(account_key).gen_cert(key=kwargs.get('cert_key'))
@Challenge.register
class RecoveryContact(ContinuityChallenge):
"""ACME "recoveryContact" challenge.
:ivar unicode activation_url:
:ivar unicode success_url:
:ivar unicode contact:
"""
typ = "recoveryContact"
activation_url = jose.Field("activationURL", omitempty=True)
success_url = jose.Field("successURL", omitempty=True)
contact = jose.Field("contact", omitempty=True)
@ChallengeResponse.register
class RecoveryContactResponse(ChallengeResponse):
"""ACME "recoveryContact" challenge response.
:ivar unicode token:
"""
typ = "recoveryContact"
token = jose.Field("token", omitempty=True)
@Challenge.register
class ProofOfPossession(ContinuityChallenge):
"""ACME "proofOfPossession" challenge.
:ivar .JWAAlgorithm alg:
:ivar bytes nonce: Random data, **not** base64-encoded.
:ivar hints: Various clues for the client (:class:`Hints`).
"""
typ = "proofOfPossession"
NONCE_SIZE = 16
class Hints(jose.JSONObjectWithFields):
"""Hints for "proofOfPossession" challenge.
:ivar JWK jwk: JSON Web Key
:ivar tuple cert_fingerprints: `tuple` of `unicode`
:ivar tuple certs: Sequence of :class:`acme.jose.ComparableX509`
certificates.
:ivar tuple subject_key_identifiers: `tuple` of `unicode`
:ivar tuple issuers: `tuple` of `unicode`
:ivar tuple authorized_for: `tuple` of `unicode`
"""
jwk = jose.Field("jwk", decoder=jose.JWK.from_json)
cert_fingerprints = jose.Field(
"certFingerprints", omitempty=True, default=())
certs = jose.Field("certs", omitempty=True, default=())
subject_key_identifiers = jose.Field(
"subjectKeyIdentifiers", omitempty=True, default=())
serial_numbers = jose.Field("serialNumbers", omitempty=True, default=())
issuers = jose.Field("issuers", omitempty=True, default=())
authorized_for = jose.Field("authorizedFor", omitempty=True, default=())
@certs.encoder
def certs(value): # pylint: disable=missing-docstring,no-self-argument
return tuple(jose.encode_cert(cert) for cert in value)
@certs.decoder
def certs(value): # pylint: disable=missing-docstring,no-self-argument
return tuple(jose.decode_cert(cert) for cert in value)
alg = jose.Field("alg", decoder=jose.JWASignature.from_json)
nonce = jose.Field(
"nonce", encoder=jose.encode_b64jose, decoder=functools.partial(
jose.decode_b64jose, size=NONCE_SIZE))
hints = jose.Field("hints", decoder=Hints.from_json)
@ChallengeResponse.register
class ProofOfPossessionResponse(ChallengeResponse):
"""ACME "proofOfPossession" challenge response.
:ivar bytes nonce: Random data, **not** base64-encoded.
:ivar acme.other.Signature signature: Sugnature of this message.
"""
typ = "proofOfPossession"
NONCE_SIZE = ProofOfPossession.NONCE_SIZE
nonce = jose.Field(
"nonce", encoder=jose.encode_b64jose, decoder=functools.partial(
jose.decode_b64jose, size=NONCE_SIZE))
signature = jose.Field("signature", decoder=other.Signature.from_json)
def verify(self):
"""Verify the challenge."""
# self.signature is not Field | pylint: disable=no-member
return self.signature.verify(self.nonce)
@Challenge.register # pylint: disable=too-many-ancestors
class DNS(_TokenDVChallenge):
class DNS(_TokenChallenge):
"""ACME "dns" challenge."""
typ = "dns"

View file

@ -9,7 +9,6 @@ from six.moves.urllib import parse as urllib_parse # pylint: disable=import-err
from acme import errors
from acme import jose
from acme import other
from acme import test_util
@ -324,233 +323,6 @@ class TLSSNI01Test(unittest.TestCase):
mock_gen_cert.assert_called_once_with(key=mock.sentinel.cert_key)
class RecoveryContactTest(unittest.TestCase):
def setUp(self):
from acme.challenges import RecoveryContact
self.msg = RecoveryContact(
activation_url='https://example.ca/sendrecovery/a5bd99383fb0',
success_url='https://example.ca/confirmrecovery/bb1b9928932',
contact='c********n@example.com')
self.jmsg = {
'type': 'recoveryContact',
'activationURL': 'https://example.ca/sendrecovery/a5bd99383fb0',
'successURL': 'https://example.ca/confirmrecovery/bb1b9928932',
'contact': 'c********n@example.com',
}
def test_to_partial_json(self):
self.assertEqual(self.jmsg, self.msg.to_partial_json())
def test_from_json(self):
from acme.challenges import RecoveryContact
self.assertEqual(self.msg, RecoveryContact.from_json(self.jmsg))
def test_from_json_hashable(self):
from acme.challenges import RecoveryContact
hash(RecoveryContact.from_json(self.jmsg))
def test_json_without_optionals(self):
del self.jmsg['activationURL']
del self.jmsg['successURL']
del self.jmsg['contact']
from acme.challenges import RecoveryContact
msg = RecoveryContact.from_json(self.jmsg)
self.assertTrue(msg.activation_url is None)
self.assertTrue(msg.success_url is None)
self.assertTrue(msg.contact is None)
self.assertEqual(self.jmsg, msg.to_partial_json())
class RecoveryContactResponseTest(unittest.TestCase):
def setUp(self):
from acme.challenges import RecoveryContactResponse
self.msg = RecoveryContactResponse(token='23029d88d9e123e')
self.jmsg = {
'resource': 'challenge',
'type': 'recoveryContact',
'token': '23029d88d9e123e',
}
def test_to_partial_json(self):
self.assertEqual(self.jmsg, self.msg.to_partial_json())
def test_from_json(self):
from acme.challenges import RecoveryContactResponse
self.assertEqual(
self.msg, RecoveryContactResponse.from_json(self.jmsg))
def test_from_json_hashable(self):
from acme.challenges import RecoveryContactResponse
hash(RecoveryContactResponse.from_json(self.jmsg))
def test_json_without_optionals(self):
del self.jmsg['token']
from acme.challenges import RecoveryContactResponse
msg = RecoveryContactResponse.from_json(self.jmsg)
self.assertTrue(msg.token is None)
self.assertEqual(self.jmsg, msg.to_partial_json())
class ProofOfPossessionHintsTest(unittest.TestCase):
def setUp(self):
jwk = KEY.public_key()
issuers = (
'C=US, O=SuperT LLC, CN=SuperTrustworthy Public CA',
'O=LessTrustworthy CA Inc, CN=LessTrustworthy But StillSecure',
)
cert_fingerprints = (
'93416768eb85e33adc4277f4c9acd63e7418fcfe',
'16d95b7b63f1972b980b14c20291f3c0d1855d95',
'48b46570d9fc6358108af43ad1649484def0debf',
)
subject_key_identifiers = ('d0083162dcc4c8a23ecb8aecbd86120e56fd24e5')
authorized_for = ('www.example.com', 'example.net')
serial_numbers = (34234239832, 23993939911, 17)
from acme.challenges import ProofOfPossession
self.msg = ProofOfPossession.Hints(
jwk=jwk, issuers=issuers, cert_fingerprints=cert_fingerprints,
certs=(CERT,), subject_key_identifiers=subject_key_identifiers,
authorized_for=authorized_for, serial_numbers=serial_numbers)
self.jmsg_to = {
'jwk': jwk,
'certFingerprints': cert_fingerprints,
'certs': (jose.encode_b64jose(OpenSSL.crypto.dump_certificate(
OpenSSL.crypto.FILETYPE_ASN1, CERT.wrapped)),),
'subjectKeyIdentifiers': subject_key_identifiers,
'serialNumbers': serial_numbers,
'issuers': issuers,
'authorizedFor': authorized_for,
}
self.jmsg_from = self.jmsg_to.copy()
self.jmsg_from.update({'jwk': jwk.to_json()})
def test_to_partial_json(self):
self.assertEqual(self.jmsg_to, self.msg.to_partial_json())
def test_from_json(self):
from acme.challenges import ProofOfPossession
self.assertEqual(
self.msg, ProofOfPossession.Hints.from_json(self.jmsg_from))
def test_from_json_hashable(self):
from acme.challenges import ProofOfPossession
hash(ProofOfPossession.Hints.from_json(self.jmsg_from))
def test_json_without_optionals(self):
for optional in ['certFingerprints', 'certs', 'subjectKeyIdentifiers',
'serialNumbers', 'issuers', 'authorizedFor']:
del self.jmsg_from[optional]
del self.jmsg_to[optional]
from acme.challenges import ProofOfPossession
msg = ProofOfPossession.Hints.from_json(self.jmsg_from)
self.assertEqual(msg.cert_fingerprints, ())
self.assertEqual(msg.certs, ())
self.assertEqual(msg.subject_key_identifiers, ())
self.assertEqual(msg.serial_numbers, ())
self.assertEqual(msg.issuers, ())
self.assertEqual(msg.authorized_for, ())
self.assertEqual(self.jmsg_to, msg.to_partial_json())
class ProofOfPossessionTest(unittest.TestCase):
def setUp(self):
from acme.challenges import ProofOfPossession
hints = ProofOfPossession.Hints(
jwk=KEY.public_key(), cert_fingerprints=(),
certs=(), serial_numbers=(), subject_key_identifiers=(),
issuers=(), authorized_for=())
self.msg = ProofOfPossession(
alg=jose.RS256, hints=hints,
nonce=b'xD\xf9\xb9\xdbU\xed\xaa\x17\xf1y|\x81\x88\x99 ')
self.jmsg_to = {
'type': 'proofOfPossession',
'alg': jose.RS256,
'nonce': 'eET5udtV7aoX8Xl8gYiZIA',
'hints': hints,
}
self.jmsg_from = {
'type': 'proofOfPossession',
'alg': jose.RS256.to_json(),
'nonce': 'eET5udtV7aoX8Xl8gYiZIA',
'hints': hints.to_json(),
}
def test_to_partial_json(self):
self.assertEqual(self.jmsg_to, self.msg.to_partial_json())
def test_from_json(self):
from acme.challenges import ProofOfPossession
self.assertEqual(
self.msg, ProofOfPossession.from_json(self.jmsg_from))
def test_from_json_hashable(self):
from acme.challenges import ProofOfPossession
hash(ProofOfPossession.from_json(self.jmsg_from))
class ProofOfPossessionResponseTest(unittest.TestCase):
def setUp(self):
# acme-spec uses a confusing example in which both signature
# nonce and challenge nonce are the same, don't make the same
# mistake here...
signature = other.Signature(
alg=jose.RS256, jwk=KEY.public_key(),
sig=b'\xa7\xc1\xe7\xe82o\xbc\xcd\xd0\x1e\x010#Z|\xaf\x15\x83'
b'\x94\x8f#\x9b\nQo(\x80\x15,\x08\xfcz\x1d\xfd\xfd.\xaap'
b'\xfa\x06\xd1\xa2f\x8d8X2>%d\xbd%\xe1T\xdd\xaa0\x18\xde'
b'\x99\x08\xf0\x0e{',
nonce=b'\x99\xc7Q\xb3f2\xbc\xdci\xfe\xd6\x98k\xc67\xdf',
)
from acme.challenges import ProofOfPossessionResponse
self.msg = ProofOfPossessionResponse(
nonce=b'xD\xf9\xb9\xdbU\xed\xaa\x17\xf1y|\x81\x88\x99 ',
signature=signature)
self.jmsg_to = {
'resource': 'challenge',
'type': 'proofOfPossession',
'nonce': 'eET5udtV7aoX8Xl8gYiZIA',
'signature': signature,
}
self.jmsg_from = {
'resource': 'challenge',
'type': 'proofOfPossession',
'nonce': 'eET5udtV7aoX8Xl8gYiZIA',
'signature': signature.to_json(),
}
def test_verify(self):
self.assertTrue(self.msg.verify())
def test_to_partial_json(self):
self.assertEqual(self.jmsg_to, self.msg.to_partial_json())
def test_from_json(self):
from acme.challenges import ProofOfPossessionResponse
self.assertEqual(
self.msg, ProofOfPossessionResponse.from_json(self.jmsg_from))
def test_from_json_hashable(self):
from acme.challenges import ProofOfPossessionResponse
hash(ProofOfPossessionResponse.from_json(self.jmsg_from))
class DNSTest(unittest.TestCase):
def setUp(self):

View file

@ -271,10 +271,8 @@ class AuthorizationTest(unittest.TestCase):
ChallengeBody(uri='http://challb2', status=STATUS_VALID,
chall=challenges.DNS(
token=b'DGyRejmCefe7v4NfDGDKfA')),
ChallengeBody(uri='http://challb3', status=STATUS_VALID,
chall=challenges.RecoveryContact()),
)
combinations = ((0, 2), (1, 2))
combinations = ((0,), (1,))
from acme.messages import Authorization
from acme.messages import Identifier
@ -300,8 +298,8 @@ class AuthorizationTest(unittest.TestCase):
def test_resolved_combinations(self):
self.assertEqual(self.authz.resolved_combinations, (
(self.challbs[0], self.challbs[2]),
(self.challbs[1], self.challbs[2]),
(self.challbs[0],),
(self.challbs[1],),
))

View file

@ -1,67 +0,0 @@
"""Other ACME objects."""
import functools
import logging
import os
from acme import jose
logger = logging.getLogger(__name__)
class Signature(jose.JSONObjectWithFields):
"""ACME signature.
:ivar .JWASignature alg: Signature algorithm.
:ivar bytes sig: Signature.
:ivar bytes nonce: Nonce.
:ivar .JWK jwk: JWK.
"""
NONCE_SIZE = 16
"""Minimum size of nonce in bytes."""
alg = jose.Field('alg', decoder=jose.JWASignature.from_json)
sig = jose.Field('sig', encoder=jose.encode_b64jose,
decoder=jose.decode_b64jose)
nonce = jose.Field(
'nonce', encoder=jose.encode_b64jose, decoder=functools.partial(
jose.decode_b64jose, size=NONCE_SIZE, minimum=True))
jwk = jose.Field('jwk', decoder=jose.JWK.from_json)
@classmethod
def from_msg(cls, msg, key, nonce=None, nonce_size=None, alg=jose.RS256):
"""Create signature with nonce prepended to the message.
:param bytes msg: Message to be signed.
:param key: Key used for signing.
:type key: `cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey`
(optionally wrapped in `.ComparableRSAKey`).
:param bytes nonce: Nonce to be used. If None, nonce of
``nonce_size`` will be randomly generated.
:param int nonce_size: Size of the automatically generated nonce.
Defaults to :const:`NONCE_SIZE`.
:param .JWASignature alg:
"""
nonce_size = cls.NONCE_SIZE if nonce_size is None else nonce_size
nonce = os.urandom(nonce_size) if nonce is None else nonce
msg_with_nonce = nonce + msg
sig = alg.sign(key, nonce + msg)
logger.debug('%r signed as %r', msg_with_nonce, sig)
return cls(alg=alg, sig=sig, nonce=nonce,
jwk=alg.kty(key=key.public_key()))
def verify(self, msg):
"""Verify the signature.
:param bytes msg: Message that was used in signing.
"""
# self.alg is not Field, but JWA | pylint: disable=no-member
return self.alg.verify(self.jwk.key, self.nonce + msg, self.sig)

View file

@ -1,94 +0,0 @@
"""Tests for acme.sig."""
import unittest
from acme import jose
from acme import test_util
KEY = test_util.load_rsa_private_key('rsa512_key.pem')
class SignatureTest(unittest.TestCase):
# pylint: disable=too-many-instance-attributes
"""Tests for acme.sig.Signature."""
def setUp(self):
self.msg = b'message'
self.sig = (b'IC\xd8*\xe7\x14\x9e\x19S\xb7\xcf\xec3\x12\xe2\x8a\x03'
b'\x98u\xff\xf0\x94\xe2\xd7<\x8f\xa8\xed\xa4KN\xc3\xaa'
b'\xb9X\xc3w\xaa\xc0_\xd0\x05$y>l#\x10<\x96\xd2\xcdr\xa3'
b'\x1b\xa1\xf5!f\xef\xc64\xb6\x13')
self.nonce = b'\xec\xd6\xf2oYH\xeb\x13\xd5#q\xe0\xdd\xa2\x92\xa9'
self.alg = jose.RS256
self.jwk = jose.JWKRSA(key=KEY.public_key())
b64sig = ('SUPYKucUnhlTt8_sMxLiigOYdf_wlOLXPI-o7aRLTsOquVjDd6r'
'AX9AFJHk-bCMQPJbSzXKjG6H1IWbvxjS2Ew')
b64nonce = '7Nbyb1lI6xPVI3Hg3aKSqQ'
self.jsig_to = {
'nonce': b64nonce,
'alg': self.alg,
'jwk': self.jwk,
'sig': b64sig,
}
self.jsig_from = {
'nonce': b64nonce,
'alg': self.alg.to_partial_json(),
'jwk': self.jwk.to_partial_json(),
'sig': b64sig,
}
from acme.other import Signature
self.signature = Signature(
alg=self.alg, sig=self.sig, nonce=self.nonce, jwk=self.jwk)
def test_attributes(self):
self.assertEqual(self.signature.nonce, self.nonce)
self.assertEqual(self.signature.alg, self.alg)
self.assertEqual(self.signature.sig, self.sig)
self.assertEqual(self.signature.jwk, self.jwk)
def test_verify_good_succeeds(self):
self.assertTrue(self.signature.verify(self.msg))
def test_verify_bad_fails(self):
self.assertFalse(self.signature.verify(self.msg + b'x'))
@classmethod
def _from_msg(cls, *args, **kwargs):
from acme.other import Signature
return Signature.from_msg(*args, **kwargs)
def test_create_from_msg(self):
signature = self._from_msg(self.msg, KEY, self.nonce)
self.assertEqual(self.signature, signature)
def test_create_from_msg_random_nonce(self):
signature = self._from_msg(self.msg, KEY)
self.assertEqual(signature.alg, self.alg)
self.assertEqual(signature.jwk, self.jwk)
self.assertTrue(signature.verify(self.msg))
def test_to_partial_json(self):
self.assertEqual(self.signature.to_partial_json(), self.jsig_to)
def test_from_json(self):
from acme.other import Signature
self.assertEqual(
self.signature, Signature.from_json(self.jsig_from))
def test_from_json_non_schema_errors(self):
from acme.other import Signature
jwk = self.jwk.to_partial_json()
self.assertRaises(
jose.DeserializationError, Signature.from_json, {
'alg': 'RS256', 'sig': 'x', 'nonce': '', 'jwk': jwk})
self.assertRaises(
jose.DeserializationError, Signature.from_json, {
'alg': 'RS256', 'sig': '', 'nonce': 'x', 'jwk': jwk})
if __name__ == '__main__':
unittest.main() # pragma: no cover

View file

@ -360,7 +360,7 @@ def _handle_subset_cert_request(config, domains, cert):
br=os.linesep)
if config.expand or config.renew_by_default or zope.component.getUtility(
interfaces.IDisplay).yesno(question, "Expand", "Cancel",
cli_flag="--expand (or in some cases, --duplicate)"):
cli_flag="--expand"):
return "renew", cert
else:
reporter_util = zope.component.getUtility(interfaces.IReporter)

View file

@ -6,6 +6,7 @@ import logging
import os
import platform
import re
import six
import socket
import stat
import subprocess
@ -310,10 +311,13 @@ def enforce_domain_sanity(domain):
# Unicode
try:
domain = domain.encode('ascii').lower()
except UnicodeDecodeError:
raise errors.ConfigurationError(
"Internationalized domain names are not presently supported: {0}"
.format(domain))
except UnicodeError:
error_fmt = (u"Internationalized domain names "
"are not presently supported: {0}")
if isinstance(domain, six.text_type):
raise errors.ConfigurationError(error_fmt.format(domain))
else:
raise errors.ConfigurationError(str(error_fmt).format(domain))
# Remove trailing dot
domain = domain[:-1] if domain.endswith('.') else domain

View file

@ -323,5 +323,21 @@ class AddDeprecatedArgumentTest(unittest.TestCase):
self.assertTrue("--old-option" not in stdout.getvalue())
class EnforceDomainSanityTest(unittest.TestCase):
"""Test enforce_domain_sanity."""
def _call(self, domain):
from letsencrypt.le_util import enforce_domain_sanity
return enforce_domain_sanity(domain)
def test_nonascii_str(self):
self.assertRaises(errors.ConfigurationError, self._call,
u"eichh\u00f6rnchen.example.com".encode("utf-8"))
def test_nonascii_unicode(self):
self.assertRaises(errors.ConfigurationError, self._call,
u"eichh\u00f6rnchen.example.com")
if __name__ == "__main__":
unittest.main() # pragma: no cover