From b4a2841cfc6d262fa4a22a23ea87d2bad6f3d44a Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 25 Feb 2026 21:43:18 -0700 Subject: [PATCH 1/2] feat: refactor location CTE for materialized views and enhance path validation --- ...a8b9c0_create_pygeoapi_supporting_views.py | 38 ++++++++----------- core/pygeoapi.py | 16 ++++++++ 2 files changed, 31 insertions(+), 23 deletions(-) diff --git a/alembic/versions/d5e6f7a8b9c0_create_pygeoapi_supporting_views.py b/alembic/versions/d5e6f7a8b9c0_create_pygeoapi_supporting_views.py index e76c6aa6..e571d8f4 100644 --- a/alembic/versions/d5e6f7a8b9c0_create_pygeoapi_supporting_views.py +++ b/alembic/versions/d5e6f7a8b9c0_create_pygeoapi_supporting_views.py @@ -45,6 +45,16 @@ ("test_wells", "test well"), ] +LATEST_LOCATION_CTE = """ + SELECT DISTINCT ON (lta.thing_id) + lta.thing_id, + lta.location_id, + lta.effective_start + FROM location_thing_association AS lta + WHERE lta.effective_end IS NULL + ORDER BY lta.thing_id, lta.effective_start DESC +""" + def _safe_view_id(view_id: str) -> str: if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", view_id): @@ -58,13 +68,7 @@ def _create_thing_view(view_id: str, thing_type: str) -> str: return f""" CREATE VIEW ogc_{safe_view_id} AS WITH latest_location AS ( - SELECT DISTINCT ON (lta.thing_id) - lta.thing_id, - lta.location_id, - lta.effective_start - FROM location_thing_association AS lta - WHERE lta.effective_end IS NULL - ORDER BY lta.thing_id, lta.effective_start DESC +{LATEST_LOCATION_CTE} ) SELECT t.id, @@ -94,16 +98,10 @@ def _create_thing_view(view_id: str, thing_type: str) -> str: def _create_latest_depth_view() -> str: - return """ + return f""" CREATE MATERIALIZED VIEW ogc_latest_depth_to_water_wells AS WITH latest_location AS ( - SELECT DISTINCT ON (lta.thing_id) - lta.thing_id, - lta.location_id, - lta.effective_start - FROM location_thing_association AS lta - WHERE lta.effective_end IS NULL - ORDER BY lta.thing_id, lta.effective_start DESC +{LATEST_LOCATION_CTE} ), ranked_obs AS ( SELECT @@ -147,16 +145,10 @@ def _create_latest_depth_view() -> str: def _create_avg_tds_view() -> str: - return """ + return f""" CREATE MATERIALIZED VIEW ogc_avg_tds_wells AS WITH latest_location AS ( - SELECT DISTINCT ON (lta.thing_id) - lta.thing_id, - lta.location_id, - lta.effective_start - FROM location_thing_association AS lta - WHERE lta.effective_end IS NULL - ORDER BY lta.thing_id, lta.effective_start DESC +{LATEST_LOCATION_CTE} ), tds_obs AS ( SELECT diff --git a/core/pygeoapi.py b/core/pygeoapi.py index 223d699e..33fa09d7 100644 --- a/core/pygeoapi.py +++ b/core/pygeoapi.py @@ -1,4 +1,5 @@ import os +import re import textwrap from importlib.util import find_spec from pathlib import Path @@ -182,6 +183,21 @@ def _mount_path() -> str: # Remove any trailing slashes so "/ogcapi/" and "ogcapi/" both become "/ogcapi". path = path.rstrip("/") + + # Disallow traversal/current-directory segments. + segments = [segment for segment in path.split("/") if segment] + if any(segment in {".", ".."} for segment in segments): + raise ValueError( + "Invalid PYGEOAPI_MOUNT_PATH: traversal segments are not allowed." + ) + + # Allow only slash-delimited segments of alphanumerics, underscore, or hyphen. + if not re.fullmatch(r"/[A-Za-z0-9_-]+(?:/[A-Za-z0-9_-]+)*", path): + raise ValueError( + "Invalid PYGEOAPI_MOUNT_PATH: only letters, numbers, underscores, " + "hyphens, and slashes are allowed." + ) + return path From 9fc0b16bbf94cf01283df40cea2382491eef2053 Mon Sep 17 00:00:00 2001 From: Jake Ross Date: Wed, 25 Feb 2026 21:50:00 -0700 Subject: [PATCH 2/2] Update alembic/versions/d5e6f7a8b9c0_create_pygeoapi_supporting_views.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- ...6f7a8b9c0_create_pygeoapi_supporting_views.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/alembic/versions/d5e6f7a8b9c0_create_pygeoapi_supporting_views.py b/alembic/versions/d5e6f7a8b9c0_create_pygeoapi_supporting_views.py index e571d8f4..afb70171 100644 --- a/alembic/versions/d5e6f7a8b9c0_create_pygeoapi_supporting_views.py +++ b/alembic/versions/d5e6f7a8b9c0_create_pygeoapi_supporting_views.py @@ -46,14 +46,14 @@ ] LATEST_LOCATION_CTE = """ - SELECT DISTINCT ON (lta.thing_id) - lta.thing_id, - lta.location_id, - lta.effective_start - FROM location_thing_association AS lta - WHERE lta.effective_end IS NULL - ORDER BY lta.thing_id, lta.effective_start DESC -""" +SELECT DISTINCT ON (lta.thing_id) + lta.thing_id, + lta.location_id, + lta.effective_start +FROM location_thing_association AS lta +WHERE lta.effective_end IS NULL +ORDER BY lta.thing_id, lta.effective_start DESC +""".strip() def _safe_view_id(view_id: str) -> str: