Esri: sanitize attribute values

Author: tomkralidis

pygeoapi/provider/base.py

@@ -294,6 +294,27 @@ def _load_and_prepare_item(self, item, identifier=None,
 
         return identifier2, json_data
 
+    def sanitize_attribute_value(self, 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("'", "''") + "'"
+
     def __repr__(self):
         return f'<BaseProvider> {self.type}'
 

pygeoapi/provider/esri.py

@@ -3,6 +3,7 @@
 # Authors: Benjamin Webb <bwebb@lincolninst.edu>
 #
 # Copyright (c) 2022 Benjamin Webb
+# Copyright (c) 2026 Tom Kralidis
 #
 # Permission is hereby granted, free of charge, to any person
 # obtaining a copy of this software and associated documentation
@@ -309,13 +310,9 @@ def _make_where(self, properties=[], datetime_=None):
 
         p = []
 
-        if properties != []:
-
+        if properties:
             for (k, v) in properties:
-                if 'String' in self.fields[k]['type']:
-                    p.append(f"{k} = '{v}'")
-                else:
-                    p.append(f"{k} = {v}")
+                p.append(f'{k} = {self.sanitize_attribute_value(v)}')
 
         if datetime_ is not None:
 

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]} = {sanitize_attribute_value(x[1])}', properties)  # noqa
+                    map(lambda x: f'{x[0]} = {self.sanitize_attribute_value(x[1])}', properties)  # noqa
                 )
 
                 LOGGER.debug(attribute_filter)
@@ -410,7 +410,7 @@ def get(self, identifier, crs_transform_spec=None, **kwargs):
             LOGGER.debug(f'Fetching identifier {identifier}')
             layer = self._get_layer()
 
-            identifier2 = sanitize_attribute_value(identifier)
+            identifier2 = self.sanitize_attribute_value(identifier)
 
             layer.SetAttributeFilter(f'{self.id_field} = {identifier2}')
 
@@ -904,25 +904,3 @@ 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("'", "''") + "'"

tests/provider/test_base_provider.py

@@ -416,3 +416,13 @@ def test_unique_subclass_query_types():
     assert BaseEDRProvider.query_types != SensorThingsEDRProvider.query_types
     assert SensorThingsEDRProvider.query_types == \
         ['items', 'locations', 'cube', 'area']
+
+
+def test_sanitize_attribute_value(basic_provider_def):
+    provider = BaseProvider(basic_provider_def)
+
+    assert provider.sanitize_attribute_value(None) == 'NULL'
+    assert provider.sanitize_attribute_value(True) == '1'
+    assert provider.sanitize_attribute_value(False) == '0'
+    assert provider.sanitize_attribute_value(11.53) == "'11.53'"
+    assert provider.sanitize_attribute_value("NAME=4' OR '1'='1") == "'NAME=4'' OR ''1''=''1'"  # noqa

tests/provider/test_esri_provider.py

@@ -3,6 +3,7 @@
 # Authors: Benjamin Webb <bwebb@lincolninst.edu>
 #
 # Copyright (c) 2022 Benjamin Webb
+# Copyright (c) 2026 Tom Kralidis
 #
 # Permission is hereby granted, free of charge, to any person
 # obtaining a copy of this software and associated documentation
@@ -168,6 +169,16 @@ def test_query_properties(config):
     results = p.query(select_properties=['GEOGSTATE', ])
     assert len(results['features'][0]['properties']) == 1
 
+    results = p.query(
+        properties=[
+            ('OBJECTID', "1' OR '1'='1")
+        ]
+    )
+
+    assert results.get('type') == 'FeatureCollection'
+    features = results.get('features')
+    assert len(features) == 0
+
 
 def test_query_sortby_datetime(config):