Detect misuse of psycopg2.sql.SQL composable

README.rst

@@ -255,6 +255,7 @@ Usage::
       B609  linux_commands_wildcard_injection
       B610  django_extra_used
       B611  django_rawsql_used
+      B612  psycopg2_sql_injection
       B701  jinja2_autoescape_false
       B702  use_of_mako_templates
       B703  django_mark_safe

bandit/plugins/psycopg2_sql_injection.py

@@ -0,0 +1,40 @@
+# -*- coding:utf-8 -*-
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import ast
+
+import bandit
+from bandit.core import test_properties as test
+
+
+@test.checks('Call')
+@test.test_id('B612')
+def psycopg2_sql_injection(context):
+    """**B612: 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.6.2
+    """
+    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,
+                    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

@@ -122,6 +122,9 @@ bandit.plugins =
     # bandit/plugins/ssh_no_host_key_verification.py
     ssh_no_host_key_verification = bandit.plugins.ssh_no_host_key_verification:ssh_no_host_key_verification
 
+    # 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

@@ -369,6 +369,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 = {