diff --git a/docs/distributedsearching.rst b/docs/distributedsearching.rst index ad40fba05..8e74e8fb7 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:`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 diff --git a/pycsw/core/util.py b/pycsw/core/util.py index f53cd6c18..2ba71dd37 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_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') diff --git a/pycsw/ogc/api/records.py b/pycsw/ogc/api/records.py index ba0693a98..dfbcc5510 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_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 diff --git a/pycsw/stac/api.py b/pycsw/stac/api.py index f2b0c3899..457cc7276 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_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']