From 01126c6458f901da369bd545a3ff1f1ceb8a5b36 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Tue, 10 Nov 2026 17:07:21 +0100 Subject: [PATCH] fix(django-spanner): prevent SQL injection in date and time functions Date and time operations (such as extracting date parts or truncating timestamps) previously formatted timezone names and lookup types directly into the SQL text as raw strings. When an application accepted timezones or date lookups from untrusted user input, this allowed an attacker to break out of quotes and execute arbitrary SQL. This change eliminates the vulnerability by passing timezones as bound query parameters rather than embedding them in the SQL statement. Date lookup types are also validated against an identifier allowlist before queries are built, following the standard approach used by other Django database engines. Additionally, fixed-offset timezones are now normalized to the format Cloud Spanner expects, and previously skipped Django SQL injection regression tests have been re-enabled and verified. --- .../django_spanner/features.py | 4 - .../django_spanner/operations.py | 118 ++++--- .../tests/mockserver_tests/models.py | 5 + .../test_datetime_expressions.py | 287 ++++++++++++++++++ .../unit/django_spanner/test_operations.py | 217 +++++++++++-- 5 files changed, 572 insertions(+), 59 deletions(-) create mode 100644 packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py diff --git a/packages/django-google-spanner/django_spanner/features.py b/packages/django-google-spanner/django_spanner/features.py index 8b182ebdb4e0..b424486b7e4a 100644 --- a/packages/django-google-spanner/django_spanner/features.py +++ b/packages/django-google-spanner/django_spanner/features.py @@ -121,15 +121,11 @@ def introspected_field_types(self): "db_functions.comparison.test_cast.CastTests.test_cast_to_decimal_field", "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_extract_func", "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_extract_iso_weekday_func", - "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_extract_lookup_name_sql_injection", - "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_trunc_lookup_name_sql_injection", "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_trunc_time_comparison", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_func", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_func_with_timezone", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_iso_weekday_func", - "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_lookup_name_sql_injection", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_func_with_timezone", - "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_lookup_name_sql_injection", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_time_comparison", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_timezone_applied_before_truncation", "db_functions.json.test_json_object.JSONObjectTests.test_basic", diff --git a/packages/django-google-spanner/django_spanner/operations.py b/packages/django-google-spanner/django_spanner/operations.py index 5a5b8e3e4612..8cf6c466cf07 100644 --- a/packages/django-google-spanner/django_spanner/operations.py +++ b/packages/django-google-spanner/django_spanner/operations.py @@ -12,9 +12,11 @@ from django.conf import settings from django.db.backends.base.operations import BaseDatabaseOperations -from django.db.utils import DatabaseError +from django.db.backends.utils import split_tzname_delta +from django.db.utils import DatabaseError, OperationalError from django.utils import timezone from django.utils.duration import duration_microseconds +from django.utils.regex_helper import _lazy_re_compile from google.cloud.spanner_dbapi.parse_utils import ( DateStr, TimestampStr, @@ -25,6 +27,12 @@ class DatabaseOperations(BaseDatabaseOperations): """A Spanner-specific version of Django database operations.""" + # EXTRACT and TRUNC date parts cannot be passed as query parameters in SQL. + # Following the practice of other Django database backends (PostgreSQL, + # MySQL, Oracle), validate lookup types with an identifier regex to prevent + # SQL injection while allowing custom user-registered lookups. + extract_trunc_lookup_pattern = _lazy_re_compile(r"^[a-zA-Z_]+$") + # Cloud Spanner only supports INT64, so all integer fields can hold # 64-bit values. integer_field_ranges = { @@ -406,6 +414,48 @@ def convert_uuidfield_value(self, value, expression, connection): value = UUID(value) return value + _well_formatted_offset_pattern = _lazy_re_compile(r"^[+-]\d{2}:\d{2}$") + _four_digit_offset_pattern = _lazy_re_compile(r"([+-])(\d{2})(\d{2})$") + _two_digit_offset_pattern = _lazy_re_compile(r"([+-])(\d{2})$") + + def _prepare_tzname_delta(self, tzname): + if not tzname or tzname == "UTC": + return "UTC" + if "+" not in tzname and "-" not in tzname: + return tzname + if self._well_formatted_offset_pattern.fullmatch(tzname): + return tzname + if tzname.startswith("UTC") and self._well_formatted_offset_pattern.fullmatch( + tzname[3:] + ): + return tzname[3:] + + tzname = self._four_digit_offset_pattern.sub(r"\1\2:\3", tzname) + tzname = self._two_digit_offset_pattern.sub(r"\1\2:00", tzname) + tzname, sign, offset = split_tzname_delta(tzname) + if offset: + if len(offset) in (7, 8) and offset.endswith(":00"): + offset = offset[:-3] + if len(offset) == 4 and offset.isdigit(): + offset = f"{offset[:2]}:{offset[2:]}" + return f"{sign}{offset}" + return tzname + + def _validate_lookup_type(self, lookup_type): + if not isinstance( + lookup_type, str + ) or not self.extract_trunc_lookup_pattern.fullmatch(lookup_type): + raise ValueError(f"Invalid lookup type: {lookup_type!r}") + + # Django's test_trunc_lookup_name_sql_injection catches only (DataError, OperationalError), + # unlike test_extract_lookup_name_sql_injection which also catches ValueError. + # OperationalError is raised here to satisfy Django's trunc exception contract. + def _validate_trunc_lookup_type(self, lookup_type): + if not isinstance( + lookup_type, str + ) or not self.extract_trunc_lookup_pattern.fullmatch(lookup_type): + raise OperationalError(f"Invalid trunc lookup type: {lookup_type!r}") + def date_extract_sql(self, lookup_type, field_name, params=None): """Extract date from the lookup. @@ -422,7 +472,8 @@ def date_extract_sql(self, lookup_type, field_name, params=None): :returns: A SQL statement for extracting. """ lookup_type = self.extract_names.get(lookup_type, lookup_type) - sql = "EXTRACT(%s FROM %s)" % (lookup_type, field_name) + self._validate_lookup_type(lookup_type) + sql = f"EXTRACT({lookup_type} FROM {field_name})" return sql, params def datetime_extract_sql(self, lookup_type, field_name, params, tzname): @@ -442,15 +493,13 @@ def datetime_extract_sql(self, lookup_type, field_name, params, tzname): :returns: A SQL statement for extracting. """ tzname = tzname if settings.USE_TZ and tzname else "UTC" + tzname_param = self._prepare_tzname_delta(tzname) lookup_type = self.extract_names.get(lookup_type, lookup_type) + self._validate_lookup_type(lookup_type) + params = () if params is None else tuple(params) return ( - 'EXTRACT(%s FROM %s AT TIME ZONE "%s")' - % ( - lookup_type, - field_name, - tzname, - ), - params, + f"EXTRACT({lookup_type} FROM {field_name} AT TIME ZONE %s)", + (*params, tzname_param), ) def time_extract_sql(self, lookup_type, field_name, params=None): @@ -468,12 +517,12 @@ def time_extract_sql(self, lookup_type, field_name, params=None): :rtype: str :returns: A SQL statement for extracting. """ + lookup_type = self.extract_names.get(lookup_type, lookup_type) + self._validate_lookup_type(lookup_type) + params = () if params is None else tuple(params) # Time is stored as TIMESTAMP with UTC time zone. - sql = 'EXTRACT(%s FROM %s AT TIME ZONE "UTC")' % ( - lookup_type, - field_name, - ) - return sql, params + sql = f"EXTRACT({lookup_type} FROM {field_name} AT TIME ZONE %s)" + return sql, (*params, "UTC") def date_trunc_sql(self, lookup_type, field_name, params, tzname=None): """Truncate date in the lookup. @@ -494,16 +543,14 @@ def date_trunc_sql(self, lookup_type, field_name, params, tzname=None): :rtype: str :returns: A SQL statement for truncating. """ + self._validate_trunc_lookup_type(lookup_type) # https://cloud.google.com/spanner/docs/functions-and-operators#date_trunc if lookup_type == "week": # Spanner truncates to Sunday but Django expects Monday. First, # subtract a day so that a Sunday will be truncated to the previous # week... field_name = "DATE_SUB(CAST(" + field_name + " AS DATE), INTERVAL 1 DAY)" - sql = "DATE_TRUNC(CAST(%s AS DATE), %s)" % ( - field_name, - lookup_type, - ) + sql = f"DATE_TRUNC(CAST({field_name} AS DATE), {lookup_type})" if lookup_type == "week": # ...then add a day to get from Sunday to Monday. sql = "DATE_ADD(CAST(" + sql + " AS DATE), INTERVAL 1 DAY)" @@ -529,20 +576,19 @@ def datetime_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"): """ # https://cloud.google.com/spanner/docs/functions-and-operators#timestamp_trunc tzname = tzname if settings.USE_TZ and tzname else "UTC" + tzname_param = self._prepare_tzname_delta(tzname) + self._validate_trunc_lookup_type(lookup_type) + params = () if params is None else tuple(params) if lookup_type == "week": # Spanner truncates to Sunday but Django expects Monday. First, # subtract a day so that a Sunday will be truncated to the previous # week... field_name = "TIMESTAMP_SUB(" + field_name + ", INTERVAL 1 DAY)" - sql = 'TIMESTAMP_TRUNC(%s, %s, "%s")' % ( - field_name, - lookup_type, - tzname, - ) + sql = f"TIMESTAMP_TRUNC({field_name}, {lookup_type}, %s)" if lookup_type == "week": # ...then add a day to get from Sunday to Monday. sql = "TIMESTAMP_ADD(" + sql + ", INTERVAL 1 DAY)" - return sql, params + return sql, (*params, tzname_param) def time_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"): """Truncate time in the lookup. @@ -564,14 +610,12 @@ def time_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"): """ # https://cloud.google.com/spanner/docs/functions-and-operators#timestamp_trunc tzname = tzname if settings.USE_TZ and tzname else "UTC" + tzname_param = self._prepare_tzname_delta(tzname) + self._validate_trunc_lookup_type(lookup_type) + params = () if params is None else tuple(params) return ( - 'TIMESTAMP_TRUNC(%s, %s, "%s")' - % ( - field_name, - lookup_type, - tzname, - ), - params, + f"TIMESTAMP_TRUNC({field_name}, {lookup_type}, %s)", + (*params, tzname_param), ) def datetime_cast_date_sql(self, field_name, params, tzname): @@ -592,7 +636,9 @@ def datetime_cast_date_sql(self, field_name, params, tzname): """ # https://cloud.google.com/spanner/docs/functions-and-operators#date tzname = tzname if settings.USE_TZ and tzname else "UTC" - return 'DATE(%s, "%s")' % (field_name, tzname), params + tzname_param = self._prepare_tzname_delta(tzname) + params = () if params is None else tuple(params) + return f"DATE({field_name}, %s)", (*params, tzname_param) def datetime_cast_time_sql(self, field_name, params, tzname): """Cast time in the lookup. @@ -611,12 +657,14 @@ def datetime_cast_time_sql(self, field_name, params, tzname): :returns: A SQL statement for casting. """ tzname = tzname if settings.USE_TZ and tzname else "UTC" + tzname_param = self._prepare_tzname_delta(tzname) + params = () if params is None else tuple(params) # Cloud Spanner doesn't have a function for converting # TIMESTAMP to another time zone. return ( - "TIMESTAMP(FORMAT_TIMESTAMP(" - "'%%Y-%%m-%%d %%R:%%E9S %%Z', %s, '%s'))" % (field_name, tzname) - ), params + f"TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', {field_name}, %s))", + (*params, tzname_param), + ) def date_interval_sql(self, timedelta): """Get a date interval in microseconds. diff --git a/packages/django-google-spanner/tests/mockserver_tests/models.py b/packages/django-google-spanner/tests/mockserver_tests/models.py index 66cafc60412f..4d3e8db68de6 100644 --- a/packages/django-google-spanner/tests/mockserver_tests/models.py +++ b/packages/django-google-spanner/tests/mockserver_tests/models.py @@ -18,3 +18,8 @@ class Singer(models.Model): first_name = models.CharField(max_length=200) last_name = models.CharField(max_length=200) + + +class Event(models.Model): + name = models.CharField(max_length=200) + start_time = models.DateTimeField() diff --git a/packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py b/packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py new file mode 100644 index 000000000000..6bde1bd4a61a --- /dev/null +++ b/packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py @@ -0,0 +1,287 @@ +# Copyright 2026 Google LLC All rights reserved. +# +# 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 zoneinfo +from datetime import datetime, timedelta, tzinfo +from datetime import timezone as dt_timezone + +from django.db.models import Value +from django.db.models.functions import ( + Coalesce, + Extract, + ExtractHour, + Trunc, + TruncDate, + TruncDay, + TruncTime, + TruncWeek, +) +from django.db.utils import OperationalError +from django.test import override_settings +from django.utils import timezone +from google.cloud.spanner_v1 import ExecuteSqlRequest, TypeCode + +from tests.mockserver_tests.mock_server_test_base import ( + MockServerTestBase, + add_single_result, +) +from tests.mockserver_tests.models import Event + + +class TestDatetimeExpressions(MockServerTestBase): + def get_last_execute_sql_request(self) -> ExecuteSqlRequest: + execute_requests = [ + request + for request in self.spanner_service.requests + if isinstance(request, ExecuteSqlRequest) + ] + self.assertTrue( + len(execute_requests) > 0, + "Expected at least one ExecuteSqlRequest in recorded requests", + ) + return execute_requests[-1] + + def test_extract_hour_parameterized(self): + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("12",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time") + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [12]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "UTC") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_extract_named_timezone_parameterized(self): + new_york_timezone = zoneinfo.ZoneInfo("America/New_York") + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("8",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time", tzinfo=new_york_timezone) + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [8]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "America/New_York") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_extract_fixed_offset_timezone_normalized(self): + fixed_offset_timezone = timezone.get_fixed_timezone(-283) + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("7",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time", tzinfo=fixed_offset_timezone) + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [7]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "-04:43") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_trunc_day_parameterized(self): + tokyo_timezone = zoneinfo.ZoneInfo("Asia/Tokyo") + sql = ( + "SELECT TIMESTAMP_TRUNC(tests_event.start_time, day, @a0) " + "AS truncated_day FROM tests_event" + ) + add_single_result( + sql, "truncated_day", TypeCode.TIMESTAMP, [("2026-11-10T00:00:00Z",)] + ) + results = list( + Event.objects.annotate( + truncated_day=TruncDay("start_time", tzinfo=tokyo_timezone) + ).values_list("truncated_day", flat=True) + ) + self.assertEqual(len(results), 1) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "Asia/Tokyo") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_trunc_week_parameterized(self): + new_york_timezone = zoneinfo.ZoneInfo("America/New_York") + sql = ( + "SELECT TIMESTAMP_ADD(TIMESTAMP_TRUNC(TIMESTAMP_SUB(tests_event.start_time, INTERVAL 1 DAY), " + "week, @a0), INTERVAL 1 DAY) AS truncated_week FROM tests_event" + ) + add_single_result( + sql, "truncated_week", TypeCode.TIMESTAMP, [("2026-11-09T00:00:00Z",)] + ) + results = list( + Event.objects.annotate( + truncated_week=TruncWeek("start_time", tzinfo=new_york_timezone) + ).values_list("truncated_week", flat=True) + ) + self.assertEqual(len(results), 1) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "America/New_York") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_trunc_date_and_time_cast_parameterized(self): + new_york_timezone = zoneinfo.ZoneInfo("America/New_York") + sql_date = "SELECT DATE(tests_event.start_time, @a0) AS truncated_date FROM tests_event" + add_single_result(sql_date, "truncated_date", TypeCode.DATE, [("2026-11-10",)]) + results_date = list( + Event.objects.annotate( + truncated_date=TruncDate("start_time", tzinfo=new_york_timezone) + ).values_list("truncated_date", flat=True) + ) + self.assertEqual(len(results_date), 1) + execute_request_date = self.get_last_execute_sql_request() + self.assertEqual(execute_request_date.sql, sql_date) + self.assertEqual(execute_request_date.params["a0"], "America/New_York") + + sql_time = ( + "SELECT TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', " + "tests_event.start_time, @a0)) AS truncated_time FROM tests_event" + ) + add_single_result( + sql_time, "truncated_time", TypeCode.TIMESTAMP, [("2026-11-10T12:00:00Z",)] + ) + results_time = list( + Event.objects.annotate( + truncated_time=TruncTime("start_time", tzinfo=new_york_timezone) + ).values_list("truncated_time", flat=True) + ) + self.assertEqual(len(results_time), 1) + execute_request_time = self.get_last_execute_sql_request() + self.assertEqual(execute_request_time.sql, sql_time) + self.assertEqual(execute_request_time.params["a0"], "America/New_York") + + def test_extract_compound_expression_preserves_parameter_order(self): + new_york_timezone = zoneinfo.ZoneInfo("America/New_York") + fallback_datetime = datetime(2026, 1, 1, 15, 0, tzinfo=dt_timezone.utc) + sql = ( + "SELECT EXTRACT(hour FROM COALESCE(tests_event.start_time, @a0) " + "AT TIME ZONE @a1) AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("10",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour( + Coalesce("start_time", Value(fallback_datetime)), + tzinfo=new_york_timezone, + ) + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [10]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "2026-01-01T15:00:00.000000Z") + self.assertEqual(execute_request.params["a1"], "America/New_York") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.TIMESTAMP) + self.assertEqual(execute_request.param_types["a1"].code, TypeCode.STRING) + + @override_settings(USE_TZ=False) + def test_extract_use_tz_false_defaults_to_utc(self): + new_york_timezone = zoneinfo.ZoneInfo("America/New_York") + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("12",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time", tzinfo=new_york_timezone) + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [12]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "UTC") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_tzname_sql_injection_travels_safely_as_parameter(self): + class MaliciousTimezone(tzinfo): + def utcoffset(self, dt): + return timedelta(0) + + def tzname(self, dt): + return "UTC'; DROP TABLE tests_event; --" + + def dst(self, dt): + return timedelta(0) + + malicious_timezone = MaliciousTimezone() + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("1",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time", tzinfo=malicious_timezone) + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [1]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertNotIn("DROP TABLE", execute_request.sql) + self.assertEqual( + execute_request.params["a0"], "UTC'; DROP TABLE tests_event; --" + ) + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_extract_rejects_malicious_lookup_type_before_rpc(self): + with self.assertRaises(ValueError): + list( + Event.objects.annotate( + injected=Extract("start_time", "hour'; DROP TABLE tests_event; --") + ).values_list("injected", flat=True) + ) + execute_requests = [ + request + for request in self.spanner_service.requests + if isinstance(request, ExecuteSqlRequest) + ] + self.assertEqual( + len(execute_requests), + 0, + "Expected zero ExecuteSqlRequest when lookup_type is invalid", + ) + + def test_trunc_rejects_malicious_lookup_type_before_rpc(self): + with self.assertRaises(OperationalError): + list( + Event.objects.annotate( + injected=Trunc("start_time", "day'; DROP TABLE tests_event; --") + ).values_list("injected", flat=True) + ) + execute_requests = [ + request + for request in self.spanner_service.requests + if isinstance(request, ExecuteSqlRequest) + ] + self.assertEqual( + len(execute_requests), + 0, + "Expected zero ExecuteSqlRequest when lookup_type is invalid", + ) diff --git a/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py b/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py index 520d268bce10..cd2c19ee0dc6 100644 --- a/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py +++ b/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py @@ -12,7 +12,7 @@ from django.conf import settings from django.core.management.color import no_style -from django.db.utils import DatabaseError +from django.db.utils import DatabaseError, OperationalError from google.cloud.spanner_dbapi.types import DateStr from tests.unit.django_spanner.simple_test import SpannerSimpleTestClass @@ -130,8 +130,8 @@ def test_datetime_extract_sql(self): "dayofweek", "dummy_field", None, "IST" ), ( - 'EXTRACT(dayofweek FROM dummy_field AT TIME ZONE "IST")', - None, + "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)", + ("IST",), ), ) @@ -142,8 +142,8 @@ def test_datetime_extract_sql_use_tz_false(self): "dayofweek", "dummy_field", None, "IST" ), ( - 'EXTRACT(dayofweek FROM dummy_field AT TIME ZONE "UTC")', - None, + "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)", + ("UTC",), ), ) settings.USE_TZ = True # reset changes. @@ -152,21 +152,21 @@ def test_time_extract_sql(self): self.assertEqual( self.db_operations.time_extract_sql("dayofweek", "dummy_field"), ( - 'EXTRACT(dayofweek FROM dummy_field AT TIME ZONE "UTC")', - None, + "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)", + ("UTC",), ), ) def test_time_trunc_sql(self): self.assertEqual( self.db_operations.time_trunc_sql("dayofweek", "dummy_field", None), - ('TIMESTAMP_TRUNC(dummy_field, dayofweek, "UTC")', None), + ("TIMESTAMP_TRUNC(dummy_field, dayofweek, %s)", ("UTC",)), ) def test_datetime_cast_date_sql(self): self.assertEqual( self.db_operations.datetime_cast_date_sql("dummy_field", None, "IST"), - ('DATE(dummy_field, "IST")', None), + ("DATE(dummy_field, %s)", ("IST",)), ) def test_datetime_cast_time_sql(self): @@ -174,8 +174,8 @@ def test_datetime_cast_time_sql(self): self.assertEqual( self.db_operations.datetime_cast_time_sql("dummy_field", None, "IST"), ( - "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', dummy_field, 'IST'))", - None, + "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', dummy_field, %s))", + ("IST",), ), ) @@ -184,8 +184,8 @@ def test_datetime_cast_time_sql_use_tz_false(self): self.assertEqual( self.db_operations.datetime_cast_time_sql("dummy_field", None, "IST"), ( - "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', dummy_field, 'UTC'))", - None, + "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', dummy_field, %s))", + ("UTC",), ), ) settings.USE_TZ = True # reset changes. @@ -327,23 +327,200 @@ def test_date_and_time_trunc_sql(self): self.assertIsNone(params) sql, params = self.db_operations.time_trunc_sql("hour", "field", None) - self.assertEqual(sql, 'TIMESTAMP_TRUNC(field, hour, "UTC")') - self.assertIsNone(params) + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, hour, %s)") + self.assertEqual(params, ("UTC",)) sql, params = self.db_operations.datetime_trunc_sql( "day", "field", None, tzname="UTC" ) - self.assertEqual(sql, 'TIMESTAMP_TRUNC(field, day, "UTC")') + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, day, %s)") + self.assertEqual(params, ("UTC",)) + + sql, params = self.db_operations.date_trunc_sql("week", "field", None) + self.assertEqual( + sql, + "DATE_ADD(CAST(DATE_TRUNC(CAST(DATE_SUB(CAST(field AS DATE), INTERVAL 1 DAY) AS DATE), week) AS DATE), INTERVAL 1 DAY)", + ) self.assertIsNone(params) + sql, params = self.db_operations.datetime_trunc_sql( + "week", "field", None, tzname="UTC" + ) + self.assertEqual( + sql, + "TIMESTAMP_ADD(TIMESTAMP_TRUNC(TIMESTAMP_SUB(field, INTERVAL 1 DAY), week, %s), INTERVAL 1 DAY)", + ) + self.assertEqual(params, ("UTC",)) + def test_datetime_cast_sql(self): sql, params = self.db_operations.datetime_cast_date_sql("field", None, "UTC") - self.assertEqual(sql, 'DATE(field, "UTC")') - self.assertIsNone(params) + self.assertEqual(sql, "DATE(field, %s)") + self.assertEqual(params, ("UTC",)) sql, params = self.db_operations.datetime_cast_time_sql("field", None, "UTC") self.assertEqual( sql, - "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', field, 'UTC'))", + "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', field, %s))", ) - self.assertIsNone(params) + self.assertEqual(params, ("UTC",)) + + def test_prepare_tzname_delta(self): + self.assertEqual( + self.db_operations._prepare_tzname_delta("UTC-04:43"), "-04:43" + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("UTC+05:00"), "+05:00" + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("UTC+05:30:00"), "+05:30" + ) + self.assertEqual(self.db_operations._prepare_tzname_delta("-0443"), "-04:43") + self.assertEqual(self.db_operations._prepare_tzname_delta("+0500"), "+05:00") + self.assertEqual(self.db_operations._prepare_tzname_delta("+03:00"), "+03:00") + self.assertEqual( + self.db_operations._prepare_tzname_delta("+05:30:00"), "+05:30" + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("-05:00:00"), "-05:00" + ) + self.assertEqual(self.db_operations._prepare_tzname_delta("+05:30"), "+05:30") + self.assertEqual(self.db_operations._prepare_tzname_delta("-04:43"), "-04:43") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC-0443"), "-04:43") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC+0500"), "+05:00") + self.assertEqual(self.db_operations._prepare_tzname_delta("+05"), "+05:00") + self.assertEqual(self.db_operations._prepare_tzname_delta("-08"), "-08:00") + self.assertEqual( + self.db_operations._prepare_tzname_delta("America/New_York"), + "America/New_York", + ) + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC"), "UTC") + self.assertEqual(self.db_operations._prepare_tzname_delta(""), "UTC") + self.assertEqual(self.db_operations._prepare_tzname_delta(None), "UTC") + + def test_extract_lookup_type_validation(self): + valid_lookups = ["year", "month", "day", "hour", "minute", "second", "isoweek"] + for lookup_type in valid_lookups: + self.db_operations._validate_lookup_type(lookup_type) + + invalid_lookups = [ + "year' OR 1=1--", + "day; DROP TABLE students;", + "foo bar", + "", + "year\x00", + None, + 123, + ] + for lookup_type in invalid_lookups: + with self.assertRaises(ValueError): + self.db_operations._validate_lookup_type(lookup_type) + with self.assertRaises(ValueError): + self.db_operations.date_extract_sql(lookup_type, "field") + with self.assertRaises(ValueError): + self.db_operations.datetime_extract_sql( + lookup_type, "field", None, "UTC" + ) + with self.assertRaises(ValueError): + self.db_operations.time_extract_sql(lookup_type, "field") + + def test_trunc_lookup_type_validation(self): + valid_lookups = [ + "year", + "quarter", + "month", + "week", + "day", + "hour", + "minute", + "second", + ] + for lookup_type in valid_lookups: + self.db_operations._validate_trunc_lookup_type(lookup_type) + + invalid_lookups = [ + "year' OR 1=1--", + "day; DROP TABLE students;", + "foo bar", + "", + "year\x00", + None, + 123, + ] + for lookup_type in invalid_lookups: + with self.assertRaises(OperationalError): + self.db_operations._validate_trunc_lookup_type(lookup_type) + with self.assertRaises(OperationalError): + self.db_operations.date_trunc_sql(lookup_type, "field", None) + with self.assertRaises(OperationalError): + self.db_operations.datetime_trunc_sql( + lookup_type, "field", None, tzname="UTC" + ) + with self.assertRaises(OperationalError): + self.db_operations.time_trunc_sql(lookup_type, "field", None) + + def test_timezone_methods_preserve_existing_params(self): + existing_params = ["param1", "param2"] + + sql, params = self.db_operations.datetime_extract_sql( + "hour", "field", existing_params, "UTC" + ) + self.assertEqual(params, ("param1", "param2", "UTC")) + + sql, params = self.db_operations.datetime_trunc_sql( + "hour", "field", existing_params, tzname="UTC" + ) + self.assertEqual(params, ("param1", "param2", "UTC")) + + sql, params = self.db_operations.time_trunc_sql( + "hour", "field", existing_params, tzname="UTC" + ) + self.assertEqual(params, ("param1", "param2", "UTC")) + + sql, params = self.db_operations.datetime_cast_date_sql( + "field", existing_params, "UTC" + ) + self.assertEqual(params, ("param1", "param2", "UTC")) + + sql, params = self.db_operations.datetime_cast_time_sql( + "field", existing_params, "UTC" + ) + self.assertEqual(params, ("param1", "param2", "UTC")) + + def test_tzname_sql_injection_defense(self): + settings.USE_TZ = True + malicious_tzname = "UTC'; DROP TABLE students; --" + + sql, params = self.db_operations.datetime_extract_sql( + "hour", "field", None, malicious_tzname + ) + self.assertNotIn(malicious_tzname, sql) + self.assertIn("%s", sql) + self.assertEqual(params, (malicious_tzname,)) + + sql, params = self.db_operations.datetime_trunc_sql( + "hour", "field", None, tzname=malicious_tzname + ) + self.assertNotIn(malicious_tzname, sql) + self.assertIn("%s", sql) + self.assertEqual(params, (malicious_tzname,)) + + sql, params = self.db_operations.time_trunc_sql( + "hour", "field", None, tzname=malicious_tzname + ) + self.assertNotIn(malicious_tzname, sql) + self.assertIn("%s", sql) + self.assertEqual(params, (malicious_tzname,)) + + sql, params = self.db_operations.datetime_cast_date_sql( + "field", None, malicious_tzname + ) + self.assertNotIn(malicious_tzname, sql) + self.assertIn("%s", sql) + self.assertEqual(params, (malicious_tzname,)) + + sql, params = self.db_operations.datetime_cast_time_sql( + "field", None, malicious_tzname + ) + self.assertNotIn(malicious_tzname, sql) + self.assertIn("%s", sql) + self.assertEqual(params, (malicious_tzname,))