DVSNIResponse.gen_cert, fix verify_cert, add tests.

acme/acme/challenges.py

@@ -6,6 +6,8 @@ import logging
 import os
 import socket
 
+from cryptography.hazmat.backends import default_backend
+from cryptography import x509
 import OpenSSL
 import requests
 
@@ -245,10 +247,25 @@ class DVSNIResponse(ChallengeResponse):
         return z.hexdigest().encode()
 
     def z_domain(self, chall):
-        """Domain name for certificate subjectAltName."""
+        """Domain name for certificate subjectAltName.
+
+        :rtype bytes:
+
+        """
         return self.z(chall) + self.DOMAIN_SUFFIX
 
-    def simple_verify(self, chall, domain, key, **kwargs):
+    def gen_cert(self, chall, domain, key):
+        """Generate DVSNI certificate.
+
+        :param .DVSNI chall: Corresponding challenge.
+        :param unicode domain:
+        :param OpenSSL.crypto.PKey
+
+        """
+        return crypto_util.gen_ss_cert(key, [
+            domain, chall.nonce_domain.decode(), self.z_domain(chall).decode()])
+
+    def simple_verify(self, chall, domain, public_key, **kwargs):
         """Simple verify.
 
         Probes DVSNI certificate and checks it using `verify_cert`;
@@ -260,37 +277,47 @@ class DVSNIResponse(ChallengeResponse):
         except errors.Error as error:
             logger.debug(error, exc_info=True)
             return False
-        # TODO: check "It is a valid self-signed certificate" and
-        # return False if not
-        return self.verify_cert(chall, domain, key, cert)
+        return self.verify_cert(chall, domain, public_key, cert)
 
-    def verify_cert(self, chall, domain, key, cert):
+    def verify_cert(self, chall, domain, public_key, cert):
         """Verify DVSNI certificate.
 
         :param .challenges.DVSNI chall: Corresponding challenge.
         :param str domain: Domain name being validated.
-        :param OpenSSL.crypto.PKey key: Public key for the key pair
+        :param public_key: Public key for the key pair
             being authorized. If ``None`` key verification is not
             performed!
+        :type public_key:
+            `~cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey`
+            or
+            `~cryptography.hazmat.primitives.asymmetric.dsa.DSAPublicKey`
+            or
+            `~cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePublicKey`
+            wrapped in `.ComparableKey
+        :param OpenSSL.crypto.X509 cert:
 
         :returns: ``True`` iff client's control of the domain has been
             verified, ``False`` otherwise.
         :rtype: bool
 
         """
+        # TODO: check "It is a valid self-signed certificate" and
+        # return False if not
+
         # pylint: disable=protected-access
         sans = crypto_util._pyopenssl_cert_or_req_san(cert)
         logging.debug('Certificate %s. SANs: %s', cert.digest('sha1'), sans)
 
-        key_filetype = OpenSSL.crypto.FILETYPE_ASN1
-        if key is None:
-            logging.warn('No key verification is performed')
-        keys_match = key is None or OpenSSL.crypto.dump_privatekey(
-            key_filetype, key) == OpenSSL.crypto.dump_privatekey(
-                key_filetype, cert.get_pubkey())
+        cert = x509.load_der_x509_certificate(
+            OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_ASN1, cert),
+            default_backend())
 
-        return (keys_match and domain in sans and
-                self.z_domain(chall).decode() in sans)
+        if public_key is None:
+            logging.warn('No key verification is performed')
+        elif public_key != jose.ComparableKey(cert.public_key()):
+            return False
+
+        return domain in sans and self.z_domain(chall).decode() in sans
 
 
 @Challenge.register

acme/acme/challenges_test.py

@@ -210,26 +210,31 @@ class DVSNIResponseTest(unittest.TestCase):
 
     def setUp(self):
         from acme.challenges import DVSNIResponse
-        self.msg = DVSNIResponse(
-            s=b'\xf5\xd6\xe3\xb2]\xe0L\x0bN\x9cKJ\x14I\xa1K\xa3#\xf9\xa8'
-              b'\xcd\x8c7\x0e\x99\x19)\xdc\xb7\xf3\x9bw')
+        # pylint: disable=invalid-name
+        s = '9dbjsl3gTAtOnEtKFEmhS6Mj-ajNjDcOmRkp3Lfzm3c'
+        self.msg = DVSNIResponse(s=jose.decode_b64jose(s))
         self.jmsg = {
             'resource': 'challenge',
             'type': 'dvsni',
-            's': '9dbjsl3gTAtOnEtKFEmhS6Mj-ajNjDcOmRkp3Lfzm3c',
+            's': s,
         }
 
-    def test_z_and_domain(self):
         from acme.challenges import DVSNI
-        challenge = DVSNI(
-            r=b"O*\xb4-\xad\xec\x95>\xed\xa9\r0\x94\xe8\x97\x9c&6"
-              b"\xbf'\xb3\xed\x9a9nX\x0f'\\m\xe7\x12",
-            nonce=int('439736375371401115242521957580409149254868992063'
-                      '44333654741504362774620418661'))
+        self.chall = DVSNI(
+            r=jose.decode_b64jose('Tyq0La3slT7tqQ0wlOiXnCY2vyez7Zo5blgPJ1xt5xI'),
+            nonce=jose.decode_b64jose('a82d5ff8ef740d12881f6d3c2277ab2e'))
+        self.z = (b'38e612b0397cc2624a07d351d7ef50e4'
+                  b'6134c0213d9ed52f7d7c611acaeed41b')
+        self.domain = 'foo.com'
+        self.key = test_util.load_pyopenssl_private_key('rsa512_key.pem')
+        self.public_key = test_util.load_rsa_private_key(
+            'rsa512_key.pem').public_key()
+
+    def test_z_and_domain(self):
         # pylint: disable=invalid-name
-        z = b'38e612b0397cc2624a07d351d7ef50e46134c0213d9ed52f7d7c611acaeed41b'
-        self.assertEqual(z, self.msg.z(challenge))
-        self.assertEqual(z + b'.acme.invalid', self.msg.z_domain(challenge))
+        self.assertEqual(self.z, self.msg.z(self.chall))
+        self.assertEqual(
+            self.z + b'.acme.invalid', self.msg.z_domain(self.chall))
 
     def test_to_partial_json(self):
         self.assertEqual(self.jmsg, self.msg.to_partial_json())
@@ -258,18 +263,28 @@ class DVSNIResponseTest(unittest.TestCase):
         chall = mock.Mock()
         chall.probe_cert.side_effect = errors.Error
         self.assertFalse(self.msg.simple_verify(
-            chall=chall, domain=None, key=None))
+            chall=chall, domain=None, public_key=None))
 
-    def test_verify_cert_postive(self):
-        pass  # XXX
+    def test_gen_verify_cert_postive_no_key(self):
+        cert = self.msg.gen_cert(self.chall, self.domain, self.key)
+        self.assertTrue(self.msg.verify_cert(
+            self.chall, self.domain, public_key=None, cert=cert))
 
-    def test_verify_cert_negative(self):
-        chall = mock.MagicMock()
+    def test_gen_verify_cert_postive_with_key(self):
+        cert = self.msg.gen_cert(self.chall, self.domain, self.key)
+        self.assertTrue(self.msg.verify_cert(
+            self.chall, self.domain, public_key=self.public_key, cert=cert))
+
+    def test_gen_verify_cert_negative_with_wrong_key(self):
+        cert = self.msg.gen_cert(self.chall, self.domain, self.key)
+        key = test_util.load_rsa_private_key('rsa256_key.pem').public_key()
         self.assertFalse(self.msg.verify_cert(
-            chall, domain="example.com", key=None,
-            cert=OpenSSL.crypto.load_certificate(
-                OpenSSL.crypto.FILETYPE_PEM,
-                test_util.load_vector('cert.pem'))))
+            self.chall, self.domain, public_key=key, cert=cert))
+
+    def test_gen_verify_cert_negative(self):
+        cert = self.msg.gen_cert(self.chall, self.domain + 'x', self.key)
+        self.assertFalse(self.msg.verify_cert(
+            self.chall, self.domain, public_key=None, cert=cert))
 
 
 class RecoveryContactTest(unittest.TestCase):

acme/acme/crypto_util.py

@@ -125,7 +125,7 @@ def _pyopenssl_cert_or_req_san(cert_or_req):
     :type cert_or_req: `OpenSSL.crypto.X509` or `OpenSSL.crypto.X509Req`.
 
     :returns: A list of Subject Alternative Names.
-    :rtype: list
+    :rtype: `list` of `unicode`
 
     """
     # constants based on implementation of
@@ -153,3 +153,43 @@ def _pyopenssl_cert_or_req_san(cert_or_req):
 
     return [part.split(part_separator)[1] for parts in san_extensions
             for part in parts if part.startswith(prefix)]
+
+
+def gen_ss_cert(key, domains, not_before=None, validity=(7 * 24 * 60 * 60)):
+    """Generate new self-signed certificate.
+
+    :type domains: `list` of `unicode`
+    :param OpenSSL.crypto.PKey key:
+
+    Uses key and contains all domains.
+
+    """
+    assert domains, "Must provide one or more hostnames for the cert."
+    cert = OpenSSL.crypto.X509()
+    cert.set_serial_number(1337)
+    cert.set_version(2)
+
+    extensions = [
+        OpenSSL.crypto.X509Extension(
+            b"basicConstraints", True, b"CA:TRUE, pathlen:0"),
+    ]
+
+    cert.get_subject().CN = domains[0]
+    # TODO: what to put into cert.get_subject()?
+    cert.set_issuer(cert.get_subject())
+
+    if len(domains) > 1:
+        extensions.append(OpenSSL.crypto.X509Extension(
+            b"subjectAltName",
+            critical=False,
+            value=b", ".join(b"DNS:" + d.encode() for d in domains)
+        ))
+
+    cert.add_extensions(extensions)
+
+    cert.gmtime_adj_notBefore(0 if not_before is None else not_before)
+    cert.gmtime_adj_notAfter(validity)
+
+    cert.set_pubkey(key)
+    cert.sign(key, "sha256")
+    return cert

acme/acme/jose/__init__.py

@@ -76,6 +76,7 @@ from acme.jose.jws import (
 
 from acme.jose.util import (
     ComparableX509,
+    ComparableKey,
     ComparableRSAKey,
     ImmutableMap,
 )

acme/acme/jose/util.py

@@ -66,14 +66,14 @@ class ComparableX509(object):  # pylint: disable=too-few-public-methods
         return '<{0}({1!r})>'.format(self.__class__.__name__, self._wrapped)
 
 
-class ComparableRSAKey(object):  # pylint: disable=too-few-public-methods
-    """Wrapper for `cryptography` RSA keys.
+class ComparableKey(object):  # pylint: disable=too-few-public-methods
+    """Comparable wrapper for `cryptography` keys.
 
-    Wraps around:
-    - `cryptography.hazmat.primitives.assymetric.RSAPrivateKey`
-    - `cryptography.hazmat.primitives.assymetric.RSAPublicKey`
+    See https://github.com/pyca/cryptography/issues/2122.
 
     """
+    __hash__ = NotImplemented
+
     def __init__(self, wrapped):
         self._wrapped = wrapped
 
@@ -85,19 +85,36 @@ class ComparableRSAKey(object):  # pylint: disable=too-few-public-methods
         if (not isinstance(other, self.__class__) or
                 self._wrapped.__class__ is not other._wrapped.__class__):
             return NotImplemented
-        # RSA*KeyWithSerialization requires cryptography>=0.8
-        if isinstance(self._wrapped, rsa.RSAPrivateKeyWithSerialization):
+        elif hasattr(self._wrapped, 'private_numbers'):
             return self.private_numbers() == other.private_numbers()
-        elif isinstance(self._wrapped, rsa.RSAPublicKeyWithSerialization):
+        elif hasattr(self._wrapped, 'public_numbers'):
             return self.public_numbers() == other.public_numbers()
         else:
-            return False  # we shouldn't reach here...
+            return NotImplemented
 
     def __ne__(self, other):
         return not self == other
 
+    def __repr__(self):
+        return '<{0}({1!r})>'.format(self.__class__.__name__, self._wrapped)
+
+    def public_key(self):
+        """Get wrapped public key."""
+        return self.__class__(self._wrapped.public_key())
+
+
+class ComparableRSAKey(ComparableKey):  # pylint: disable=too-few-public-methods
+    """Wrapper for `cryptography` RSA keys.
+
+    Wraps around:
+    - `cryptography.hazmat.primitives.assymetric.RSAPrivateKey`
+    - `cryptography.hazmat.primitives.assymetric.RSAPublicKey`
+
+    """
+
     def __hash__(self):
         # public_numbers() hasn't got stable hash!
+        # https://github.com/pyca/cryptography/issues/2143
         if isinstance(self._wrapped, rsa.RSAPrivateKeyWithSerialization):
             priv = self.private_numbers()
             pub = priv.public_numbers
@@ -107,13 +124,6 @@ class ComparableRSAKey(object):  # pylint: disable=too-few-public-methods
             pub = self.public_numbers()
             return hash((self.__class__, pub.n, pub.e))
 
-    def __repr__(self):
-        return '<{0}({1!r})>'.format(self.__class__.__name__, self._wrapped)
-
-    def public_key(self):
-        """Get wrapped public key."""
-        return self.__class__(self._wrapped.public_key())
-
 
 class ImmutableMap(collections.Mapping, collections.Hashable):
     # pylint: disable=too-few-public-methods

acme/acme/test_util.py

@@ -55,3 +55,9 @@ def load_rsa_private_key(*names):
                            serialization.load_der_private_key)
     return jose.ComparableRSAKey(loader(
         load_vector(*names), password=None, backend=default_backend()))
+
+def load_pyopenssl_private_key(*names):
+    """Load pyOpenSSL private key."""
+    loader = _guess_loader(
+        names[-1], OpenSSL.crypto.FILETYPE_PEM, OpenSSL.crypto.FILETYPE_ASN1)
+    return OpenSSL.crypto.load_privatekey(loader, load_vector(*names))

letsencrypt-nginx/letsencrypt_nginx/configurator.py

@@ -7,9 +7,11 @@ import socket
 import subprocess
 import sys
 
+import OpenSSL
 import zope.interface
 
 from acme import challenges
+from acme import crypto_util as acme_crypto_util
 
 from letsencrypt import achallenges
 from letsencrypt import constants as core_constants
@@ -271,14 +273,17 @@ class NginxConfigurator(common.Plugin):
     def _get_snakeoil_paths(self):
         # TODO: generate only once
         tmp_dir = os.path.join(self.config.work_dir, "snakeoil")
-        key = crypto_util.init_save_key(
+        le_key = crypto_util.init_save_key(
             key_size=1024, key_dir=tmp_dir, keyname="key.pem")
-        cert_pem = crypto_util.make_ss_cert(
-            key.pem, domains=[socket.gethostname()])
-        cert = os.path.join(tmp_dir, "cert.pem")
-        with open(cert, 'w') as cert_file:
+        key = OpenSSL.crypto.load_privatekey(
+            OpenSSL.crypto.FILETYPE_PEM, le_key.pem)
+        cert = acme_crypto_util.gen_ss_cert(key, domains=[socket.gethostname()])
+        cert_path = os.path.join(tmp_dir, "cert.pem")
+        cert_pem = OpenSSL.crypto.dump_certificate(
+            OpenSSL.crypto.FILETYPE_PEM, cert)
+        with open(cert_path, 'w') as cert_file:
             cert_file.write(cert_pem)
-        return cert, key.file
+        return cert_path, le_key.file
 
     def _make_server_ssl(self, vhost):
         """Make a server SSL.

letsencrypt/achallenges.py

@@ -17,6 +17,8 @@ Note, that all annotated challenges act as a proxy objects::
   achall.token == challb.token
 
 """
+import OpenSSL
+
 from acme import challenges
 from acme.jose import util as jose_util
 
@@ -48,7 +50,7 @@ class DVSNI(AnnotatedChallenge):
     acme_type = challenges.DVSNI
 
     def gen_cert_and_response(self, s=None):  # pylint: disable=invalid-name
-        """Generate a DVSNI cert and save it to filepath.
+        """Generate a DVSNI cert and response.
 
         :returns: ``(cert_pem, response)`` tuple,  where ``cert_pem`` is the PEM
             encoded  certificate and ``response`` is an instance
@@ -56,9 +58,12 @@ class DVSNI(AnnotatedChallenge):
         :rtype: tuple
 
         """
+        key = crypto_util.private_jwk_to_pyopenssl(self.key)
         response = challenges.DVSNIResponse(s=s)
-        cert_pem = crypto_util.make_ss_cert(self.key, [
-            self.domain, self.nonce_domain, response.z_domain(self.challb)])
+        cert = response.gen_cert(self.challb.chall, self.domain, key)
+        cert_pem = OpenSSL.crypto.dump_certificate(
+            OpenSSL.crypto.FILETYPE_PEM, cert)
+
         return cert_pem, response
 
 

letsencrypt/crypto_util.py

@@ -12,7 +12,6 @@ from cryptography.hazmat.primitives import serialization
 import OpenSSL
 
 from acme import crypto_util as acme_crypto_util
-from acme import jose
 
 from letsencrypt import errors
 from letsencrypt import le_util
@@ -216,49 +215,13 @@ def pyopenssl_load_certificate(data):
     return _pyopenssl_load(data, OpenSSL.crypto.load_certificate)
 
 
-def make_ss_cert(key, domains, not_before=None,
-                 validity=(7 * 24 * 60 * 60)):
-    """Returns new self-signed cert in PEM form.
-
-    Uses key and contains all domains.
-
-    """
-    if isinstance(key, jose.JWK):
-        key = key.key.private_bytes(
-            encoding=serialization.Encoding.PEM,
-            format=serialization.PrivateFormat.TraditionalOpenSSL,
-            encryption_algorithm=serialization.NoEncryption())
-
-    assert domains, "Must provide one or more hostnames for the cert."
-    pkey = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, key)
-    cert = OpenSSL.crypto.X509()
-    cert.set_serial_number(1337)
-    cert.set_version(2)
-
-    extensions = [
-        OpenSSL.crypto.X509Extension(
-            "basicConstraints", True, 'CA:TRUE, pathlen:0'),
-    ]
-
-    cert.get_subject().CN = domains[0]
-    # TODO: what to put into cert.get_subject()?
-    cert.set_issuer(cert.get_subject())
-
-    if len(domains) > 1:
-        extensions.append(OpenSSL.crypto.X509Extension(
-            "subjectAltName",
-            critical=False,
-            value=", ".join("DNS:%s" % d for d in domains)
-        ))
-
-    cert.add_extensions(extensions)
-
-    cert.gmtime_adj_notBefore(0 if not_before is None else not_before)
-    cert.gmtime_adj_notAfter(validity)
-
-    cert.set_pubkey(pkey)
-    cert.sign(pkey, "sha256")
-    return OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, cert)
+def private_jwk_to_pyopenssl(jwk):
+    """Convert private JWK to pyOpenSSL key."""
+    key_pem = jwk.key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.TraditionalOpenSSL,
+        encryption_algorithm=serialization.NoEncryption())
+    return OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, key_pem)
 
 
 def _get_sans_from_cert_or_req(

letsencrypt/tests/crypto_util_test.py

@@ -160,15 +160,6 @@ class ValidPrivkeyTest(unittest.TestCase):
         self.assertFalse(self._call('foo bar'))
 
 
-class MakeSSCertTest(unittest.TestCase):
-    # pylint: disable=too-few-public-methods
-    """Tests for letsencrypt.crypto_util.make_ss_cert."""
-
-    def test_it(self):  # pylint: disable=no-self-use
-        from letsencrypt.crypto_util import make_ss_cert
-        make_ss_cert(RSA512_KEY, ['example.com', 'www.example.com'])
-
-
 class GetSANsFromCertTest(unittest.TestCase):
     """Tests for letsencrypt.crypto_util.get_sans_from_cert."""