From 1a871cf70b49cd9b561bffb908c2244dd52250d7 Mon Sep 17 00:00:00 2001 From: Angelos Tzotsos Date: Sun, 26 Jul 2026 11:32:25 -0400 Subject: [PATCH 1/2] add auth support for federated catalogues --- docs/distributedsearching.rst | 38 ++++++++++++++++- pycsw/core/util.py | 79 ++++++++++++++++++++++++++++------- pycsw/ogc/api/records.py | 36 ++++++++++++---- pycsw/stac/api.py | 30 +++++++++++-- 4 files changed, 154 insertions(+), 29 deletions(-) diff --git a/docs/distributedsearching.rst b/docs/distributedsearching.rst index ad40fba05..d8c0390fd 100644 --- a/docs/distributedsearching.rst +++ b/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:`auth`. + 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:`auth`. 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 + +.. _auth: + +Authentication and Authorization support +---------------------------------------- + +Distributed search support includes the ability to query federated catalogues which are access controlled. The various types of access control are defined below. + +Identity and Access Management (IAM) +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +To add an IAM definition as part of a distributed catalogue, define am ``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: iam + client_id: foo + client_secret: bar + realm: realm123 + grant_type: client_credentials # optional, defaults to 'client_credentials' + url: https://example.org/iam diff --git a/pycsw/core/util.py b/pycsw/core/util.py index f53cd6c18..16273e6ba 100644 --- a/pycsw/core/util.py +++ b/pycsw/core/util.py @@ -5,7 +5,7 @@ # Angelos Tzotsos # Ricardo Garcia Silva # -# 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,7 +519,7 @@ 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: @@ -526,7 +527,7 @@ def load_custom_repo_mappings(repository_mappings: str) -> typing.Optional[typin 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_iam_access_token(iam: dict) -> str: + """ + Helper function to return access token from IAM + + :param iam: `dict` of IAM settings: + - client_id: client id + - client_secret: client secret + - realm: realm + - url: URL of IAM + - grant_type: grant type (optional) + + :returns: `str` of access token or `None` + """ + + client_id = iam['client_id'] + client_secret = iam['client_secret'] + url = iam['url'] + realm = iam['realm'] + grant_type = iam.get('grant_type', 'client_credentials') + + if None in [client_id, client_secret, realm, url]: + LOGGER.warning('Missing IAM 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'IAM 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') diff --git a/pycsw/ogc/api/records.py b/pycsw/ogc/api/records.py index ba0693a98..d5220f088 100644 --- a/pycsw/ogc/api/records.py +++ b/pycsw/ogc/api/records.py @@ -38,9 +38,9 @@ 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 @@ -48,7 +48,9 @@ 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_iam_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', '') == 'iam': + LOGGER.debug('Getting IAM acces token') + ds_access_token = get_iam_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', '') == 'iam': + LOGGER.debug('Getting IAM acces token') + ds_access_token = get_iam_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 diff --git a/pycsw/stac/api.py b/pycsw/stac/api.py index f2b0c3899..279090341 100644 --- a/pycsw/stac/api.py +++ b/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_iam_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', '') == 'iam': + LOGGER.debug('Getting IAM acces token') + ds_access_token = get_iam_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'] From 6aa1ac5a2ced42ac00dc11510b15bcd358582e88 Mon Sep 17 00:00:00 2001 From: Tom Kralidis Date: Sun, 26 Jul 2026 14:42:39 -0400 Subject: [PATCH 2/2] s/iam/oidc/i --- docs/distributedsearching.rst | 20 ++++++++++---------- pycsw/core/util.py | 30 +++++++++++++++--------------- pycsw/ogc/api/records.py | 14 +++++++------- pycsw/stac/api.py | 8 ++++---- 4 files changed, 36 insertions(+), 36 deletions(-) diff --git a/docs/distributedsearching.rst b/docs/distributedsearching.rst index d8c0390fd..8e74e8fb7 100644 --- a/docs/distributedsearching.rst +++ b/docs/distributedsearching.rst @@ -124,7 +124,7 @@ http://localhost/collections/metadata:main/items?distributedSearch=true .. note:: - To define access control as part of a distributed catalogue, see :ref:`auth`. + To define access control as part of a distributed catalogue, see :ref:`access-control`. Merging results ^^^^^^^^^^^^^^^ @@ -157,7 +157,7 @@ Experimental support for distibuted searching is available in pycsw's STAC API s .. note:: - To define access control as part of a distributed catalogue, see :ref:`auth`. + 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: @@ -171,17 +171,17 @@ 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 -.. _auth: +.. _access-control: -Authentication and Authorization support ----------------------------------------- +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. -Identity and Access Management (IAM) -^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +OpenID Connect (OIDC) +^^^^^^^^^^^^^^^^^^^^^ -To add an IAM definition as part of a distributed catalogue, define am ``auth`` object per below: +To add an OIDC definition as part of a distributed catalogue, define an ``auth`` object per below: .. code-block:: yaml @@ -192,9 +192,9 @@ To add an IAM definition as part of a distributed catalogue, define am ``auth`` title: My OGC API - Records endpoint url: https://example.org/ogc/api/collections/my-catalogue auth: - type: iam + type: oidc client_id: foo client_secret: bar realm: realm123 grant_type: client_credentials # optional, defaults to 'client_credentials' - url: https://example.org/iam + url: https://example.org/oidc diff --git a/pycsw/core/util.py b/pycsw/core/util.py index 16273e6ba..2ba71dd37 100644 --- a/pycsw/core/util.py +++ b/pycsw/core/util.py @@ -575,28 +575,28 @@ def remove_url_auth(url: str) -> str: return url.replace(auth, '') -def get_iam_access_token(iam: dict) -> str: +def get_oidc_access_token(oidc: dict) -> str: """ - Helper function to return access token from IAM + Helper function to return access token from OIDC - :param iam: `dict` of IAM settings: - - client_id: client id - - client_secret: client secret - - realm: realm - - url: URL of IAM - - grant_type: grant type (optional) + :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 = iam['client_id'] - client_secret = iam['client_secret'] - url = iam['url'] - realm = iam['realm'] - grant_type = iam.get('grant_type', 'client_credentials') + 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 IAM credential information in environment') + LOGGER.warning('Missing OIDC credential information in environment') payload = { 'client_id': client_id, @@ -615,7 +615,7 @@ def get_iam_access_token(iam: dict) -> str: response = requests.post(url, data=payload, headers=headers) response.raise_for_status() except requests.exceptions.HTTPError as err: - LOGGER.warning(f'IAM auth error: {err}') + LOGGER.warning(f'OIDC auth error: {err}') return None except requests.exceptions.MissingSchema as err: LOGGER.warning(f'Invalid URL: {err}') diff --git a/pycsw/ogc/api/records.py b/pycsw/ogc/api/records.py index d5220f088..dfbcc5510 100644 --- a/pycsw/ogc/api/records.py +++ b/pycsw/ogc/api/records.py @@ -48,7 +48,7 @@ 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_iam_access_token, get_today_and_now, +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 @@ -887,9 +887,9 @@ def items(self, headers_, json_post_data, args, collection='metadata:main'): } try: ds_headers = {} - if 'auth' in fc and fc['auth'].get('type', '') == 'iam': - LOGGER.debug('Getting IAM acces token') - ds_access_token = get_iam_access_token(fc['auth']) + 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', @@ -1052,9 +1052,9 @@ def item(self, headers_, args, collection, item): continue LOGGER.debug(f"Running distributed item search against {fc['url']}") try: - if 'auth' in fc and fc['auth'].get('type', '') == 'iam': - LOGGER.debug('Getting IAM acces token') - ds_access_token = get_iam_access_token(fc['auth']) + 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', diff --git a/pycsw/stac/api.py b/pycsw/stac/api.py index 279090341..457cc7276 100644 --- a/pycsw/stac/api.py +++ b/pycsw/stac/api.py @@ -39,7 +39,7 @@ 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, get_iam_access_token, +from pycsw.core.util import (geojson_geometry2bbox, get_oidc_access_token, str2bool, wkt2geom) LOGGER = logging.getLogger(__name__) @@ -662,9 +662,9 @@ def items(self, headers_, json_post_data, args, collection='metadata:main'): LOGGER.debug(f'Querying STAC API search: {url}') ds_headers = {} - if 'auth' in fc and fc['auth'].get('type', '') == 'iam': - LOGGER.debug('Getting IAM acces token') - ds_access_token = get_iam_access_token(fc['auth']) + 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',