Merge lp:~mandel/ubuntu-sso-client/pinned-certs into lp:ubuntu-sso-client

[Manuel de la Peña]

- Added a ssl cert vault that will allow to remember pinned ssl certificates (LP: #955339)

- Added a ssl cert vault that will allow to remember pinned ssl certificates (LP: #955339)

[Roberto Alsina (ralsina)]

Looks good to me, but testing this IRL is a pain.

review: Approve

[dobey (dobey)]

Looks ok.

review: Approve

[Manuel de la Peña (mandel)]

Rejecting since we won't use this code.

review: Disapprove

Unmerged revisions

952. By Manuel de la Peña on 2012-03-15

Remove unused function.

951. By Manuel de la Peña on 2012-03-15

Merged libsoup-ssl-dialog into pinned-certs.

950. By Manuel de la Peña on 2012-03-14

Merged libsoup-ssl-dialog into pinned-certs.

949. By Manuel de la Peña on 2012-03-14

Added the required code to store the pinned certificates in order to not promp the ssl dialog constantly.

948. By Manuel de la Peña on 2012-03-13

Pass the cert pem from the child WebClient implementations to the base class.

947. By Manuel de la Peña on 2012-03-12

Reduced diff size.

946. By Manuel de la Peña on 2012-03-12

Reduced diff size.

945. By Manuel de la Peña on 2012-03-12

Merged with trunk.

944. By Manuel de la Peña on 2012-03-12

Added the required code to let the libsoup webclient implementation deal with ssl cert errors.

943. By Manuel de la Peña on 2012-03-12

Made changes to the tests so that they work correctly.

=== modified file 'ubuntu_sso/utils/webclient/common.py'
--- ubuntu_sso/utils/webclient/common.py	2012-03-15 11:43:19 +0000
+++ ubuntu_sso/utils/webclient/common.py	2012-03-15 11:43:19 +0000
@@ -21,6 +21,7 @@
 
 from httplib2 import iri2uri
 from oauth import oauth
+from OpenSSL.crypto import load_certificate, FILETYPE_PEM
 from twisted.internet import defer
 from urlparse import urlparse
 
@@ -28,6 +29,8 @@
 from ubuntu_sso.logger import setup_logging
 from ubuntu_sso.utils.runner import spawn_program
 from ubuntu_sso.utils.ui import SSL_DETAILS_TEMPLATE
+from ubuntu_sso.utils.ui import SSL_DETAILS_TEMPLATE
+from ubuntu_sso.utils.webclient.ssl_certs import PinnedSSLCertsVault
 from ubuntu_sso.utils.webclient.timestamp import TimestampChecker
 
 SSL_DIALOG = 'ubuntu-sso-ssl-certificate-qt'
@@ -109,6 +112,7 @@
         self.proxy_username = None
         self.proxy_password = None
         self.oauth_sign_plain = oauth_sign_plain
+        self.ssl_vault = PinnedSSLCertsVault()
 
     def request(self, iri, method="GET", extra_headers=None,
                 oauth_credentials=None, post_content=None):
@@ -238,9 +242,16 @@
                          return_code)
             defer.returnValue(False)
 
-    def format_ssl_details(self, details):
+    def get_ssl_subject_details(self, cert):
         """Return a formatted string with the details."""
-        return SSL_DETAILS_TEMPLATE % details
+        cert_subject = cert.get_subject()
+        details = dict(organization=cert_subject.O,
+                       common_name=cert_subject.CN,
+                       locality_name=cert_subject.L,
+                       unit=cert_subject.OU,
+                       country_name=cert_subject.C,
+                       state_name=cert_subject.ST)
+        return details
 
     def _launch_ssl_dialog(self, domain, details):
         """Launch a dialog used to approve the ssl cert."""
@@ -252,16 +263,18 @@
                                                    '--appname', self.appname)
         return spawn_program(args)
 
-    def _was_ssl_accepted(self, cert_details):
-        """Return if the cert was already accepted."""
-        # TODO: Ensure that we look at pinned certs in a following branch
-        return False
-
     @defer.inlineCallbacks
-    def request_ssl_cert_approval(self, domain, details):
+    def request_ssl_cert_approval(self, cert_pem):
         """Request the user for ssl approval."""
-        if self._was_ssl_accepted(details):
+        cert = load_certificate(FILETYPE_PEM, cert_pem)
+        if self.ssl_vault.is_certificate_pinned(cert):
             defer.returnValue(True)
 
-        return_code = yield self._launch_ssl_dialog(domain, details)
+        subject_details = self.get_ssl_subject_details(cert)
+        details = SSL_DETAILS_TEMPLATE % subject_details
+        return_code = yield self._launch_ssl_dialog(
+                                               subject_details['common_name'],
+                                               details)
+        if return_code == USER_SUCCESS:
+            self.ssl_vault.store_certificate(cert)
         defer.returnValue(return_code == USER_SUCCESS)

=== modified file 'ubuntu_sso/utils/webclient/libsoup.py'
--- ubuntu_sso/utils/webclient/libsoup.py	2012-03-15 11:43:19 +0000
+++ ubuntu_sso/utils/webclient/libsoup.py	2012-03-15 11:43:19 +0000
@@ -17,7 +17,6 @@
 
 import httplib
 
-from OpenSSL.crypto import load_certificate, FILETYPE_PEM
 from twisted.internet import defer
 
 from ubuntu_sso.logger import setup_logging
@@ -67,17 +66,9 @@
         if message.status_code == httplib.OK:
             response = self._get_response(message)
             if is_ssl and errors.real != 0:
-                cert = load_certificate(FILETYPE_PEM,
-                                    raw_cert.props.certificate_pem)
-                cert_subject = cert.get_subject()
-                ssl_details = dict(organization=cert_subject.O,
-                                   common_name=cert_subject.CN,
-                                   locality_name=cert_subject.L,
-                                   unit=cert_subject.OU,
-                                   country_name=cert_subject.C,
-                                   state_name=cert_subject.ST)
                 e = SSLCertError(message.reason_phrase,
-                        ssl_details=ssl_details, response=response)
+                        ssl_details=raw_cert.props.certificate_pem,
+                        response=response)
                 d.errback(e)
             else:
                 d.callback(response)
@@ -136,8 +127,7 @@
         failure.trap(SSLCertError)
         logger.debug('SSL cert errors found.')
         trust_cert = yield self.request_ssl_cert_approval(
-                           failure.value.ssl_details['common_name'],
-                           self.format_ssl_details(failure.value.ssl_details))
+                                                    failure.value.ssl_details)
         if trust_cert:
             defer.returnValue(failure.value.response)
 

=== modified file 'ubuntu_sso/utils/webclient/qtnetwork.py'
--- ubuntu_sso/utils/webclient/qtnetwork.py	2012-03-09 12:04:31 +0000
+++ ubuntu_sso/utils/webclient/qtnetwork.py	2012-03-15 11:43:19 +0000
@@ -30,7 +30,6 @@
     QNetworkProxyFactory,
     QNetworkReply,
     QNetworkRequest,
-    QSslCertificate,
 )
 from twisted.internet import defer
 
@@ -266,31 +265,12 @@
                 exception = WebClientError(error_string, content)
             d.errback(exception)
 
-    def _get_certificate_details(self, cert):
-        """Return an string with the details of the certificate."""
-        detail_titles = {QSslCertificate.Organization: 'organization',
-                         QSslCertificate.CommonName: 'common_name',
-                         QSslCertificate.LocalityName: 'locality_name',
-                         QSslCertificate.OrganizationalUnitName: 'unit',
-                         QSslCertificate.CountryName: 'country_name',
-                         QSslCertificate.StateOrProvinceName: 'state_name'}
-        details = {}
-        for info, title in detail_titles.iteritems():
-            details[title] = str(cert.issuerInfo(info))
-        return self.format_ssl_details(details)
-
-    def _get_certificate_host(self, cert):
-        """Return the host of the cert."""
-        return str(cert.issuerInfo(QSslCertificate.CommonName))
-
     @defer.inlineCallbacks
     def _handle_ssl_errors(self, reply, errors):
         """Handle the case in which we got an ssl error."""
         # ask the user if the cer should be trusted
-        cert = errors[0].certificate()
-        trust_cert = yield self.request_ssl_cert_approval(
-                                    self._get_certificate_host(cert),
-                                    self._get_certificate_details(cert))
+        cert_pem = errors[0].certificate().toPem()
+        trust_cert = yield self.request_ssl_cert_approval(cert_pem)
         if trust_cert:
             reply.ignoreSslErrors()
 

=== added file 'ubuntu_sso/utils/webclient/ssl_certs.py'
--- ubuntu_sso/utils/webclient/ssl_certs.py	1970-01-01 00:00:00 +0000
+++ ubuntu_sso/utils/webclient/ssl_certs.py	2012-03-15 11:43:19 +0000
@@ -0,0 +1,105 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright 2011 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License version 3, as published
+# by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""Store and retrieve certificates that have been pinned."""
+
+import base64
+import os
+import tempfile
+
+from ubuntu_sso import xdg_base_directory
+from ubuntu_sso.logger import setup_logging
+
+PINNED_CERT_LINE_FORMAT = ('%(domain)s:%(port)s\t%(hash_algorithm)s\t'
+                           '%(fingerprint)s\t%(override_type)s\t'
+                           '%(serial_number)s\n')
+HTTPS_PORT = 443
+HASH_ALGORITHM = ('sha1', 'OID.2.16.840.1.101.3.4.2.1')
+DEFAULT_OVERRIDE_TYPE = 'MUT'
+PINNED_CERT_DIR = os.path.join(xdg_base_directory.xdg_cache_home, 'sso',
+                                                                  'ssl-certs')
+
+logger = setup_logging("ubuntu_sso.utils.webclient.ssl_certs")
+
+
+class PinnedSSLCertsVault(object):
+    """Represents a vault that will store the pinned certs."""
+
+    def __init__(self, certs_path=PINNED_CERT_DIR):
+        """Create a new instance."""
+        self.certs_path = certs_path
+        if not os.path.exists(self.certs_path):
+            os.makedirs(self.certs_path)
+
+    def _get_pinned_certificate_line(self, cert):
+        """Return the line of a pinned cert.
+
+        The following format is used:
+        Fields are separated by a tab character. Each line is terminated
+        by a line feed character (UNIX format).
+
+        1. domainname:port : port 443 for HTTPS (SSL)
+        2. hash algorithm OID:
+            SHA1-256: OID.2.16.840.1.101.3.4.2.1 (most used)
+            SHA-384: OID.2.16.840.1.101.3.4.2.2
+            SHA-512: OID.2.16.840.1.101.3.4.2.3
+        3. Certificate fingerprint using previous hash algorithm
+        4. One or more characters for override type:
+            M : allow mismatches in the hostname
+            U : allow untrusted certs (whether it's self signed cert or
+                a missing or invalid issuer cert)
+            T : allow errors in the validity time, for example, for expired or
+                not yet valid certs
+        5. Certificate's serial number and the issuer name as a base64
+            encoded string
+
+        more details can be found at:
+        https://developer.mozilla.org/En/Cert_override.txt
+        """
+        cert_subject = cert.get_subject()
+        serial_number = '%s@%s' % (cert.get_serial_number(), cert.get_issuer())
+        details = dict(domain=cert_subject.CN,
+                       port=HTTPS_PORT,
+                       hash_algorithm=HASH_ALGORITHM[1],
+                       fingerprint=cert.digest(HASH_ALGORITHM[0]),
+                       override_type=DEFAULT_OVERRIDE_TYPE,
+                       serial_number=base64.b64encode(serial_number))
+        return PINNED_CERT_LINE_FORMAT % details
+
+    def _get_cert_file_name(self, cert):
+        """Return the file name to use with the cert."""
+        fingerprint = cert.digest(HASH_ALGORITHM[0])
+        return fingerprint.replace(':', '-')
+
+    def store_certificate(self, cert):
+        """Store a new certificate."""
+        _, temp_path = tempfile.mkstemp(prefix='tmp-', dir=self.certs_path)
+        with open(temp_path, 'wb') as fd:
+            fd.write(self._get_pinned_certificate_line(cert))
+        logger.debug('Temp file %s written', temp_path)
+        fingerprint = self._get_cert_file_name(cert)
+        os.rename(temp_path, os.path.join(self.certs_path, fingerprint))
+        logger.debug('Cert with fingerprint %s added.', fingerprint)
+
+    def is_certificate_pinned(self, cert):
+        """Return if a certificate was already pinned."""
+        fingerprint = self._get_cert_file_name(cert)
+        return fingerprint in self.pinned_certificates
+
+    @property
+    def pinned_certificates(self):
+        """Return all the fingerprints of the stored certs."""
+        return [name for name in os.listdir(self.certs_path) if not
+                name.startswith('tmp-')]

=== added file 'ubuntu_sso/utils/webclient/tests/test_ssl_certs.py'
--- ubuntu_sso/utils/webclient/tests/test_ssl_certs.py	1970-01-01 00:00:00 +0000
+++ ubuntu_sso/utils/webclient/tests/test_ssl_certs.py	2012-03-15 11:43:19 +0000
@@ -0,0 +1,151 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright 2011 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License version 3, as published
+# by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""Tests for the ssl certs operations."""
+
+import os
+import tempfile
+
+from OpenSSL import crypto
+from socket import gethostname
+from twisted.internet import defer, reactor
+from twisted.trial.unittest import TestCase
+
+from ubuntu_sso.utils.webclient import ssl_certs
+
+
+class PinnedSSLCertsVaultTestCase(TestCase):
+    """Test the addition of a new pinned cert."""
+
+    @defer.inlineCallbacks
+    def setUp(self):
+        """Set the diff tests."""
+        yield super(PinnedSSLCertsVaultTestCase, self).setUp()
+        self.cert_details = dict(organization='Canonical',
+                                 common_name=gethostname(),
+                                 locality_name='London',
+                                 unit='Ubuntu One',
+                                 country_name='UK',
+                                 state_name='London',)
+        self.cert = self._generate_cert(self.cert_details)
+
+        self.certs_path = self.mktemp()
+        os.mkdir(self.certs_path)
+        self.vault = ssl_certs.PinnedSSLCertsVault(self.certs_path)
+
+    def _generate_cert(self, cert_details):
+        """Return a cert using the given details."""
+        # create a key pair
+        key = crypto.PKey()
+        key.generate_key(crypto.TYPE_RSA, 1024)
+
+        # create a self-signed cert
+        cert = crypto.X509()
+        cert.get_subject().C = cert_details['country_name']
+        cert.get_subject().ST = cert_details['state_name']
+        cert.get_subject().L = cert_details['locality_name']
+        cert.get_subject().O = cert_details['organization']
+        cert.get_subject().OU = cert_details['unit']
+        cert.get_subject().CN = cert_details['common_name']
+        cert.set_serial_number(1000)
+        cert.gmtime_adj_notBefore(0)
+        cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60)
+        cert.set_issuer(cert.get_subject())
+        cert.set_pubkey(key)
+        cert.sign(key, 'sha1')
+
+        return cert
+
+    def test_missing_dir(self):
+        """Test addind a pinned cert when the dir is missing."""
+        os.rmdir(self.certs_path)
+        self.vault = ssl_certs.PinnedSSLCertsVault(self.certs_path)
+        self.assertTrue(os.path.exists(self.certs_path))
+
+    def test_pinned_certificates(self):
+        """Test getting the list of pinned certificates."""
+        # create number of certs and ensure that the list is there.
+        added = []
+
+        for index in (1, 2, 3):
+            cert_details = self.cert_details.copy()
+            cert_details['common_name'] = \
+                    cert_details['common_name'] + str(index)
+            cert = self._generate_cert(cert_details)
+            self.vault.store_certificate(cert)
+            added.append(
+                    cert.digest(ssl_certs.HASH_ALGORITHM[0]).replace(':', '-'))
+
+        for fingerprint in added:
+            self.assertIn(fingerprint, self.vault.pinned_certificates)
+
+    def test_pinned_certificates_temp_files(self):
+        """Test getting the list of pinned certificates."""
+        # create a number of tmp files and assert that they are not returned.
+        tempfiles = []
+        added = []
+
+        for index in (1, 2, 3):
+            _, temp_path = tempfile.mkstemp(prefix='tmp-', dir=self.certs_path)
+            with open(temp_path, 'w') as fd:
+                fd.write("I'm a test!")
+            tempfiles.append(temp_path)
+
+        for index in (1, 2, 3):
+            cert_details = self.cert_details.copy()
+            cert_details['common_name'] = \
+                    cert_details['common_name'] + str(index)
+            cert = self._generate_cert(cert_details)
+            self.vault.store_certificate(cert)
+            added.append(
+                    cert.digest(ssl_certs.HASH_ALGORITHM[0]).replace(':', '-'))
+
+        for fingerprint in added:
+            self.assertIn(fingerprint, self.vault.pinned_certificates)
+
+        for temp in tempfiles:
+            self.assertNotIn(temp, self.vault.pinned_certificates)
+
+    def test_adding_certificated(self):
+        """Test adding a new certificate."""
+        self.vault.store_certificate(self.cert)
+        fingerprint = self.cert.digest(
+                ssl_certs.HASH_ALGORITHM[0]).replace(':', '-')
+        self.assertIn(fingerprint, os.listdir(self.certs_path))
+
+    def _assert_added(self):
+        """Assert the addition with multiple threads."""
+        vault = ssl_certs.PinnedSSLCertsVault(self.certs_path)
+        vault.store_certificate(self.cert)
+        fingerprint = self.cert.digest(
+                ssl_certs.HASH_ALGORITHM[0]).replace(':', '-')
+        self.assertIn(fingerprint, vault.pinned_certificates)
+        self.assertIn(fingerprint, os.listdir(self.certs_path))
+
+    def test_stepping_on_each_other(self):
+        """Test using several vaults in threads."""
+        # pylint: disable=E1101
+        for _ in (1, 2, 3, 4, 5):
+            reactor.callInThread(self._assert_added)
+        # pylint: enable=E1101
+
+    def test_get_cert_filename(self):
+        """Assert that the filename is correct."""
+        expected = self.cert.digest(
+                ssl_certs.HASH_ALGORITHM[0]).replace(':', '-')
+        # pylint: disable=W0212
+        filename = self.vault._get_cert_file_name(self.cert)
+        # pylint: enable=W0212
+        self.assertEqual(expected, filename)

=== modified file 'ubuntu_sso/utils/webclient/tests/test_webclient.py'
--- ubuntu_sso/utils/webclient/tests/test_webclient.py	2012-03-15 11:43:19 +0000
+++ ubuntu_sso/utils/webclient/tests/test_webclient.py	2012-03-15 11:43:19 +0000
@@ -892,7 +892,7 @@
         with open(key_path, 'wt') as fd:
             fd.write(crypto.dump_privatekey(crypto.FILETYPE_PEM, key))
 
-        return dict(key=key_path, cert=cert_path)
+        return dict(key=key_path, cert=cert_path, cert_info=cert)
 
 
 class CorrectProxyTestCase(BaseSSLTestCase):
@@ -1012,6 +1012,8 @@
         details = SSL_DETAILS_TEMPLATE % self.cert_details
         self.assertIn(('_launch_ssl_dialog', gethostname(), details),
                       self.called)
+        self.assertTrue(self.wc.ssl_vault.is_certificate_pinned(
+            self.ssl_settings['cert_info']))
 
     def test_ssl_fail_dialog_user_accepts(self):
         """Test showing the dialog and accepting."""
@@ -1026,12 +1028,24 @@
         """Test showing the dialog and rejecting."""
         self.failUnlessFailure(self.wc.request(self.base_iri + SIMPLERESOURCE),
                                      webclient.WebClientError)
-
-    def test_format_ssl_details(self):
+        # assert the cert was not pinned
+        self.assertFalse(self.wc.ssl_vault.is_certificate_pinned(
+            self.ssl_settings['cert_info']))
+
+    @defer.inlineCallbacks
+    def test_ssl_fail_pinned_cert(self):
+        """Test when the cert is pinned."""
+        self.patch(self.wc.ssl_vault, 'is_certificate_pinned', lambda _: True)
+        result = yield self.wc.request(self.base_ssl_iri + SIMPLERESOURCE)
+        details = SSL_DETAILS_TEMPLATE % self.cert_details
+        self.assertNotIn(('_launch_ssl_dialog', gethostname(), details),
+                      self.called)
+        self.assertEqual(SAMPLE_RESOURCE, result.content)
+
+    def test_get_ssl_subject_details(self):
         """Assert that details are correctly formatted"""
-        details = SSL_DETAILS_TEMPLATE % self.cert_details
-        self.assertEqual(details,
-                self.wc.format_ssl_details(self.cert_details))
+        self.assertEqual(self.cert_details,
+              self.wc.get_ssl_subject_details(self.ssl_settings['cert_info']))
 
     if WEBCLIENT_MODULE_NAME.endswith(".txweb"):
         reason = 'SSL support has not yet been implemented.'