diff --git a/.ci/travis/linux/blacklist.txt b/.ci/travis/linux/blacklist.txt
index 039d19370c4..bb4e646cc6c 100755
--- a/.ci/travis/linux/blacklist.txt
+++ b/.ci/travis/linux/blacklist.txt
@@ -29,4 +29,5 @@ PyQgsServerAccessControl
PyQgsAuthManagerPKIPostgresTest
PyQgsAuthManagerPasswordPostgresTest
PyQgsAuthManagerOgrPostgresTest
+PyQgsDbManagerPostgis
diff --git a/python/plugins/db_manager/db_plugins/postgis/connector.py b/python/plugins/db_manager/db_plugins/postgis/connector.py
index db8e4abf19a..24ec9de5215 100644
--- a/python/plugins/db_manager/db_plugins/postgis/connector.py
+++ b/python/plugins/db_manager/db_plugins/postgis/connector.py
@@ -26,7 +26,7 @@ from builtins import range
from functools import cmp_to_key
-from qgis.PyQt.QtCore import QRegExp
+from qgis.PyQt.QtCore import QRegExp, QFile
from qgis.core import Qgis, QgsCredentials, QgsDataSourceUri
from ..connector import DBConnector
@@ -66,6 +66,7 @@ class PostGisDBConnector(DBConnector):
try:
self.connection = psycopg2.connect(expandedConnInfo)
except self.connection_error_types() as e:
+ # get credentials if cached or asking to the user no more than 3 times
err = str(e)
uri = self.uri()
conninfo = uri.connectionInfo(False)
@@ -88,44 +89,13 @@ class PostGisDBConnector(DBConnector):
except self.connection_error_types() as e:
if i == 2:
raise ConnectionError(e)
-
err = str(e)
finally:
- # remove certs (if any) of the expanded connectionInfo
- expandedUri = QgsDataSourceUri(newExpandedConnInfo)
-
- sslCertFile = expandedUri.param("sslcert")
- if sslCertFile:
- sslCertFile = sslCertFile.replace("'", "")
- os.remove(sslCertFile)
-
- sslKeyFile = expandedUri.param("sslkey")
- if sslKeyFile:
- sslKeyFile = sslKeyFile.replace("'", "")
- os.remove(sslKeyFile)
-
- sslCAFile = expandedUri.param("sslrootcert")
- if sslCAFile:
- sslCAFile = sslCAFile.replace("'", "")
- os.remove(sslCAFile)
+ # clear certs for each time trying to connect
+ self._clearSslTempCertsIfAny(newExpandedConnInfo)
finally:
- # remove certs (if any) of the expanded connectionInfo
- expandedUri = QgsDataSourceUri(expandedConnInfo)
-
- sslCertFile = expandedUri.param("sslcert")
- if sslCertFile:
- sslCertFile = sslCertFile.replace("'", "")
- os.remove(sslCertFile)
-
- sslKeyFile = expandedUri.param("sslkey")
- if sslKeyFile:
- sslKeyFile = sslKeyFile.replace("'", "")
- os.remove(sslKeyFile)
-
- sslCAFile = expandedUri.param("sslrootcert")
- if sslCAFile:
- sslCAFile = sslCAFile.replace("'", "")
- os.remove(sslCAFile)
+ # clear certs of the first connection try
+ self._clearSslTempCertsIfAny(expandedConnInfo)
self.connection.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
@@ -141,6 +111,33 @@ class PostGisDBConnector(DBConnector):
def _connectionInfo(self):
return str(self.uri().connectionInfo(True))
+ def _clearSslTempCertsIfAny(self, connectionInfo):
+ # remove certs (if any) of the connectionInfo
+ expandedUri = QgsDataSourceUri(connectionInfo)
+
+ def removeCert(certFile):
+ certFile = certFile.replace("'", "")
+ file = QFile(certFile)
+ # set permission to allow removing on Win.
+ # On linux and Mac if file is set with QFile::>ReadUser
+ # does not create problem removing certs
+ if not file.setPermissions(QFile.WriteOwner):
+ raise Exception('Cannot change permissions on {}: error code: {}'.format(file.fileName(), file.error()))
+ if not file.remove():
+ raise Exception('Cannot remove {}: error code: {}'.format(file.fileName(), file.error()))
+
+ sslCertFile = expandedUri.param("sslcert")
+ if sslCertFile:
+ removeCert(sslCertFile)
+
+ sslKeyFile = expandedUri.param("sslkey")
+ if sslKeyFile:
+ removeCert(sslKeyFile)
+
+ sslCAFile = expandedUri.param("sslrootcert")
+ if sslCAFile:
+ removeCert(sslCAFile)
+
def _checkSpatial(self):
""" check whether postgis_version is present in catalog """
c = self._execute(None, u"SELECT COUNT(*) FROM pg_proc WHERE proname = 'postgis_version'")
diff --git a/src/providers/postgres/qgspostgresconn.cpp b/src/providers/postgres/qgspostgresconn.cpp
index bda652650ec..8a9d56b9787 100644
--- a/src/providers/postgres/qgspostgresconn.cpp
+++ b/src/providers/postgres/qgspostgresconn.cpp
@@ -242,7 +242,26 @@ QgsPostgresConn::QgsPostgresConn( const QString &conninfo, bool readOnly, bool s
{
QString fileName = expandedUri.param( param );
fileName.remove( QStringLiteral( "'" ) );
- QFile::remove( fileName );
+ QFile file( fileName );
+ // set minimal permission to allow removing on Win.
+ // On linux and Mac if file is set with QFile::ReadUser
+ // does not create problem removing certs
+ if ( !file.setPermissions( QFile::WriteOwner ) )
+ {
+ QString errorMsg = tr( "Cannot set WriteOwner permission to cert: %0 to allow removing it" ).arg( file.fileName() );
+ PQfinish();
+ QgsMessageLog::logMessage( tr( "Client security failure" ) + '\n' + errorMsg, tr( "PostGIS" ) );
+ mRef = 0;
+ return;
+ }
+ if ( !file.remove() )
+ {
+ QString errorMsg = tr( "Cannot remove cert: %0" ).arg( file.fileName() );
+ PQfinish();
+ QgsMessageLog::logMessage( tr( "Client security failure" ) + '\n' + errorMsg, tr( "PostGIS" ) );
+ mRef = 0;
+ return;
+ }
}
}
diff --git a/tests/src/python/CMakeLists.txt b/tests/src/python/CMakeLists.txt
index 1197ea96dea..c3a85b2d7f3 100644
--- a/tests/src/python/CMakeLists.txt
+++ b/tests/src/python/CMakeLists.txt
@@ -235,6 +235,7 @@ IF (ENABLE_PGTEST)
ADD_PYTHON_TEST(PyQgsAuthManagerPKIPostgresTest test_authmanager_pki_postgres.py)
ADD_PYTHON_TEST(PyQgsAuthManagerPasswordPostgresTest test_authmanager_password_postgres.py)
ADD_PYTHON_TEST(PyQgsAuthManagerOgrPostgresTest test_authmanager_ogr_postgres.py)
+ ADD_PYTHON_TEST(PyQgsDbManagerPostgis test_db_manager_postgis.py)
ENDIF (ENABLE_PGTEST)
IF (ENABLE_MSSQLTEST)
diff --git a/tests/src/python/test_authmanager_pki_postgres.py b/tests/src/python/test_authmanager_pki_postgres.py
index e39ea354f50..6c48bdd5c0e 100644
--- a/tests/src/python/test_authmanager_pki_postgres.py
+++ b/tests/src/python/test_authmanager_pki_postgres.py
@@ -30,6 +30,7 @@ import signal
import stat
import subprocess
import tempfile
+import glob
from shutil import rmtree
@@ -44,6 +45,7 @@ from qgis.core import (
)
from qgis.PyQt.QtNetwork import QSslCertificate
+from qgis.PyQt.QtCore import QFile
from qgis.testing import (
start_app,
@@ -234,6 +236,28 @@ class TestAuthManager(unittest.TestCase):
pg_layer = self._getPostGISLayer('testlayer_èé')
self.assertFalse(pg_layer.isValid())
+ def testRemoveTemporaryCerts(self):
+ """
+ Check that no temporary cert remain after connection with
+ postgres provider
+ """
+ def cleanTempPki():
+ pkies = glob.glob(os.path.join(tempfile.gettempdir(), 'tmp*_{*}.pem'))
+ for fn in pkies:
+ f = QFile(fn)
+ f.setPermissions(QFile.WriteOwner)
+ f.remove()
+
+ # remove any temppki in temprorary path to check that no
+ # other pki remain after connection
+ cleanTempPki()
+ # connect using postgres provider
+ pg_layer = self._getPostGISLayer('testlayer_èé', authcfg=self.auth_config.id())
+ self.assertTrue(pg_layer.isValid())
+ # do test no certs remained
+ pkies = glob.glob(os.path.join(tempfile.gettempdir(), 'tmp*_{*}.pem'))
+ self.assertEqual(len(pkies), 0)
+
if __name__ == '__main__':
unittest.main()
diff --git a/tests/src/python/test_db_manager_postgis.py b/tests/src/python/test_db_manager_postgis.py
new file mode 100644
index 00000000000..71bd216a3ae
--- /dev/null
+++ b/tests/src/python/test_db_manager_postgis.py
@@ -0,0 +1,299 @@
+# -*- coding: utf-8 -*-
+"""QGIS Unit tests for the DBManager GPKG plugin
+
+.. 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
+the Free Software Foundation; either version 2 of the License, or
+(at your option) any later version.
+"""
+__author__ = 'Luigi Pirelli'
+__date__ = '2017-11-02'
+__copyright__ = 'Copyright 2017, Boundless Spatial Inc'
+# This will get replaced with a git SHA1 when you do a git archive
+__revision__ = '$Format:%H$'
+
+import os
+import shutil
+import time
+import signal
+import stat
+import subprocess
+import tempfile
+import glob
+from shutil import rmtree
+
+from qgis.core import (
+ QgsApplication,
+ QgsAuthManager,
+ QgsAuthMethodConfig,
+ QgsVectorLayer,
+ QgsDataSourceUri,
+ QgsSettings,
+ QgsProviderRegistry,
+ QgsWkbTypes,
+)
+from qgis.PyQt.QtCore import QCoreApplication, QFile
+from qgis.testing import start_app, unittest
+from qgis.PyQt.QtNetwork import QSslCertificate
+
+from plugins.db_manager.db_plugins import supportedDbTypes, createDbPlugin
+from plugins.db_manager.db_plugins.plugin import TableField
+
+from utilities import unitTestDataPath
+
+QGIS_POSTGRES_SERVER_PORT = os.environ.get('QGIS_POSTGRES_SERVER_PORT', '55432')
+QGIS_POSTGRES_EXECUTABLE_PATH = os.environ.get('QGIS_POSTGRES_EXECUTABLE_PATH', '/usr/lib/postgresql/9.4/bin')
+
+assert os.path.exists(QGIS_POSTGRES_EXECUTABLE_PATH)
+
+QGIS_AUTH_DB_DIR_PATH = tempfile.mkdtemp()
+
+# Postgres test path
+QGIS_PG_TEST_PATH = tempfile.mkdtemp()
+
+os.environ['QGIS_AUTH_DB_DIR_PATH'] = QGIS_AUTH_DB_DIR_PATH
+
+qgis_app = start_app()
+
+QGIS_POSTGRES_CONF_TEMPLATE = """
+hba_file = '%(tempfolder)s/pg_hba.conf'
+listen_addresses = '*'
+port = %(port)s
+max_connections = 100
+unix_socket_directories = '%(tempfolder)s'
+ssl = true
+ssl_ciphers = 'DEFAULT:!LOW:!EXP:!MD5:@STRENGTH' # allowed SSL ciphers
+ssl_cert_file = '%(server_cert)s'
+ssl_key_file = '%(server_key)s'
+ssl_ca_file = '%(sslrootcert_path)s'
+password_encryption = on
+"""
+
+QGIS_POSTGRES_HBA_TEMPLATE = """
+hostssl all all 0.0.0.0/0 cert clientcert=1
+hostssl all all ::1/0 cert clientcert=1
+host all all 127.0.0.1/32 trust
+host all all ::1/32 trust
+"""
+
+TEST_CONNECTION_NAME = 'test_connection'
+
+
+class TestPyQgsDBManagerPostgis(unittest.TestCase):
+
+ @classmethod
+ def setUpAuth(cls):
+ """Run before all tests and set up authentication"""
+ authm = QgsApplication.authManager()
+ assert (authm.setMasterPassword('masterpassword', True))
+ cls.pg_conf = os.path.join(cls.tempfolder, 'postgresql.conf')
+ cls.pg_hba = os.path.join(cls.tempfolder, 'pg_hba.conf')
+ # Client side
+ cls.sslrootcert_path = os.path.join(cls.certsdata_path, 'chains_subissuer-issuer-root_issuer2-root2.pem')
+ cls.sslcert = os.path.join(cls.certsdata_path, 'gerardus_cert.pem')
+ cls.sslkey = os.path.join(cls.certsdata_path, 'gerardus_key.pem')
+ assert os.path.isfile(cls.sslcert)
+ assert os.path.isfile(cls.sslkey)
+ assert os.path.isfile(cls.sslrootcert_path)
+ os.chmod(cls.sslcert, stat.S_IRUSR)
+ os.chmod(cls.sslkey, stat.S_IRUSR)
+ os.chmod(cls.sslrootcert_path, stat.S_IRUSR)
+ cls.auth_config = QgsAuthMethodConfig("PKI-Paths")
+ cls.auth_config.setConfig('certpath', cls.sslcert)
+ cls.auth_config.setConfig('keypath', cls.sslkey)
+ cls.auth_config.setName('test_pki_auth_config')
+ cls.username = 'Gerardus'
+ cls.sslrootcert = QSslCertificate.fromPath(cls.sslrootcert_path)
+ assert cls.sslrootcert is not None
+ authm.storeCertAuthorities(cls.sslrootcert)
+ authm.rebuildCaCertsCache()
+ authm.rebuildTrustedCaCertsCache()
+ authm.rebuildCertTrustCache()
+ assert (authm.storeAuthenticationConfig(cls.auth_config)[0])
+ assert cls.auth_config.isValid()
+
+ # Server side
+ cls.server_cert = os.path.join(cls.certsdata_path, 'localhost_ssl_cert.pem')
+ cls.server_key = os.path.join(cls.certsdata_path, 'localhost_ssl_key.pem')
+ cls.server_rootcert = cls.sslrootcert_path
+ os.chmod(cls.server_cert, stat.S_IRUSR)
+ os.chmod(cls.server_key, stat.S_IRUSR)
+ os.chmod(cls.server_rootcert, stat.S_IRUSR)
+
+ # Place conf in the data folder
+ with open(cls.pg_conf, 'w+') as f:
+ f.write(QGIS_POSTGRES_CONF_TEMPLATE % {
+ 'port': cls.port,
+ 'tempfolder': cls.tempfolder,
+ 'server_cert': cls.server_cert,
+ 'server_key': cls.server_key,
+ 'sslrootcert_path': cls.sslrootcert_path,
+ })
+
+ with open(cls.pg_hba, 'w+') as f:
+ f.write(QGIS_POSTGRES_HBA_TEMPLATE)
+
+ @classmethod
+ def setUpServer(cls):
+ """Run before all tests:
+ Creates an auth configuration"""
+ cls.port = QGIS_POSTGRES_SERVER_PORT
+ cls.dbname = 'test_pki'
+ cls.tempfolder = QGIS_PG_TEST_PATH
+ cls.certsdata_path = os.path.join(unitTestDataPath('auth_system'), 'certs_keys')
+ cls.hostname = 'localhost'
+ cls.data_path = os.path.join(cls.tempfolder, 'data')
+ os.mkdir(cls.data_path)
+
+ cls.setUpAuth()
+ subprocess.check_call([os.path.join(QGIS_POSTGRES_EXECUTABLE_PATH, 'initdb'), '-D', cls.data_path])
+
+ cls.server = subprocess.Popen([os.path.join(QGIS_POSTGRES_EXECUTABLE_PATH, 'postgres'), '-D',
+ cls.data_path, '-c',
+ "config_file=%s" % cls.pg_conf],
+ env=os.environ,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE)
+ # Wait max 10 secs for the server to start
+ end = time.time() + 10
+ while True:
+ line = cls.server.stderr.readline()
+ print(line)
+ if line.find(b"database system is ready to accept") != -1:
+ break
+ if time.time() > end:
+ raise Exception("Timeout connecting to PostgreSQL")
+ # Create a DB
+ subprocess.check_call([os.path.join(QGIS_POSTGRES_EXECUTABLE_PATH, 'createdb'), '-h', 'localhost', '-p', cls.port, 'test_pki'])
+ # Inject test SQL from test path
+ test_sql = os.path.join(unitTestDataPath('provider'), 'testdata_pg.sql')
+ subprocess.check_call([os.path.join(QGIS_POSTGRES_EXECUTABLE_PATH, 'psql'), '-h', 'localhost', '-p', cls.port, '-f', test_sql, cls.dbname])
+ # Create a role
+ subprocess.check_call([os.path.join(QGIS_POSTGRES_EXECUTABLE_PATH, 'psql'), '-h', 'localhost', '-p', cls.port, '-c', 'CREATE ROLE "%s" WITH SUPERUSER LOGIN' % cls.username, cls.dbname])
+
+ @classmethod
+ def setUpProvider(cls, authId):
+ cls.dbconn = 'dbname=\'qgis_test\''
+ if 'QGIS_PGTEST_DB' in os.environ:
+ cls.dbconn = os.environ['QGIS_PGTEST_DB']
+ uri = QgsDataSourceUri()
+ uri.setConnection("localhost", cls.port, cls.dbname, "", "", QgsDataSourceUri.SslVerifyFull, authId)
+ uri.setKeyColumn('pk')
+ uri.setSrid('EPSG:4326')
+ uri.setDataSource('qgis_test', 'someData', "geom", "", "pk")
+ provider = QgsProviderRegistry.instance().createProvider('postgres', uri.uri(False))
+ if provider is None:
+ raise Exception("cannot create postgres provider")
+ if not provider.isValid():
+ raise Exception("Created postgres provider is not valid: {}".format(str(provider.errors())))
+ # save provider config that is the way how db_manager is aware of a PG connection
+ cls.addConnectionConfig(TEST_CONNECTION_NAME, uri)
+
+ @classmethod
+ def addConnectionConfig(cls, conn_name, uri):
+ """Necessary to allow db_manager to have the list of connections get from settings."""
+ uri = QgsDataSourceUri(uri)
+
+ settings = QgsSettings()
+ baseKey = "/PostgreSQL/connections/"
+ baseKey += conn_name
+ settings.setValue(baseKey + "/service", uri.service())
+ settings.setValue(baseKey + "/host", uri.host())
+ settings.setValue(baseKey + "/port", uri.port())
+ settings.setValue(baseKey + "/database", uri.database())
+ if uri.username():
+ settings.setValue(baseKey + "/username", uri.username())
+ if uri.password():
+ settings.setValue(baseKey + "/password", uri.password())
+ if uri.authConfigId():
+ settings.setValue(baseKey + "/authcfg", uri.authConfigId())
+ if uri.sslMode():
+ settings.setValue(baseKey + "/sslmode", uri.sslMode())
+
+ @classmethod
+ def setUpClass(cls):
+ """Run before all tests"""
+ # start ans setup server
+ cls.setUpServer()
+
+ # start a standalone qgis application
+ QCoreApplication.setOrganizationName("QGIS_Test")
+ QCoreApplication.setOrganizationDomain("TestPyQgsDBManagerPostgis.com")
+ QCoreApplication.setApplicationName("TestPyQgsDBManagerPostgis")
+ QgsSettings().clear()
+ start_app()
+
+ @classmethod
+ def tearDownClass(cls):
+ """Run after all tests"""
+ cls.server.terminate()
+ os.kill(cls.server.pid, signal.SIGABRT)
+ del cls.server
+ time.sleep(2)
+ rmtree(QGIS_AUTH_DB_DIR_PATH)
+ rmtree(cls.tempfolder)
+ QgsSettings().clear()
+
+ ###########################################
+
+ def testSupportedDbTypes(self):
+ self.assertIn('postgis', supportedDbTypes())
+
+ def testCreateDbPlugin(self):
+ plugin = createDbPlugin('postgis')
+ self.assertIsNotNone(plugin)
+
+ def testConnect(self):
+ # create a PKI postgis connection
+ # that will be listed in postgis connection of db_manager
+ self.setUpProvider(self.auth_config.id())
+
+ plugin = createDbPlugin('postgis')
+ connections = plugin.connections()
+ self.assertEqual(len(connections), 1)
+ # test connection
+ postgisConnPlugin = connections[0]
+ self.assertTrue(postgisConnPlugin.connect())
+ # test removing connection
+ self.assertTrue(postgisConnPlugin.remove())
+ self.assertEqual(len(plugin.connections()), 0)
+ # test without connection params => fail
+ connection = createDbPlugin('postgis', 'conn name')
+ connection_succeeded = False
+ try:
+ connection.connect()
+ connection_succeeded = True
+ except:
+ pass
+ self.assertFalse(connection_succeeded, 'exception should have been raised')
+
+ def testRemoveTemporaryCerts(self):
+ """
+ Check that no temporary cert remain after connection with
+ db_manager postgis plugin
+ """
+ def cleanTempPki():
+ pkies = glob.glob(os.path.join(tempfile.gettempdir(), 'tmp*_{*}.pem'))
+ for fn in pkies:
+ f = QFile(fn)
+ f.setPermissions(QFile.WriteOwner)
+ f.remove()
+
+ # remove any temppki in temprorary path to check that no
+ # other pki remain after connection
+ cleanTempPki()
+ # connect
+ self.setUpProvider(self.auth_config.id())
+ plugin = createDbPlugin('postgis')
+ connections = plugin.connections()
+ self.assertEqual(len(connections), 1)
+ # test connection
+ postgisConnPlugin = connections[0]
+ self.assertTrue(postgisConnPlugin.connect())
+ # do test no certs remained
+ pkies = glob.glob(os.path.join(tempfile.gettempdir(), 'tmp*_{*}.pem'))
+ self.assertEqual(len(pkies), 0)
+
+
+if __name__ == '__main__':
+ unittest.main()