From e5001b10dfc3a58ee6712d1bcdfafc132608dcda Mon Sep 17 00:00:00 2001 From: BAyotte Date: Wed, 29 Apr 2026 08:18:28 -0400 Subject: [PATCH 1/2] checking input characters validity and escaping html --- ogc/servers.py | 30 +++++++++- ogc/settings.py | 3 + ogc/test/test_input_security.py | 97 +++++++++++++++++++++++++++++++++ ogc/wcs_response_1_0_0.py | 20 ++++--- ogc/wms_response_1_3_0.py | 13 +++-- 5 files changed, 147 insertions(+), 16 deletions(-) create mode 100644 ogc/test/test_input_security.py diff --git a/ogc/servers.py b/ogc/servers.py index b0844bb..691bd9d 100755 --- a/ogc/servers.py +++ b/ogc/servers.py @@ -7,6 +7,7 @@ """ import re +import xml.sax.saxutils from flask import Flask, request, Response, make_response, send_file import six import logging @@ -21,6 +22,16 @@ logger = logging.getLogger(__name__) +def _check_query_string(raw_qs: bytes) -> None: + """Raise WCSException if the raw query string exceeds the maximum allowed length or contains invalid UTF-8.""" + if len(raw_qs) > settings.MAX_QUERY_STRING_BYTES: + raise WCSException("Request query string exceeds maximum allowed length.") + try: + raw_qs.decode("utf-8") + except UnicodeDecodeError: + raise WCSException("Request contains invalid UTF-8 encoding.") + + def respond_xml(doc, status=200): # First, validate that XML can be parsed. from lxml import etree @@ -231,6 +242,12 @@ def ogc_render(self, ogc_idx): return respond_xml("

Only GET supported

", status=405) ogc = self.ogcs[ogc_idx] + + try: + _check_query_string(request.query_string) + except WCSException as e: + return respond_xml(e.to_xml(), status=400) + if not request.args: return self.home_func(ogc.endpoint) try: @@ -253,7 +270,7 @@ def ogc_render(self, ogc_idx): } if request.base_url: - args["base_url"] = request.base_url + "?" + args["base_url"] = xml.sax.saxutils.escape(request.base_url, {'"': '"'}) + "?" else: args["base_url"] = None ogc_response = None @@ -310,6 +327,12 @@ def wrapper(*args, **kwargs) -> Response: logger.info("OGC server.edr_render") if request.method != "GET": return respond_xml("

Only GET supported

", status=405) + + try: + _check_query_string(request.query_string) + except WCSException as e: + return respond_xml(e.to_xml(), status=400) + try: # We'll filter out any characters from URl parameter values that # are not in the allowlist. @@ -336,7 +359,7 @@ def wrapper(*args, **kwargs) -> Response: if format_argument is not None: filtered_args["f"] = format_argument.lower() - filtered_args["base_url"] = request.base_url + filtered_args["base_url"] = xml.sax.saxutils.escape(request.base_url, {'"': '"'}) if request.base_url else None # Replace the arguments with the filtered option request.args = ImmutableMultiDict(filtered_args) @@ -347,6 +370,9 @@ def wrapper(*args, **kwargs) -> Response: if headers: response.headers = headers return response + except WCSException as e: + logger.error("OGC: server.edr_render WCSException: %s", str(e), exc_info=True) + return respond_xml(e.to_xml(), status=400) except Exception as e: logger.error("OGC: server.edr_render Exception: %s", str(e), exc_info=True) ee = WCSException() diff --git a/ogc/settings.py b/ogc/settings.py index 52a92e8..d1b09c3 100755 --- a/ogc/settings.py +++ b/ogc/settings.py @@ -62,6 +62,9 @@ # Max WCS/WMS response size MAX_GRID_COORDS_REQUEST_SIZE = 1024 * 1024 +# HTTP input size limits +MAX_QUERY_STRING_BYTES = 8192 + # WMS Capabilities limit layers WMS_LIMIT_LAYERS = False diff --git a/ogc/test/test_input_security.py b/ogc/test/test_input_security.py new file mode 100644 index 0000000..0a9fbce --- /dev/null +++ b/ogc/test/test_input_security.py @@ -0,0 +1,97 @@ +""" +Tests verifying that the server rejects oversized query strings. +""" + +import pytest +import numpy as np +import podpac + +from ogc import servers, core, settings +from ogc import podpac as pogc +from ogc.ogc_common import WCSException +from ogc.servers import _check_query_string + + +# --------------------------------------------------------------------------- +# Shared fixture (mirrors test_servers.py) +# --------------------------------------------------------------------------- + + +@pytest.fixture +def client(): + lat = np.linspace(90, -90, 11) + lon = np.linspace(-180, 180, 21) + data = np.random.default_rng(1).random((11, 21)) + coords = podpac.Coordinates([lat, lon], dims=["lat", "lon"]) + node = podpac.data.Array(source=data, coordinates=coords) + layer = pogc.Layer( + node=node, + identifier="layer1", + title="Layer 1", + abstract="Layer 1 Data", + group="Layers", + ) + ogc_instance = core.OGC(layers=[layer]) + app = servers.FlaskServer(__name__, ogcs=[ogc_instance]) + app.config["TESTING"] = True + yield app.test_client() + + +# --------------------------------------------------------------------------- +# 1. _check_query_string unit tests — exercises the real production function +# --------------------------------------------------------------------------- + + +def test_check_query_string_overflow(): + """Raises WCSException when byte length exceeds MAX_QUERY_STRING_BYTES.""" + oversized = b"A" * (settings.MAX_QUERY_STRING_BYTES + 1) + with pytest.raises(WCSException, match="maximum allowed length"): + _check_query_string(oversized) + + +def test_check_query_string_valid(): + """Does not raise for a normal ASCII query string.""" + _check_query_string(b"SERVICE=WCS&REQUEST=GetCapabilities&VERSION=1.0.0") + + +def test_check_query_string_exactly_at_limit(): + """Does not raise when the query string is exactly at the limit.""" + at_limit = b"A" * settings.MAX_QUERY_STRING_BYTES + _check_query_string(at_limit) + + +def test_check_query_string_invalid_utf8(): + """Raises WCSException when the query string contains raw non-UTF-8 bytes.""" + with pytest.raises(WCSException, match="invalid UTF-8 encoding"): + _check_query_string(b"SERVICE=WCS&COVERAGE=\xff\xfe") + + +def test_check_query_string_valid_percent_encoded(): + """Does not raise for percent-encoded non-ASCII (all bytes are ASCII in the raw QS).""" + _check_query_string(b"SERVICE=WCS&COVERAGE=%C3%A9") + + +# --------------------------------------------------------------------------- +# 2. HTTP integration — overflow rejected via test client +# --------------------------------------------------------------------------- + + +def test_ogc_render_overflow_returns_400(client): + """A query string over the limit returns a 400 WCSException response.""" + oversized = "A=" + "B" * (settings.MAX_QUERY_STRING_BYTES + 1) + response = client.get("/ogc", environ_overrides={"QUERY_STRING": oversized}) + assert response.status_code == 400 + assert b"ExceptionReport" in response.data + assert b"maximum allowed length" in response.data + + +# --------------------------------------------------------------------------- +# 3. Regression — valid WCS request still works end-to-end +# --------------------------------------------------------------------------- + + +def test_valid_wcs_get_capabilities_unchanged(client): + """Normal GetCapabilities request is unaffected by the new guards.""" + response = client.get("/ogc?SERVICE=WCS&REQUEST=GetCapabilities&VERSION=1.0.0") + assert response.status_code == 200 + assert b"WCS_Capabilities" in response.data diff --git a/ogc/wcs_response_1_0_0.py b/ogc/wcs_response_1_0_0.py index d90813f..0b657db 100755 --- a/ogc/wcs_response_1_0_0.py +++ b/ogc/wcs_response_1_0_0.py @@ -1,5 +1,6 @@ import logging import datetime +from xml.sax.saxutils import escape import traitlets as tl from ogc import GridCoordinates @@ -89,11 +90,11 @@ class CoverageDescription(ogc_common.XMLNode): def coverage_offering(self, coverage): xml = """ """ if coverage.identifier: - xml += " {coverage.identifier}\n".format(coverage=coverage) + xml += " {}\n".format(escape(coverage.identifier)) if coverage.title: - xml += " {coverage.title}\n".format(coverage=coverage) + xml += " {}\n".format(escape(coverage.title)) if coverage.abstract: - xml += " {coverage.abstract}\n".format(coverage=coverage) + xml += " {}\n".format(escape(coverage.abstract)) temporal_domain = "" if hasattr(coverage.layer, "valid_times"): if coverage.layer.all_times_valid: @@ -226,14 +227,15 @@ class Capabilities(ogc_common.XMLNode): version = tl.Unicode(default_value=SERVICE_VERSION) def service(self): + title = escape(self.service_title) if self.service_title else "" return """\ - {self.service_title} - {self.service_title} + {title} + {title} UNAVAILABLE {constraints} -""".format(self=self, constraints=settings.CONSTRAINTS) +""".format(title=title, constraints=settings.CONSTRAINTS) base_url = tl.Unicode(default_value=None, allow_none=True) # e.g., http://hostname:port/path? @@ -296,13 +298,13 @@ def contents(self): for coverage in self.coverages: xml += " \n" if coverage.abstract: - xml += " {coverage.abstract}\n".format(coverage=coverage) + xml += " {}\n".format(escape(coverage.abstract)) if coverage.identifier: # required - xml += " {coverage.identifier}\n".format(coverage=coverage) + xml += " {}\n".format(escape(coverage.identifier)) else: logger.info("Invalid layer. Missing name.") if coverage.title: # required - xml += " {coverage.title}\n".format(coverage=coverage) + xml += " {}\n".format(escape(coverage.title)) else: logger.info("Invalid layer. Missing label.") if coverage.wgs84_bounding_box_lower_corner_lat_lon or coverage.wgs84_bounding_box_upper_corner_lat_lon: diff --git a/ogc/wms_response_1_3_0.py b/ogc/wms_response_1_3_0.py index 4e195f1..5edadde 100755 --- a/ogc/wms_response_1_3_0.py +++ b/ogc/wms_response_1_3_0.py @@ -1,4 +1,5 @@ import logging +from xml.sax.saxutils import escape import traitlets as tl import numpy as np @@ -43,10 +44,11 @@ class Capabilities(ogc_common.XMLNode): version = tl.Unicode(default_value=SERVICE_VERSION) def service(self): + title = escape(self.service_title) if self.service_title else "" return """\ WMS - {self.service_title} + {title} {constraints} 1 @@ -54,6 +56,7 @@ def service(self): {maxHeightWMS} """.format( + title=title, self=self, constraints=settings.CONSTRAINTS, maxWidthWMS=int(np.sqrt(settings.MAX_GRID_COORDS_REQUEST_SIZE)), @@ -121,13 +124,13 @@ def coverage_times_list(self, coverage, default_time): def coverage_layer(self, coverage): xml = """ \n""" if coverage.identifier: - xml += f" {coverage.identifier}\n" + xml += f" {escape(coverage.identifier)}\n" if coverage.title: - xml += f" {coverage.title}\n" + xml += f" {escape(coverage.title)}\n" else: logger.info("Invalid layer. Missing title.") if coverage.abstract: - xml += f" {coverage.abstract}\n" + xml += f" {escape(coverage.abstract)}\n" xml += self._get_CRS_and_BoundingBox() @@ -199,7 +202,7 @@ def coverage_layer(self, coverage): def layers(self): xml = " \n" - xml += " {}\n".format(self.service_group_title) + xml += " {}\n".format(escape(self.service_group_title) if self.service_group_title else "") xml += self._get_CRS_and_BoundingBox(depth=2) # If configured, trim layers list to layers specified in settings From ba7e14b181454307397ef3f9a708934449df87b5 Mon Sep 17 00:00:00 2001 From: BAyotte Date: Wed, 29 Apr 2026 14:17:49 +0000 Subject: [PATCH 2/2] updating black formatting --- ogc/servers.py | 6 ++++-- ogc/test/test_input_security.py | 1 - ogc/wms_response_1_3_0.py | 4 +++- 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/ogc/servers.py b/ogc/servers.py index 691bd9d..2c1093f 100755 --- a/ogc/servers.py +++ b/ogc/servers.py @@ -270,7 +270,7 @@ def ogc_render(self, ogc_idx): } if request.base_url: - args["base_url"] = xml.sax.saxutils.escape(request.base_url, {'"': '"'}) + "?" + args["base_url"] = xml.sax.saxutils.escape(request.base_url, {'"': """}) + "?" else: args["base_url"] = None ogc_response = None @@ -359,7 +359,9 @@ def wrapper(*args, **kwargs) -> Response: if format_argument is not None: filtered_args["f"] = format_argument.lower() - filtered_args["base_url"] = xml.sax.saxutils.escape(request.base_url, {'"': '"'}) if request.base_url else None + filtered_args["base_url"] = ( + xml.sax.saxutils.escape(request.base_url, {'"': """}) if request.base_url else None + ) # Replace the arguments with the filtered option request.args = ImmutableMultiDict(filtered_args) diff --git a/ogc/test/test_input_security.py b/ogc/test/test_input_security.py index 0a9fbce..3093681 100644 --- a/ogc/test/test_input_security.py +++ b/ogc/test/test_input_security.py @@ -11,7 +11,6 @@ from ogc.ogc_common import WCSException from ogc.servers import _check_query_string - # --------------------------------------------------------------------------- # Shared fixture (mirrors test_servers.py) # --------------------------------------------------------------------------- diff --git a/ogc/wms_response_1_3_0.py b/ogc/wms_response_1_3_0.py index 5edadde..813a487 100755 --- a/ogc/wms_response_1_3_0.py +++ b/ogc/wms_response_1_3_0.py @@ -202,7 +202,9 @@ def coverage_layer(self, coverage): def layers(self): xml = " \n" - xml += " {}\n".format(escape(self.service_group_title) if self.service_group_title else "") + xml += " {}\n".format( + escape(self.service_group_title) if self.service_group_title else "" + ) xml += self._get_CRS_and_BoundingBox(depth=2) # If configured, trim layers list to layers specified in settings