Merge pull request #233 from Samweli/auth_support_api_header

Authentication support  for the QGIS API header method

src/qgis_stac/api/base.py

@@ -119,6 +119,7 @@ def get_items(
             api_capability=self.capability,
             response_handler=self.handle_items,
             error_handler=self.handle_error,
+            auth_config=self.auth_config,
         )
 
         QgsApplication.taskManager().addTask(self.content_task)
@@ -135,6 +136,7 @@ def get_collections(
             resource_type=ResourceType.COLLECTION,
             response_handler=self.handle_collections,
             error_handler=self.handle_error,
+            auth_config=self.auth_config,
         )
 
         QgsApplication.taskManager().addTask(self.content_task)
@@ -156,6 +158,7 @@ def get_collection(
             resource_type=ResourceType.COLLECTION,
             response_handler=self.handle_collection,
             error_handler=self.handle_error,
+            auth_config=self.auth_config,
         )
 
         QgsApplication.taskManager().addTask(self.content_task)
@@ -172,6 +175,7 @@ def get_conformance(
             resource_type=ResourceType.CONFORMANCE,
             response_handler=self.handle_conformance,
             error_handler=self.handle_error,
+            auth_config=self.auth_config,
         )
 
         QgsApplication.taskManager().addTask(self.content_task)

src/qgis_stac/api/models.py

@@ -143,6 +143,11 @@ class ResourceType(enum.Enum):
     CONFORMANCE = "Conformance"
 
 
+class QgsAuthMethods(enum.Enum):
+    """ Represents the QGIS authentication method names."""
+    API_HEADER = 'APIHeader'
+
+
 class QueryableFetchType(enum.Enum):
     """Queryable fetch types"""
     CATALOG = "Catalog"

src/qgis_stac/api/network.py

@@ -10,6 +10,7 @@
 
 from qgis.core import (
     QgsApplication,
+    QgsAuthMethodConfig,
     QgsNetworkContentFetcherTask,
     QgsTask,
 )
@@ -37,6 +38,7 @@
     ResourceType,
     SpatialExtent,
     TemporalExtent,
+    QgsAuthMethods,
     Queryable,
     QueryableProperty,
     QueryableFetchType
@@ -82,6 +84,7 @@ def __init__(
             api_capability: ApiCapability = None,
             response_handler: typing.Callable = None,
             error_handler: typing.Callable = None,
+            auth_config = None,
     ):
         super().__init__()
         self.url = url
@@ -90,6 +93,7 @@ def __init__(
         self.api_capability = api_capability
         self.response_handler = response_handler
         self.error_handler = error_handler
+        self.auth_config = auth_config
 
     def run(self):
         """
@@ -98,8 +102,13 @@ def run(self):
         :returns: Whether the task completed successfully
         :rtype: bool
         """
+        pystac_auth = {}
+        if self.auth_config:
+            pystac_auth = self.prepare_auth_properties(
+                self.auth_config
+            )
         try:
-            self.client = Client.open(self.url)
+            self.client = Client.open(self.url, **pystac_auth)
             if self.resource_type == \
                     ResourceType.FEATURE:
                 if self.search_params:
@@ -149,6 +158,31 @@ def run(self):
 
         return self.response is not None
 
+    def prepare_auth_properties(self, auth_config_id):
+        """ Fetches the required headers and parameters
+         from the QGIS Authentication method with the passed configuration id
+         and return their values in a dictionary
+
+         :param auth_config_id: Authentication method configuration id
+         :type auth_config_id: str
+
+         :returns: Authentication properties
+         :rtype: dict
+         """
+        auth_props = {}
+        auth_mgr = QgsApplication.authManager()
+        auth_cfg = QgsAuthMethodConfig()
+        auth_mgr.loadAuthenticationConfig(
+            auth_config_id,
+            auth_cfg,
+            True
+        )
+        if auth_cfg.method() == QgsAuthMethods.API_HEADER.value:
+            auth_props["headers"] = auth_cfg.configMap()
+            auth_props["parameters"] = auth_cfg.configMap()
+
+        return auth_props
+
     def prepare_collection_result(
             self,
             collection_response

test/mock/mock_http_server.py

@@ -7,14 +7,20 @@
 from threading import Thread
 
 from .stac_api_server_app import app
+from .stac_api_auth_server_app import app as auth_app
 
 
 class MockSTACApiServer(Thread):
     """ Mock a live """
-    def __init__(self, port=5000):
+    def __init__(self, port=5000, auth=False):
         super().__init__()
         self.port = port
-        self.app = app
+
+        if not auth:
+            self.app = app
+        else:
+            self.app = auth_app
+
         self.url = "http://localhost:%s" % self.port
 
         try:

test/mock/stac_api_auth_server_app.py

@@ -0,0 +1,30 @@
+import json
+import sys
+
+from pathlib import Path
+
+from flask import Flask, jsonify, request
+
+app = Flask(__name__)
+
+DATA_PATH = Path(__file__).parent / "data"
+
+
+@app.route("/")
+def catalog():
+    catalog = DATA_PATH / "catalog.json"
+
+    with catalog.open() as fl:
+        return json.load(fl)
+
+
+@app.route("/collections")
+def collections():
+    headers = request.headers
+    auth = headers.get("API_HEADER_KEY")
+    if auth == 'test_api_header_key':
+        collections = DATA_PATH / "collections.json"
+        with collections.open() as fl:
+            return json.load(fl)
+    else:
+        return jsonify({"message": "Unauthorized"}), 401

test/test_stac_api_client_auth.py

@@ -0,0 +1,111 @@
+# coding=utf-8
+"""Tests for the plugin STAC API client.
+
+"""
+import unittest
+import logging
+
+from multiprocessing import Process
+
+from mock.mock_http_server import MockSTACApiServer
+from qgis.PyQt.QtTest import QSignalSpy
+
+from qgis_stac.api.client import Client
+from qgis.core import QgsApplication
+from qgis.core import QgsAuthMethodConfig
+
+
+class STACApiClientAuthTest(unittest.TestCase):
+
+    def setUp(self):
+        self.app_server = MockSTACApiServer(auth=True)
+
+        self.server = Process(target=self.app_server.run)
+        self.server.start()
+
+        self.api_client = Client(self.app_server.url)
+        self.response = None
+        self.error = None
+
+    def set_auth_method(
+            self,
+            config_name,
+            config_method,
+            config_map
+    ):
+        AUTHDB_MASTERPWD = "password"
+
+        auth_manager = QgsApplication.authManager()
+        if not auth_manager.masterPasswordHashInDatabase():
+            auth_manager.setMasterPassword(AUTHDB_MASTERPWD, True)
+            # Create config
+            auth_manager.authenticationDatabasePath()
+            auth_manager.masterPasswordIsSet()
+
+        cfg = QgsAuthMethodConfig()
+        cfg.setName(config_name)
+        cfg.setMethod(config_method)
+        cfg.setConfigMap(config_map)
+        auth_manager.storeAuthenticationConfig(cfg)
+
+        return cfg.id()
+
+    def test_auth_collections_fetch(self):
+        # check if auth works correctly
+        cfg_id = self.set_auth_method(
+            "STAC_API_AUTH_TEST",
+            "APIHeader",
+            {"API_HEADER_KEY": "test_api_header_key"}
+        )
+
+        api_client = Client(
+            self.app_server.url,
+            auth_config=cfg_id
+        )
+
+        spy = QSignalSpy(api_client.collections_received)
+        api_client.collections_received.connect(self.app_response)
+        api_client.error_received.connect(self.error_response)
+
+        api_client.get_collections()
+        result = spy.wait(timeout=1000)
+
+        self.assertTrue(result)
+        self.assertIsNotNone(self.response)
+        self.assertIsNone(self.error)
+        self.assertEqual(len(self.response), 2)
+
+        cfg_id = self.set_auth_method(
+            "STAC_API_AUTH_TEST",
+            "APIHeader",
+            {"API_HEADER_KEY": "unauthorized_api_header_key"}
+        )
+
+        api_client = Client(
+            self.app_server.url,
+            auth_config=cfg_id
+        )
+
+        spy = QSignalSpy(api_client.collections_received)
+        api_client.collections_received.connect(self.app_response)
+        api_client.error_received.connect(self.error_response)
+
+        api_client.get_collections()
+        result = spy.wait(timeout=1000)
+
+        self.assertFalse(result)
+        self.assertIsNotNone(self.error)
+
+    def app_response(self, *response_args):
+        self.response = response_args
+
+    def error_response(self, *response_args):
+        self.error = response_args
+
+    def tearDown(self):
+        self.server.terminate()
+        self.server.join()
+
+
+if __name__ == '__main__':
+    unittest.main()