Finished basic POP challenge with tests

letsencrypt/client/client.py

@@ -68,7 +68,8 @@ class Client(object):
         self.config = config
 
         if dv_auth is not None:
-            cont_auth = continuity_auth.ContinuityAuthenticator(config)
+            cont_auth = continuity_auth.ContinuityAuthenticator(config,
+                                                                installer)
             self.auth_handler = auth_handler.AuthHandler(
                 dv_auth, cont_auth, self.network, self.account)
         else:

letsencrypt/client/continuity_auth.py

@@ -6,6 +6,7 @@ from letsencrypt.acme import challenges
 from letsencrypt.client import achallenges
 from letsencrypt.client import errors
 from letsencrypt.client import interfaces
+from letsencrypt.client import proof_of_possession
 from letsencrypt.client import recovery_token
 
 
@@ -13,22 +14,30 @@ class ContinuityAuthenticator(object):
     """IAuthenticator for
     :const:`~letsencrypt.acme.challenges.ContinuityChallenge` class challenges.
 
-    :ivar rec_token: Performs "recoveryToken" challenges
+    :ivar rec_token: Performs "recoveryToken" challenges.
     :type rec_token: :class:`letsencrypt.client.recovery_token.RecoveryToken`
 
+    :ivar proof_of_pos: Performs "proofOfPossession" challenges.
+    :type proof_of_pos:
+        :class:`letsencrypt.client.proof_of_possession.Proof_of_Possession`
+
     """
     zope.interface.implements(interfaces.IAuthenticator)
 
     # This will have an installer soon for get_key/cert purposes
-    def __init__(self, config):
+    def __init__(self, config, installer):
         """Initialize Client Authenticator.
 
         :param config: Configuration.
         :type config: :class:`letsencrypt.client.interfaces.IConfig`
 
+        :param installer: Let's Encrypt Installer.
+        :type installer: :class:`letsencrypt.client.interfaces.IInstaller`
+
         """
         self.rec_token = recovery_token.RecoveryToken(
             config.server, config.rec_token_dir)
+        self.proof_of_pos = proof_of_possession.ProofOfPossession(installer)
 
     def get_chall_pref(self, unused_domain):  # pylint: disable=no-self-use
         """Return list of challenge preferences."""
@@ -38,7 +47,9 @@ class ContinuityAuthenticator(object):
         """Perform client specific challenges for IAuthenticator"""
         responses = []
         for achall in achalls:
-            if isinstance(achall, achallenges.RecoveryToken):
+            if isinstance(achall, achallenges.ProofOfPossession):
+                responses.append(self.proof_of_pos.perform(achall))
+            elif isinstance(achall, achallenges.RecoveryToken):
                 responses.append(self.rec_token.perform(achall))
             else:
                 raise errors.LetsEncryptContAuthError("Unexpected Challenge")
@@ -49,5 +60,5 @@ class ContinuityAuthenticator(object):
         for achall in achalls:
             if isinstance(achall, achallenges.RecoveryToken):
                 self.rec_token.cleanup(achall)
-            else:
+            elif not isinstance(achall, achallenges.ProofOfPossession):
                 raise errors.LetsEncryptContAuthError("Unexpected Challenge")

letsencrypt/client/proof_of_possession.py

@@ -1,8 +1,4 @@
-"""Proof of Possession Identifier Validation Challenge.
-
-Based on draft-barnes-acme, section 6.5.
-
-"""
+"""Proof of Possession Identifier Validation Challenge."""
 import M2Crypto
 import os
 import zope.component
@@ -14,22 +10,17 @@ from letsencrypt.client import interfaces
 from letsencrypt.client.display import util as display_util
 
 
-class ProofOfPossession(object):
+class ProofOfPossession(object): # pylint: disable=too-few-public-methods
     """Proof of Possession Identifier Validation Challenge.
 
     Based on draft-barnes-acme, section 6.5.
 
+    :ivar installer: Installer object
+    :type installer: :class:`~letsencrypt.client.interfaces.IInstaller`
+
     """
-    def __init__(self, certs_keys):
-        """Initializes the object with known certificates and keys.
-
-        :param list certs_keys: tuples with form `[(cert, key, path)]`, where:
-            - `cert` - str path to certificate file
-            - `key` - str path to associated key file
-            - `path` - file path to configuration file
-
-        """
-        self.certs_keys = certs_keys
+    def __init__(self, installer):
+        self.installer = installer
 
     def perform(self, achall):
         """Perform the Proof of Possession Challenge.
@@ -46,18 +37,19 @@ class ProofOfPossession(object):
                 achall.challb.alg in [jose.HS256, jose.HS384, jose.HS512]):
             return None
 
-        for cert, prv_key, _ in self.certs_keys:
-            der_key = M2Crypto.X509.load_cert(cert).get_pubkey().as_der()
-            cert_key = challb.alg.kty.load(der_key)
-            if cert_key == challb.hints.jwk:
-                return _gen_response(achall, key)
+        # This will work regardless of how JWKES is implemented
+        for cert, key, _ in self.installer.get_all_certs_keys():
+            der_cert_key = M2Crypto.X509.load_cert(cert).get_pubkey().as_der()
+            cert_key = achall.challb.alg.kty.load(der_cert_key)
+            if cert_key == achall.challb.hints.jwk:
+                return self._gen_response(achall, key)
 
         # Is there are different prompt we should give the user?
-        code, prv_key = zope.component.getUtility(
-            interfaces.IDsiplay).input(
+        code, key = zope.component.getUtility(
+            interfaces.IDisplay).input(
                 "Path to private key for identifier: %s " % achall.domain)
         if code != display_util.CANCEL:
-            return _gen_response(achall, prv_key)
+            return self._gen_response(achall, key)
 
         # If we get here, the key wasn't found
         return False
@@ -68,8 +60,8 @@ class ProofOfPossession(object):
         :param challb: Proof of Possession Challenge
         :type challb: :class:`letsencrypt.acme.challenges.ProofOfPossession`
 
-        :param str key_path: Path to the private key corresponding to the
-            hinted to public key
+        :param str key_path: Path to the key corresponding to the hinted to
+            public key.
 
         :returns: Response or None/False if the challenge cannot be completed
         :rtype: :class:`letsencrypt.acme.challenges.ProofOfPossessionResponse'
@@ -78,11 +70,13 @@ class ProofOfPossession(object):
         """
 
         if os.path.isfile(key_path):
-            with key as open(key_path, 'rb'):
+            with open(key_path, 'rb') as key:
                 try:
                     jwk = challb.alg.kty.load(key.read())
                 except (IndexError, ValueError, TypeError):
-                    return False    
-            sig = other.Signature.from_msg(challb.nonce, jwk, alg=challb.alg)
+                    return False
+            # If JWKES doesn't have a key attribute, this needs to be modified
+            sig = other.Signature.from_msg(challb.nonce, jwk.key,
+                                           alg=challb.alg)
             return challenges.ProofOfPossessionResponse(nonce=challb.nonce,
                                                         signature=sig)

letsencrypt/client/tests/continuity_auth_test.py

@@ -16,9 +16,11 @@ class PerformTest(unittest.TestCase):
         from letsencrypt.client.continuity_auth import ContinuityAuthenticator
 
         self.auth = ContinuityAuthenticator(
-            mock.MagicMock(server="demo_server.org"))
+            mock.MagicMock(server="demo_server.org"), None)
         self.auth.rec_token.perform = mock.MagicMock(
             name="rec_token_perform", side_effect=gen_client_resp)
+        self.auth.proof_of_pos.perform = mock.MagicMock(
+            name="proof_of_pos_perform", side_effect=gen_client_resp)
 
     def test_rec_token1(self):
         token = achallenges.RecoveryToken(challb=None, domain="0")
@@ -36,6 +38,24 @@ class PerformTest(unittest.TestCase):
         for i in xrange(5):
             self.assertEqual(responses[i], "RecoveryToken%d" % i)
 
+    def test_pop_and_rec_token(self):
+        achalls = []
+        for i in xrange(4):
+            if i % 2 == 0:
+                achalls.append(achallenges.RecoveryToken(challb=None,
+                                                         domain=str(i)))
+            else:
+                achalls.append(achallenges.ProofOfPossession(challb=None,
+                                                             domain=str(i)))
+        responses = self.auth.perform(achalls)
+
+        self.assertEqual(len(responses), 4)
+        for i in xrange(4):
+            if i % 2 == 0:
+                self.assertEqual(responses[i], "RecoveryToken%d" % i)
+            else:
+                self.assertEqual(responses[i], "ProofOfPossession%d" % i)
+
     def test_unexpected(self):
         self.assertRaises(
             errors.LetsEncryptContAuthError, self.auth.perform, [
@@ -43,7 +63,8 @@ class PerformTest(unittest.TestCase):
 
     def test_chall_pref(self):
         self.assertEqual(
-            self.auth.get_chall_pref("example.com"), [challenges.RecoveryToken])
+            self.auth.get_chall_pref("example.com"),
+            [challenges.ProofOfPossession, challenges.RecoveryToken])
 
 
 class CleanupTest(unittest.TestCase):
@@ -53,7 +74,7 @@ class CleanupTest(unittest.TestCase):
         from letsencrypt.client.continuity_auth import ContinuityAuthenticator
 
         self.auth = ContinuityAuthenticator(
-            mock.MagicMock(server="demo_server.org"))
+            mock.MagicMock(server="demo_server.org"), None)
         self.mock_cleanup = mock.MagicMock(name="rec_token_cleanup")
         self.auth.rec_token.cleanup = self.mock_cleanup
 

letsencrypt/client/tests/proof_of_possession_test.py

@@ -0,0 +1,78 @@
+"""Tests for proof_of_possession.py"""
+import Crypto.PublicKey.RSA
+import os
+import pkg_resources
+import unittest
+
+import mock
+
+from letsencrypt.acme import challenges
+from letsencrypt.acme import jose
+from letsencrypt.client import achallenges
+from letsencrypt.client import proof_of_possession
+from letsencrypt.client.display import util as display_util
+
+
+BASE_PACKAGE = "letsencrypt.client.tests"
+CERT0_PATH = pkg_resources.resource_filename(
+    BASE_PACKAGE, os.path.join("testdata", "cert.pem"))
+CERT1_PATH = pkg_resources.resource_filename(
+    BASE_PACKAGE, os.path.join("testdata", "cert-san.pem"))
+CERT2_PATH = pkg_resources.resource_filename(
+    BASE_PACKAGE, os.path.join("testdata", "matching_cert.pem"))
+KEY_PATH = pkg_resources.resource_filename(
+    BASE_PACKAGE, os.path.join("testdata", "rsa512_key.pem"))
+KEY = Crypto.PublicKey.RSA.importKey(pkg_resources.resource_string(
+    BASE_PACKAGE, os.path.join('testdata', 'rsa512_key.pem'))).publickey()
+
+
+class ProofOfPossessionTest(unittest.TestCase):
+    def setUp(self):
+        self.installer = mock.MagicMock()
+        self.installer.get_all_certs_keys.return_value = zip(
+            [CERT0_PATH, CERT1_PATH, CERT2_PATH], 3 * [KEY_PATH], 3 * [None])
+        self.proof_of_pos = proof_of_possession.ProofOfPossession(
+            self.installer)
+
+        hints = challenges.ProofOfPossession.Hints(
+            jwk=jose.JWKRSA(key=KEY), cert_fingerprints=(),
+            certs=(), serial_numbers=(), subject_key_identifiers=(),
+            issuers=(), authorized_for=())
+        challenge = challenges.ProofOfPossession(
+            alg=jose.RS256, nonce='zczv4HMLVe_0kimJ25Juig', hints=hints)
+        self.achall = achallenges.ProofOfPossession(
+            challb=challenge, domain="example.com")
+
+    def test_perform_no_input(self):
+        response = self.proof_of_pos.perform(self.achall)
+        self.assertTrue(response.verify())
+
+    @mock.patch("letsencrypt.client.recovery_token.zope.component.getUtility")
+    def test_perform_with_input(self, mock_input):
+        # Remove the matching certificate
+        self.installer.get_all_certs_keys.return_value.pop()
+        mock_input().input.side_effect = [(display_util.CANCEL, ""),
+                                          (display_util.OK, CERT0_PATH),
+                                          (display_util.OK, KEY_PATH)]
+
+        response = self.proof_of_pos.perform(self.achall)
+        self.assertFalse(response)
+
+        response = self.proof_of_pos.perform(self.achall)
+        self.assertFalse(response)
+
+        response = self.proof_of_pos.perform(self.achall)
+        self.assertTrue(response.verify())
+
+    def test_perform_bad_challenge(self):
+        hints = challenges.ProofOfPossession.Hints(
+            jwk=jose.jwk.JWKOct(key=KEY), cert_fingerprints=(),
+            certs=(), serial_numbers=(), subject_key_identifiers=(),
+            issuers=(), authorized_for=())
+        challenge = challenges.ProofOfPossession(
+            alg=jose.HS512, nonce='zczv4HMLVe_0kimJ25Juig', hints=hints)
+        self.achall = achallenges.ProofOfPossession(
+            challb=challenge, domain="example.com")
+
+        response = self.proof_of_pos.perform(self.achall)
+        self.assertEqual(response, None)

letsencrypt/client/tests/testdata/matching_cert.pem

@@ -0,0 +1,14 @@
+-----BEGIN CERTIFICATE-----
+MIICNzCCAeGgAwIBAgIJALizm9Y3q620MA0GCSqGSIb3DQEBCwUAMHcxCzAJBgNV
+BAYTAlVTMREwDwYDVQQIDAhNaWNoaWdhbjESMBAGA1UEBwwJQW5uIEFyYm9yMSsw
+KQYDVQQKDCJVbml2ZXJzaXR5IG9mIE1pY2hpZ2FuIGFuZCB0aGUgRUZGMRQwEgYD
+VQQDDAtleGFtcGxlLmNvbTAeFw0xNTA1MDkwMDI0NTJaFw0xNjA1MDgwMDI0NTJa
+MHcxCzAJBgNVBAYTAlVTMREwDwYDVQQIDAhNaWNoaWdhbjESMBAGA1UEBwwJQW5u
+IEFyYm9yMSswKQYDVQQKDCJVbml2ZXJzaXR5IG9mIE1pY2hpZ2FuIGFuZCB0aGUg
+RUZGMRQwEgYDVQQDDAtleGFtcGxlLmNvbTBcMA0GCSqGSIb3DQEBAQUAA0sAMEgC
+QQD0thFxUTc2v6qV55wRxfwnBUOeN4bVfu5ywJqy65kzR7T1yZi5TPEiQyM7/3Hg
+BVy9ddFc8RX4vNZaR+ROXNEzAgMBAAGjUDBOMB0GA1UdDgQWBBRJieHEVSHKmBk0
+mTExx1erzlylCjAfBgNVHSMEGDAWgBRJieHEVSHKmBk0mTExx1erzlylCjAMBgNV
+HRMEBTADAQH/MA0GCSqGSIb3DQEBCwUAA0EABT/nlpqOaanFSLZmWIrKv0zt63k4
+bmWNMA8fYT45KYpLomsW8qXdpC82IlVKfNk7fW0UYT3HOeDSJRcycxNCTQ==
+-----END CERTIFICATE-----