OGR: sanitize attribute filter values

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("'", "''") + "'"

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