Merge pull request #3716 from elpaso/auth_tests_more_2_18

More Authentication Tests

ci/travis/linux/qt5/blacklist.txt

@@ -6,7 +6,7 @@ PyQgsMapUnitScale
 PyQgsPalLabelingServer
 PyQgsRelationEditWidget
 PyQgsServer
-PyQgsAuthManagerEndpointTest
+PyQgsAuthManagerPasswordOWSTest
 PyQgsServerAccessControl
 PyQgsSipCoverage
 PyQgsSpatialiteProvider

tests/src/python/CMakeLists.txt

@@ -150,5 +150,7 @@ IF (WITH_SERVER)
   ADD_PYTHON_TEST(PyQgsServerAccessControl test_qgsserver_accesscontrol.py)
   ADD_PYTHON_TEST(PyQgsServerWFST test_qgsserver_wfst.py)
   ADD_PYTHON_TEST(PyQgsOfflineEditingWFS test_offline_editing_wfs.py)
-  ADD_PYTHON_TEST(PyQgsAuthManagerEndpointTest test_authmanager_endpoint.py)
+  ADD_PYTHON_TEST(PyQgsAuthManagerPasswordOWSTest test_authmanager_password_ows.py)
+  #ADD_PYTHON_TEST(PyQgsAuthManagerPKIOWSTest test_authmanager_pki_ows.py)
+  ADD_PYTHON_TEST(PyQgsAuthManagerPKIPostgresTest test_authmanager_pki_postgres.py)
 ENDIF (WITH_SERVER)

tests/src/python/qgis_wrapped_server.py

@@ -13,6 +13,21 @@
   * QGIS_SERVER_USERNAME (default ="username")
   * QGIS_SERVER_PASSWORD (default ="password")
 
+PKI authentication with HTTPS can be enabled with:
+
+  * QGIS_SERVER_PKI_CERTIFICATE (server certificate)
+  * QGIS_SERVER_PKI_KEY (server private key)
+  * QGIS_SERVER_PKI_AUTHORITY (root CA)
+  * QGIS_SERVER_PKI_USERNAME (valid username)
+
+ Sample run:
+
+ QGIS_SERVER_PKI_USERNAME=Gerardus QGIS_SERVER_PORT=47547 QGIS_SERVER_HOST=localhost \
+    QGIS_SERVER_PKI_KEY=/home/dev/QGIS/tests/testdata/auth_system/certs_keys/localhost_ssl_key.pem \
+    QGIS_SERVER_PKI_CERTIFICATE=/home/dev/QGIS/tests/testdata/auth_system/certs_keys/localhost_ssl_cert.pem \
+    QGIS_SERVER_PKI_AUTHORITY=/home/dev/QGIS/tests/testdata/auth_system/certs_keys/chains_subissuer-issuer-root_issuer2-root2.pem \
+    python /home/dev/QGIS/tests/src/python/qgis_wrapped_server.py
+
 .. 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
@@ -31,12 +46,27 @@
 
 import os
 import sys
+import ssl
 import urllib.parse
 from http.server import BaseHTTPRequestHandler, HTTPServer
 from qgis.server import QgsServer, QgsServerFilter
 
 QGIS_SERVER_PORT = int(os.environ.get('QGIS_SERVER_PORT', '8081'))
 QGIS_SERVER_HOST = os.environ.get('QGIS_SERVER_HOST', '127.0.0.1')
+# PKI authentication
+QGIS_SERVER_PKI_CERTIFICATE = os.environ.get('QGIS_SERVER_PKI_CERTIFICATE')
+QGIS_SERVER_PKI_KEY = os.environ.get('QGIS_SERVER_PKI_KEY')
+QGIS_SERVER_PKI_AUTHORITY = os.environ.get('QGIS_SERVER_PKI_AUTHORITY')
+QGIS_SERVER_PKI_USERNAME = os.environ.get('QGIS_SERVER_PKI_USERNAME')
+
+# Check if PKI - https is enabled
+https = (QGIS_SERVER_PKI_CERTIFICATE is not None and
+         os.path.isfile(QGIS_SERVER_PKI_CERTIFICATE) and
+         QGIS_SERVER_PKI_KEY is not None and
+         os.path.isfile(QGIS_SERVER_PKI_KEY) and
+         QGIS_SERVER_PKI_AUTHORITY is not None and
+         os.path.isfile(QGIS_SERVER_PKI_AUTHORITY) and
+         QGIS_SERVER_PKI_USERNAME)
 
 qgs_server = QgsServer()
 
@@ -66,8 +96,20 @@ def responseComplete(self):
 class Handler(BaseHTTPRequestHandler):
 
     def do_GET(self):
+        # For PKI: check the username from client certificate
+        if https:
+            try:
+                ssl.match_hostname(self.connection.getpeercert(), QGIS_SERVER_PKI_USERNAME)
+            except Exception as ex:
+                print("SSL Exception %s" % ex)
+                self.send_response(401)
+                self.end_headers()
+                self.wfile.write('UNAUTHORIZED')
+                return
         # CGI vars:
         for k, v in self.headers.items():
+            # Uncomment to print debug info about env vars passed into QGIS Server env
+            #print('Setting ENV var %s to %s' % ('HTTP_%s' % k.replace(' ', '-').replace('-', '_').replace(' ', '-').upper(), v))
             qgs_server.putenv('HTTP_%s' % k.replace(' ', '-').replace('-', '_').replace(' ', '-').upper(), v)
         qgs_server.putenv('SERVER_PORT', str(self.server.server_port))
         qgs_server.putenv('SERVER_NAME', self.server.server_name)
@@ -96,7 +138,19 @@ def do_POST(self):
 
 if __name__ == '__main__':
     server = HTTPServer((QGIS_SERVER_HOST, QGIS_SERVER_PORT), Handler)
-    print('Starting server on %s:%s, use <Ctrl-C> to stop' %
-          (QGIS_SERVER_HOST, server.server_port))
-    sys.stdout.flush()
+    if https:
+        server.socket = ssl.wrap_socket(server.socket,
+                                        certfile=QGIS_SERVER_PKI_CERTIFICATE,
+                                        keyfile=QGIS_SERVER_PKI_KEY,
+                                        ca_certs=QGIS_SERVER_PKI_AUTHORITY,
+                                        cert_reqs=ssl.CERT_REQUIRED,
+                                        server_side=True,
+                                        ssl_version=ssl.PROTOCOL_TLSv1)
+    message = 'Starting server on %s://%s:%s, use <Ctrl-C> to stop' % \
+              ('https' if https else 'http', QGIS_SERVER_HOST, server.server_port)
+    try:
+        print(message, flush=True)
+    except:
+        print(message)
+        sys.stdout.flush()
     server.serve_forever()

tests/src/python/test_authmanager_endpoint.py → tests/src/python/test_authmanager_password_ows.py

@@ -8,7 +8,7 @@
 configuration to access an HTTP Basic protected endpoint.
 
 
-From build dir, run: ctest -R PyQgsAuthManagerEnpointTest -V
+From build dir, run: ctest -R PyQgsAuthManagerPasswordOWSTest -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
@@ -22,7 +22,10 @@
 import tempfile
 import random
 import string
-import urllib
+try:
+    from urllib.parse import quote
+except:
+    from urllib import quote
 
 __author__ = 'Alessandro Pasotti'
 __date__ = '18/09/2016'
@@ -47,7 +50,7 @@
 try:
     QGIS_SERVER_ENDPOINT_PORT = os.environ['QGIS_SERVER_ENDPOINT_PORT']
 except:
-    QGIS_SERVER_ENDPOINT_PORT = '0' # Auto
+    QGIS_SERVER_ENDPOINT_PORT = '0'  # Auto
 
 
 QGIS_AUTH_DB_DIR_PATH = tempfile.mkdtemp()
@@ -84,11 +87,14 @@ def setUpClass(cls):
         cls.auth_config.setConfig('username', cls.username)
         cls.auth_config.setConfig('password', cls.password)
         assert (authm.storeAuthenticationConfig(cls.auth_config)[0])
+        cls.hostname = '127.0.0.1'
+        cls.protocol = 'http'
 
         os.environ['QGIS_SERVER_HTTP_BASIC_AUTH'] = '1'
         os.environ['QGIS_SERVER_USERNAME'] = cls.username
         os.environ['QGIS_SERVER_PASSWORD'] = cls.password
         os.environ['QGIS_SERVER_PORT'] = str(cls.port)
+        os.environ['QGIS_SERVER_HOST'] = cls.hostname
         server_path = os.path.dirname(os.path.realpath(__file__)) + \
             '/qgis_wrapped_server.py'
         cls.server = subprocess.Popen([sys.executable, server_path],
@@ -98,7 +104,7 @@ def setUpClass(cls):
         cls.port = int(re.findall(b':(\d+)', line)[0])
         assert cls.port != 0
         # Wait for the server process to start
-        assert waitServer('http://127.0.0.1:%s' % cls.port), "Server is not responding! http://127.0.0.1:%s" % cls.port
+        assert waitServer('%s://%s:%s' % (cls.protocol, cls.hostname, cls.port)), "Server is not responding! '%s://%s:%s" % (cls.protocol, cls.hostname, cls.port)
 
     @classmethod
     def tearDownClass(cls):
@@ -125,13 +131,16 @@ def _getWFSLayer(cls, type_name, layer_name=None, authcfg=None):
         parms = {
             'srsname': 'EPSG:4326',
             'typename': type_name,
-            'url': 'http://127.0.0.1:%s/?map=%s' % (cls.port, cls.project_path),
+            'url': '%s://%s:%s/?map=%s' % (cls.protocol, cls.hostname, cls.port, cls.project_path),
             'version': 'auto',
             'table': '',
         }
         if authcfg is not None:
             parms.update({'authcfg': authcfg})
-        uri = ' '.join([("%s='%s'" % (k, v.decode('utf-8'))) for k, v in list(parms.items())])
+        try: # Py2
+            uri = ' '.join([("%s='%s'" % (k, v.decode('utf-8'))) for k, v in list(parms.items())])
+        except AttributeError: # Py3
+            uri = ' '.join([("%s='%s'" % (k, v)) for k, v in list(parms.items())])
         wfs_layer = QgsVectorLayer(uri, layer_name, 'WFS')
         return wfs_layer
 
@@ -144,11 +153,11 @@ def _getWMSLayer(cls, layers, layer_name=None, authcfg=None):
             layer_name = 'wms_' + layers.replace(',', '')
         parms = {
             'crs': 'EPSG:4326',
-            'url': 'http://127.0.0.1:%s/?map=%s' % (cls.port, cls.project_path),
+            'url': '%s://%s:%s/?map=%s' % (cls.protocol, cls.hostname, cls.port, cls.project_path),
             'format': 'image/png',
             # This is needed because of a really weird implementation in QGIS Server, that
             # replaces _ in the the real layer name with spaces
-            'layers': urllib.quote(layers.replace('_', ' ')),
+            'layers': quote(layers.replace('_', ' ')),
             'styles': '',
             'version': 'auto',
             #'sql': '',