Skip to content

Binary parameter handling for GeoDjango #2169

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 7 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 26 additions & 3 deletions debug_toolbar/panels/sql/forms.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import base64
import json

from django import forms
Expand All @@ -10,6 +11,23 @@
from debug_toolbar.toolbar import DebugToolbar


def _reconstruct_params(params):
"""
Reconstruct parameters that were encoded for JSON storage,
especially binary data that was base64 encoded.
"""
if isinstance(params, list):
return [_reconstruct_params(param) for param in params]
elif isinstance(params, dict):
if "__djdt_binary__" in params:
# Reconstruct binary data from base64
return base64.b64decode(params["__djdt_binary__"])
else:
return {key: _reconstruct_params(value) for key, value in params.items()}
else:
return params


class SQLSelectForm(forms.Form):
"""
Validate params
Expand Down Expand Up @@ -69,10 +87,15 @@ def clean(self):
cleaned_data["query"] = query
return cleaned_data

def _get_query_params(self):
"""Get reconstructed parameters for the current query"""
query = self.cleaned_data["query"]
return _reconstruct_params(json.loads(query["params"]))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you think this would be improved by making use of the json.loads(..., cls) argument?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If I understand you correctly, @tim-schilling, you're suggesting that we create our own DebugToolbarJSONDecoder and use it like this: json.loads(query["params"], cls=DebugToolbarJSONDecoder), so we can encapsulate all the base64 logic inside of _reconstruct_params in custom JSON decoder class?


def select(self):
query = self.cleaned_data["query"]
sql = query["raw_sql"]
params = json.loads(query["params"])
params = self._get_query_params()
with self.cursor as cursor:
cursor.execute(sql, params)
headers = [d[0] for d in cursor.description]
Expand All @@ -82,7 +105,7 @@ def select(self):
def explain(self):
query = self.cleaned_data["query"]
sql = query["raw_sql"]
params = json.loads(query["params"])
params = self._get_query_params()
vendor = query["vendor"]
with self.cursor as cursor:
if vendor == "sqlite":
Expand All @@ -101,7 +124,7 @@ def explain(self):
def profile(self):
query = self.cleaned_data["query"]
sql = query["raw_sql"]
params = json.loads(query["params"])
params = self._get_query_params()
with self.cursor as cursor:
cursor.execute("SET PROFILING=1") # Enable profiling
cursor.execute(sql, params) # Execute SELECT
Expand Down
6 changes: 6 additions & 0 deletions debug_toolbar/panels/sql/tracking.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import base64
import contextlib
import contextvars
import datetime
Expand Down Expand Up @@ -126,6 +127,11 @@ def _decode(self, param):
if isinstance(param, dict):
return {key: self._decode(value) for key, value in param.items()}

# Handle binary data (e.g., GeoDjango EWKB geometry data)
if isinstance(param, (bytes, bytearray)):
# Mark as binary data for later reconstruction
return {"__djdt_binary__": base64.b64encode(param).decode("ascii")}

# make sure datetime, date and time are converted to string by force_str
CONVERT_TYPES = (datetime.datetime, datetime.date, datetime.time)
try:
Expand Down
8 changes: 8 additions & 0 deletions docs/changes.rst
Original file line number Diff line number Diff line change
@@ -1,6 +1,14 @@
Change log
==========

Pending
-------

* Fixed SQL Explain functionality for GeoDjango queries with binary parameters.
Binary data (such as EWKB geometry) is now properly handled through base64
encoding, preventing "parse error - invalid geometry" errors when using
Explain on spatial queries.

6.0.0 (2025-07-22)
------------------

Expand Down
173 changes: 173 additions & 0 deletions tests/panels/test_sql_geodjango_fix.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,173 @@
"""
Tests for GeoDjango binary parameter handling fix
"""

import base64
import json

from debug_toolbar.panels.sql.forms import _reconstruct_params
from debug_toolbar.panels.sql.tracking import NormalCursorMixin

from ..base import BaseTestCase


class MockCursor:
"""Mock cursor for testing"""


class MockConnection:
"""Mock database connection for testing"""

vendor = "postgresql"
alias = "default"


class MockLogger:
"""Mock logger for testing"""

def record(self, **kwargs):
pass


class TestCursor(NormalCursorMixin):
"""Test cursor that can be instantiated"""

def __init__(self):
# Initialize with mock objects
self.cursor = MockCursor()
self.db = MockConnection()
self.logger = MockLogger()


class GeoDjangoBinaryParameterTest(BaseTestCase):
"""Test cases for GeoDjango binary parameter handling"""

def test_binary_parameter_encoding_decoding(self):
"""Test that binary parameters are properly encoded and decoded"""
# Create a test cursor with the _decode method
cursor = TestCursor()

# Test binary data similar to GeoDjango EWKB geometry
binary_data = b"\x01\x01\x00\x00\x20\xe6\x10\x00\x00\xff\xfe\xfd"

# Test encoding (what happens when query is logged)
encoded = cursor._decode(binary_data)

# Should be marked as binary data
self.assertIsInstance(encoded, dict)
self.assertIn("__djdt_binary__", encoded)

# Should be base64 encoded
expected_b64 = base64.b64encode(binary_data).decode("ascii")
self.assertEqual(encoded["__djdt_binary__"], expected_b64)

# Test JSON serialization (what happens in tracking.py)
json_params = json.dumps([encoded])

# Test parsing back from JSON
parsed = json.loads(json_params)

# Test reconstruction (what happens in forms.py)
reconstructed = _reconstruct_params(parsed)

# Should recover original binary data
self.assertEqual(len(reconstructed), 1)
self.assertEqual(reconstructed[0], binary_data)
self.assertIsInstance(reconstructed[0], bytes)

def test_mixed_parameter_types(self):
"""Test that mixed parameter types are handled correctly"""
cursor = TestCursor()

# Test with mixed types including binary data
params = [
"string_param",
42,
b"\x01\x02\x03", # binary data
None,
["nested", "list"],
]

# Encode each parameter
encoded_params = [cursor._decode(p) for p in params]

# Serialize to JSON
json_str = json.dumps(encoded_params)

# Parse and reconstruct
parsed = json.loads(json_str)
reconstructed = _reconstruct_params(parsed)

# Check each parameter
self.assertEqual(reconstructed[0], "string_param") # string unchanged
self.assertEqual(reconstructed[1], 42) # int unchanged
self.assertEqual(reconstructed[2], b"\x01\x02\x03") # binary restored
self.assertIsNone(reconstructed[3]) # None unchanged
self.assertEqual(reconstructed[4], ["nested", "list"]) # list unchanged

def test_nested_binary_data(self):
"""Test binary data nested in lists and dicts"""
cursor = TestCursor()

# Test nested structures with binary data
nested_params = [
[b"\x01\x02", "string", b"\x03\x04"],
{"key": b"\x05\x06", "other": "value"},
]

# Encode
encoded = [cursor._decode(p) for p in nested_params]

# Serialize and parse
json_str = json.dumps(encoded)
parsed = json.loads(json_str)
reconstructed = _reconstruct_params(parsed)

# Check nested list
self.assertEqual(reconstructed[0][0], b"\x01\x02")
self.assertEqual(reconstructed[0][1], "string")
self.assertEqual(reconstructed[0][2], b"\x03\x04")

# Check nested dict
self.assertEqual(reconstructed[1]["key"], b"\x05\x06")
self.assertEqual(reconstructed[1]["other"], "value")

def test_empty_binary_data(self):
"""Test handling of empty binary data"""
cursor = TestCursor()

# Test empty bytes
empty_bytes = b""
encoded = cursor._decode(empty_bytes)

# Should still be marked as binary
self.assertIsInstance(encoded, dict)
self.assertIn("__djdt_binary__", encoded)

# Reconstruct
json_str = json.dumps([encoded])
parsed = json.loads(json_str)
reconstructed = _reconstruct_params(parsed)

self.assertEqual(reconstructed[0], empty_bytes)

def test_bytearray_support(self):
"""Test that bytearray is also handled as binary data"""
cursor = TestCursor()

# Test bytearray
byte_array = bytearray(b"\x01\x02\x03\x04")
encoded = cursor._decode(byte_array)

# Should be marked as binary
self.assertIn("__djdt_binary__", encoded)

# Reconstruct (should become bytes, not bytearray)
json_str = json.dumps([encoded])
parsed = json.loads(json_str)
reconstructed = _reconstruct_params(parsed)

# Should be equal in content (bytes vs bytearray comparison works)
self.assertEqual(reconstructed[0], byte_array)
# Should be bytes type after reconstruction
self.assertIsInstance(reconstructed[0], bytes)
Loading