From aae88bc66cf5b4d146c030b79e90e43b61d5be73 Mon Sep 17 00:00:00 2001
From: Tom Kralidis <tomkralidis@gmail.com>
Date: Mon, 13 Jul 2026 22:42:42 -0400
Subject: [PATCH] OGR: sanitize attribute filter values

---
 pygeoapi/provider/ogr.py                 | 28 ++++++++++++++++++++++--
 tests/provider/test_ogr_gpkg_provider.py | 28 ++++++++++++++++++++++++
 2 files changed, 54 insertions(+), 2 deletions(-)

diff --git a/pygeoapi/provider/ogr.py b/pygeoapi/provider/ogr.py
index 7fee6bb24..5b4245bf0 100644
--- a/pygeoapi/provider/ogr.py
+++ b/pygeoapi/provider/ogr.py
@@ -322,7 +322,7 @@ def query(self, offset=0, limit=10, resulttype='results',
                 LOGGER.debug('processing properties')
 
                 attribute_filter = ' and '.join(
-                    map(lambda x: f'{x[0]} = \'{x[1]}\'', properties)
+                    map(lambda x: f'{x[0]} = {sanitize_attribute_value(x[1])}', properties)  # noqa
                 )
 
                 LOGGER.debug(attribute_filter)
@@ -410,7 +410,9 @@ def get(self, identifier, crs_transform_spec=None, **kwargs):
             LOGGER.debug(f'Fetching identifier {identifier}')
             layer = self._get_layer()
 
-            layer.SetAttributeFilter(f"{self.id_field} = '{identifier}'")
+            identifier2 = sanitize_attribute_value(identifier)
+
+            layer.SetAttributeFilter(f'{self.id_field} = {identifier2}')
 
             ogr_feature = self._get_next_feature(layer, identifier)
             result = self._ogr_feature_to_json(
@@ -902,3 +904,25 @@ def _ignore_gdal_error(inst, fn, *args, **kwargs) -> Any:
     """
     value = getattr(inst, fn)(*args, **kwargs)
     return value
+
+
+def sanitize_attribute_value(value) -> str:
+    """
+    Sanitize an attribute value used in an
+    OGR layer SetAttributeFilter function
+
+    :param value: `str` of attribute value
+
+    :returns: `str` of sanitized attribute value
+    """
+
+    if value is None:
+        return 'NULL'
+
+    if isinstance(value, bool):
+        return '1' if value else '0'
+
+    if isinstance(value, (int, float)):
+        return f"'{value}'"
+
+    return "'" + str(value).replace("'", "''") + "'"
diff --git a/tests/provider/test_ogr_gpkg_provider.py b/tests/provider/test_ogr_gpkg_provider.py
index 4217ba525..4f619ed44 100644
--- a/tests/provider/test_ogr_gpkg_provider.py
+++ b/tests/provider/test_ogr_gpkg_provider.py
@@ -80,6 +80,14 @@ def test_get(config_poi_portugal):
     assert result['id'] == 536678593
     assert 'cafe' in result['properties']['fclass']
 
+    with pytest.raises(ProviderItemNotFoundError):
+        item_id = 'foo%27%20OR%20%271%27%3D%271'
+        p.get(item_id)
+
+    with pytest.raises(ProviderItemNotFoundError):
+        item_id = "x' OR (SELECT substr(sql,1,1) FROM sqlite_master WHERE name='secret_table')='C"  # noqa
+        p.get(item_id)
+
 
 def test_get_not_existing_feature_raise_exception(
     config_poi_portugal
@@ -397,3 +405,23 @@ def test_query_with_property_filtering(config_gpkg_4326):
         assert 'straatnaam' in feature['properties']
 
         assert feature['properties']['straatnaam'] == 'Arnhemseweg'
+
+    feature_collection = p.query(
+        properties=[
+            ('straatnaam', "Arnhemseweg' OR '1'='1")
+        ]
+    )
+
+    assert feature_collection.get('type') == 'FeatureCollection'
+    features = feature_collection.get('features')
+    assert len(features) == 0
+
+    feature_collection = p.query(
+        properties=[
+            ('straatnaam', "doesnotexist' OR '1'='1")
+        ]
+    )
+
+    assert feature_collection.get('type') == 'FeatureCollection'
+    features = feature_collection.get('features')
+    assert len(features) == 0