Add validation of query parameter for server requests

CHANGES.md

@@ -5,7 +5,9 @@
   reducing visual offsets between rendered map tiles and dataset coordinates. (#1216)
 * Ensure compatibility with matplotlib 3.11.0 (#1219)
 * Fixed duplicated `tornado` log entries emitted by xcube Server. (#1224)
-
+* Added validation of allowed query parameters for static routes and 
+  configured allowed query parameters for `api/viewer/`. (#1225)
+
 ### Other changes
 * Pinned libjxl <=0.11.2 because libjxl >=0.12.0 causes CI failures due to 
   rasterio/GDAL binary incompatibilities.

test/webapi/viewer/test_routes.py

@@ -22,6 +22,20 @@ def test_viewer(self):
         response = self.fetch("/viewer/images/logo.png")
         self.assertResponseOK(response)
 
+    def test_viewer_known_query_params(self):
+        response = self.fetch(
+            "/viewer/?serverUrl=http://localhost:8080"
+            "&serverId=local&serverName=Local&compact=1"
+        )
+        self.assertResponseOK(response)
+
+    def test_viewer_unknown_query_param(self):
+        response = self.fetch("/viewer?rest_route=/wp/v2/users/")
+        self.assertEqual(404, response.status)
+
+        response = self.fetch("/viewer/?rest_route=/wp/v2/users/")
+        self.assertEqual(404, response.status)
+
 
 class ViewerConfigRoutesTest(RoutesTestCase):
     def test_viewer_config(self):

xcube/server/api.py

@@ -186,7 +186,11 @@ def optional_apis(self) -> tuple[str]:
         return self._optional_apis
 
     def static_route(
-        self, path: str, default_filename: Optional[str] = None, **openapi_metadata
+        self,
+        path: str,
+        default_filename: Optional[str] = None,
+        allowed_query_params: Optional[Sequence[str]] = None,
+        **openapi_metadata,
     ):
         """Decorator that adds static route to this API.
 
@@ -198,6 +202,9 @@ def static_route(
             path: The route path.
             default_filename: Optional default filename, e.g.,
                 "index.html".
+            allowed_query_params: Optional names of query parameters allowed
+                for this static route. If given, other query parameters should
+                be rejected by web framework implementations.
             **openapi_metadata: Optional OpenAPI GET operation metadata.
         """
 
@@ -210,6 +217,7 @@ def decorator_func(get_root_path: Callable[[], Optional[str]]):
                         str(root_path),
                         api_name=self.name,
                         default_filename=default_filename,
+                        allowed_query_params=allowed_query_params,
                         openapi_metadata=openapi_metadata,
                     )
                 )
@@ -825,6 +833,9 @@ class ApiStaticRoute:
         default_filename: Optional default filename, e.g., "index.html".
         api_name: Optional name of the API to which this route belongs
             to.
+        allowed_query_params: Optional names of query parameters allowed
+            for this static route. If given, other query parameters should
+            be rejected by web framework implementations.
         openapi_metadata: Optional OpenAPI operation metadata.
     """
 
@@ -834,6 +845,7 @@ def __init__(
         dir_path: str,
         default_filename: Optional[str] = None,
         api_name: Optional[str] = None,
+        allowed_query_params: Optional[Sequence[str]] = None,
         openapi_metadata: Optional[dict[str, Any]] = None,
     ):
         assert_instance(path, str, name="path")
@@ -843,11 +855,17 @@ def __init__(
         )
         assert_instance(default_filename, (type(None), str), name="default_filename")
         assert_instance(api_name, (type(None), str), name="api_name")
+        if allowed_query_params is not None:
+            assert_true(
+                all(isinstance(param, str) for param in allowed_query_params),
+                message="allowed_query_params must contain strings",
+            )
         assert_instance(openapi_metadata, (type(None), dict), name="openapi_metadata")
         self.path = path
         self.dir_path = dir_path
         self.default_filename = default_filename
         self.api_name = api_name
+        self.allowed_query_params = tuple(allowed_query_params or ())
         self.openapi_metadata = openapi_metadata
 
 

xcube/server/webservers/tornado.py

@@ -42,6 +42,45 @@
 
 SERVER_CTX_ATTR_NAME = "__xcube_server_ctx"
 
+def _assert_allowed_query_params(
+    request: tornado.httputil.HTTPServerRequest, allowed_query_params: Sequence[str]
+):
+    unknown_query_params = set(request.query_arguments) - set(allowed_query_params)
+    if unknown_query_params:
+        raise tornado.web.HTTPError(
+            404,
+            reason=(
+                f"Unknown query parameter(s): {', '.join(sorted(unknown_query_params))}"
+            ),
+        )
+
+class QueryValidatingRedirectHandler(tornado.web.RedirectHandler):
+    def initialize(
+        self,
+        url: str,
+        permanent: bool = True,
+        allowed_query_params: Sequence[str] = (),
+    ):
+        super().initialize(url, permanent=permanent)
+        self._allowed_query_params = allowed_query_params
+
+    def prepare(self):
+        _assert_allowed_query_params(self.request, self._allowed_query_params)
+
+
+class QueryValidatingStaticFileHandler(tornado.web.StaticFileHandler):
+    def initialize(
+        self,
+        path: str,
+        default_filename: Optional[str] = None,
+        allowed_query_params: Sequence[str] = (),
+    ):
+        super().initialize(path=path, default_filename=default_filename)
+        self._allowed_query_params = allowed_query_params
+
+    def prepare(self):
+        _assert_allowed_query_params(self.request, self._allowed_query_params)
+
 
 class TornadoFramework(Framework):
     """
@@ -103,15 +142,30 @@ def add_static_routes(self, api_routes: Sequence[ApiStaticRoute], url_prefix: st
         for api_route in api_routes:
             base_url = f"{url_prefix}{api_route.path}"
             default_filename = api_route.default_filename
+            allowed_query_params = api_route.allowed_query_params
+            if allowed_query_params:
+                redirect_handler = QueryValidatingRedirectHandler
+                static_file_handler = QueryValidatingStaticFileHandler
+                redirect_kwargs = {
+                    "url": f"{base_url}/",
+                    "allowed_query_params": allowed_query_params,
+                }
+                static_file_kwargs = {
+                    "path": api_route.dir_path,
+                    "default_filename": default_filename,
+                    "allowed_query_params": allowed_query_params,
+                }
+            else:
+                redirect_handler = tornado.web.RedirectHandler
+                static_file_handler = tornado.web.StaticFileHandler
+                redirect_kwargs = {"url": f"{base_url}/"}
+                static_file_kwargs = {
+                    "path": api_route.dir_path,
+                    "default_filename": default_filename,
+                }
+            handlers.append((f"{base_url}", redirect_handler, redirect_kwargs))
             handlers.append(
-                (f"{base_url}", tornado.web.RedirectHandler, {"url": f"{base_url}/"})
-            )
-            handlers.append(
-                (
-                    f"{base_url}/(.*)",
-                    tornado.web.StaticFileHandler,
-                    {"path": api_route.dir_path, "default_filename": default_filename},
-                )
+                (f"{base_url}/(.*)", static_file_handler, static_file_kwargs)
             )
             LOG.log(
                 LOG_LEVEL_DETAIL,

xcube/webapi/viewer/routes.py

@@ -24,6 +24,17 @@
 _viewer_module = "xcube.webapi.viewer"
 _data_dir = "dist"
 _default_filename = "index.html"
+_allowed_query_params = frozenset(
+    (
+        "compact",
+        "dataset",
+        "serverId",
+        "serverName",
+        "serverUrl",
+        "stateKey",
+        "variable",
+    )
+)
 
 _responses = {
     200: {
@@ -40,6 +51,7 @@
 @api.static_route(
     "/viewer",
     default_filename=_default_filename,
+    allowed_query_params=_allowed_query_params,
     summary="Brings up the xcube Viewer webpage",
     responses=_responses,
 )