Fix SQL injection vulnerability. Fix bug with schema prefixed tables. (#889)

* Do not use str.format() for database queries. Ensure that pg columns returned respect schema prefix.

* quotes not needed with %s

* use sorted on test array comparison

* linting

pygeoapi/provider/postgresql.py

@@ -114,21 +114,20 @@ class DatabaseConnection:
         if self.context == 'query':
             # Get table column names and types, excluding geometry and
             # transaction ID columns
-            query_cols = "SELECT attr.attname, tp.typname \
-            FROM pg_catalog.pg_class as cls \
-            INNER JOIN pg_catalog.pg_attribute as attr \
-                ON cls.oid = attr.attrelid \
-            INNER JOIN pg_catalog.pg_type as tp \
-                ON tp.oid = attr.atttypid \
-            WHERE cls.relname = '{}' \
-                AND tp.typname != 'geometry' \
-                AND tp.typname != 'cid' \
-                AND tp.typname != 'oid' \
-                AND tp.typname != 'tid' \
-                AND tp.typname != 'xid';".format(
-                self.table)
+            query_cols = """
+            SELECT
+                attr.attname,
+                tp.typname
+            FROM pg_catalog.pg_attribute as attr
+            INNER JOIN pg_catalog.pg_type as tp
+                ON tp.oid = attr.atttypid
+            WHERE
+                attr.attrelid = %s::regclass::oid
+                AND tp.typname != 'geometry'
+                AND attnum > 0
+            """
 
-            self.cur.execute(query_cols)
+            self.cur.execute(query_cols, (self.table,))
             result = self.cur.fetchall()
             if self.properties:
                 result = [res for res in result if res[0] in self.properties]

tests/test_postgresql_provider.py

@@ -91,7 +91,9 @@ def test_query_materialised_view(config, config_materialised_view):
     features = p.query(limit=14776).get("features", None)
     properties = features[0].get("properties", None)
     # Only width and depth properties should be available
-    assert list(properties.keys()) == ["osm_id", "width", "depth"]
+    assert sorted(list(properties.keys())) == sorted(
+        ["osm_id", "width", "depth"]
+    )
     p_full = PostgreSQLProvider(config)
     full_features = p_full.query(limit=14776).get("features", None)
     drain_features = [