diff --git a/CHANGELOG.md b/CHANGELOG.md index 701e134..3f2c011 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,13 @@ # Changelog +## 0.6.1 +### Introduction +Covered additional security-relevant cases. + +### Maintenance +* Added input validation +* Added exception sanitization +* Added checks for invalid relative paths and symlinks + ## 0.6.0 ### Introduction diff --git a/ogc/edr/edr_api.py b/ogc/edr/edr_api.py index 902e65c..b385a64 100644 --- a/ogc/edr/edr_api.py +++ b/ogc/edr/edr_api.py @@ -1,14 +1,17 @@ import json +import logging import pyproj import numpy as np import pygeoapi.api import pygeoapi.api.environmental_data_retrieval as pygeoedr +from functools import wraps from http import HTTPStatus from datetime import datetime, timezone -from typing import Tuple, List, Dict, Any, Union +from typing import Tuple, List, Dict, Any, Union, Callable from traitlets import TraitError from ogc import podpac as pogc +from ogc.ogc_common import EDRException from pygeoapi.plugin import load_plugin from pygeoapi.util import filter_dict_by_key_value, to_json, get_provider_by_type from pygeoapi.api import API, APIRequest @@ -16,6 +19,8 @@ from .edr_provider import EdrProvider from .. import settings +logger = logging.getLogger(__file__) + class EdrAPI: """Used to modify the default responses before returning data to the user.""" @@ -29,7 +34,56 @@ class EdrAPI: ) SCHEMA_CLASS = "https://schemas.opengis.net/ogcapi/edr/1.1/openapi" + @staticmethod + def raise_edr_exception(func: Callable[..., Tuple[dict, int, str]]) -> Callable[..., Tuple[dict, int, str]]: + """Decorator that raises an EDRException for non-success HTTP status codes. + + The decorated function must return a tuple containing (headers, status_code, contents). + An internal error is raised if the return does not match the expected tuple. + This decorator is useful to ensure output sanitization occurs before returning responses to client. + + Parameters + ---------- + func : Callable[..., Tuple[dict, int, str]] + Function returning a tuple containing (headers, status_code, contents). + + Returns + ------- + Callable[..., Tuple[dict, int, str]] + A wrapped function that validates the response status code. + + Raises + ------ + EDRException + If the returned response has a non-success status code. + EDRException + If the function did not return the expected tuple. + """ + + @wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Any: + output = func(*args, **kwargs) + + # Handle (headers, status_code, contents) + if isinstance(output, tuple) and len(output) == 3: + _, status_code, content = output + + if not (200 <= status_code < 300): + logger.warning(f"Unsuccessful Response ({status_code}): {content}") + status = HTTPStatus(status_code) + raise EDRException( + status_code=status.value, exception_code=status.phrase, exception_text=status.description + ) + else: + logger.warning("Unhandled return type.") + raise EDRException() + + return output + + return wrapper + @jsonldify + @raise_edr_exception @staticmethod def landing_page(api: API, request: APIRequest) -> Tuple[dict, int, str]: """Provide the API landing page. @@ -48,6 +102,7 @@ def landing_page(api: API, request: APIRequest) -> Tuple[dict, int, str]: """ return pygeoapi.api.landing_page(api, request) + @raise_edr_exception @staticmethod def openapi_(api: API, request: APIRequest) -> Tuple[dict, int, str]: """Provide the OpenAPI documentation. @@ -64,6 +119,9 @@ def openapi_(api: API, request: APIRequest) -> Tuple[dict, int, str]: Tuple[dict, int, str] Headers, HTTP Status, and Content returned as a tuple. """ + if request._args.get("ui") is not None and request._args.get("ui") != "redoc": + raise EDRException(status_code=400, exception_code="InvalidQuery", exception_text="") + html_path = "openapi/redoc.html" if request._args.get("ui") == "redoc" else "openapi/swagger.html" headers = request.get_response_headers(**api.api_headers) @@ -83,6 +141,7 @@ def openapi_(api: API, request: APIRequest) -> Tuple[dict, int, str]: else: return headers, HTTPStatus.OK, api.openapi + @raise_edr_exception @staticmethod def conformance(api: API, request: APIRequest) -> Tuple[dict, int, str]: """Provide the conformance definition. @@ -113,6 +172,7 @@ def conformance(api: API, request: APIRequest) -> Tuple[dict, int, str]: return headers, HTTPStatus.OK, to_json(conformance, api.pretty_print) @jsonldify + @raise_edr_exception @staticmethod def describe_collections(api: API, request: APIRequest, dataset: str | None = None) -> Tuple[dict, int, str]: """Provide the collection/collections metadata. @@ -174,6 +234,7 @@ def describe_collections(api: API, request: APIRequest, dataset: str | None = No return headers, status, to_json(collection_description, api.pretty_print) + @raise_edr_exception @staticmethod def get_collection_edr_instances( api: API, request: APIRequest, dataset: str, instance_id: str | None = None @@ -232,6 +293,7 @@ def get_collection_edr_instances( return headers, status, to_json(instance_description, api.pretty_print) + @raise_edr_exception @staticmethod def get_collection_edr_query( api: API, @@ -494,7 +556,7 @@ def _instance_parameters( "label": {"en": value["title"]}, "symbol": { "value": value["x-ogc-unit"], - "type": "http://www.opengis.net/def/uom/UCUM/", + "type": "http://www.opengis.net/def/uom/UCUM/", # NOSONAR(S5332) - This is part of the EDR specification }, }, } diff --git a/ogc/edr/edr_routes.py b/ogc/edr/edr_routes.py index b03435b..69853b6 100644 --- a/ogc/edr/edr_routes.py +++ b/ogc/edr/edr_routes.py @@ -7,6 +7,7 @@ import pygeoapi.plugin import pygeoapi.api from typing import Tuple, Any, Dict, Generator +from werkzeug.security import safe_join from http import HTTPStatus from copy import deepcopy from pygeoapi.openapi import get_oas @@ -107,11 +108,13 @@ def static_files(self, request: pygeoapi.api.APIRequest, file_path: str) -> Tupl static_path = os.path.join(os.path.dirname(pygeoapi.__file__), "static") if "templates" in self.api.config["server"]: static_path = self.api.config["server"]["templates"].get("static", static_path) - file_path = os.path.join(static_path, file_path) - if os.path.isfile(file_path): - mime_type, _ = mimetypes.guess_type(file_path) + + # Use a safe join to ensure the untrusted path is a subpath of the trusted static directory + safe_path = safe_join(static_path, file_path) + if safe_path is not None and os.path.isfile(safe_path) and not os.path.islink(safe_path): + mime_type, _ = mimetypes.guess_type(safe_path) mime_type = mime_type or "application/octet-stream" - with open(file_path, "rb") as f: + with open(safe_path, "rb") as f: content = f.read() return {"Content-Type": mime_type}, HTTPStatus.OK, content else: diff --git a/ogc/edr/test/test_edr_routes.py b/ogc/edr/test/test_edr_routes.py index 4931f99..9b245a1 100644 --- a/ogc/edr/test/test_edr_routes.py +++ b/ogc/edr/test/test_edr_routes.py @@ -1,7 +1,9 @@ import os import json +import pytest import numpy as np import tempfile +from unittest.mock import patch from pygeoapi.api import APIRequest from http import HTTPStatus from typing import Dict, List, Any @@ -9,6 +11,7 @@ from werkzeug.wrappers import Request from werkzeug.datastructures import ImmutableMultiDict from ogc import podpac as pogc +from ogc.ogc_common import EDRException from ogc.edr.edr_routes import EdrRoutes @@ -27,7 +30,7 @@ def mock_request(request_args: Dict[str, Any] | None = None) -> APIRequest: Mock API request for route testing. """ request_args = request_args if request_args is not None else {} - environ = create_environ(base_url="http://127.0.0.1:5000/ogc/edr") + environ = create_environ(base_url="https://127.0.0.1:5000/ogc/edr") request = Request(environ) request.args = ImmutableMultiDict(request_args.items()) return APIRequest(request, ["en"]) @@ -44,6 +47,29 @@ def test_edr_routes_static_files_valid_path(): assert headers["Content-Type"] == "image/png" +def test_edr_routes_static_files_prevents_path_traversal(): + """Test the EDR static routes prevents path traversal.""" + request = mock_request() + edr_routes = EdrRoutes(layers=[]) + + static_path = os.path.join(os.path.dirname(__file__), "..", "static") + file_path = os.path.join(os.path.dirname(__file__), "..") + with tempfile.NamedTemporaryFile(dir=file_path) as temp_file: + relative_path = os.path.relpath(temp_file.name, static_path) + _, status, _ = edr_routes.static_files(request, relative_path) + assert os.path.exists(temp_file.name) + assert status == HTTPStatus.NOT_FOUND + + +def test_edr_routes_static_files_prevents_following_symlinks(): + """Test the EDR static routes prevents following symlinks.""" + request = mock_request() + edr_routes = EdrRoutes(layers=[]) + with patch("os.path.islink", returns=True): + _, status, _ = edr_routes.static_files(request, "img/logo.png") + assert status == HTTPStatus.NOT_FOUND + + def test_edr_routes_static_files_invalid_path(): """Test the EDR static routes with an invalid static file path.""" request = mock_request() @@ -321,14 +347,15 @@ def test_edr_routes_collection_query_invalid_type(layers: List[pogc.Layer], sing request = mock_request(single_layer_cube_args) edr_routes = EdrRoutes(layers=layers) - _, status, _ = edr_routes.collection_query( - request, - collection_id=collection_id, - instance_id=instance_id, - query_type="corridor", - ) + with pytest.raises(EDRException) as exception_info: + edr_routes.collection_query( + request, + collection_id=collection_id, + instance_id=instance_id, + query_type="corridor", + ) - assert status == HTTPStatus.BAD_REQUEST + assert exception_info.value.status_code == 400 def test_edr_routes_collection_query_invalid_bbox(layers: List[pogc.Layer], single_layer_cube_args: Dict[str, Any]): @@ -346,14 +373,15 @@ def test_edr_routes_collection_query_invalid_bbox(layers: List[pogc.Layer], sing request = mock_request(single_layer_cube_args) edr_routes = EdrRoutes(layers=layers) - _, status, _ = edr_routes.collection_query( - request, - collection_id=layers[0].group, - instance_id=next(iter(layers[0].time_instances())), - query_type="cube", - ) + with pytest.raises(EDRException) as exception_info: + edr_routes.collection_query( + request, + collection_id=layers[0].group, + instance_id=next(iter(layers[0].time_instances())), + query_type="cube", + ) - assert status == HTTPStatus.BAD_REQUEST + assert exception_info.value.status_code == 400 def test_edr_routes_collection_query_missing_parameter( @@ -387,8 +415,8 @@ def test_edr_routes_collection_query_missing_parameter( def test_edr_routes_request_url_updates_configuration_url(): """Test the EDR routes request base URL updates the configuration URL.""" - request_url = "http://test:5000/ogc/edr/static/img/logo.png" - expected_config_url = "http://test:5000/ogc/edr" + request_url = "https://test:5000/ogc/edr/static/img/logo.png" + expected_config_url = "https://test:5000/ogc/edr" request = mock_request({"base_url": request_url}) edr_routes = EdrRoutes(layers=[]) diff --git a/ogc/ogc_common.py b/ogc/ogc_common.py index db091ad..8e17dbb 100755 --- a/ogc/ogc_common.py +++ b/ogc/ogc_common.py @@ -1,6 +1,6 @@ import logging import string - +import json import lxml import lxml.etree import numpy as np @@ -16,6 +16,8 @@ "urn:ogc:def:crs:EPSG::26910", # used in an example, "urn:ogc:def:crs:EPSG::4326", # used in an example, ) +INTERNAL_APPLICATION_ERROR = "Internal application error." +NO_APPLICABLE_CODE = "NoApplicableCode" class EscapeFormatter(string.Formatter): @@ -69,12 +71,16 @@ class OutputFormat(XMLNode): value = tl.Unicode(default_value=None, allow_none=True) + # Allowed values of None mean that all values are allowed, an empty list means no values allowed + allowed_values = tl.List(tl.Unicode(), default_value=None, allow_none=True) + def validate(self): assert bool(self.value) is True, "error validating output format" - # Can check here for specific allowed formats if desired, prob. not necessary. - def to_xml(self): - return "%s" % self.value + if self.allowed_values is not None: + assert self.value is not None and self.value.lower() in [ + allowed_value.lower() for allowed_value in self.allowed_values + ], "error validating output format, value not in allowed values" class BoundingBox(XMLNode): @@ -121,8 +127,8 @@ def to_xml(self): class WCSException(Exception): def __init__( self, - exception_text="Internal application error.", - exception_code="NoApplicableCode", + exception_text=INTERNAL_APPLICATION_ERROR, + exception_code=NO_APPLICABLE_CODE, locator="", ): """ @@ -161,8 +167,8 @@ class WMTSException(WCSException): def __init__( self, - exception_text="Internal application error.", - exception_code="NoApplicableCode", + exception_text=INTERNAL_APPLICATION_ERROR, + exception_code=NO_APPLICABLE_CODE, locator="", ): """ @@ -171,3 +177,36 @@ def __init__( 'OperationNotSupported', 'TileOutOfRange' """ super().__init__(exception_text, exception_code, locator) + + +class EDRException(Exception): + def __init__( + self, + status_code=500, + exception_code=NO_APPLICABLE_CODE, + exception_text=INTERNAL_APPLICATION_ERROR, + ): + """ + exception_code: 'NoApplicableCode', 'NotFound', 'InvalidParameterValue', 'InvalidQuery' + """ + super().__init__(status_code, exception_text, exception_code) + + self.status_code = status_code + self.exception_code = exception_code + self.exception_text = exception_text + + def to_json(self) -> str: + """Return JSON string for the exception. + + Returns + ------- + str + The exception in JSON string format. + """ + return json.dumps( + { + "code": self.status_code, + "type": self.exception_code, + "description": self.exception_text, + } + ) diff --git a/ogc/servers.py b/ogc/servers.py index 51b2a1b..71a75d8 100755 --- a/ogc/servers.py +++ b/ogc/servers.py @@ -14,22 +14,50 @@ from typing import Callable from werkzeug.datastructures import ImmutableMultiDict -from ogc.ogc_common import WCSException +from ogc.ogc_common import WCSException, EDRException from pygeoapi.api import APIRequest from pygeoapi.util import get_api_rules from . import settings logger = logging.getLogger(__name__) +INVALID_ARGUMENTS = "Invalid arguments" 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.""" + """Checks the query string for malicious and invalid content. + + Parameters + ---------- + raw_qs : bytes + The raw query string bytes. + + Raises + ------ + ValueError + Raised if the query string exceeds a maximum allowed size. + ValueError + Raised if the query string contains null bytes. + ValueError + Raised if the query string contains invalid UTF-8 character. + """ if len(raw_qs) > settings.MAX_QUERY_STRING_BYTES: - raise WCSException("Request query string exceeds maximum allowed length.") + raise ValueError("Request query string exceeds maximum allowed length.") + if b"%00" in raw_qs: + raise ValueError("Request contains null bytes.") try: raw_qs.decode("utf-8") except UnicodeDecodeError: - raise WCSException("Request contains invalid UTF-8 encoding.") + raise ValueError("Request contains invalid UTF-8 encoding.") + + +def _disable_caching(response: Response) -> Response: + """Set headers so browsers/proxies never cache a response, and a later user of the same + machine or a shared cache can never be served a stale copy of it. Applied to every + response uniformly (not by URL/path pattern) to close off Web Cache Deception vectors.""" + response.headers["Cache-Control"] = "no-cache, no-store, max-age=0, must-revalidate" + response.headers["Pragma"] = "no-cache" + response.headers["Expires"] = "-1" + return response def respond_xml(doc, status=200): @@ -138,6 +166,8 @@ def __init__(self, *args, ogcs=None, home_func=None): """ super().__init__(*args) + self.after_request(_disable_caching) + self.home_func = home_func if self.home_func is None: self.home_func = home @@ -259,8 +289,9 @@ def ogc_render(self, ogc_idx): try: _check_query_string(request.query_string) - except WCSException as e: - return respond_xml(e.to_xml(), status=400) + except ValueError as e: + ee = WCSException(exception_code="InvalidParameterValue", exception_text=str(e)) + return respond_xml(ee.to_xml(), status=400) if not request.args: return self.home_func(ogc.endpoint) @@ -346,8 +377,9 @@ def wrapper(*args, **kwargs) -> Response: try: _check_query_string(request.query_string) - except WCSException as e: - return respond_xml(e.to_xml(), status=400) + except ValueError as e: + ee = EDRException(status_code=400, exception_code="InvalidQuery", exception_text=str(e)) + return Response(ee.to_json(), status=ee.status_code) try: # We'll filter out any characters from URl parameter values that @@ -369,11 +401,21 @@ def wrapper(*args, **kwargs) -> Response: # Replace format with its lowercase version to match pygeoapi expectations query_type = kwargs.get("query_type") default_format = settings.JSON + query_formats = [settings.HTML, settings.JSON] + if query_type is not None: default_format = settings.EDR_QUERY_DEFAULTS.get(query_type, default_format) - format_argument = filtered_args.get("f", default_format) - if format_argument is not None: - filtered_args["f"] = format_argument.lower() + query_formats = settings.EDR_QUERY_FORMATS.get(query_type, []) + + format_argument = filtered_args.get("f", default_format).lower() + filtered_args["f"] = format_argument + + if format_argument not in [item.lower() for item in query_formats]: + raise EDRException( + status_code=400, + exception_code="InvalidQuery", + exception_text=INVALID_ARGUMENTS, + ) filtered_args["base_url"] = ( xml.sax.saxutils.escape(request.base_url, {'"': """}) if request.base_url else None @@ -388,13 +430,13 @@ def wrapper(*args, **kwargs) -> Response: if headers: response.headers = headers return response - except WCSException as e: - logger.exception("OGC: server.edr_render WCSException: %s", str(e)) - return respond_xml(e.to_xml(), status=400) + except EDRException as e: + logger.exception("OGC: server.edr_render EDRException: %s", str(e)) + return Response(e.to_json(), status=e.status_code) except Exception as e: # noqa: B902 logger.exception("OGC: server.edr_render Exception: %s", str(e)) - ee = WCSException() - return respond_xml(ee.to_xml(), status=500) + ee = EDRException() + return Response(ee.to_json(), status=ee.status_code) return wrapper diff --git a/ogc/settings.py b/ogc/settings.py index f2f324c..f585f7f 100755 --- a/ogc/settings.py +++ b/ogc/settings.py @@ -11,9 +11,15 @@ # Settings applied around the OGC server package. crs_84 = "crs:84" epsg_4326 = "epsg:4326" -crs_84_uri_format = "http://www.opengis.net/def/crs/OGC/1.3/CRS84" -crs_84h_uri_format = "http://www.opengis.net/def/crs/OGC/0/CRS84h" -epsg_4326_uri_format = "http://www.opengis.net/def/crs/EPSG/0/4326" +crs_84_uri_format = ( + "http://www.opengis.net/def/crs/OGC/1.3/CRS84" # NOSONAR(S5332) - This is part of the EDR specification +) +crs_84h_uri_format = ( + "http://www.opengis.net/def/crs/OGC/0/CRS84h" # NOSONAR(S5332) - This is part of the EDR specification +) +epsg_4326_uri_format = ( + "http://www.opengis.net/def/crs/EPSG/0/4326" # NOSONAR(S5332) - This is part of the EDR specification +) # Default/Supported WMS CRS/SRS WMS_CRS = { diff --git a/ogc/test/conftest.py b/ogc/test/conftest.py new file mode 100644 index 0000000..770f8f6 --- /dev/null +++ b/ogc/test/conftest.py @@ -0,0 +1,15 @@ +import pytest +import importlib +from unittest.mock import patch +from ogc import settings + + +@pytest.fixture(scope="module", autouse=True) +def set_env_vars(): + """Setup the environmental variables for the module to support WMS and WCS.""" + with patch.dict("os.environ", {"OGC_SUPPORTED_FORMATS": "wms,wcs"}): + importlib.reload(settings) + yield + + # Fix imports after patching for test + importlib.reload(settings) diff --git a/ogc/test/test_input_security.py b/ogc/test/test_input_security.py index 3093681..e8f231b 100644 --- a/ogc/test/test_input_security.py +++ b/ogc/test/test_input_security.py @@ -8,7 +8,6 @@ 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 # --------------------------------------------------------------------------- @@ -42,9 +41,9 @@ def client(): def test_check_query_string_overflow(): - """Raises WCSException when byte length exceeds MAX_QUERY_STRING_BYTES.""" + """Raises ValueError 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"): + with pytest.raises(ValueError, match="maximum allowed length"): _check_query_string(oversized) @@ -60,11 +59,17 @@ def test_check_query_string_exactly_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"): + """Raises ValueError when the query string contains raw non-UTF-8 bytes.""" + with pytest.raises(ValueError, match="invalid UTF-8 encoding"): _check_query_string(b"SERVICE=WCS&COVERAGE=\xff\xfe") +def test_check_query_string_null_byte_injection(): + """Raises ValueError when the query string contains null bytes.""" + with pytest.raises(ValueError, match="null bytes"): + _check_query_string(b"SERVICE=WCS&COVERAGE=%00") + + 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") diff --git a/ogc/test/test_input_validation.py b/ogc/test/test_input_validation.py new file mode 100644 index 0000000..0874d3e --- /dev/null +++ b/ogc/test/test_input_validation.py @@ -0,0 +1,614 @@ +import pytest +import importlib +import podpac +import datetime +import numpy as np +from flask.testing import FlaskClient +from itertools import chain +from collections.abc import Iterator +from unittest.mock import patch +from ogc import core +from ogc import servers +from ogc import settings +from ogc import podpac as pogc +from ogc.settings import EDR_TIME_INSTANCE_DIMENSION + +lat = np.linspace(90, -90, 11) +lon = np.linspace(-180, 180, 21) +time = np.array(["2025-10-24T12:00:00"], dtype="datetime64") +instance = np.array(["2025-10-24T00:00:00"], dtype="datetime64") +data_static = np.random.default_rng(1).random((11, 21)) +coords_static = podpac.Coordinates([lat, lon], dims=["lat", "lon"]) +data_with_time = np.random.default_rng(1).random((11, 21, 1)) +coords_with_time = podpac.Coordinates([lat, lon, time], dims=["lat", "lon", "time"]) +data_with_instance = np.random.default_rng(1).random((11, 21, 1, 1)) +coords_with_instance = podpac.Coordinates( + [lat, lon, time, instance], dims=["lat", "lon", "time", EDR_TIME_INSTANCE_DIMENSION] +) + +# Define a layer which does not include temporal coordinates +node_static = podpac.data.Array(source=data_static, coordinates=coords_static) +layer_static = pogc.Layer( + node=node_static, + identifier="layerStatic", + title="Layer Static", + abstract="Layer Static", + group="Layers", +) + +# Define a layer which includes time coordinates +node_time = podpac.data.Array(source=data_with_time, coordinates=coords_with_time) +layer_time = pogc.Layer( + node=node_time, + identifier="layerTime", + title="Layer Time", + abstract="Layer Time", + group="Layers", + valid_times=[dt.astype(datetime.datetime) for dt in time], +) + +# Define a layer which includes both time coordinates and instance coordinates +node_instance = podpac.data.Array(source=data_with_instance, coordinates=coords_with_instance) +layer_instance = pogc.Layer( + node=node_instance, + identifier="layerInstance", + title="Layer Instance", + abstract="Layer Instance", + group="Layers", + valid_times=[dt.astype(datetime.datetime) for dt in time], +) + + +@pytest.fixture +def enable_all_formats_in_env(): + """Test client for FlaskServer with all formats enabled.""" + with patch.dict("os.environ", {"OGC_SUPPORTED_FORMATS": "wms,wcs,wmts,edr"}): + importlib.reload(settings) + yield + importlib.reload(settings) + + +@pytest.fixture +def client(): + """ + Create a test client for the Flask server. + + Yields + ------ + client : FlaskClient + A test client for the Flask server. + """ + # Create an OGC instance with the test layers + ogc = core.OGC(layers=[layer_static, layer_time, layer_instance]) + + # Create a FlaskServer instance + app = servers.FlaskServer(__name__, ogcs=[ogc]) + app.config.update({"TESTING": True}) + yield app.test_client() + + +def make_valid_ogc_wms_get_capabilities_args() -> dict: + """Valid argument dictionary for WMS get capabilities. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "SERVICE": "WMS", + "REQUEST": "GetCapabilities", + } + + +def make_valid_ogc_wms_get_feature_info_args() -> dict: + """Valid argument dictionary for WMS get feature info. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "SERVICE": "WMS", + "REQUEST": "GetFeatureInfo", + "VERSION": "1.3.0", + } + + +def make_valid_ogc_wms_get_legend_graphic_args(layer: str) -> dict: + """Valid argument dictionary for WMS get legend graphic. + + Parameters + ---------- + layer : str + Identifier for the layer. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "SERVICE": "WMS", + "REQUEST": "GetLegendGraphic", + "VERSION": "1.3.0", + "LAYER": layer, + } + + +def make_valid_ogc_wms_get_map_args(layer: str, time: str) -> dict: + """Valid argument dictionary for WMS get map. + + Parameters + ---------- + layer : str + Identifier for the layer. + time : str + Available time for the layer. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "SERVICE": "WMS", + "REQUEST": "GetMap", + "VERSION": "1.3.0", + "LAYERS": [layer], + "CRS": "EPSG:4326", + "BBOX": "-180,-90,180,90", + "FORMAT": "image/png", + "TIME": time, + "HEIGHT": 512, + "WIDTH": 512, + } + + +def make_valid_ogc_wcs_describe_coverage_args(layer: str) -> dict: + """Valid argument dictionary for WCS describe coverage. + + Parameters + ---------- + layer : str + Identifier for the layer. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "SERVICE": "WCS", + "REQUEST": "DescribeCoverage", + "VERSION": "1.0.0", + "COVERAGE": layer, + } + + +def make_valid_ogc_wcs_get_capabilities_args() -> dict: + """Valid argument dictionary for WCS get capabilities. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "SERVICE": "WCS", + "REQUEST": "GetCapabilities", + } + + +def make_valid_ogc_wcs_get_coverage_args(layer: str, time: str) -> dict: + """Valid argument dictionary for WCS get coverage. + + Parameters + ---------- + layer : str + Identifier for the layer. + time : str + Available time for the layer. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "SERVICE": "WCS", + "REQUEST": "GetCoverage", + "VERSION": "1.0.0", + "COVERAGE": layer, + "REQUEST_CRS": "EPSG:4326", + "CRS": "EPSG:4326", + "BBOX": "-180,-90,180,90", + "FORMAT": "geotiff", + "TIME": time, + "HEIGHT": 512, + "WIDTH": 512, + } + + +def make_valid_ogc_wmts_get_capabilities_args() -> dict: + """Valid argument dictionary for WMTS get capabilities. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "SERVICE": "WMTS", + "REQUEST": "GetCapabilities", + } + + +def make_valid_ogc_wmts_get_feature_info_args() -> dict: + """Valid argument dictionary for WMTS get feature info. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "SERVICE": "WMTS", + "REQUEST": "GetFeatureInfo", + "VERSION": "1.0.0", + } + + +def make_valid_ogc_wmts_get_tile_args(layer: str, time: str) -> dict: + """Valid argument dictionary for WMTS get tile. + + Parameters + ---------- + layer : str + Identifier for the layer. + time : str + Available time for the layer. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "SERVICE": "WMTS", + "REQUEST": "GetTile", + "VERSION": "1.0.0", + "LAYER": layer, + "TILEROW": "0", + "TILECOL": "0", + "TILEMATRIX": "0", + "TILEMATRIXSET": "WebMercatorQuad", + "FORMAT": "image/png", + "TIME": time, + } + + +def make_valid_ogc_edr_format_args() -> dict: + """Valid argument dictionary for EDR requests using only format. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "f": "json", + } + + +def make_valid_ogc_edr_api_args() -> dict: + """Valid argument dictionary for EDR api requests. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "f": "json", + "ui": "redoc", + } + + +def make_valid_ogc_edr_static_cube_args(layer: str) -> dict: + """Valid argument dictionary for EDR cube query without time or instances. + + Parameters + ---------- + layer : str + Identifier for the layer. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "f": "CoverageJSON", + "bbox": "-180,-90,180,90", + "crs": "http://www.opengis.net/def/crs/OGC/1.3/CRS84", + "parameter-name": layer, + "resolution-x": 512, + "resolution-y": 512, + } + + +def make_valid_ogc_edr_static_area_args(layer: str) -> dict: + """Valid argument dictionary for EDR area query without time or instances. + + Parameters + ---------- + layer : str + Identifier for the layer. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "f": "CoverageJSON", + "coords": "POLYGON((-180 90, -180 -90, 180 -90, 180 90, -180 90))", + "crs": "http://www.opengis.net/def/crs/OGC/1.3/CRS84", + "parameter-name": layer, + "resolution-x": 512, + "resolution-y": 512, + } + + +def make_valid_ogc_edr_static_position_args(layer: str) -> dict: + """Valid argument dictionary for EDR position query without time or instances. + + Parameters + ---------- + layer : str + Identifier for the layer. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "f": "CoverageJSON", + "coords": "POINT(40 50)", + "crs": "http://www.opengis.net/def/crs/OGC/1.3/CRS84", + "parameter-name": layer, + } + + +def make_valid_ogc_edr_instance_cube_args(layer: str, time: str) -> dict: + """Valid argument dictionary for EDR cube query with time and instances. + + Parameters + ---------- + layer : str + Identifier for the layer. + time : str + Available time for the layer. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "f": "CoverageJSON", + "bbox": "-180,-90,180,90", + "crs": "http://www.opengis.net/def/crs/OGC/1.3/CRS84", + "datetime": time, + "parameter-name": layer, + "resolution-x": 512, + "resolution-y": 512, + } + + +def make_valid_ogc_edr_instance_area_args(layer: str, time: str) -> dict: + """Valid argument dictionary for EDR area query with time and instances. + + Parameters + ---------- + layer : str + Identifier for the layer. + time : str + Available time for the layer. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "f": "CoverageJSON", + "coords": "POLYGON((-180 90, -180 -90, 180 -90, 180 90, -180 90))", + "crs": "http://www.opengis.net/def/crs/OGC/1.3/CRS84", + "datetime": time, + "parameter-name": layer, + "resolution-x": 512, + "resolution-y": 512, + } + + +def make_valid_ogc_edr_instance_position_args(layer: str, time: str) -> dict: + """Valid argument dictionary for EDR position query with time and instances. + + Parameters + ---------- + layer : str + Identifier for the layer. + time : str + Available time for the layer. + + Returns + ------- + dict + The argument dictionary. + """ + return { + "f": "CoverageJSON", + "coords": "POINT(40 50)", + "crs": "http://www.opengis.net/def/crs/OGC/1.3/CRS84", + "datetime": time, + "parameter-name": layer, + } + + +class BaseValidation: + @staticmethod + def generate_cases(url: str, params: dict) -> Iterator[tuple[str, dict, bool]]: + """Generate cases for each argument group. + The first case uses all valid arguments and returns true. + The remaining cases change a single argument to "invalid" and return false. + + Parameters + ---------- + url: str + The URL to request from. + params : dict + The valid argument group. + + Yields + ------ + Iterator[tuple[dict, bool]] + An iterator containing the URL, updated arguments, and a boolean whether it is valid or not. + """ + yield url, params, True + + for key in params: + invalid = params.copy() + invalid[key] = "invalid" + yield url, invalid, False + + @staticmethod + def input_validation( + client: FlaskClient, + url: str, + params: dict, + should_pass: bool, + ): + """Check that the application validates input properly for the provided query arguments. + + Parameters + ---------- + client: FlaskClient + The client used to make requests. + url : str + The URL to request from. + params: dict + The arguments for the request. + should_pass: bool + Whether the test should pass or fail. + """ + response = client.get(url, query_string=params) + + if should_pass: + assert response.status_code == 200 + else: + assert response.status_code == 400 + + +class TestWcsValidation: + @pytest.mark.parametrize( + "url, params, should_pass", + chain( + BaseValidation.generate_cases("/ogc?", make_valid_ogc_wcs_describe_coverage_args(layer_static.identifier)), + BaseValidation.generate_cases("/ogc?", make_valid_ogc_wcs_get_capabilities_args()), + BaseValidation.generate_cases( + "/ogc?", make_valid_ogc_wcs_get_coverage_args(layer_time.identifier, str(time[0])) + ), + ), + ) + def test_input_validation( + self, enable_all_formats_in_env, client: FlaskClient, url: str, params: dict, should_pass: bool + ): + BaseValidation.input_validation(client, url, params, should_pass) + + +class TestWmsValidation: + # Ignore the following until implemented + # BaseValidation.generate_cases("/ogc?", make_valid_ogc_wms_get_feature_info_args()) + @pytest.mark.parametrize( + "url, params, should_pass", + chain( + BaseValidation.generate_cases("/ogc?", make_valid_ogc_wms_get_capabilities_args()), + BaseValidation.generate_cases("/ogc?", make_valid_ogc_wms_get_legend_graphic_args(layer_static.identifier)), + BaseValidation.generate_cases( + "/ogc?", make_valid_ogc_wms_get_map_args(layer_time.identifier, str(time[0])) + ), + ), + ) + def test_input_validation( + self, enable_all_formats_in_env, client: FlaskClient, url: str, params: dict, should_pass: bool + ): + BaseValidation.input_validation(client, url, params, should_pass) + + +class TestWmtsValidation: + # Ignore the following until implemented + # BaseValidation.generate_cases("/ogc?", make_valid_ogc_wmts_get_feature_info_args()) + @pytest.mark.parametrize( + "url, params, should_pass", + chain( + BaseValidation.generate_cases("/ogc?", make_valid_ogc_wmts_get_capabilities_args()), + BaseValidation.generate_cases( + "/ogc?", make_valid_ogc_wmts_get_tile_args(layer_time.identifier, str(time[0])) + ), + ), + ) + def test_input_validation( + self, enable_all_formats_in_env, client: FlaskClient, url: str, params: dict, should_pass: bool + ): + BaseValidation.input_validation(client, url, params, should_pass) + + +class TestEdrValidation: + @pytest.mark.parametrize( + "url, params, should_pass", + chain( + BaseValidation.generate_cases("/ogc/edr?", make_valid_ogc_edr_format_args()), + BaseValidation.generate_cases("/ogc/edr/api?", make_valid_ogc_edr_api_args()), + BaseValidation.generate_cases("/ogc/edr/openapi?", make_valid_ogc_edr_api_args()), + BaseValidation.generate_cases("/ogc/edr/conformance?", make_valid_ogc_edr_format_args()), + BaseValidation.generate_cases("/ogc/edr/collections?", make_valid_ogc_edr_format_args()), + BaseValidation.generate_cases( + f"/ogc/edr/collections/{layer_static.group}?", make_valid_ogc_edr_format_args() + ), + BaseValidation.generate_cases( + f"/ogc/edr/collections/{layer_static.group}/instances?", make_valid_ogc_edr_format_args() + ), + BaseValidation.generate_cases( + f"/ogc/edr/collections/{layer_static.group}/cube?", + make_valid_ogc_edr_static_cube_args(layer_static.identifier), + ), + BaseValidation.generate_cases( + f"/ogc/edr/collections/{layer_static.group}/area?", + make_valid_ogc_edr_static_area_args(layer_static.identifier), + ), + BaseValidation.generate_cases( + f"/ogc/edr/collections/{layer_static.group}/position?", + make_valid_ogc_edr_static_position_args(layer_static.identifier), + ), + BaseValidation.generate_cases( + f"/ogc/edr/collections/{layer_instance.group}/instances/{instance[0]}/cube?", + make_valid_ogc_edr_instance_cube_args(layer_instance.identifier, str(time[0])), + ), + BaseValidation.generate_cases( + f"/ogc/edr/collections/{layer_instance.group}/instances/{instance[0]}/area?", + make_valid_ogc_edr_instance_area_args(layer_instance.identifier, str(time[0])), + ), + BaseValidation.generate_cases( + f"/ogc/edr/collections/{layer_instance.group}/instances/{instance[0]}/position?", + make_valid_ogc_edr_instance_position_args(layer_instance.identifier, str(time[0])), + ), + ), + ) + def test_input_validation( + self, enable_all_formats_in_env, client: FlaskClient, url: str, params: dict, should_pass: bool + ): + BaseValidation.input_validation(client, url, params, should_pass) diff --git a/ogc/test/test_servers.py b/ogc/test/test_servers.py index 10871cb..a1df634 100644 --- a/ogc/test/test_servers.py +++ b/ogc/test/test_servers.py @@ -1,11 +1,12 @@ +import os +import tempfile from ogc import servers from ogc import core from ogc import podpac as pogc from ogc import settings -from ogc.ogc_common import WCSException +from ogc.ogc_common import EDRException from pygeoapi.api import APIRequest from unittest.mock import patch -from typing import Callable, Generator import importlib import podpac @@ -13,33 +14,6 @@ import numpy as np -@pytest.fixture -def supported_formats() -> Generator[Callable[[str], None], None, None]: - """Fixture used to patch OGC supported formats. - - Returns - ------- - Generator[Callable[[str], None], None, None] - A generator which yields a function which patches the OGC supported formats based on input string. - """ - - def _supported_formats(formats: str): - """Patch the supported formats setting. - - Parameters - ---------- - formats : str - The formats which should be supported by the server as a string. - """ - with patch.dict("os.environ", {"OGC_SUPPORTED_FORMATS": formats}): - importlib.reload(settings) - - yield _supported_formats - - # Fix imports after patching for test - importlib.reload(settings) - - @pytest.fixture def client(): """ @@ -73,6 +47,15 @@ def client(): yield app.test_client() +@pytest.fixture +def disable_all_formats_in_env(): + """Setup the environmental variables for no supported formats.""" + with patch.dict("os.environ", {"OGC_SUPPORTED_FORMATS": ""}): + importlib.reload(settings) + yield + importlib.reload(settings) + + def test_server_construction(client): """ Test the construction of the server. @@ -148,43 +131,21 @@ def test_server_with_default_supported_services(client): assert response.status_code == 404 -def test_server_without_wcs_supported_service(supported_formats, client): +def test_server_without_wcs_supported_service(disable_all_formats_in_env, client): """ Test the WCS service is unavailable when WCS is not a supported format. """ - supported_formats("wms") - - response = client.get("/ogc?service=WMS&request=GetCapabilities") - assert response.status_code == 200 - response = client.get("/ogc?service=WCS&request=GetCapabilities") assert response.status_code == 400 - response = client.get("/ogc?service=WMTS&request=GetCapabilities") - assert response.status_code == 400 - - response = client.get("/ogc/edr") - assert response.status_code == 404 - -def test_server_without_wms_supported_service(supported_formats, client): +def test_server_without_wms_supported_service(disable_all_formats_in_env, client): """ Test the WMS service is unavailable when WMS is not a supported format. """ - supported_formats("wcs") - - response = client.get("/ogc?service=WCS&request=GetCapabilities") - assert response.status_code == 200 - response = client.get("/ogc?service=WMS&request=GetCapabilities") assert response.status_code == 400 - response = client.get("/ogc?service=WMTS&request=GetCapabilities") - assert response.status_code == 400 - - response = client.get("/ogc/edr") - assert response.status_code == 404 - # --------------------------------------------------------------------------- # edr_render tests @@ -220,6 +181,26 @@ def test_edr_render_post_returns_405(enable_edr_in_env, client): assert response.status_code == 405 +def test_edr_render_static_file_returns_200(enable_edr_in_env, client): + static_path = os.path.join(os.path.dirname(__file__), "..", "edr", "static") + file_path = static_path + with tempfile.NamedTemporaryFile(dir=file_path) as temp_file: + relative_path = os.path.relpath(temp_file.name, static_path) + response = client.get(f"/ogc/edr/static/{relative_path}") + assert os.path.exists(temp_file.name) + assert response.status_code == 200 + + +def test_edr_render_static_file_path_traversal_returns_404(enable_edr_in_env, client): + static_path = os.path.join(os.path.dirname(__file__), "..", "edr", "static") + file_path = os.path.join(os.path.dirname(__file__), "..", "edr") + with tempfile.NamedTemporaryFile(dir=file_path) as temp_file: + relative_path = os.path.relpath(temp_file.name, static_path) + response = client.get(f"/ogc/edr/static/{relative_path}") + assert os.path.exists(temp_file.name) + assert response.status_code == 404 + + def test_edr_render_query_string_too_long_returns_400(enable_edr_in_env, client): oversized = "f=json&" + "A=" + "B" * settings.MAX_QUERY_STRING_BYTES response = client.get("/ogc/edr", environ_overrides={"QUERY_STRING": oversized}) @@ -262,19 +243,19 @@ def capturing(request, locales): assert captured_args.get("f") == "json" -def test_edr_render_wcs_exception_returns_400(enable_edr_in_env, client): +def test_edr_render_edr_exception_returns_400(enable_edr_in_env, client): """WCSException raised by a handler is returned as a 400 XML response.""" app = client.application - def raises_wcs_exception(api_request, *args, **kwargs): - raise WCSException("test error") + def raises_edr_exception(api_request, *args, **kwargs): + raise EDRException(status_code=400, exception_code="InvalidQuery", exception_text="") - wrapper = app.edr_render(raises_wcs_exception) + wrapper = app.edr_render(raises_edr_exception) app.add_url_rule("/test_edr_wcs", endpoint="test_edr_wcs", view_func=wrapper, methods=["GET"]) response = client.get("/test_edr_wcs") assert response.status_code == 400 - assert "ExceptionReport" in response.get_data(as_text=True) + assert "InvalidQuery" in response.get_data(as_text=True) def test_edr_render_exception_returns_500(enable_edr_in_env, client): @@ -289,4 +270,4 @@ def raises_runtime_error(api_request, *args, **kwargs): response = client.get("/test_edr_exc") assert response.status_code == 500 - assert "ExceptionReport" in response.get_data(as_text=True) + assert "NoApplicableCode" in response.get_data(as_text=True) diff --git a/ogc/version.py b/ogc/version.py index c104e60..4bf07b9 100755 --- a/ogc/version.py +++ b/ogc/version.py @@ -17,7 +17,7 @@ ####################### MAJOR = 0 MINOR = 6 -HOTFIX = 0 +HOTFIX = 1 ####################### diff --git a/ogc/wcs_request_1_0_0.py b/ogc/wcs_request_1_0_0.py index 0f7fcc9..49b4a23 100755 --- a/ogc/wcs_request_1_0_0.py +++ b/ogc/wcs_request_1_0_0.py @@ -206,7 +206,7 @@ def _load_from_kv(self, args): upper_corner=(float(bbox[3]), float(bbox[2])), ) - self.output_format = ogc_common.OutputFormat(value=args["format"]) + self.output_format = ogc_common.OutputFormat(value=args["format"], allowed_values=["geotiff"]) if "time" in args: # TIME : time1, time2,... # or diff --git a/ogc/wmts/wmts_routes.py b/ogc/wmts/wmts_routes.py index 94ea5cc..35efbb0 100644 --- a/ogc/wmts/wmts_routes.py +++ b/ogc/wmts/wmts_routes.py @@ -134,7 +134,7 @@ def get_capabilities(self, args: Dict[str, Any]) -> str: try: get_capabilities.load_from_kv(args) get_capabilities.validate() - except AssertionError: + except Exception: # noqa: B902 logger.exception(LOAD_FAILURE) raise WMTSException(exception_text=INVALID_ARGUMENTS) @@ -175,7 +175,7 @@ def get_tile(self, args: Dict[str, Any]) -> Dict[str, Any]: try: get_tile.load_from_kv(args) get_tile.validate() - except AssertionError: + except Exception: # noqa: B902 logger.exception(LOAD_FAILURE) raise WMTSException(exception_text=INVALID_ARGUMENTS)