Skip to content
Permalink
Browse files

Merge pull request #6647 from luipir/certs_not_removed_with_pki_postg…

…is_on_win_port3

Certs not removed with pki postgis on win port3
  • Loading branch information
luipir committed Jun 11, 2018
2 parents 5567223 + 1743451 commit f4065d8aed1cc308456a3362b203bb1146caebfc
@@ -29,4 +29,5 @@ PyQgsServerAccessControl
PyQgsAuthManagerPKIPostgresTest
PyQgsAuthManagerPasswordPostgresTest
PyQgsAuthManagerOgrPostgresTest
PyQgsDbManagerPostgis

@@ -26,7 +26,7 @@

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 @@ def __init__(self, uri):
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 @@ def __init__(self, uri):
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 @@ def __init__(self, uri):
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'")
@@ -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;
}
}
}

@@ -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)
@@ -30,6 +30,7 @@
import stat
import subprocess
import tempfile
import glob

from shutil import rmtree

@@ -44,6 +45,7 @@
)

from qgis.PyQt.QtNetwork import QSslCertificate
from qgis.PyQt.QtCore import QFile

from qgis.testing import (
start_app,
@@ -234,6 +236,28 @@ def testInvalidAuthAccess(self):
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()

0 comments on commit f4065d8

Please sign in to comment.
You can’t perform that action at this time.