Safely handle single feature SQL requests (#2288)

* Update test_mysql_provider.py

* Update sql.py

* Fix import order

* Update sql.py

Fix assertion

fix flake8

Author: webb-ben

pygeoapi/provider/sql.py

@@ -64,7 +64,8 @@
 from sqlalchemy.exc import (
     ConstraintColumnNotFoundError,
     InvalidRequestError,
-    OperationalError
+    OperationalError,
+    SQLAlchemyError
 )
 from sqlalchemy.ext.automap import automap_base
 from sqlalchemy.orm import Session, load_only
@@ -312,8 +313,12 @@ def get(self, identifier, crs_transform_spec=None, **kwargs):
         # Execute query within self-closing database Session context
         with Session(self._engine) as session:
             # Retrieve data from database as feature
-            item = session.get(self.table_model, identifier)
-            if item is None:
+            try:
+                item = session.get(self.table_model, identifier)
+                # Ensure that item is not None
+                assert item is not None
+            except (AssertionError, SQLAlchemyError) as e:
+                LOGGER.debug(e, exc_info=True)
                 msg = f'No such item: {self.id_field}={identifier}.'
                 raise ProviderItemNotFoundError(msg)
             crs_transform_out = get_transform_from_spec(crs_transform_spec)
@@ -827,3 +832,67 @@ def _get_bbox_filter(self, bbox: list[float]):
             func.ST_GeomFromText(polygon_wkt), geom_column
         )
         return bbox_filter
+
+    def get(self, identifier, crs_transform_spec=None, **kwargs):
+        """
+        Query the provider for a specific
+        feature id e.g: /collections/hotosm_bdi_waterways/items/13990765
+
+        :param identifier: feature id
+        :param crs_transform_spec: `CrsTransformSpec` instance, optional
+
+        :returns: GeoJSON FeatureCollection
+        """
+        LOGGER.debug(f'Get item by ID: {identifier}')
+
+        # Execute query within self-closing database Session context
+        with Session(self._engine) as session:
+            # Retrieve data from database as feature
+            try:
+                item = session.get(self.table_model, identifier)
+                # Ensure that item is not None
+                assert item is not None
+                # Ensure returned row has exact match
+                feature_id = getattr(item, self.id_field)
+                assert str(feature_id) == identifier
+            except (AssertionError, SQLAlchemyError) as e:
+                LOGGER.debug(e, exc_info=True)
+                msg = f'No such item: {self.id_field}={identifier}.'
+                raise ProviderItemNotFoundError(msg)
+            crs_transform_out = get_transform_from_spec(crs_transform_spec)
+            feature = self._sqlalchemy_to_feature(item, crs_transform_out)
+
+            # Drop non-defined properties
+            if self.properties:
+                props = feature['properties']
+                dropping_keys = deepcopy(props).keys()
+                for item in dropping_keys:
+                    if item not in self.properties:
+                        props.pop(item)
+
+            # Add fields for previous and next items
+            id_field = getattr(self.table_model, self.id_field)
+            prev_item = (
+                session.query(self.table_model)
+                .order_by(id_field.desc())
+                .filter(id_field < feature_id)
+                .first()
+            )
+            next_item = (
+                session.query(self.table_model)
+                .order_by(id_field.asc())
+                .filter(id_field > feature_id)
+                .first()
+            )
+            feature['prev'] = (
+                getattr(prev_item, self.id_field)
+                if prev_item is not None
+                else feature_id
+            )
+            feature['next'] = (
+                getattr(next_item, self.id_field)
+                if next_item is not None
+                else feature_id
+            )
+
+        return feature

tests/provider/test_mysql_provider.py

@@ -166,6 +166,19 @@ def test_query_skip_geometry(config):
     assert feature['geometry'] is None
 
 
+def test_get_with_injection(config):
+    """Testing query for injection attack string"""
+    p = MySQLProvider(config)
+    feature = p.get('1')
+    assert feature.get('type') == 'Feature'
+
+    with pytest.raises(ProviderItemNotFoundError):
+        p.get('1; DROP TABLE location;')
+
+    with pytest.raises(ProviderItemNotFoundError):
+        p.get('1<script>alert("We love pygeoapi")</script>')
+
+
 def test_get_not_existing_item_raise_exception(config):
     """Testing query for a not existing object"""
     p = MySQLProvider(config)

tests/provider/test_postgresql_provider.py

@@ -354,6 +354,19 @@ def test_get_simple(config, id_, prev, next_):
     assert result['next'] == next_
 
 
+def test_get_with_injection(config):
+    """Testing query for injection attack string"""
+    p = PostgreSQLProvider(config)
+    feature = p.get('29701937')
+    assert feature.get('type') == 'Feature'
+
+    with pytest.raises(ProviderItemNotFoundError):
+        p.get('29701937; DROP TABLE location;')
+
+    with pytest.raises(ProviderItemNotFoundError):
+        p.get('29701937<script>alert("We love pygeoapi")</script>')
+
+
 def test_get_with_config_properties(config):
     """
     Test that get is restricted by properties in the config.