OAProc: secure subscriber URLs in requests

Author: tomkralidis

docs/source/configuration.rst

@@ -292,6 +292,10 @@ default.
           type: process  # REQUIRED (collection, process, or stac-collection)
           processor:
               name: HelloWorld  # Python path of process definition
+          # optional, allow for internal HTTP request execution
+          # if set to True, enables requests to link local ranges and loopback
+          # default: False
+          allow_internal_requests: True
 
 
 .. seealso::

pygeoapi/process/base.py

@@ -3,7 +3,7 @@
 # Authors: Tom Kralidis <tomkralidis@gmail.com>
 #          Francesco Martinelli <francesco.martinelli@ingv.it>
 #
-# Copyright (c) 2022 Tom Kralidis
+# Copyright (c) 2026 Tom Kralidis
 # Copyright (c) 2024 Francesco Martinelli
 #
 # Permission is hereby granted, free of charge, to any person
@@ -53,6 +53,8 @@ def __init__(self, processor_def: dict, process_metadata: dict):
         self.name = processor_def['name']
         self.metadata = process_metadata
         self.supports_outputs = False
+        self.allow_internal_requests = processor_def.get(
+            'allow_internal_requests', False)
 
     def set_job_id(self, job_id: str) -> None:
         """

pygeoapi/process/manager/base.py

@@ -46,10 +46,12 @@
     BaseProcessor,
     JobNotFoundError,
     JobResultNotFoundError,
+    ProcessorExecuteError,
     UnknownProcessError,
 )
 from pygeoapi.util import (
     get_current_datetime,
+    is_request_allowed,
     JobStatus,
     ProcessExecutionMode,
     RequestedProcessExecutionMode,
@@ -105,7 +107,11 @@ def get_processor(self, process_id: str) -> BaseProcessor:
         except KeyError as err:
             raise UnknownProcessError('Invalid process identifier') from err
         else:
-            return load_plugin('process', process_conf['processor'])
+            pp = load_plugin('process', process_conf['processor'])
+            pp.allow_internal_requests = process_conf.get(
+                'allow_internal_requests', False)
+
+            return pp
 
     def get_jobs(self,
                  status: JobStatus = None,
@@ -394,13 +400,13 @@ def execute_process(
         """
 
         job_id = str(uuid.uuid1())
-        processor = self.get_processor(process_id)
-        processor.set_job_id(job_id)
+        self.processor = self.get_processor(process_id)
+        self.processor.set_job_id(job_id)
         extra_execute_handler_parameters = {
             'requested_response': requested_response
         }
 
-        job_control_options = processor.metadata.get(
+        job_control_options = self.processor.metadata.get(
             'jobControlOptions', [])
 
         if execution_mode == RequestedProcessExecutionMode.respond_async:
@@ -473,7 +479,7 @@ def execute_process(
         # TODO: handler's response could also be allowed to include more HTTP
         # headers
         mime_type, outputs, status = handler(
-            processor,
+            self.processor,
             job_id,
             data_dict,
             requested_outputs,
@@ -483,26 +489,37 @@ def execute_process(
 
     def _send_in_progress_notification(self, subscriber: Optional[Subscriber]):
         if subscriber and subscriber.in_progress_uri:
-            response = requests.post(subscriber.in_progress_uri, json={})
-            LOGGER.debug(
-                f'In progress notification response: {response.status_code}'
-            )
+            self.__do_subscriber_request(subscriber.in_progress_uri)
 
     def _send_success_notification(
             self, subscriber: Optional[Subscriber], outputs: Any
     ):
-        if subscriber:
-            response = requests.post(subscriber.success_uri, json=outputs)
-            LOGGER.debug(
-                f'Success notification response: {response.status_code}'
-            )
+        if subscriber and subscriber.success_uri:
+            self.__do_subscriber_request(subscriber.success_uri, outputs)
 
     def _send_failed_notification(self, subscriber: Optional[Subscriber]):
         if subscriber and subscriber.failed_uri:
-            response = requests.post(subscriber.failed_uri, json={})
-            LOGGER.debug(
-                f'Failed notification response: {response.status_code}'
-            )
+            self.__do_subscriber_request(subscriber.failed_uri)
+
+    def __do_subscriber_request(self, url: str, data: dict = {}) -> None:
+        """
+        Helper function to execute a subscriber URL via HTTP POST
+
+        :param url: `str` of URL
+        :param data: `dict` of request payload
+
+        :returns: `None`
+        """
+
+        if not is_request_allowed(url, self.processor.allow_internal_requests):
+            msg = 'URL not allowed'
+            LOGGER.error(f'{msg}: {url}')
+            raise ProcessorExecuteError(msg)
+
+        response = requests.post(url, json=data)
+        LOGGER.debug(
+            f'Response: {response.status_code}'
+        )
 
     def __repr__(self):
         return f'<BaseManager> {self.name}'

pygeoapi/provider/filesystem.py

@@ -78,7 +78,7 @@ def get_data_path(self, baseurl, urlpath, dirpath):
         child_links = []
 
         if '..' in dirpath:
-            msg = f'Invalid path requested'
+            msg = 'Invalid path requested'
             LOGGER.error(f'{msg}: {dirpath}')
             raise ProviderInvalidQueryError(msg)
 

pygeoapi/resources/schemas/config/pygeoapi-config-0.x.yml

@@ -682,7 +682,11 @@ properties:
                                           For custom built plugins, use the import path (e.g. `mypackage.provider.MyProvider`)
                               required:
                                   - name
-                      required:
+                          allow_internal_requests:
+                              type: boolean
+                              description: whether to allow internal HTTP requests
+                              default: false
+                      requred:
                           - type
                           - processor
 definitions:

pygeoapi/util.py

@@ -36,13 +36,15 @@
 from decimal import Decimal
 from enum import Enum
 from heapq import heappush
+import ipaddress
 import json
 import logging
 import mimetypes
 import os
 import pathlib
 from pathlib import Path
 import re
+import socket
 from typing import Any, IO, Union, List, Optional
 from urllib.parse import urlparse
 from urllib.request import urlopen
@@ -754,3 +756,30 @@ def remove_url_auth(url: str) -> str:
     u = urlparse(url)
     auth = f'{u.username}:{u.password}@'
     return url.replace(auth, '')
+
+
+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

tests/other/test_util.py

@@ -2,7 +2,7 @@
 #
 # Authors: Tom Kralidis <tomkralidis@gmail.com>
 #
-# 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
@@ -321,3 +321,21 @@ def test_get_choice_from_headers():
                                         'accept') == 'application/ld+json'
     assert util.get_choice_from_headers(
         {'accept-language': 'en_US', 'accept': '*/*'}, 'accept') == '*/*'
+
+
+@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://pygeoapi.io', False, True],
+    ['https://pygeoapi.io', True, True]
+])
+def test_is_request_allowed(url, allow_internal, result):
+    assert util.is_request_allowed(url, allow_internal) is result