acme.verify.simple_http_simple_verify

acme/challenges.py

@@ -72,6 +72,8 @@ class SimpleHTTPResponse(ChallengeResponse):
            [RFC4648]", base64.b64decode ignores those characters
 
         """
+        # TODO: check that path combined with uri does not go above
+        # URI_ROOT_PATH!
         return len(self.path) <= 25
 
     @property

acme/verify.py

@@ -0,0 +1,33 @@
+"""Simple challenges verification utilities."""
+import logging
+
+import requests
+
+
+logger = logging.getLogger(__name__)
+
+
+def simple_http_simple_verify(response, chall, domain):
+    """Verify SimpleHTTP.
+
+    According to the ACME specification, "the ACME server MUST ignore
+    the certificate provided by the HTTPS server", so ``requests.get``
+    is called with ``verify=False``.
+
+    """
+    uri = response.uri(domain)
+    logger.debug("Verifying %s at %s...", chall.typ, uri)
+    try:
+        http_response = requests.get(uri, verify=False)
+    except requests.exceptions.RequestException as error:
+        logger.error("Unable to verify %s: %s", uri, error)
+        return False
+    logger.debug(
+        'Received %s. Headers: %s', http_response, http_response.headers)
+
+    good_token = http_response.text == chall.token
+    if not good_token:
+        logger.error(
+            "Unable to verify %s! Expected: %r, returned: %r.",
+            uri, chall.token, http_response.text)
+    return response.good_path and http_response and good_token

acme/verify_test.py

@@ -0,0 +1,43 @@
+"""Tests for acme.verify."""
+import unittest
+
+import mock
+import requests
+
+from acme import challenges
+
+
+class SimpleHTTPSimpleVerifyTest(unittest.TestCase):
+    """Tests for acme.verify.simple_http_simple_verify."""
+
+    def setUp(self):
+        self.chall = challenges.SimpleHTTP(token="foo")
+        self.resp_http = challenges.SimpleHTTPResponse(path="bar", tls=False)
+        self.resp_https = challenges.SimpleHTTPResponse(path="bar", tls=True)
+
+    @classmethod
+    def _call(cls, *args, **kwargs):
+        from acme.verify import simple_http_simple_verify
+        return simple_http_simple_verify(*args, **kwargs)
+
+    @mock.patch("acme.verify.requests.get")
+    def test_good_token(self, mock_get):
+        for resp in self.resp_http, self.resp_https:
+            mock_get.reset_mock()
+            mock_get.return_value = mock.MagicMock(text=self.chall.token)
+            self.assertTrue(self._call(resp, self.chall, "local"))
+            mock_get.assert_called_once_with(resp.uri("local"), verify=False)
+
+    @mock.patch("acme.verify.requests.get")
+    def test_bad_token(self, mock_get):
+        mock_get().text = self.chall.token + "!"
+        self.assertFalse(self._call(self.resp_http, self.chall, "local"))
+
+    @mock.patch("acme.verify.requests.get")
+    def test_connection_error(self, mock_get):
+        mock_get.side_effect = requests.exceptions.RequestException
+        self.assertFalse(self._call(self.resp_http, self.chall, "local"))
+
+
+if __name__ == '__main__':
+    unittest.main()  # pragma: no cover

letsencrypt/plugins/manual.py

@@ -1,22 +1,18 @@
 """Manual plugin."""
-import logging
 import os
 import sys
 
-import requests
 import zope.component
 import zope.interface
 
 from acme import challenges
 from acme import jose
+from acme import verify as acme_verify
 
 from letsencrypt import interfaces
 from letsencrypt.plugins import common
 
 
-logger = logging.getLogger(__name__)
-
-
 class ManualAuthenticator(common.Plugin):
     """Manual Authenticator.
 
@@ -61,9 +57,7 @@ s.serve_forever()" """
 
     According to the ACME specification, "the ACME server MUST ignore
     the certificate provided by the HTTPS server", so the first command
-    generates temporary self-signed certificate. For the same reason
+    generates temporary self-signed certificate.
-    ``requests.get`` in `_verify` sets ``verify=False``. Python HTTPS
-    server command serves the ``token`` on all URIs.
 
     """
 
@@ -109,7 +103,8 @@ binary for temporary key/certificate generation.""".replace("\n", "")
             uri=response.uri(achall.domain),
             command=self.template.format(achall=achall, response=response)))
 
-        if self._verify(achall, response):
+        if acme_verify.simple_http_simple_verify(
+                response, achall.challb, achall.domain):
             return response
         else:
             return None
@@ -121,21 +116,5 @@ binary for temporary key/certificate generation.""".replace("\n", "")
         sys.stdout.write(message)
         raw_input("Press ENTER to continue")
 
-    def _verify(self, achall, chall_response):  # pylint: disable=no-self-use
-        uri = chall_response.uri(achall.domain)
-        logger.debug("Verifying %s...", uri)
-        try:
-            response = requests.get(uri, verify=False)
-        except requests.exceptions.ConnectionError as error:
-            logger.exception(error)
-            return False
-
-        ret = response.text == achall.token
-        if not ret:
-            logger.error("Unable to verify %s! Expected: %r, returned: %r.",
-                         uri, achall.token, response.text)
-
-        return ret
-
     def cleanup(self, achalls):  # pylint: disable=missing-docstring,no-self-use
         pass  # pragma: no cover

letsencrypt/plugins/manual_test.py

@@ -2,7 +2,6 @@
 import unittest
 
 import mock
-import requests
 
 from acme import challenges
 
@@ -32,28 +31,25 @@ class ManualAuthenticatorTest(unittest.TestCase):
 
     @mock.patch("letsencrypt.plugins.manual.sys.stdout")
     @mock.patch("letsencrypt.plugins.manual.os.urandom")
-    @mock.patch("letsencrypt.plugins.manual.requests.get")
+    @mock.patch("acme.verify.simple_http_simple_verify")
     @mock.patch("__builtin__.raw_input")
-    def test_perform(self, mock_raw_input, mock_get, mock_urandom, mock_stdout):
+    def test_perform(self, mock_raw_input, mock_verify, mock_urandom,
+                     mock_stdout):
         mock_urandom.return_value = "foo"
-        mock_get().text = self.achalls[0].token
+        mock_verify.return_value = True
 
-        self.assertEqual(
+        resp = challenges.SimpleHTTPResponse(tls=False, path='Zm9v')
-            [challenges.SimpleHTTPResponse(tls=False, path='Zm9v')],
+        self.assertEqual([resp], self.auth.perform(self.achalls))
-            self.auth.perform(self.achalls))
         mock_raw_input.assert_called_once()
-        mock_get.assert_called_with(
+        mock_verify.assert_called_with(resp, self.achalls[0].challb, "foo.com")
-            "http://foo.com/.well-known/acme-challenge/Zm9v", verify=False)
 
         message = mock_stdout.write.mock_calls[0][1][0]
         self.assertTrue(self.achalls[0].token in message)
         self.assertTrue('Zm9v' in message)
 
-        mock_get().text = self.achalls[0].token + '!'
+        mock_verify.return_value = False
         self.assertEqual([None], self.auth.perform(self.achalls))
 
-        mock_get.side_effect = requests.exceptions.ConnectionError
-        self.assertEqual([None], self.auth.perform(self.achalls))
 
 if __name__ == "__main__":
     unittest.main()  # pragma: no cover

setup.py

@@ -61,7 +61,6 @@ letsencrypt_install_requires = [
     'pyrfc3339',
     'python2-pythondialog>=3.2.2rc1',  # Debian squeeze support, cf. #280
     'pytz',
-    'requests',
     'zope.component',
     'zope.interface',
     'M2Crypto',