Django sql injection

README.rst

@@ -240,6 +240,8 @@ Usage::
       B607  start_process_with_partial_path
       B608  hardcoded_sql_expressions
       B609  linux_commands_wildcard_injection
+      B610  django_extra_used
+      B611  django_rawsql_used
       B701  jinja2_autoescape_false
       B702  use_of_mako_templates
 

bandit/plugins/django_sql_injection.py

@@ -0,0 +1,116 @@
+# -*- coding:utf-8 -*-
+#
+# Copyright (C) 2018 [Victor Torre](https://github.com/ehooo)
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+
+import ast
+import bandit
+from bandit.core import test_properties as test
+
+
+def keywords2dict(keywords):
+    kwargs = {}
+    for node in keywords:
+        if isinstance(node, ast.keyword):
+            kwargs[node.arg] = node.value
+    return kwargs
+
+
+@test.checks('Call')
+@test.test_id('B610')
+def django_extra_used(context):
+    """**B610: Potential SQL injection on extra function**
+
+    .. seealso::
+
+     - https://docs.djangoproject.com/en/dev/topics/
+        security/#sql-injection-protection
+
+    .. versionadded:: X.X.X
+
+    """
+    description = "Use of extra potential SQL attack vector."
+    if context.call_function_name == 'extra':
+        kwargs = keywords2dict(context.node.keywords)
+        args = context.node.args
+        if args:
+            if len(args) >= 1:
+                kwargs['select'] = args[0]
+            if len(args) >= 2:
+                kwargs['where'] = args[1]
+            if len(args) >= 3:
+                kwargs['params'] = args[2]
+            if len(args) >= 4:
+                kwargs['tables'] = args[3]
+            if len(args) >= 5:
+                kwargs['order_by'] = args[4]
+            if len(args) >= 6:
+                kwargs['select_params'] = args[5]
+        insecure = False
+        for key in ['where', 'tables']:
+            if key in kwargs:
+                if isinstance(kwargs[key], ast.List):
+                    for val in kwargs[key].elts:
+                        if not isinstance(val, ast.Str):
+                            insecure = True
+                            break
+                else:
+                    insecure = True
+                    break
+        if not insecure and 'select' in kwargs:
+            if isinstance(kwargs['select'], ast.Dict):
+                for k in kwargs['select'].keys:
+                    if not isinstance(k, ast.Str):
+                        insecure = True
+                        break
+                if not insecure:
+                    for v in kwargs['select'].values:
+                        if not isinstance(v, ast.Str):
+                            insecure = True
+                            break
+            else:
+                insecure = True
+
+        if insecure:
+            return bandit.Issue(
+                severity=bandit.MEDIUM,
+                confidence=bandit.MEDIUM,
+                text=description
+            )
+
+
+@test.checks('Call')
+@test.test_id('B611')
+def django_rawsql_used(context):
+    """**B611: Potential SQL injection on RawSQL function**
+
+    .. seealso::
+
+     - https://docs.djangoproject.com/en/dev/topics/
+        security/#sql-injection-protection
+
+    .. versionadded:: X.X.X
+
+    """
+    description = "Use of RawSQL potential SQL attack vector."
+    if context.is_module_imported_like('django.db.models'):
+        if context.call_function_name == 'RawSQL':
+            sql = context.node.args[0]
+            if not isinstance(sql, ast.Str):
+                return bandit.Issue(
+                    severity=bandit.MEDIUM,
+                    confidence=bandit.MEDIUM,
+                    text=description
+                )

examples/django_sql_injection_extra.py

@@ -0,0 +1,29 @@
+from django.contrib.auth.models import User
+
+User.objects.filter(username='admin').extra(
+    select={'test': 'secure'},
+    where=['secure'],
+    tables=['secure']
+)
+User.objects.filter(username='admin').extra({'test': 'secure'})
+User.objects.filter(username='admin').extra(select={'test': 'secure'})
+User.objects.filter(username='admin').extra(where=['secure'])
+
+User.objects.filter(username='admin').extra(dict(could_be='insecure'))
+User.objects.filter(username='admin').extra(select=dict(could_be='insecure'))
+query = '"username") AS "username", * FROM "auth_user" WHERE 1=1 OR "username"=? --'
+User.objects.filter(username='admin').extra(select={'test': query})
+User.objects.filter(username='admin').extra(select={'test': '%secure' % 'nos'})
+User.objects.filter(username='admin').extra(select={'test': '{}secure'.format('nos')})
+
+where_var = ['1=1) OR 1=1 AND (1=1']
+User.objects.filter(username='admin').extra(where=where_var)
+where_str = '1=1) OR 1=1 AND (1=1'
+User.objects.filter(username='admin').extra(where=[where_str])
+User.objects.filter(username='admin').extra(where=['%secure' % 'nos'])
+User.objects.filter(username='admin').extra(where=['{}secure'.format('no')])
+
+tables_var = ['django_content_type" WHERE "auth_user"."username"="admin']
+User.objects.all().extra(tables=tables_var).distinct()
+tables_str = 'django_content_type" WHERE "auth_user"."username"="admin'
+User.objects.all().extra(tables=[tables_str]).distinct()

examples/django_sql_injection_raw.py

@@ -0,0 +1,11 @@
+from django.db.models.expressions import RawSQL
+from django.contrib.auth.models import User
+
+User.objects.annotate(val=RawSQL('secure', []))
+User.objects.annotate(val=RawSQL('%secure' % 'nos', []))
+User.objects.annotate(val=RawSQL('{}secure'.format('no'), []))
+raw = '"username") AS "val" FROM "auth_user" WHERE "username"="admin" --'
+User.objects.annotate(val=RawSQL(raw, []))
+raw = '"username") AS "val" FROM "auth_user"' \
+      ' WHERE "username"="admin" OR 1=%s --'
+User.objects.annotate(val=RawSQL(raw, [0]))

setup.cfg

@@ -89,6 +89,10 @@ bandit.plugins =
     # bandit/plugins/injection_wildcard.py
     linux_commands_wildcard_injection = bandit.plugins.injection_wildcard:linux_commands_wildcard_injection
 
+    # bandit/plugins/django_sql_injection.py
+    django_extra_used = bandit.plugins.django_sql_injection:django_extra_used
+    django_rawsql_used = bandit.plugins.django_sql_injection:django_rawsql_used
+
     # bandit/plugins/insecure_ssl_tls.py
     ssl_with_bad_version = bandit.plugins.insecure_ssl_tls:ssl_with_bad_version
     ssl_with_bad_defaults = bandit.plugins.insecure_ssl_tls:ssl_with_bad_defaults

tests/functional/test_functional.py

@@ -446,6 +446,24 @@ def test_wildcard_injection(self):
         }
         self.check_example('wildcard-injection.py', expect)
 
+    def test_django_sql_injection(self):
+        """Test insecure extra functions on Django."""
+
+        expect = {
+            'SEVERITY': {'UNDEFINED': 0, 'LOW': 0, 'MEDIUM': 11, 'HIGH': 0},
+            'CONFIDENCE': {'UNDEFINED': 0, 'LOW': 0, 'MEDIUM': 11, 'HIGH': 0}
+        }
+        self.check_example('django_sql_injection_extra.py', expect)
+
+    def test_django_sql_injection_raw(self):
+        """Test insecure raw functions on Django."""
+
+        expect = {
+            'SEVERITY': {'UNDEFINED': 0, 'LOW': 0, 'MEDIUM': 4, 'HIGH': 0},
+            'CONFIDENCE': {'UNDEFINED': 0, 'LOW': 0, 'MEDIUM': 4, 'HIGH': 0}
+        }
+        self.check_example('django_sql_injection_raw.py', expect)
+
     def test_yaml(self):
         '''Test for `yaml.load`.'''
         expect = {