Detect misuse of psycopg2.sql.SQL composable

bandit/plugins/psycopg2_sql_injection.py

@@ -0,0 +1,39 @@
+# SPDX-License-Identifier: Apache-2.0
+import ast
+
+import bandit
+from bandit.core import issue
+from bandit.core import test_properties as test
+
+
+@test.checks("Call")
+@test.test_id("B613")
+def psycopg2_sql_injection(context):
+    """**B613: Potential SQL injection on psycopg2 raw SQL composable object **
+
+    The `psycopg2.sql.SQL` composable object should not be used to represent
+    variable identifiers or values that may be controlled by an attacker since
+    the argument that is passed to the `SQL` constructor is not escaped when
+    the SQL statement is composed. Instead, `SQL` should only be used to
+    represent constant strings.
+
+    .. seealso::
+
+     - https://www.psycopg.org/docs/sql.html
+
+    .. versionadded:: 1.7.6
+    """
+    if context.is_module_imported_like("psycopg2.sql"):
+        if context.call_function_name == "SQL":
+            argument = context.node.args[0]
+            if not isinstance(argument, ast.Str):
+                return bandit.Issue(
+                    severity=bandit.MEDIUM,
+                    confidence=bandit.MEDIUM,
+                    cwe=issue.Cwe.SQL_INJECTION,
+                    text=(
+                        "Possible SQL injection vector through instantiation "
+                        "of psycopg2.sql.SQL composable object on an argument "
+                        "other than a string literal."
+                    ),
+                )

examples/psycopg2_sql_injection.py

@@ -0,0 +1,4 @@
+from psycopg2 import sql
+
+table = 'users; drop table users; --'
+sql.SQL('select * from {}').format(sql.SQL(table))

setup.cfg

@@ -146,6 +146,9 @@ bandit.plugins =
     #bandit/plugins/tarfile_unsafe_members.py
     tarfile_unsafe_members = bandit.plugins.tarfile_unsafe_members:tarfile_unsafe_members
 
+    # bandit/plugins/psycopg2_sql_injection.py
+    psycopg2_sql_injection = bandit.plugins.psycopg2_sql_injection:psycopg2_sql_injection
+
 [build_sphinx]
 all_files = 1
 build-dir = doc/build

tests/functional/test_functional.py

@@ -393,6 +393,14 @@ def test_popen_wrappers(self):
         }
         self.check_example("popen_wrappers.py", expect)
 
+    def test_psycopg2_sql_injection(self):
+        """Test the `psycopg2` SQL injection example."""
+        expect = {
+            "SEVERITY": {"UNDEFINED": 0, "LOW": 0, "MEDIUM": 1, "HIGH": 0},
+            "CONFIDENCE": {"UNDEFINED": 0, "LOW": 0, "MEDIUM": 1, "HIGH": 0},
+        }
+        self.check_example("psycopg2_sql_injection.py", expect)
+
     def test_random_module(self):
         """Test for the `random` module."""
         expect = {