pip install psycopg2 numpy nose2 pyyaml mock future termcolor oauthlib

termcolor \

oauthlib

IF(WITH_OAUTH2_PLUGIN)

ADD_PYTHON_TEST(PyQgsAuthManagerOAuth2OWSTest test_authmanager_oauth2_ows.py)

ENDIF()

import copy

QGIS_SERVER_HTTP_BASIC_AUTH = os.environ.get('QGIS_SERVER_HTTP_BASIC_AUTH', False)

QGIS_SERVER_USERNAME = os.environ.get('QGIS_SERVER_USERNAME', 'username')

QGIS_SERVER_PASSWORD = os.environ.get('QGIS_SERVER_PASSWORD', 'password')

QGIS_SERVER_OAUTH2_CERTIFICATE = os.environ.get('QGIS_SERVER_OAUTH2_CERTIFICATE')

QGIS_SERVER_OAUTH2_KEY = os.environ.get('QGIS_SERVER_OAUTH2_KEY')

QGIS_SERVER_OAUTH2_AUTHORITY = os.environ.get('QGIS_SERVER_OAUTH2_AUTHORITY')

QGIS_SERVER_OAUTH2_USERNAME = os.environ.get('QGIS_SERVER_OAUTH2_USERNAME', 'username')

QGIS_SERVER_OAUTH2_PASSWORD = os.environ.get('QGIS_SERVER_OAUTH2_PASSWORD', 'password')

QGIS_SERVER_OAUTH2_TOKEN_EXPIRES_IN = os.environ.get('QGIS_SERVER_OAUTH2_TOKEN_EXPIRES_IN', 3600)

QGIS_SERVER_PKI_AUTH = (

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)

QGIS_SERVER_OAUTH2_AUTH = (

QGIS_SERVER_OAUTH2_CERTIFICATE is not None and

os.path.isfile(QGIS_SERVER_OAUTH2_CERTIFICATE) and

QGIS_SERVER_OAUTH2_KEY is not None and

os.path.isfile(QGIS_SERVER_OAUTH2_KEY) and

QGIS_SERVER_OAUTH2_AUTHORITY is not None and

os.path.isfile(QGIS_SERVER_OAUTH2_AUTHORITY) and

QGIS_SERVER_OAUTH2_USERNAME and QGIS_SERVER_OAUTH2_PASSWORD)

HTTPS_ENABLED = QGIS_SERVER_PKI_AUTH or QGIS_SERVER_OAUTH2_AUTH

if QGIS_SERVER_HTTP_BASIC_AUTH:

auth = handler.requestHeader('HTTP_AUTHORIZATION')

if QGIS_SERVER_OAUTH2_AUTH:

from qgis.server import QgsServerFilter

from oauthlib.oauth2 import RequestValidator, LegacyApplicationServer

import base64

from datetime import datetime

# Naive token storage implementation

_tokens = {}

class SimpleValidator(RequestValidator):

"""Validate username and password

def validate_client_id(self, client_id, request):

return True

def authenticate_client(self, request, *args, **kwargs):

"""Wide open"""

request.client = type("Client", (), {'client_id': 'my_id'})

return True

def validate_user(self, username, password, client, request, *args, **kwargs):

if username == QGIS_SERVER_OAUTH2_USERNAME and password == QGIS_SERVER_OAUTH2_PASSWORD:

return True

return False

def validate_grant_type(self, client_id, grant_type, client, request, *args, **kwargs):

# Clients should only be allowed to use one type of grant.

return grant_type in ('password', 'refresh_token')

def get_default_scopes(self, client_id, request, *args, **kwargs):

# Scopes a client will authorize for if none are supplied in the

# authorization request.

return ('my_scope', )

def validate_scopes(self, client_id, scopes, client, request, *args, **kwargs):

"""Wide open"""

return True

def save_bearer_token(self, token, request, *args, **kwargs):

# Remember to associate it with request.scopes, request.user and

# request.client. The two former will be set when you validate

# the authorization code. Don't forget to save both the

# access_token and the refresh_token and set expiration for the

# access_token to now + expires_in seconds.

_tokens[token['access_token']] = copy.copy(token)

_tokens[token['access_token']]['expiration'] = datetime.now().timestamp() + int(token['expires_in'])

def validate_bearer_token(self, token, scopes, request):

"""Check the token"""

return token in _tokens and _tokens[token]['expiration'] > datetime.now().timestamp()

def validate_refresh_token(self, refresh_token, client, request, *args, **kwargs):

"""Ensure the Bearer token is valid and authorized access to scopes."""

for t in _tokens.values():

if t['refresh_token'] == refresh_token:

return True

return False

def get_original_scopes(self, refresh_token, request, *args, **kwargs):

"""Get the list of scopes associated with the refresh token."""

return []

validator = SimpleValidator()

oauth_server = LegacyApplicationServer(validator, token_expires_in=QGIS_SERVER_OAUTH2_TOKEN_EXPIRES_IN)

class OAuth2Filter(QgsServerFilter):

"""This filter provides testing endpoint for OAuth2 Resource Owner Grant Flow

def responseComplete(self):

handler = self.serverInterface().requestHandler()

def _token(ttl):

"""Common code for new and refresh token"""

handler.clear()

body = bytes(handler.data()).decode('utf8')

old_expires_in = oauth_server.default_token_type.expires_in

# Hacky way to dynamically set token expiration time

oauth_server.default_token_type.expires_in = ttl

headers, payload, code = oauth_server.create_token_response('/token', 'post', body, {})

oauth_server.default_token_type.expires_in = old_expires_in

for k, v in headers.items():

handler.setResponseHeader(k, v)

handler.setStatusCode(code)

handler.appendBody(payload.encode('utf-8'))

# Token expiration

ttl = handler.parameterMap().get('TTL', QGIS_SERVER_OAUTH2_TOKEN_EXPIRES_IN)

# Issue a new token

if handler.url().find('/token') != -1:

_token(ttl)

return

# Refresh token

if handler.url().find('/refresh') != -1:

_token(ttl)

return

# Check for valid token

auth = handler.requestHeader('HTTP_AUTHORIZATION')

if auth:

result, response = oauth_server.verify_request(handler.url(), 'post', '', {'Authorization': auth})

if result:

# This is a test endpoint for OAuth2, it requires a valid token

if handler.url().find('/result') != -1:

handler.clear()

handler.appendBody(b'Valid Token: enjoy OAuth2')

# Standard flow

return

else:

# Wrong token, default response 401

pass

# No auth ...

handler.clear()

handler.setStatusCode(401)

handler.setResponseHeader('Status', '401 Unauthorized')

handler.setResponseHeader('WWW-Authenticate', 'Bearer realm="QGIS Server"')

handler.appendBody(b'Invalid Token: Authorization required.')

filter = OAuth2Filter(qgs_server.serverInterface())

qgs_server.serverInterface().registerFilter(filter)

# HTTPS is enabled if any of PKI or OAuth2 are enabled too

if HTTPS_ENABLED:

if QGIS_SERVER_OAUTH2_AUTH:

server.socket = ssl.wrap_socket(

server.socket,

certfile=QGIS_SERVER_OAUTH2_CERTIFICATE,

ca_certs=QGIS_SERVER_OAUTH2_AUTHORITY,

keyfile=QGIS_SERVER_OAUTH2_KEY,

server_side=True,

#cert_reqs=ssl.CERT_REQUIRED, # No certs for OAuth2

ssl_version=ssl.PROTOCOL_TLSv1)

else:

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)

('https' if HTTPS_ENABLED else 'http', QGIS_SERVER_HOST, server.server_port), flush=True)

import os

import sys

import re

import subprocess

import tempfile

import urllib

import stat

import json

import time

import random

__author__ = 'Alessandro Pasotti'

__date__ = '20/04/2017'

__copyright__ = 'Copyright 2017, The QGIS Project'

__revision__ = '$Format:%H$'

from shutil import rmtree

from utilities import unitTestDataPath, waitServer

from qgis.core import (

QgsAuthManager,

QgsAuthMethodConfig,

QgsVectorLayer,

QgsRasterLayer,

)

from qgis.PyQt.QtNetwork import QSslCertificate

from qgis.testing import (

start_app,

unittest,

)

try:

QGIS_SERVER_ENDPOINT_PORT = os.environ['QGIS_SERVER_ENDPOINT_PORT']

except:

QGIS_SERVER_ENDPOINT_PORT = '0' # Auto

QGIS_AUTH_DB_DIR_PATH = tempfile.mkdtemp()

os.environ['QGIS_AUTH_DB_DIR_PATH'] = QGIS_AUTH_DB_DIR_PATH

qgis_app = start_app()

def setup_oauth(username, password, token_uri, refresh_token_uri='', authcfg_id='oauth-2', authcfg_name='OAuth2 test configuration'):

"""Setup oauth configuration to access OAuth API,

cfgjson = {

"accessMethod": 0,

"apiKey": "",

"clientId": "",

"clientSecret": "",

"configType": 1,

"grantFlow": 2,

"password": password,

"persistToken": False,

"redirectPort": '7070',

"redirectUrl": "",

"refreshTokenUrl": refresh_token_uri,

"requestTimeout": '30',

"requestUrl": "",

"scope": "",

"state": "",

"tokenUrl": token_uri,

"username": username,

"version": 1

}

if authcfg_id not in QgsAuthManager.instance().availableAuthMethodConfigs():

authConfig = QgsAuthMethodConfig('OAuth2')

authConfig.setId(authcfg_id)

authConfig.setName(authcfg_name)

authConfig.setConfig('oauth2config', json.dumps(cfgjson))

if QgsAuthManager.instance().storeAuthenticationConfig(authConfig):

return authcfg_id

else:

authConfig = QgsAuthMethodConfig()

QgsAuthManager.instance().loadAuthenticationConfig(authcfg_id, authConfig, True)

authConfig.setName(authcfg_name)

authConfig.setConfig('oauth2config', json.dumps(cfgjson))

if QgsAuthManager.instance().updateAuthenticationConfig(authConfig):

return authcfg_id

return None

class TestAuthManager(unittest.TestCase):

@classmethod

def setUpAuth(cls):

"""Run before all tests and set up authentication"""

authm = QgsAuthManager.instance()

assert (authm.setMasterPassword('masterpassword', True))

cls.sslrootcert_path = os.path.join(cls.certsdata_path, 'chains_subissuer-issuer-root_issuer2-root2.pem')

assert os.path.isfile(cls.sslrootcert_path)

os.chmod(cls.sslrootcert_path, stat.S_IRUSR)

cls.sslrootcert = QSslCertificate.fromPath(cls.sslrootcert_path)

assert cls.sslrootcert is not None

authm.storeCertAuthorities(cls.sslrootcert)

authm.rebuildCaCertsCache()

authm.rebuildTrustedCaCertsCache()

cls.server_cert = os.path.join(cls.certsdata_path, '127_0_0_1_ssl_cert.pem')

cls.server_key = os.path.join(cls.certsdata_path, '127_0_0_1_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)

os.environ['QGIS_SERVER_HOST'] = cls.hostname

os.environ['QGIS_SERVER_PORT'] = str(cls.port)

os.environ['QGIS_SERVER_OAUTH2_KEY'] = cls.server_key

os.environ['QGIS_SERVER_OAUTH2_CERTIFICATE'] = cls.server_cert

os.environ['QGIS_SERVER_OAUTH2_USERNAME'] = cls.username

os.environ['QGIS_SERVER_OAUTH2_PASSWORD'] = cls.password

os.environ['QGIS_SERVER_OAUTH2_AUTHORITY'] = cls.server_rootcert

# Set default token expiration to 2 seconds, note that this can be

# also controlled when issuing token requests by adding ttl=<int>

# to the query string

os.environ['QGIS_SERVER_OAUTH2_TOKEN_EXPIRES_IN'] = '2'

@classmethod

def setUpClass(cls):

"""Run before all tests:

cls.port = QGIS_SERVER_ENDPOINT_PORT

# Clean env just to be sure

env_vars = ['QUERY_STRING', 'QGIS_PROJECT_FILE']

for ev in env_vars:

try:

del os.environ[ev]

except KeyError:

pass

cls.testdata_path = unitTestDataPath('qgis_server')

cls.certsdata_path = os.path.join(unitTestDataPath('auth_system'), 'certs_keys')

cls.project_path = os.path.join(cls.testdata_path, "test_project.qgs")

# cls.hostname = 'localhost'

cls.protocol = 'https'

cls.hostname = '127.0.0.1'

cls.username = 'username'

cls.password = 'password'

cls.setUpAuth()

server_path = os.path.join(os.path.dirname(os.path.realpath(__file__)),

'qgis_wrapped_server.py')

cls.server = subprocess.Popen([sys.executable, server_path],

env=os.environ, stdout=subprocess.PIPE)

line = cls.server.stdout.readline()

cls.port = int(re.findall(b':(\d+)', line)[0])

assert cls.port != 0

# We need a valid port before we setup the oauth configuration

cls.token_uri = '%s://%s:%s/token' % (cls.protocol, cls.hostname, cls.port)

cls.refresh_token_uri = '%s://%s:%s/refresh' % (cls.protocol, cls.hostname, cls.port)

# Need a random authcfg or the cache will bites us back!

cls.authcfg_id = setup_oauth(cls.username, cls.password, cls.token_uri, cls.refresh_token_uri, str(random.randint(0, 10000000)))

# This is to test wrong credentials

cls.wrong_authcfg_id = setup_oauth('wrong', 'wrong', cls.token_uri, cls.refresh_token_uri, str(random.randint(0, 10000000)))

# Get the authentication configuration instance:

cls.auth_config = QgsAuthManager.instance().availableAuthMethodConfigs()[cls.authcfg_id]

assert cls.auth_config.isValid()

# Wait for the server process to start

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):

"""Run after all tests"""

cls.server.kill()

rmtree(QGIS_AUTH_DB_DIR_PATH)

del cls.server

def setUp(self):

"""Run before each test."""

pass

def tearDown(self):

"""Run after each test."""

pass

@classmethod

def _getWFSLayer(cls, type_name, layer_name=None, authcfg=None):

"""

if layer_name is None:

layer_name = 'wfs_' + type_name

parms = {

'srsname': 'EPSG:4326',

'typename': type_name,

'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)) for k, v in list(parms.items())])

wfs_layer = QgsVectorLayer(uri, layer_name, 'WFS')

return wfs_layer

@classmethod

def _getWMSLayer(cls, layers, layer_name=None, authcfg=None):

"""

if layer_name is None:

layer_name = 'wms_' + layers.replace(',', '')

parms = {

'crs': 'EPSG:4326',

'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.parse.quote(layers.replace('_', ' ')),

'styles': '',

'version': 'auto',

# 'sql': '',

}

if authcfg is not None:

parms.update({'authcfg': authcfg})

uri = '&'.join([("%s=%s" % (k, v.replace('=', '%3D'))) for k, v in list(parms.items())])