Merge pull request #106 from untitaker/tls_fingerprints

TLS fingerprints (2.1)

AUTHORS.rst

@@ -7,3 +7,4 @@ In alphabetical order:
 - Clément Mondon
 - Julian Mehne
 - Markus Unterwaditzer
+- Thomas Weißschuh

CONTRIBUTING.rst

@@ -20,6 +20,6 @@
 
   * But not because you wrote too few tests.
 
-  * Add yourself to ``CONTRIBUTORS.rst``. Don't add anything to
+  * Add yourself to ``AUTHORS.rst``. Don't add anything to
     ``CHANGELOG.rst``, I do that myself shortly before the release. You can
     help by writing meaningful commit messages.

build.sh

@@ -33,7 +33,7 @@ _davserver() {
 }
 
 command__install_tests() {
-    $PIP_INSTALL pytest pytest-xprocess
+    $PIP_INSTALL pytest pytest-xprocess git+https://github.com/mitsuhiko/werkzeug/ pytest-localserver
     _optimize_pip
     _davserver $DAV_SERVER
     [ "$TRAVIS" != "true" ] || $PIP_INSTALL coverage coveralls

tests/storage/dav/servers/radicale/__init__.py

@@ -8,8 +8,6 @@ import pytest
 import wsgi_intercept
 import wsgi_intercept.requests_intercept
 
-wsgi_intercept.requests_intercept.install()
-
 
 RADICALE_SCHEMA = '''
 create table collection (
@@ -93,10 +91,12 @@ class ServerMixin(object):
         do_the_radicale_dance(str(tmpdir))
         from radicale import Application
 
+        wsgi_intercept.requests_intercept.install()
         wsgi_intercept.add_wsgi_intercept('127.0.0.1', 80, Application)
 
         def teardown():
             wsgi_intercept.remove_wsgi_intercept('127.0.0.1', 80)
+            wsgi_intercept.requests_intercept.uninstall()
         request.addfinalizer(teardown)
 
     def get_storage_args(self, collection='test'):

tests/storage/test_http.py

@@ -41,7 +41,7 @@ def test_list(monkeypatch):
         u'\n'.join([u'BEGIN:VCALENDAR'] + items + [u'END:VCALENDAR'])
     ] * 2
 
-    def get(method, url, *a, **kw):
+    def get(self, method, url, *a, **kw):
         assert method == 'GET'
         assert url == collection_url
         r = Response()
@@ -52,7 +52,7 @@ def test_list(monkeypatch):
         r.encoding = 'ISO-8859-1'
         return r
 
-    monkeypatch.setattr('requests.request', get)
+    monkeypatch.setattr('requests.sessions.Session.request', get)
 
     s = HttpStorage(url=collection_url)
 

tests/utils/test_main.py

@@ -11,6 +11,8 @@ import click
 
 from click.testing import CliRunner
 import pytest
+import requests
+
 import vdirsyncer.utils as utils
 import vdirsyncer.doubleclick as doubleclick
 from vdirsyncer.utils.vobject import split_collection
@@ -178,3 +180,18 @@ def test_get_class_init_args_on_storage():
     all, required = utils.get_class_init_args(MemoryStorage)
     assert all == set(['collection', 'read_only', 'instance_name'])
     assert not required
+
+
+def test_request_ssl(httpsserver):
+    sha1 = '94:FD:7A:CB:50:75:A4:69:82:0A:F8:23:DF:07:FC:69:3E:CD:90:CA'
+    md5 = '19:90:F7:23:94:F2:EF:AB:2B:64:2D:57:3D:25:95:2D'
+
+    httpsserver.serve_content('')  # we need to serve something
+
+    with pytest.raises(requests.exceptions.SSLError) as excinfo:
+        utils.request('GET', httpsserver.url)
+    assert 'certificate verify failed' in str(excinfo.value)
+    utils.request('GET', httpsserver.url, verify=False)
+    utils.request('GET', httpsserver.url, verify=False,
+                  verify_fingerprint=sha1)
+    utils.request('GET', httpsserver.url, verify=False, verify_fingerprint=md5)

vdirsyncer/storage/dav.py

@@ -162,13 +162,15 @@ class DavSession(object):
     '''
 
     def __init__(self, url, username='', password='', verify=True, auth=None,
-                 useragent=USERAGENT, dav_header=None):
+                 useragent=USERAGENT, verify_fingerprint=None,
+                 dav_header=None):
         if username and not password:
             password = utils.get_password(username, url)
 
         self._settings = {
             'verify': prepare_verify(verify),
-            'auth': prepare_auth(auth, username, password)
+            'auth': prepare_auth(auth, username, password),
+            'verify_fingerprint': verify_fingerprint,
         }
         self.useragent = useragent
         self.url = url.rstrip('/') + '/'
@@ -222,6 +224,8 @@ class DavStorage(Storage):
     :param password: Password for authentication.
     :param verify: Verify SSL certificate, default True. This can also be a
         local path to a self-signed SSL certificate.
+    :param verify_fingerprint: Optional. SHA1 or MD5 fingerprint of the
+        expected server certificate.
     :param auth: Optional. Either ``basic``, ``digest`` or ``guess``. Default
         ``guess``. If you know yours, consider setting it explicitly for
         performance.
@@ -248,14 +252,15 @@ class DavStorage(Storage):
 
     def __init__(self, url, username='', password='', collection=None,
                  verify=True, auth=None, useragent=USERAGENT,
-                 unsafe_href_chars='@', **kwargs):
+                 unsafe_href_chars='@', verify_fingerprint=None, **kwargs):
         super(DavStorage, self).__init__(**kwargs)
 
         url = url.rstrip('/') + '/'
         if collection is not None:
             url = utils.urlparse.urljoin(url, collection)
         self.session = DavSession(url, username, password, verify, auth,
-                                  useragent, dav_header=self.dav_header)
+                                  useragent, verify_fingerprint,
+                                  dav_header=self.dav_header)
         self.collection = collection
         self.unsafe_href_chars = unsafe_href_chars
 
@@ -268,7 +273,8 @@ class DavStorage(Storage):
         if kwargs.pop('collection', None) is not None:
             raise TypeError('collection argument must not be given.')
         discover_args, _ = utils.split_dict(kwargs, lambda key: key in (
-            'username', 'password', 'verify', 'auth', 'useragent'
+            'username', 'password', 'verify', 'auth', 'useragent',
+            'verify_fingerprint',
         ))
         d = cls.discovery_class(DavSession(
             url=url, dav_header=cls.dav_header, **discover_args))

vdirsyncer/storage/http.py

@@ -54,6 +54,8 @@ class HttpStorage(Storage):
     :param password: Password for authentication.
     :param verify: Verify SSL certificate, default True. This can also be a
         local path to a self-signed SSL certificate.
+    :param verify_fingerprint: Optional. SHA1 or MD5 fingerprint of the
+        expected server certificate.
     :param auth: Optional. Either ``basic``, ``digest`` or ``guess``. Default
         ``guess``. If you know yours, consider setting it explicitly for
         performance.
@@ -81,7 +83,8 @@ class HttpStorage(Storage):
     _items = None
 
     def __init__(self, url, username='', password='', collection=None,
-                 verify=True, auth=None, useragent=USERAGENT, **kwargs):
+                 verify=True, auth=None, useragent=USERAGENT,
+                 verify_fingerprint=None, **kwargs):
         super(HttpStorage, self).__init__(**kwargs)
 
         if username and not password:
@@ -89,6 +92,7 @@ class HttpStorage(Storage):
 
         self._settings = {
             'verify': prepare_verify(verify),
+            'verify_fingerprint': verify_fingerprint,
             'auth': prepare_auth(auth, username, password),
             'latin1_fallback': False
         }

vdirsyncer/utils/__init__.py

@@ -11,6 +11,7 @@ import os
 import threading
 
 import requests
+from requests.packages.urllib3.poolmanager import PoolManager
 
 from .. import exceptions, log
 from ..doubleclick import click, ctx
@@ -165,13 +166,27 @@ def _password_from_keyring(username, host):
     return keyring.get_password(password_key_prefix + host, username)
 
 
+class _FingerprintAdapter(requests.adapters.HTTPAdapter):
+    def __init__(self, fingerprint=None, **kwargs):
+        self.fingerprint = str(fingerprint)
+        super(_FingerprintAdapter, self).__init__(**kwargs)
+
+    def init_poolmanager(self, connections, maxsize, block=False):
+        self.poolmanager = PoolManager(num_pools=connections,
+                                       maxsize=maxsize,
+                                       block=block,
+                                       assert_fingerprint=self.fingerprint)
+
+
 def request(method, url, data=None, headers=None, auth=None, verify=None,
-            session=None, latin1_fallback=True):
+            session=None, latin1_fallback=True, verify_fingerprint=None):
     '''
     Wrapper method for requests, to ease logging and mocking. Parameters should
     be the same as for ``requests.request``, except:
 
     :param session: A requests session object to use.
+    :param verify_fingerprint: Optional. SHA1 or MD5 fingerprint of the
+        expected server certificate.
     :param latin1_fallback: RFC-2616 specifies the default Content-Type of
         text/* to be latin1, which is not always correct, but exactly what
         requests is doing. Setting this parameter to False will use charset
@@ -181,9 +196,16 @@ def request(method, url, data=None, headers=None, auth=None, verify=None,
     '''
 
     if session is None:
-        func = requests.request
+        session = requests.Session()
-    else:
+
-        func = session.request
+    if verify_fingerprint is not None:
+        https_prefix = 'https://'
+
+        if not isinstance(session.adapters[https_prefix], _FingerprintAdapter):
+            fingerprint_adapter = _FingerprintAdapter(verify_fingerprint)
+            session.mount(https_prefix, fingerprint_adapter)
+
+    func = session.request
 
     logger.debug(u'{} {}'.format(method, url))
     logger.debug(headers)