Distributed search auth

docs/distributedsearching.rst

@@ -122,10 +122,14 @@ With the above configured, a distributed search can be invoked as follows:
 
 http://localhost/collections/metadata:main/items?distributedSearch=true
 
+.. note::
+
+   To define access control as part of a distributed catalogue, see :ref:`access-control`.
+
 Merging results
 ^^^^^^^^^^^^^^^
 
-When `distributedsearch.merge_results` exists and is set to ``true``, pycsw will merge all results in ``federatedSearchResults``.  To prevent identifier collision, merged federated search results will have identifiers prefixed by their catalogue ``id`` (as defined in ``distributedsearch.catalogues[*].id``.  In addition, a ``federatedCatalogueId`` property is added to the feature with the catalogue id.
+When ``distributedsearch.merge_results`` exists and is set to ``true``, pycsw will merge all results in ``federatedSearchResults``.  To prevent identifier collision, merged federated search results will have identifiers prefixed by their catalogue ``id`` (as defined in ``distributedsearch.catalogues[*].id``.  In addition, a ``federatedCatalogueId`` property is added to the feature with the catalogue id.
 
 STAC API
 --------
@@ -147,11 +151,13 @@ Experimental support for distibuted searching is available in pycsw's STAC API s
             collections:
                 - daymet-annual-pr
 
-
 .. note::
 
    To constrain STAC API distributed search to specific collections, define one to many in the `collections` (array) directive.
 
+.. note::
+
+   To define access control as part of a distributed catalogue, see :ref:`access-control`.
 
 With the above configured, a distributed search can be invoked as follows:
 
@@ -164,3 +170,31 @@ Merging behaviour is implemented in the same manner as OGC API - Records support
 
 
 .. _`OGC API - Records - Part 4: Federated Search`: https://github.com/opengeospatial/ogcapi-records/blob/master/extensions/federated-search/document.adoc
+
+.. _access-control:
+
+Access control
+--------------
+
+Distributed search support includes the ability to query federated catalogues which are access controlled.  The various types of access control are defined below.
+
+OpenID Connect (OIDC)
+^^^^^^^^^^^^^^^^^^^^^
+
+To add an OIDC definition as part of a distributed catalogue, define an ``auth`` object per below:
+
+.. code-block:: yaml
+
+    distributedsearch:
+        catalogues:
+            - id: fedcat04
+              type: OARec
+              title: My OGC API - Records endpoint
+              url: https://example.org/ogc/api/collections/my-catalogue
+              auth:
+                  type: oidc
+                  client_id: foo
+                  client_secret: bar
+                  realm: realm123
+                  grant_type: client_credentials  # optional, defaults to 'client_credentials'
+                  url: https://example.org/oidc

pycsw/core/util.py

@@ -5,7 +5,7 @@
 #          Angelos Tzotsos <tzotsos@gmail.com>
 #          Ricardo Garcia Silva <ricardo.garcia.silva@gmail.com>
 #
-# Copyright (c) 2025 Tom Kralidis
+# Copyright (c) 2026 Tom Kralidis
 # Copyright (c) 2015 Angelos Tzotsos
 # Copyright (c) 2017 Ricardo Garcia Silva
 #
@@ -33,23 +33,24 @@
 # =================================================================
 
 from configparser import BasicInterpolation, ConfigParser
-from pathlib import Path
+import datetime
 import importlib
 import importlib.util
 import json
+import logging
 import os
+from pathlib import Path
 import re
-import datetime
-import logging
 import sys
 import time
 import typing
 
-from urllib.request import Request, urlopen
-from urllib.parse import urlparse
+from owslib.util import http_post
 from shapely.geometry import shape
 from shapely.wkt import loads
-from owslib.util import http_post
+import requests
+from urllib.request import Request, urlopen
+from urllib.parse import urlparse
 
 from pycsw.core.etree import etree, PARSER
 
@@ -175,7 +176,7 @@ def nspath_eval(xpath, nsmap):
 def wktenvelope2bbox(envelope):
     """returns bbox string of WKT ENVELOPE definition"""
 
-    tmparr = [x.strip() for x in envelope.split('(')[1].split(')')[0].split(',')]
+    tmparr = [x.strip() for x in envelope.split('(')[1].split(')')[0].split(',')]  # noqa
     bbox = '%s,%s,%s,%s' % (tmparr[0], tmparr[3], tmparr[1], tmparr[2])
     return bbox
 
@@ -237,7 +238,7 @@ def bbox2wktpolygon(bbox):
     precision = int(os.environ.get('COORDINATE_PRECISION', 2))
     if bbox.startswith('ENVELOPE'):
         bbox = wktenvelope2bbox(bbox)
-    minx, miny, maxx, maxy = [f"{float(coord):.{precision}f}" for coord in bbox.split(",")]
+    minx, miny, maxx, maxy = [f"{float(coord):.{precision}f}" for coord in bbox.split(",")]  # noqa
     wktGeometry = 'POLYGON((%s %s, %s %s, %s %s, %s %s, %s %s))' \
         % (minx, miny, minx, maxy, maxx, maxy, maxx, miny, minx, miny)
     return wktGeometry
@@ -500,9 +501,9 @@ def programmatic_import(target_module: str) -> typing.Optional[typing.Any]:
     target_module_path = Path(target_module)
     if target_module_path.is_file():
         module_name = target_module_path.stem
-        # this is an adaptation of the Python docs on using importlib to import a
-        # filepath:
-        # https://docs.python.org/3/library/importlib.html#importing-a-source-file-directly
+        # this is an adaptation of the Python docs on using importlib
+        # to import a filepath:
+        # https://docs.python.org/3/library/importlib.html#importing-a-source-file-directly  # noqa
         spec = importlib.util.spec_from_file_location(
             module_name, target_module_path)
         if spec is not None:
@@ -518,15 +519,15 @@ def programmatic_import(target_module: str) -> typing.Optional[typing.Any]:
     return result
 
 
-def load_custom_repo_mappings(repository_mappings: str) -> typing.Optional[typing.Dict]:
+def load_custom_repo_mappings(repository_mappings: str) -> typing.Optional[typing.Dict]:  # noqa
     imported_mappings_module = programmatic_import(repository_mappings)
     result = None
     if imported_mappings_module is not None:
         result = getattr(imported_mappings_module, "MD_CORE_MODEL", None)
     return result
 
 
-def sanitize_db_connect (url):
+def sanitize_db_connect(url):
     """
     helper function to remove user:pw from db connect for logging purposes
 
@@ -539,6 +540,7 @@ def sanitize_db_connect (url):
     else:
         return url
 
+
 def str2bool(value: typing.Union[bool, str]) -> bool:
     """
     helper function to return Python boolean
@@ -571,3 +573,52 @@ def remove_url_auth(url: str) -> str:
     u = urlparse(url)
     auth = f'{u.username}:{u.password}@'
     return url.replace(auth, '')
+
+
+def get_oidc_access_token(oidc: dict) -> str:
+    """
+    Helper function to return access token from OIDC
+
+    :param oidc: `dict` of OIDC settings:
+                 - client_id: client id
+                 - client_secret: client secret
+                 - realm: realm
+j                - url: URL of OIDC
+                 - grant_type: grant type (optional)
+
+    :returns: `str` of access token or `None`
+    """
+
+    client_id = oidc['client_id']
+    client_secret = oidc['client_secret']
+    url = oidc['url']
+    realm = oidc['realm']
+    grant_type = oidc.get('grant_type', 'client_credentials')
+
+    if None in [client_id, client_secret, realm, url]:
+        LOGGER.warning('Missing OIDC credential information in environment')
+
+    payload = {
+        'client_id': client_id,
+        'client_secret': client_secret
+    }
+
+    if grant_type is not None:
+        payload['grant_type'] = grant_type
+
+    url = f'{url}/realms/{realm}/protocol/openid-connect/token'
+    headers = {
+        'Content-Type': 'application/x-www-form-urlencoded'
+    }
+
+    try:
+        response = requests.post(url, data=payload, headers=headers)
+        response.raise_for_status()
+    except requests.exceptions.HTTPError as err:
+        LOGGER.warning(f'OIDC auth error: {err}')
+        return None
+    except requests.exceptions.MissingSchema as err:
+        LOGGER.warning(f'Invalid URL: {err}')
+        return None
+
+    return response.json().get('access_token')

pycsw/ogc/api/records.py

@@ -38,17 +38,19 @@
 from urllib.parse import urlencode, quote
 
 import json_merge_patch
-from owslib.ogcapi.records import Records
 from pygeofilter.parsers.ecql import parse as parse_ecql
 from pygeofilter.parsers.cql2_json import parse as parse_cql2_json
+import requests
 
 from pycsw import __version__
 from pycsw.broker import load_client
 from pycsw.core import log
 from pycsw.core.config import StaticContext
 from pycsw.core.metadata import parse_record
 from pycsw.core.pygeofilter_evaluate import to_filter
-from pycsw.core.util import bind_url, get_today_and_now, jsonify_links, load_custom_repo_mappings, str2bool, wkt2geom
+from pycsw.core.util import (bind_url, get_oidc_access_token, get_today_and_now,
+                             jsonify_links, load_custom_repo_mappings,
+                             str2bool, wkt2geom)
 from pycsw.ogc.api.oapi import gen_oapi
 from pycsw.ogc.api.util import match_env_var, render_j2_template, to_json, to_rfc3339
 from pycsw.ogc.pubsub import publish_message
@@ -879,14 +881,22 @@ def items(self, headers_, json_post_data, args, collection='metadata:main'):
                     LOGGER.debug(f"Federated catalogue type {fc['type']} not supported; skipping")
                     continue
                 LOGGER.debug(f"Running distributed search against {fc['url']}")
-                fc_url, _, fc_collection = fc['url'].rsplit('/', 2)
                 response['federatedSearchResults'][fc['id']] = {
                     'type': 'FeatureCollection',
                     'features': []
                 }
                 try:
-                    w = Records(fc_url)
-                    fc_results = w.collection_items(fc_collection, **args)
+                    ds_headers = {}
+                    if 'auth' in fc and fc['auth'].get('type', '') == 'oidc':
+                        LOGGER.debug('Getting OIDC acces token')
+                        ds_access_token = get_oidc_access_token(fc['auth'])
+                        if ds_access_token is not None:
+                            ds_headers = {
+                                'Content-Type': 'application/json',
+                                'Authorization': f'Bearer {ds_access_token}'
+                            }
+                    items_url = f"{fc['url']}/items"
+                    fc_results = requests.get(items_url, params=args, headers=ds_headers).json()
                     for feature in fc_results['features']:
                         if merge_results:
                             feature['id'] = f"{fc['id']}::{feature['id']}"
@@ -1041,11 +1051,19 @@ def item(self, headers_, args, collection, item):
                         LOGGER.debug(f"Federated catalogue type {fc['type']} not supported; skipping")
                         continue
                     LOGGER.debug(f"Running distributed item search against {fc['url']}")
-                    fc_url, _, fc_collection = fc.rsplit('/', 2)
                     try:
-                        w = Records(fc_url)
-                        response = record = w.collection_item(fc_collection, item)
-                        LOGGER.debug(f'Found item from {fc}')
+                        if 'auth' in fc and fc['auth'].get('type', '') == 'oidc':
+                            LOGGER.debug('Getting OIDC acces token')
+                            ds_access_token = get_oidc_access_token(fc['auth'])
+                            if ds_access_token is not None:
+                                ds_headers = {
+                                    'Content-Type': 'application/json',
+                                    'Authorization': f'Bearer {ds_access_token}'
+                                }
+
+                        item_url = f"{fc['url']}/items/{item}"
+                        response = record = requests.get(item_url, headers=ds_headers).json()
+                        LOGGER.debug(f"Found item {item} from {fc['url']}")
                         break
                     except RuntimeError:
                         continue

pycsw/stac/api.py

@@ -39,7 +39,8 @@
 from pycsw.core.pygeofilter_evaluate import to_filter
 from pycsw.ogc.api.oapi import gen_oapi
 from pycsw.ogc.api.records import API, build_anytext
-from pycsw.core.util import geojson_geometry2bbox, str2bool, wkt2geom
+from pycsw.core.util import (geojson_geometry2bbox, get_oidc_access_token,
+                             str2bool, wkt2geom)
 
 LOGGER = logging.getLogger(__name__)
 
@@ -424,9 +425,16 @@ def items(self, headers_, json_post_data, args, collection='metadata:main'):
         json_post_data2 = {}
         distributed_search_args = {}
 
-        distributed = str2bool(args.get('distributedsearch', False))
+        distributed = str2bool(args.get('distributedSearch', False))
 
         if distributed:
+
+            distributed_endpoints = [c for c in self.config['distributedsearch']['catalogues'] if c['type'] == 'STAC-API']
+
+            if not distributed_endpoints:
+                LOGGER.debug('Did not find any STAC-API endpoints; skipping')
+                distributed = False
+
             LOGGER.debug('Setting distributed search args')
             args.pop('distributedSearch', None)
             distributed_search_args = deepcopy(args)
@@ -652,8 +660,22 @@ def items(self, headers_, json_post_data, args, collection='metadata:main'):
 
                 try:
                     LOGGER.debug(f'Querying STAC API search: {url}')
-                    stac_search_results = requests.get(url, params=distributed_search_args2).json()
-                    for feature in stac_search_results['features']:
+                    ds_headers = {}
+
+                    if 'auth' in fc and fc['auth'].get('type', '') == 'oidc':
+                        LOGGER.debug('Getting OIDC acces token')
+                        ds_access_token = get_oidc_access_token(fc['auth'])
+                        if ds_access_token is not None:
+                            ds_headers = {
+                                'Content-Type': 'application/json',
+                                'Authorization': f'Bearer {ds_access_token}'
+                            }
+
+                    stac_search_results = requests.get(
+                        url, params=distributed_search_args2,
+                        headers=ds_headers).json()
+
+                    for feature in stac_search_results.get('features', []):
                         if merge_results:
                             feature['id'] = f"{fc['id']}::{feature['id']}"
                             feature['federatedCatalogueId'] = fc['id']