add support for internal request blocking on harvest

docs/configuration.rst

@@ -27,6 +27,7 @@ pycsw's runtime configuration is defined by ``default.yml``.  pycsw ships with a
 - **smtp_ssl**: Option to choose between SMTP and SMTP_SSL. To enable it, set the value to ``true`` (default is ``false``)
 - **spatial_ranking**: parameter that enables (``true`` or ``false``) ranking of spatial query results as per `K.J. Lanfear 2006 - A Spatial Overlay Ranking Method for a Geospatial Search of Text Objects  <https://pubs.usgs.gov/of/2006/1279/2006-1279.pdf>`_.
 - **workers**: set the number of workers used by the wsgi server when lunching pycsw using the provided docker/entrypoint.py. If not set, it will use 2 workers as Default.
+- **allow_internal_requests**: whether to allow for internal HTTP requests to be invoked (default is ``false``)
 
 **profiles**
 

pycsw/core/util.py

@@ -36,20 +36,20 @@
 import datetime
 import importlib
 import importlib.util
+import ipaddress
 import json
 import logging
 import os
 from pathlib import Path
 import re
+import socket
 import sys
 import time
 import typing
 
-from owslib.util import http_post
 from shapely.geometry import shape
 from shapely.wkt import loads
 import requests
-from urllib.request import Request, urlopen
 from urllib.parse import urlparse
 
 from pycsw.core.etree import etree, PARSER
@@ -290,14 +290,23 @@ def getqattr(obj, name):
     return result
 
 
-def http_request(method, url, request=None, timeout=30):
+def http_request(method, url, request=None, timeout=30,
+                 allow_internal_requests=False):
     """Perform HTTP request"""
+
+    if not is_request_allowed(url, allow_internal_requests):
+        raise ValueError('URL not allowed')
+
+    headers = {
+        'User-Agent': 'pycsw (https://pycsw.org/)'
+    }
+
     if method == 'POST':
-        return http_post(url, request, timeout=timeout).text
+        return requests.post(url, headers=headers, data=request,
+                             timeout=timeout, allow_redirects=False).text
     else:  # GET
-        request = Request(url)
-        request.add_header('User-Agent', 'pycsw (https://pycsw.org/)')
-        return urlopen(request, timeout=timeout).read()
+        return requests.get(url, headers=headers, timeout=timeout,
+                            allow_redirects=False).text
 
 
 def bind_url(url):
@@ -622,3 +631,30 @@ def get_oidc_access_token(oidc: dict) -> str:
         return None
 
     return response.json().get('access_token')
+
+
+def is_request_allowed(url: str, allow_internal: bool = False) -> bool:
+    """
+    Test whether an HTTP request is allowed to be executed
+
+    :param url: `str` of URL
+    :param allow_internal: `bool` of whether internal requests are
+                           allowed (default `False`)
+
+    :returns: `bool` of whether HTTP request execution is allowed
+    """
+
+    is_allowed = False
+
+    u = urlparse(url)
+
+    ip = socket.gethostbyname(u.hostname)
+
+    is_private = ipaddress.ip_address(ip).is_private
+
+    if not is_private:
+        is_allowed = True
+    if is_private and allow_internal:
+        is_allowed = True
+
+    return is_allowed

pycsw/ogc/csw/csw2.py

@@ -1287,7 +1287,7 @@ def harvest(self):
             # fetch content-based resource
             LOGGER.debug('Fetching resource %s', self.parent.kvp['source'])
             try:
-                content = util.http_request('GET', self.parent.kvp['source'])
+                content = util.http_request('GET', self.parent.kvp['source'], self.parent.config['server'].get('allow_internal_requests', False))
             except Exception as err:
                 errortext = 'Error fetching resource %s.\nError: %s.' % \
                 (self.parent.kvp['source'], str(err))

pycsw/ogc/csw/csw3.py

@@ -1346,7 +1346,8 @@ def harvest(self):
             # fetch content-based resource
             LOGGER.info('Fetching resource %s', self.parent.kvp['source'])
             try:
-                content = util.http_request('GET', self.parent.kvp['source'])
+                content = util.http_request('GET', self.parent.kvp['source'], self.parent.config['server'].get('allow_internal_     requests', False))
+
             except Exception as err:
                 errortext = 'Error fetching resource %s.\nError: %s.' % \
                 (self.parent.kvp['source'], str(err))

pycsw/server.py

@@ -112,6 +112,7 @@ def __init__(self, rtconfig=None, env=None, version='3.0.0'):
         # set server.home safely
         # TODO: make this more abstract
         self.config['server']['home'] = os.path.dirname(os.path.join(os.path.dirname(__file__), '..'))
+        self.config['server']['allow_internal_requests'] = self.config['server'].get('allow_internal_requests', False)
 
         if 'PYCSW_IS_CSW' in self.environ and self.environ['PYCSW_IS_CSW']:
             self.config['server']['url'] = self.config['server']['url'].rstrip('/') + '/csw'

tests/unittests/test_util.py

@@ -4,7 +4,7 @@
 # Authors: Tom Kralidis
 #
 # Copyright (c) 2017 Ricardo Garcia Silva
-# Copyright (c) 2025 Tom Kralidis
+# Copyright (c) 2026 Tom Kralidis
 #
 # Permission is hereby granted, free of charge, to any person
 # obtaining a copy of this software and associated documentation
@@ -202,24 +202,6 @@ def test_getqattr_invalid():
     assert result is None
 
 
-def test_http_request_post():
-    # here we replace owslib.util.http_post with a mock object
-    # because we are not interested in testing owslib
-    method = "POST"
-    url = "some_phony_url"
-    request = "some_phony_request"
-    timeout = 40
-    with mock.patch("pycsw.core.util.http_post",
-                    autospec=True) as mock_http_post:
-        util.http_request(
-            method=method,
-            url=url,
-            request=request,
-            timeout=timeout
-        )
-        mock_http_post.assert_called_with(url, request, timeout=timeout)
-
-
 @pytest.mark.parametrize("url, expected", [
     ("http://host/wms", "http://host/wms?"),
     ("http://host/wms?foo=bar&", "http://host/wms?foo=bar&"),
@@ -437,3 +419,21 @@ def test_str2bool():
 def test_geojson_geometry2bbox(geometry, expected):
     bounds = util.geojson_geometry2bbox(geometry)
     assert bounds == expected
+
+
+@pytest.mark.parametrize('url,allow_internal,result', [
+    ['http://127.0.0.1/test', False, False],
+    ['http://127.0.0.1/test', True, True],
+    ['http://192.168.0.12/test', False, False],
+    ['http://192.168.0.12/test', True, True],
+    ['http://169.254.0.11/test', False, False],
+    ['http://169.254.0.11/test', True, True],
+    ['http://0.0.0.0/test', True, True],
+    ['http://0.0.0.0/test', False, False],
+    ['http://localhost:5000/test', False, False],
+    ['http://localhost:5000/test', True, True],
+    ['https://pycsw.org', False, True],
+    ['https://pycsw.org', True, True]
+])
+def test_is_request_allowed(url, allow_internal, result):
+    assert util.is_request_allowed(url, allow_internal) is result