Release/0.6.1

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

ogc/edr/edr_api.py

@@ -1,21 +1,26 @@
 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
 from pygeoapi.linked_data import jsonldify
 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
                         },
                     },
                 }

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:

ogc/edr/test/test_edr_routes.py

@@ -1,14 +1,17 @@
 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
 from werkzeug.test import create_environ
 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 @@
         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 @@
     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 @@
     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 @@
     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_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=[])
 

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 "<OutputFormat>%s</OutputFormat>" % 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,
+            }
+        )