Form for SQL validation

debug_toolbar/forms.py

@@ -0,0 +1,92 @@
+from django import forms
+from django.conf import settings
+from django.core.exceptions import ValidationError
+from django.utils.functional import cached_property
+
+try:
+    import json
+except ImportError:
+    from django.utils import simplejson as json
+
+try:
+    from hashlib import sha1
+except ImportError:
+    from django.utils.hashcompat import sha_constructor as sha1
+
+from debug_toolbar.utils.compat.db import connections
+
+
+class SQLSelectForm(forms.Form):
+    """
+    Validate params
+
+        sql: urlencoded sql with positional arguments
+        params: JSON encoded parameter values
+        duration: time for SQL to execute passed in from toolbar just for redisplay
+        hash: the hash of (secret + sql + params) for tamper checking
+    """
+    sql = forms.CharField()
+    params = forms.CharField()
+    alias = forms.CharField(required=False, initial='default')
+    duration = forms.FloatField()
+    hash = forms.CharField()
+
+    def __init__(self, *args, **kwargs):
+        initial = kwargs.get('initial', None)
+
+        if initial is not None:
+            initial['hash'] = self.make_hash(initial)
+
+        super(SQLSelectForm, self).__init__(*args, **kwargs)
+
+        for name in self.fields:
+            self.fields[name].widget = forms.HiddenInput()
+
+    def clean_sql(self):
+        value = self.cleaned_data['sql']
+
+        if not value.lower().strip().startswith('select'):
+            raise ValidationError("Only 'select' queries are allowed.")
+
+        return value
+
+    def clean_params(self):
+        value = self.cleaned_data['params']
+
+        try:
+            return json.loads(value)
+        except ValueError:
+            raise ValidationError('Is not valid JSON')
+
+    def clean_alias(self):
+        value = self.cleaned_data['alias']
+
+        if value not in connections:
+            raise ValidationError("Database alias '%s' not found" % value)
+
+        return value
+
+    def clean_hash(self):
+        hash = self.cleaned_data['hash']
+
+        if hash != self.make_hash(self.data):
+            raise ValidationError('Tamper alert')
+
+        return hash
+
+    def reformat_sql(self):
+        from debug_toolbar.panels.sql import reformat_sql
+        sql, params = self.cleaned_data['sql'], self.cleaned_data['params']
+        return reformat_sql(self.cursor.db.ops.last_executed_query(self.cursor, sql, params))
+
+    def make_hash(self, data):
+        params = settings.SECRET_KEY + data['sql'] + data['params']
+        return sha1(params).hexdigest()
+
+    @property
+    def connection(self):
+        return connections[self.cleaned_data['alias']]
+
+    @cached_property
+    def cursor(self):
+        return self.connection.cursor()

debug_toolbar/panels/sql.py

@@ -1,10 +1,12 @@
 import re
 import uuid
+from copy import copy
 
 from django.db.backends import BaseDatabaseWrapper
 from django.utils.html import escape
 from django.utils.translation import ugettext_lazy as _, ungettext_lazy as __
 
+from debug_toolbar.forms import SQLSelectForm
 from debug_toolbar.utils.compat.db import connections
 from debug_toolbar.middleware import DebugToolbarMiddleware
 from debug_toolbar.panels import DebugPanel
@@ -170,6 +172,9 @@ def process_response(self, request, response):
                     query['iso_level'] = get_isolation_level_display(query['engine'], query['iso_level'])
                 if 'trans_status' in query:
                     query['trans_status'] = get_transaction_status_display(query['engine'], query['trans_status'])
+
+                query['form'] = SQLSelectForm(auto_id=None, initial=copy(query))
+
                 if query['sql']:
                     query['sql'] = reformat_sql(query['sql'])
                 query['rgb_color'] = self._databases[alias]['rgb_color']

debug_toolbar/templates/debug_toolbar/panels/sql.html

@@ -44,21 +44,18 @@
 
 					{% if query.params %}
 						{% if query.is_select %}
+							<form method="post">
+								{% for field in query.form.hidden_fields %}
+									{{ field }}
+								{% endfor %}
 
-                            <form method="post">
-                                <input type="hidden" name="sql" value="{{ query.raw_sql }}" />
-                                <input type="hidden" name="params" value="{{ query.params }}" />
-                                <input type="hidden" name="duration" value="{{ query.duration|floatformat:"2" }}" />
-                                <input type="hidden" name="hash" value="{{ query.hash }}" />
-                                <input type="hidden" name="alias" value="{{ query.alias }}" />
+								<button formaction="/__debug__/sql_select/" class="remoteCall">Sel</button>
+								<button formaction="/__debug__/sql_explain/" class="remoteCall">Expl</button>
 
-                                <button formaction="/__debug__/sql_select/" class="remoteCall">Sel</button>
-                                <button formaction="/__debug__/sql_explain/" class="remoteCall">Expl</button>
-
-                                {% ifequal query.engine 'mysql' %}
-                                    <button formaction="/__debug__/sql_profile/" class="remoteCall">Prof</button>
-                                {% endifequal %}
-                            </form>
+								{% ifequal query.engine 'mysql' %}
+									<button formaction="/__debug__/sql_profile/" class="remoteCall">Prof</button>
+								{% endifequal %}
+							</form>
 						{% endif %}
 					{% endif %}
 					</td>

debug_toolbar/utils/tracking/db.py

@@ -139,9 +139,6 @@ def execute(self, sql, params=()):
                 'duration': duration,
                 'raw_sql': sql,
                 'params': _params,
-                'hash': sha1(settings.SECRET_KEY \
-                                        + smart_str(sql) \
-                                        + _params).hexdigest(),
                 'stacktrace': stacktrace,
                 'start_time': start,
                 'stop_time': stop,

debug_toolbar/views.py

@@ -4,87 +4,55 @@
 views in any other way is generally not advised.
 """
 
-from django.conf import settings
 from django.http import HttpResponseBadRequest
 from django.shortcuts import render_to_response
 from django.utils import simplejson
 from django.views.decorators.csrf import csrf_exempt
 
-from debug_toolbar.utils.compat.db import connections
 
 try:
     from hashlib import sha1
 except ImportError:
     from django.utils.hashcompat import sha_constructor as sha1
 
-
-class InvalidSQLError(Exception):
-    def __init__(self, value):
-        self.value = value
-
-    def __str__(self):
-        return repr(self.value)
+from debug_toolbar.forms import SQLSelectForm
 
 
 @csrf_exempt
 def sql_select(request):
-    """
-    Returns the output of the SQL SELECT statement.
+    """Returns the output of the SQL SELECT statement"""
+    form = SQLSelectForm(request.POST or None)
 
-    Expected GET variables:
-        sql: urlencoded sql with positional arguments
-        params: JSON encoded parameter values
-        duration: time for SQL to execute passed in from toolbar just for redisplay
-        hash: the hash of (secret + sql + params) for tamper checking
-    """
-    from debug_toolbar.panels.sql import reformat_sql
-    sql = request.REQUEST.get('sql', '')
-    params = request.REQUEST.get('params', '')
-    alias = request.REQUEST.get('alias', 'default')
-    hash = sha1(settings.SECRET_KEY + sql + params).hexdigest()
-    if hash != request.REQUEST.get('hash', ''):
-        return HttpResponseBadRequest('Tamper alert')  # SQL Tampering alert
-    if sql.lower().strip().startswith('select'):
-        params = simplejson.loads(params)
-        cursor = connections[alias].cursor()
+    if form.is_valid():
+        sql = form.cleaned_data['sql']
+        params = form.cleaned_data['params']
+        cursor = form.cursor
         cursor.execute(sql, params)
         headers = [d[0] for d in cursor.description]
         result = cursor.fetchall()
         cursor.close()
         context = {
             'result': result,
-            'sql': reformat_sql(cursor.db.ops.last_executed_query(cursor, sql, params)),
-            'duration': request.REQUEST.get('duration', 0.0),
+            'sql': form.reformat_sql(),
+            'duration': form.cleaned_data['duration'],
             'headers': headers,
-            'alias': alias,
+            'alias': form.cleaned_data['alias'],
         }
         return render_to_response('debug_toolbar/panels/sql_select.html', context)
-    raise InvalidSQLError("Only 'select' queries are allowed.")
+    return HttpResponseBadRequest('Form errors')
 
 
 @csrf_exempt
 def sql_explain(request):
-    """
-    Returns the output of the SQL EXPLAIN on the given query.
+    """Returns the output of the SQL EXPLAIN on the given query"""
+    form = SQLSelectForm(request.POST or None)
 
-    Expected GET variables:
-        sql: urlencoded sql with positional arguments
-        params: JSON encoded parameter values
-        duration: time for SQL to execute passed in from toolbar just for redisplay
-        hash: the hash of (secret + sql + params) for tamper checking
-    """
-    from debug_toolbar.panels.sql import reformat_sql
-    sql = request.REQUEST.get('sql', '')
-    params = request.REQUEST.get('params', '')
-    alias = request.REQUEST.get('alias', 'default')
-    hash = sha1(settings.SECRET_KEY + sql + params).hexdigest()
-    if hash != request.REQUEST.get('hash', ''):
-        return HttpResponseBadRequest('Tamper alert')  # SQL Tampering alert
-    if sql.lower().strip().startswith('select'):
-        params = simplejson.loads(params)
-        cursor = connections[alias].cursor()
-
-        conn = connections[alias].connection
+    if form.is_valid():
+        sql = form.cleaned_data['sql']
+        params = form.cleaned_data['params']
+        cursor = form.cursor
+
+        conn = form.connection
         engine = conn.__class__.__module__.split('.', 1)[0]
 
         if engine == "sqlite3":
@@ -100,36 +68,24 @@ def sql_explain(request):
         cursor.close()
         context = {
             'result': result,
-            'sql': reformat_sql(cursor.db.ops.last_executed_query(cursor, sql, params)),
-            'duration': request.REQUEST.get('duration', 0.0),
+            'sql': form.reformat_sql(),
+            'duration': form.cleaned_data['duration'],
             'headers': headers,
-            'alias': alias,
+            'alias': form.cleaned_data['alias'],
         }
         return render_to_response('debug_toolbar/panels/sql_explain.html', context)
-    raise InvalidSQLError("Only 'select' queries are allowed.")
+    return HttpResponseBadRequest('Form errors')
 
 
 @csrf_exempt
 def sql_profile(request):
-    """
-    Returns the output of running the SQL and getting the profiling statistics.
+    """Returns the output of running the SQL and getting the profiling statistics"""
+    form = SQLSelectForm(request.POST or None)
 
-    Expected GET variables:
-        sql: urlencoded sql with positional arguments
-        params: JSON encoded parameter values
-        duration: time for SQL to execute passed in from toolbar just for redisplay
-        hash: the hash of (secret + sql + params) for tamper checking
-    """
-    from debug_toolbar.panels.sql import reformat_sql
-    sql = request.REQUEST.get('sql', '')
-    params = request.REQUEST.get('params', '')
-    alias = request.REQUEST.get('alias', 'default')
-    hash = sha1(settings.SECRET_KEY + sql + params).hexdigest()
-    if hash != request.REQUEST.get('hash', ''):
-        return HttpResponseBadRequest('Tamper alert')  # SQL Tampering alert
-    if sql.lower().strip().startswith('select'):
-        params = simplejson.loads(params)
-        cursor = connections[alias].cursor()
+    if form.is_valid():
+        sql = form.cleaned_data['sql']
+        params = form.cleaned_data['params']
+        cursor = form.cursor
         result = None
         headers = None
         result_error = None
@@ -147,13 +103,13 @@ def sql_profile(request):
         context = {
             'result': result,
             'result_error': result_error,
-            'sql': reformat_sql(cursor.db.ops.last_executed_query(cursor, sql, params)),
-            'duration': request.REQUEST.get('duration', 0.0),
+            'sql': form.reformat_sql(),
+            'duration': form.cleaned_data['duration'],
             'headers': headers,
-            'alias': alias,
+            'alias': form.cleaned_data['alias'],
         }
         return render_to_response('debug_toolbar/panels/sql_profile.html', context)
-    raise InvalidSQLError("Only 'select' queries are allowed.")
+    return HttpResponseBadRequest('Form errors')
 
 
 def template_source(request):