Merge pull request #5463 from boundlessgeo/BD-2507-pki-bundle-validation

[auth] PKI bundle validation utility function

python/core/auth/qgsauthcertutils.sip

@@ -294,6 +294,30 @@ Get short strings describing an SSL error
 %End
 
 
+    static QList<QSslError> validateCertChain( const QList<QSslCertificate> &certificateChain,
+        const QString &hostName = QString(),
+        bool trustRootCa = false ) ;
+%Docstring
+ validateCertChain validates the given ``certificateChain``
+ \param certificateChain list of certificates to be checked, with leaf first and with optional root CA last
+ \param hostName (optional) name of the host to be verified
+ \param trustRootCa if true the CA will be added to the trusted CAs for this validation check
+ :return: list of QSslError, if the list is empty then the cert chain is valid
+ :rtype: list of QSslError
+%End
+
+    static QStringList validatePKIBundle( QgsPkiBundle &bundle, bool useIntermediates = true, bool trustRootCa = false );
+%Docstring
+ validatePKIBundle validate the PKI bundle by checking the certificate chain, the
+ expiration and effective dates, optionally trusts the root CA
+ \param bundle
+ \param useIntermediates if true the intermediate certs are also checked
+ \param trustRootCa if true the CA will be added to the trusted CAs for this validation check (if useIntermediates is false)
+ this option is ignored and set to false
+ :return: a list of error strings, if the list is empty then the PKI bundle is valid
+ :rtype: list of str
+%End
+
 };
 
 /************************************************************************

python/core/auth/qgsauthconfig.sip

@@ -323,6 +323,7 @@ Whether the bundle is valid
 Authentication method configuration
  :rtype: QgsAuthMethodConfig
 %End
+
     void setConfig( const QgsAuthMethodConfig &config );
 %Docstring
 Set authentication method configuration
@@ -333,6 +334,7 @@ Set authentication method configuration
 Client certificate object
  :rtype: QSslCertificate
 %End
+
     void setClientCert( const QSslCertificate &cert );
 %Docstring
 Set client certificate object
@@ -343,6 +345,7 @@ Set client certificate object
 Private key object
  :rtype: QSslKey
 %End
+
     void setClientCertKey( const QSslKey &certkey );
 %Docstring
 Set private key object

src/core/auth/qgsauthcertutils.cpp

@@ -1241,3 +1241,111 @@ QList<QPair<QSslError::SslError, QString> > QgsAuthCertUtils::sslErrorEnumString
                          QgsAuthCertUtils::sslErrorEnumString( QSslError::CertificateBlacklisted ) );
   return errenums;
 }
+
+QList<QSslError> QgsAuthCertUtils::validateCertChain( const QList<QSslCertificate> &certificateChain,
+    const QString &hostName,
+    bool trustRootCa )
+{
+  QList<QSslError> sslErrors;
+  QList<QSslCertificate> trustedChain;
+  // Filter out all CAs that are not trusted from QgsAuthManager
+  for ( const auto &cert : certificateChain )
+  {
+    bool untrusted = false;
+    for ( const auto &untrustedCert : QgsApplication::authManager()->getUntrustedCaCerts() )
+    {
+      if ( cert.digest( ) == untrustedCert.digest( ) )
+      {
+        untrusted = true;
+        break;
+      }
+    }
+    if ( ! untrusted )
+    {
+      trustedChain << cert;
+    }
+  }
+
+  // Check that no certs in the chain are expired or not yet valid or blacklisted
+  const QList<QSslCertificate> constTrustedChain( trustedChain );
+  for ( const auto &cert : constTrustedChain )
+  {
+    // TODO: move all the checks to QgsAuthCertUtils::certIsViable( )
+    const QDateTime currentTime = QDateTime::currentDateTime();
+    if ( cert.expiryDate() <= currentTime )
+    {
+      sslErrors << QSslError( QSslError::SslError::CertificateExpired, cert );
+    }
+    if ( cert.effectiveDate() >= QDateTime::currentDateTime() )
+    {
+      sslErrors << QSslError( QSslError::SslError::CertificateNotYetValid, cert );
+    }
+    if ( cert.isBlacklisted() )
+    {
+      sslErrors << QSslError( QSslError::SslError::CertificateBlacklisted, cert );
+    }
+  }
+
+  // Merge in the root CA if present and asked for
+  if ( trustRootCa && trustedChain.count() > 1 && trustedChain.last().isSelfSigned() )
+  {
+    static QMutex sMutex;
+    QMutexLocker lock( &sMutex );
+    QSslConfiguration oldSslConfig( QSslConfiguration::defaultConfiguration() );
+    QSslConfiguration sslConfig( oldSslConfig );
+    sslConfig.setCaCertificates( casMerge( sslConfig.caCertificates(), QList<QSslCertificate>() << trustedChain.last() ) );
+    QSslConfiguration::setDefaultConfiguration( sslConfig );
+    sslErrors = QSslCertificate::verify( trustedChain, hostName );
+    QSslConfiguration::setDefaultConfiguration( oldSslConfig );
+  }
+  else
+  {
+    sslErrors = QSslCertificate::verify( trustedChain, hostName );
+  }
+  return sslErrors;
+}
+
+QStringList QgsAuthCertUtils::validatePKIBundle( QgsPkiBundle &bundle, bool useIntermediates, bool trustRootCa )
+{
+  QStringList errors;
+  QList<QSslError> sslErrors;
+  if ( useIntermediates )
+  {
+    QList<QSslCertificate> certsList( bundle.caChain() );
+    certsList.insert( 0, bundle.clientCert( ) );
+    sslErrors = QgsAuthCertUtils::validateCertChain( certsList, QString(), trustRootCa );
+  }
+  else
+  {
+    sslErrors = QSslCertificate::verify( QList<QSslCertificate>() << bundle.clientCert() );
+  }
+  const QList<QSslError> constSslErrors( sslErrors );
+  for ( const auto &sslError : constSslErrors )
+  {
+    if ( sslError.error() != QSslError::NoError )
+    {
+      errors << sslError.errorString();
+    }
+  }
+  // Now check the key with QCA!
+  QCA::PrivateKey pvtKey( QCA::PrivateKey::fromPEM( bundle.clientKey().toPem() ) );
+  QCA::PublicKey pubKey( QCA::PublicKey::fromPEM( bundle.clientCert().publicKey().toPem( ) ) );
+  bool keyValid( ! pvtKey.isNull() );
+  if ( keyValid && !( pubKey.toRSA().isNull( ) || pvtKey.toRSA().isNull( ) ) )
+  {
+    keyValid = pubKey.toRSA().n() == pvtKey.toRSA().n();
+  }
+  else if ( keyValid && !( pubKey.toDSA().isNull( ) || pvtKey.toDSA().isNull( ) ) )
+  {
+    keyValid = pubKey == QCA::DSAPublicKey( pvtKey.toDSA() );
+  }
+  else
+  {
+    QgsDebugMsg( "Key is not DSA, RSA: validation is not supported by QCA" );
+  }
+  if ( ! keyValid )
+  {
+    errors << QObject::tr( "Private key does not match client certificate public key." );
+  }
+  return errors;
+}

src/core/auth/qgsauthcertutils.h

@@ -24,6 +24,7 @@
 #include <QSslCertificate>
 #include <QSslError>
 
+#include "qgsauthconfig.h"
 #include "qgis_core.h"
 
 class QgsAuthConfigSslServer;
@@ -329,6 +330,28 @@ class CORE_EXPORT QgsAuthCertUtils
      */
     static QList<QPair<QSslError::SslError, QString> > sslErrorEnumStrings() SIP_SKIP;
 
+    /**
+     * \brief validateCertChain validates the given \a certificateChain
+     * \param certificateChain list of certificates to be checked, with leaf first and with optional root CA last
+     * \param hostName (optional) name of the host to be verified
+     * \param trustRootCa if true the CA will be added to the trusted CAs for this validation check
+     * \return list of QSslError, if the list is empty then the cert chain is valid
+     */
+    static QList<QSslError> validateCertChain( const QList<QSslCertificate> &certificateChain,
+        const QString &hostName = QString(),
+        bool trustRootCa = false ) ;
+
+    /**
+     * \brief validatePKIBundle validate the PKI bundle by checking the certificate chain, the
+     * expiration and effective dates, optionally trusts the root CA
+     * \param bundle
+     * \param useIntermediates if true the intermediate certs are also checked
+     * \param trustRootCa if true the CA will be added to the trusted CAs for this validation check (if useIntermediates is false)
+     * this option is ignored and set to false
+     * \return a list of error strings, if the list is empty then the PKI bundle is valid
+     */
+    static QStringList validatePKIBundle( QgsPkiBundle &bundle, bool useIntermediates = true, bool trustRootCa = false );
+
   private:
     static void appendDirSegment_( QStringList &dirname, const QString &segment, QString value );
 };

src/core/auth/qgsauthconfig.h

@@ -276,16 +276,19 @@ class CORE_EXPORT QgsPkiConfigBundle
 
     //! Authentication method configuration
     const QgsAuthMethodConfig config() const { return mConfig; }
+
     //! Set authentication method configuration
     void setConfig( const QgsAuthMethodConfig &config ) { mConfig = config; }
 
     //! Client certificate object
     const QSslCertificate clientCert() const { return mCert; }
+
     //! Set client certificate object
     void setClientCert( const QSslCertificate &cert ) { mCert = cert; }
 
     //! Private key object
     const QSslKey clientCertKey() const { return mCertKey; }
+
     //! Set private key object
     void setClientCertKey( const QSslKey &certkey ) { mCertKey = certkey; }
 

tests/src/python/test_qgsauthsystem.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 """QGIS Unit tests for bindings to core authentication system classes
 
-From build dir: ctest -R PyQgsAuthenticationSystem -V
+From build dir: LC_ALL=en_US.UTF-8 ctest -R PyQgsAuthenticationSystem -V
 
 .. note:: This program is free software; you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
@@ -17,19 +17,13 @@
 import os
 import tempfile
 
-from qgis.core import QgsAuthManager, QgsAuthCertUtils, QgsPkiBundle, QgsAuthMethodConfig, QgsAuthMethod, QgsAuthConfigSslServer, QgsApplication
+from qgis.core import QgsAuthCertUtils, QgsPkiBundle, QgsAuthMethodConfig, QgsAuthMethod, QgsAuthConfigSslServer, QgsApplication
 from qgis.gui import QgsAuthEditorWidgets
-
-
 from qgis.PyQt.QtCore import QFileInfo, qDebug
-from qgis.PyQt.QtWidgets import QDialog, QVBoxLayout, QDialogButtonBox
-from qgis.PyQt.QtTest import QTest
 from qgis.PyQt.QtNetwork import QSsl, QSslError, QSslSocket
-
-from qgis.testing import (
-    start_app,
-    unittest,
-)
+from qgis.PyQt.QtTest import QTest
+from qgis.PyQt.QtWidgets import QDialog, QDialogButtonBox, QVBoxLayout
+from qgis.testing import start_app, unittest
 
 from utilities import unitTestDataPath
 
@@ -618,6 +612,82 @@ def test_140_cas_remove_self_signed(self):
         for c in filtered:
             self.assertFalse(c.isSelfSigned())
 
+    def test_150_verify_keychain(self):
+        """Test the verify keychain function"""
+
+        def testChain(path):
+
+            # Test that a chain with an untrusted CA is not valid
+            self.assertTrue(len(QgsAuthCertUtils.validateCertChain(QgsAuthCertUtils.certsFromFile(path))) > 0)
+
+            # Test that a chain with an untrusted CA is valid when the addRootCa argument is true
+            self.assertTrue(len(QgsAuthCertUtils.validateCertChain(QgsAuthCertUtils.certsFromFile(path), None, True)) == 0)
+
+            # Test that a chain with an untrusted CA is not valid when the addRootCa argument is true
+            # and a wrong domainis true
+            self.assertTrue(len(QgsAuthCertUtils.validateCertChain(QgsAuthCertUtils.certsFromFile(path), 'my.wrong.domain', True)) > 0)
+
+        testChain(PKIDATA + '/chain_subissuer-issuer-root.pem')
+        testChain(PKIDATA + '/localhost_ssl_w-chain.pem')
+        testChain(PKIDATA + '/fra_w-chain.pem')
+
+        path = PKIDATA + '/localhost_ssl_w-chain.pem'
+
+        # Test that a chain with an untrusted CA is not valid when the addRootCa argument is true
+        # and a wrong domain is set
+        self.assertTrue(len(QgsAuthCertUtils.validateCertChain(QgsAuthCertUtils.certsFromFile(path), 'my.wrong.domain', True)) > 0)
+
+        # Test that a chain with an untrusted CA is valid when the addRootCa argument is true
+        # and a right domain is set
+        self.assertTrue(len(QgsAuthCertUtils.validateCertChain(QgsAuthCertUtils.certsFromFile(path), 'localhost', True)) == 0)
+
+        # Test that a chain with an untrusted CA is not valid when the addRootCa argument is false
+        # and a right domain is set
+        self.assertTrue(len(QgsAuthCertUtils.validateCertChain(QgsAuthCertUtils.certsFromFile(path), 'localhost', False)) > 0)
+
+    def test_validate_pki_bundle(self):
+        """Text the pki bundle validation"""
+
+        def mkPEMBundle(client_cert, client_key, password, chain):
+            return QgsPkiBundle.fromPemPaths(PKIDATA + '/' + client_cert,
+                                             PKIDATA + '/' + client_key,
+                                             password,
+                                             QgsAuthCertUtils.certsFromFile(
+                                                 PKIDATA + '/' + chain
+                                             ))
+
+        # Valid bundle:
+        bundle = mkPEMBundle('fra_cert.pem', 'fra_key.pem', 'password', 'chain_subissuer-issuer-root.pem')
+
+        # Test valid bundle with intermediates and without trusted root
+        self.assertEqual(QgsAuthCertUtils.validatePKIBundle(bundle), ['The root certificate of the certificate chain is self-signed, and untrusted'])
+        # Test valid without intermediates
+        self.assertEqual(QgsAuthCertUtils.validatePKIBundle(bundle, False), ['The issuer certificate of a locally looked up certificate could not be found', 'No certificates could be verified'])
+        # Test valid with intermediates and trusted root
+        self.assertEqual(QgsAuthCertUtils.validatePKIBundle(bundle, True, True), [])
+
+        # Wrong chain
+        bundle = mkPEMBundle('fra_cert.pem', 'fra_key.pem', 'password', 'chain_issuer2-root2.pem')
+        # Test invalid bundle with intermediates and without trusted root
+        self.assertEqual(QgsAuthCertUtils.validatePKIBundle(bundle), ['The issuer certificate of a locally looked up certificate could not be found', 'No certificates could be verified'])
+        # Test valid without intermediates
+        self.assertEqual(QgsAuthCertUtils.validatePKIBundle(bundle, False), ['The issuer certificate of a locally looked up certificate could not be found', 'No certificates could be verified'])
+        # Test valid with intermediates and trusted root
+        self.assertEqual(QgsAuthCertUtils.validatePKIBundle(bundle, True, True), ['The issuer certificate of a locally looked up certificate could not be found', 'No certificates could be verified'])
+
+        # Wrong key
+        bundle = mkPEMBundle('fra_cert.pem', 'ptolemy_key.pem', 'password', 'chain_subissuer-issuer-root.pem')
+        # Test invalid bundle with intermediates and without trusted root
+        self.assertEqual(QgsAuthCertUtils.validatePKIBundle(bundle), ['The root certificate of the certificate chain is self-signed, and untrusted', 'Private key does not match client certificate public key.'])
+        # Test invalid without intermediates
+        self.assertEqual(QgsAuthCertUtils.validatePKIBundle(bundle, False), ['The issuer certificate of a locally looked up certificate could not be found', 'No certificates could be verified', 'Private key does not match client certificate public key.'])
+        # Test invalid with intermediates and trusted root
+        self.assertEqual(QgsAuthCertUtils.validatePKIBundle(bundle, True, True), ['Private key does not match client certificate public key.'])
+
+        # TODO: Wrong root CA
+        # TODO: expired/not-yet-valid cert
+        # TODO: expired/not-yet-valid intermediate (is it possible to build a cert from one of those?)
+
 
 if __name__ == '__main__':
     unittest.main()