From 15e7deabed7dbcb48a77241d43f1d1ebe62d4ed4 Mon Sep 17 00:00:00 2001
From: David Bitner <bitner@dbspatial.com>
Date: Mon, 25 Apr 2022 19:37:02 -0500
Subject: [PATCH] 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   | 27 +++++++++++++--------------
 tests/test_postgresql_provider.py |  4 +++-
 2 files changed, 16 insertions(+), 15 deletions(-)

diff --git a/pygeoapi/provider/postgresql.py b/pygeoapi/provider/postgresql.py
index 0fa0e39..024e35b 100644
--- a/pygeoapi/provider/postgresql.py
+++ b/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]
diff --git a/tests/test_postgresql_provider.py b/tests/test_postgresql_provider.py
index 44f5e0b..580a68f 100644
--- a/tests/test_postgresql_provider.py
+++ b/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 = [