From 4ee67f7db157af5b17ac806d90942f4175a404b6 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, truncating timestamps, and casting dates/times) 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: 1. Passing timezones as bound query parameters rather than embedding them in the SQL statement. 2. Validating date lookup types against an identifier allowlist before queries are built, following the standard approach used by other Django database engines. 3. Normalizing fixed-offset timezones (e.g. +02:00, UTC+02:00) to the format Cloud Spanner expects while preserving named timezones (such as Etc/GMT+5). Additionally, when no timezone is specified (or when USE_TZ = False), timezone arguments are now omitted from the generated SQL functions (EXTRACT, TIMESTAMP_TRUNC, DATE, FORMAT_TIMESTAMP) rather than falling back to a hardcoded "UTC". This allows Cloud Spanner to evaluate expressions using the database-level default timezone (ALTER DATABASE ... SET OPTIONS (default_time_zone = ...)). Unit, mock server, and emulator system tests have been added and verified, and previously skipped Django SQL injection regression tests have been re-enabled. --- .../django_spanner/features.py | 4 - .../django_spanner/operations.py | 161 ++++--- .../tests/mockserver_tests/models.py | 6 + .../test_datetime_expressions.py | 430 ++++++++++++++++++ .../tests/system/django_spanner/models.py | 2 + .../test_datetime_expressions.py | 267 +++++++++++ .../unit/django_spanner/test_operations.py | 430 ++++++++++++++++-- 7 files changed, 1211 insertions(+), 89 deletions(-) create mode 100644 packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py create mode 100644 packages/django-google-spanner/tests/system/django_spanner/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..32d983b23ac6 100644 --- a/packages/django-google-spanner/django_spanner/operations.py +++ b/packages/django-google-spanner/django_spanner/operations.py @@ -12,9 +12,10 @@ from django.conf import settings from django.db.backends.base.operations import BaseDatabaseOperations -from django.db.utils import DatabaseError +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 +26,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 +413,48 @@ def convert_uuidfield_value(self, value, expression, connection): value = UUID(value) return value + def _prepare_tzname_delta(self, tzname): + if not tzname or tzname == "UTC": + return "UTC" + # Only parse pure offset representations (e.g. +05:30, -08:00, UTC+5). + # Named timezones (including Etc/GMT+5, Etc/GMT-10) should be passed as-is to Spanner. + if tzname.startswith(("UTC+", "UTC-")): + tzname = tzname[3:] + elif not tzname.startswith(("+", "-")): + return tzname + # Spanner does not support seconds in timezone offsets (e.g. +05:30:00, -05:00:00). + if len(tzname) in (8, 9) and tzname.endswith(":00"): + return tzname[:-3] + return tzname + + def _get_timezone_param(self, tzname): + """Determine the timezone parameter according to Django USE_TZ settings. + + - When USE_TZ is True: use the application's timezone; if none is provided, + return None so Cloud Spanner uses the database default timezone. + - When USE_TZ is False: default to 'UTC' so naive timestamps (stored on the + UTC timeline) are evaluated without server timezone drift, unless an explicit + timezone is supplied. + """ + if settings.USE_TZ: + return self._prepare_tzname_delta(tzname) if tzname else None + return self._prepare_tzname_delta(tzname) if tzname else "UTC" + + 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 +471,9 @@ 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) + params = () if params is None else tuple(params) + sql = f"EXTRACT({lookup_type} FROM {field_name})" return sql, params def datetime_extract_sql(self, lookup_type, field_name, params, tzname): @@ -441,17 +492,16 @@ def datetime_extract_sql(self, lookup_type, field_name, params, tzname): :rtype: str :returns: A SQL statement for extracting. """ - tzname = tzname if settings.USE_TZ and tzname else "UTC" lookup_type = self.extract_names.get(lookup_type, lookup_type) - return ( - 'EXTRACT(%s FROM %s AT TIME ZONE "%s")' - % ( - lookup_type, - field_name, - tzname, - ), - params, - ) + self._validate_lookup_type(lookup_type) + params = () if params is None else tuple(params) + tzname_param = self._get_timezone_param(tzname) + if tzname_param: + return ( + f"EXTRACT({lookup_type} FROM {field_name} AT TIME ZONE %s)", + (*params, tzname_param), + ) + return f"EXTRACT({lookup_type} FROM {field_name})", params def time_extract_sql(self, lookup_type, field_name, params=None): """Extract time from the lookup. @@ -468,12 +518,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,22 +544,21 @@ 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) + params = () if params is None else tuple(params) # 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, - ) + field_name = f"DATE_SUB(CAST({field_name} AS DATE), INTERVAL 1 DAY)" + 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)" + sql = f"DATE_ADD(CAST({sql} AS DATE), INTERVAL 1 DAY)" return sql, params - def datetime_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"): + def datetime_trunc_sql(self, lookup_type, field_name, params, tzname=None): """Truncate datetime in the lookup. :type lookup_type: str @@ -528,23 +577,25 @@ def datetime_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"): :returns: A SQL statement for truncating. """ # https://cloud.google.com/spanner/docs/functions-and-operators#timestamp_trunc - tzname = tzname if settings.USE_TZ and tzname else "UTC" + 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, - ) + field_name = f"TIMESTAMP_SUB({field_name}, INTERVAL 1 DAY)" + tzname_param = self._get_timezone_param(tzname) + if tzname_param: + sql = f"TIMESTAMP_TRUNC({field_name}, {lookup_type}, %s)" + params = (*params, tzname_param) + else: + sql = f"TIMESTAMP_TRUNC({field_name}, {lookup_type})" if lookup_type == "week": # ...then add a day to get from Sunday to Monday. - sql = "TIMESTAMP_ADD(" + sql + ", INTERVAL 1 DAY)" + sql = f"TIMESTAMP_ADD({sql}, INTERVAL 1 DAY)" return sql, params - def time_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"): + def time_trunc_sql(self, lookup_type, field_name, params, tzname=None): """Truncate time in the lookup. :type lookup_type: str @@ -557,21 +608,20 @@ def time_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"): :param params: list of query params. :type tzname: str - :param tzname: The name of the timezone. Defaults to 'UTC' For backward compatability. + :param tzname: The name of the timezone. :rtype: str :returns: A SQL statement for truncating. """ # https://cloud.google.com/spanner/docs/functions-and-operators#timestamp_trunc + self._validate_trunc_lookup_type(lookup_type) + params = () if params is None else tuple(params) + # Time is stored as TIMESTAMP with UTC time zone. tzname = tzname if settings.USE_TZ and tzname else "UTC" + tzname_param = self._prepare_tzname_delta(tzname) 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): @@ -584,15 +634,17 @@ def datetime_cast_date_sql(self, field_name, params, tzname): :param params: list of query params. :type tzname: str - :param tzname: The time zone name. If using of time zone is not - allowed in settings default will be UTC. + :param tzname: The time zone name. :rtype: str :returns: A SQL statement for casting. """ # 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 + params = () if params is None else tuple(params) + tzname_param = self._get_timezone_param(tzname) + if tzname_param: + return f"DATE({field_name}, %s)", (*params, tzname_param) + return f"DATE({field_name})", params def datetime_cast_time_sql(self, field_name, params, tzname): """Cast time in the lookup. @@ -604,19 +656,22 @@ def datetime_cast_time_sql(self, field_name, params, tzname): :param params: list of query params. :type tzname: str - :param tzname: The time zone name. If using of time zone is not - allowed in settings default will be UTC. + :param tzname: The time zone name. :rtype: str :returns: A SQL statement for casting. """ - tzname = tzname if settings.USE_TZ and tzname else "UTC" - # Cloud Spanner doesn't have a function for converting - # TIMESTAMP to another time zone. + params = () if params is None else tuple(params) + tzname_param = self._get_timezone_param(tzname) + if tzname_param: + return ( + f"TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', {field_name}, %s), 'UTC')", + (*params, tzname_param), + ) return ( - "TIMESTAMP(FORMAT_TIMESTAMP(" - "'%%Y-%%m-%%d %%R:%%E9S %%Z', %s, '%s'))" % (field_name, tzname) - ), params + f"TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', {field_name}), 'UTC')", + params, + ) 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..db6c951a8d16 100644 --- a/packages/django-google-spanner/tests/mockserver_tests/models.py +++ b/packages/django-google-spanner/tests/mockserver_tests/models.py @@ -18,3 +18,9 @@ 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() + start_time_of_day = models.TimeField(null=True) 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..3f14081d7ae3 --- /dev/null +++ b/packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py @@ -0,0 +1,430 @@ +# 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, + TruncHour, + 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(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"], "-0443") + 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('1970-01-01 %H:%M:%E9S', " + "tests_event.start_time, @a0), 'UTC') 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): + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour_use_tz_false FROM tests_event" + ) + add_single_result(sql, "extracted_hour_use_tz_false", TypeCode.INT64, [("12",)]) + results = list( + Event.objects.annotate( + extracted_hour_use_tz_false=ExtractHour("start_time") + ).values_list("extracted_hour_use_tz_false", 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") + + 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", + ) + + def test_trunc_timefield_defaults_to_utc(self): + sql = ( + "SELECT TIMESTAMP_TRUNC(tests_event.start_time_of_day, hour, @a0) " + "AS truncated_hour FROM tests_event" + ) + add_single_result( + sql, "truncated_hour", TypeCode.TIMESTAMP, [("1970-01-01T10:00:00Z",)] + ) + results = list( + Event.objects.annotate( + truncated_hour=TruncHour("start_time_of_day") + ).values_list("truncated_hour", 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"], "UTC") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_extract_timefield_defaults_to_utc(self): + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time_of_day AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("10",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time_of_day") + ).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"], "UTC") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_trunc_with_fixed_offset_parameterized(self): + kolkata_tz = dt_timezone(timedelta(hours=5, minutes=30)) + sql = ( + "SELECT TIMESTAMP_TRUNC(tests_event.start_time, hour, @a0) " + "AS truncated_hour FROM tests_event" + ) + add_single_result( + sql, "truncated_hour", TypeCode.TIMESTAMP, [("2026-11-10T10:00:00Z",)] + ) + results = list( + Event.objects.annotate( + truncated_hour=TruncHour("start_time", tzinfo=kolkata_tz) + ).values_list("truncated_hour", 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"], "+05:30") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + @override_settings(USE_TZ=False) + def test_cast_date_and_time_use_tz_false(self): + 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")).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"], "UTC") + + sql_time = ( + "SELECT TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', " + "tests_event.start_time, @a0), 'UTC') 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")).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"], "UTC") + + @override_settings(USE_TZ=False) + def test_trunc_hour_use_tz_false(self): + sql = ( + "SELECT TIMESTAMP_TRUNC(tests_event.start_time, hour, @a0) " + "AS truncated_hour FROM tests_event" + ) + add_single_result( + sql, "truncated_hour", TypeCode.TIMESTAMP, [("2026-11-10T12:00:00Z",)] + ) + results = list( + Event.objects.annotate(truncated_hour=TruncHour("start_time")).values_list( + "truncated_hour", 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"], "UTC") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + @override_settings(USE_TZ=False) + def test_extract_and_trunc_use_tz_false_with_explicit_timezone(self): + # Under USE_TZ=False, Django ORM's get_tzname() returns None even if tzinfo + # is specified on the expression. Our backend then defaults the timezone parameter to 'UTC'. + fixed_offset_timezone = timezone.get_fixed_timezone(-300) + sql_extract = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql_extract, "extracted_hour", TypeCode.INT64, [("12",)]) + results_extract = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time", tzinfo=fixed_offset_timezone) + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results_extract, [12]) + execute_request_extract = self.get_last_execute_sql_request() + self.assertEqual(execute_request_extract.sql, sql_extract) + self.assertEqual(execute_request_extract.params["a0"], "UTC") + + sql_trunc = ( + "SELECT TIMESTAMP_TRUNC(tests_event.start_time, hour, @a0) " + "AS truncated_hour FROM tests_event" + ) + add_single_result( + sql_trunc, "truncated_hour", TypeCode.TIMESTAMP, [("2026-11-10T12:00:00Z",)] + ) + results_trunc = list( + Event.objects.annotate( + truncated_hour=TruncHour("start_time", tzinfo=fixed_offset_timezone) + ).values_list("truncated_hour", flat=True) + ) + self.assertEqual(len(results_trunc), 1) + execute_request_trunc = self.get_last_execute_sql_request() + self.assertEqual(execute_request_trunc.sql, sql_trunc) + self.assertEqual(execute_request_trunc.params["a0"], "UTC") diff --git a/packages/django-google-spanner/tests/system/django_spanner/models.py b/packages/django-google-spanner/tests/system/django_spanner/models.py index ac7b523fb1b9..0d228eb2441a 100644 --- a/packages/django-google-spanner/tests/system/django_spanner/models.py +++ b/packages/django-google-spanner/tests/system/django_spanner/models.py @@ -27,6 +27,8 @@ def __str__(self): class Event(models.Model): start_date = models.DateTimeField() end_date = models.DateTimeField() + event_date = models.DateField(null=True) + event_time = models.TimeField(null=True) class Meta: constraints = [ diff --git a/packages/django-google-spanner/tests/system/django_spanner/test_datetime_expressions.py b/packages/django-google-spanner/tests/system/django_spanner/test_datetime_expressions.py new file mode 100644 index 000000000000..dd0ca1adaeaf --- /dev/null +++ b/packages/django-google-spanner/tests/system/django_spanner/test_datetime_expressions.py @@ -0,0 +1,267 @@ +# Copyright 2026 Google LLC +# +# Use of this source code is governed by a BSD-style +# license that can be found in the LICENSE file or at +# https://developers.google.com/open-source/licenses/bsd + +import zoneinfo +from datetime import date, datetime, time, timezone + +from django.db import connection +from django.db.models.functions import ( + ExtractDay, + ExtractHour, + ExtractMinute, + ExtractMonth, + ExtractSecond, + ExtractYear, + TruncDate, + TruncDay, + TruncHour, + TruncMonth, + TruncTime, + TruncYear, +) +from django.test import TransactionTestCase, override_settings + +from .models import Event +from .utils import ( + setup_database, + setup_instance, + teardown_database, + teardown_instance, +) + + +class TestDatetimeExpressions(TransactionTestCase): + @classmethod + def setUpClass(cls): + setup_instance() + setup_database() + with connection.schema_editor() as editor: + editor.create_model(Event) + + @classmethod + def tearDownClass(cls): + with connection.schema_editor() as editor: + editor.delete_model(Event) + teardown_database() + teardown_instance() + + def setUp(self): + super().setUp() + Event.objects.all().delete() + + def test_extract_and_trunc_with_timezone(self): + new_york_timezone = zoneinfo.ZoneInfo("America/New_York") + # 2026-01-01 03:00:00 UTC is 2025-12-31 22:00:00 in America/New_York (UTC-5) + event = Event.objects.create( + start_date=datetime(2026, 1, 1, 3, 0, tzinfo=timezone.utc), + end_date=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc), + ) + + # 1. Extract hour in America/New_York (should be 22) + ny_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(hour=ExtractHour("start_date", tzinfo=new_york_timezone)) + .values_list("hour", flat=True) + .get() + ) + self.assertEqual(ny_hour, 22) + + # 2. Extract day in America/New_York (should be 31) + ny_day = ( + Event.objects.filter(pk=event.pk) + .annotate(day=ExtractDay("start_date", tzinfo=new_york_timezone)) + .values_list("day", flat=True) + .get() + ) + self.assertEqual(ny_day, 31) + + # 3. Extract in UTC (should be 3) + utc_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(hour=ExtractHour("start_date", tzinfo=timezone.utc)) + .values_list("hour", flat=True) + .get() + ) + self.assertEqual(utc_hour, 3) + + # 4. TruncDate in America/New_York (should be 2025-12-31) + ny_date = ( + Event.objects.filter(pk=event.pk) + .annotate(date=TruncDate("start_date", tzinfo=new_york_timezone)) + .values_list("date", flat=True) + .get() + ) + self.assertEqual(ny_date, date(2025, 12, 31)) + + # 5. TruncDay in America/New_York + ny_day_trunc = ( + Event.objects.filter(pk=event.pk) + .annotate(day_trunc=TruncDay("start_date", tzinfo=new_york_timezone)) + .values_list("day_trunc", flat=True) + .get() + ) + self.assertEqual( + ny_day_trunc, datetime(2025, 12, 31, 5, 0, tzinfo=new_york_timezone) + ) + + # 6. TruncTime in America/New_York (should be 22:00:00) + ny_time = ( + Event.objects.filter(pk=event.pk) + .annotate(time=TruncTime("start_date", tzinfo=new_york_timezone)) + .values_list("time", flat=True) + .get() + ) + self.assertEqual(str(ny_time), "22:00:00") + + def test_extract_and_trunc_without_timezone(self): + event = Event.objects.create( + start_date=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc), + end_date=datetime(2026, 1, 1, 15, 0, tzinfo=timezone.utc), + ) + + # Under USE_TZ=True, omitting tzinfo causes Django to use the current + # active timezone (settings.TIME_ZONE, which defaults to UTC). + extracted_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(hour=ExtractHour("start_date")) + .values_list("hour", flat=True) + .get() + ) + self.assertEqual(extracted_hour, 12) + + truncated_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncHour("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual( + truncated_hour, datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc) + ) + + truncated_date = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncDate("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual(truncated_date, date(2026, 1, 1)) + + truncated_time = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncTime("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual(truncated_time, time(12, 0, 0)) + + @override_settings(USE_TZ=False) + def test_extract_and_trunc_use_tz_false(self): + event = Event.objects.create( + start_date=datetime(2026, 1, 1, 12, 0), + end_date=datetime(2026, 1, 1, 15, 0), + ) + + # When USE_TZ is False, operations default to UTC timezone arguments so that + # naive timestamps (stored on the UTC timeline) do not drift based on the + # database's default timezone (e.g. America/Los_Angeles). + extracted_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(hour=ExtractHour("start_date")) + .values_list("hour", flat=True) + .get() + ) + self.assertEqual(extracted_hour, 12) + + truncated_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncHour("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual(truncated_hour, datetime(2026, 1, 1, 12, 0)) + + truncated_date = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncDate("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual(truncated_date, date(2026, 1, 1)) + + truncated_time = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncTime("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual(truncated_time, time(12, 0, 0)) + + def test_date_field_roundtrip_and_expressions(self): + event = Event.objects.create( + start_date=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc), + end_date=datetime(2026, 1, 1, 15, 0, tzinfo=timezone.utc), + event_date=date(2026, 5, 15), + ) + refetched = Event.objects.get(pk=event.pk) + self.assertEqual(refetched.event_date, date(2026, 5, 15)) + + extracted = ( + Event.objects.filter(pk=event.pk) + .annotate( + year=ExtractYear("event_date"), + month=ExtractMonth("event_date"), + day=ExtractDay("event_date"), + ) + .values("year", "month", "day") + .get() + ) + self.assertEqual(extracted["year"], 2026) + self.assertEqual(extracted["month"], 5) + self.assertEqual(extracted["day"], 15) + + truncated = ( + Event.objects.filter(pk=event.pk) + .annotate( + month=TruncMonth("event_date"), + year=TruncYear("event_date"), + ) + .values("month", "year") + .get() + ) + self.assertEqual(truncated["month"], date(2026, 5, 1)) + self.assertEqual(truncated["year"], date(2026, 1, 1)) + + def test_time_field_roundtrip_and_expressions(self): + event = Event.objects.create( + start_date=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc), + end_date=datetime(2026, 1, 1, 15, 0, tzinfo=timezone.utc), + event_time=time(14, 30, 45), + ) + refetched = Event.objects.get(pk=event.pk) + self.assertEqual(refetched.event_time, time(14, 30, 45)) + + extracted = ( + Event.objects.filter(pk=event.pk) + .annotate( + hour=ExtractHour("event_time"), + minute=ExtractMinute("event_time"), + second=ExtractSecond("event_time"), + ) + .values("hour", "minute", "second") + .get() + ) + self.assertEqual(extracted["hour"], 14) + self.assertEqual(extracted["minute"], 30) + self.assertEqual(extracted["second"], 45) + + truncated_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(hour=TruncHour("event_time")) + .values_list("hour", flat=True) + .get() + ) + self.assertEqual(truncated_hour, time(14, 0, 0)) 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..20f75d0d13bf 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 @@ -10,9 +10,9 @@ from decimal import Decimal from unittest import mock -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 django.test import override_settings from google.cloud.spanner_dbapi.types import DateStr from tests.unit.django_spanner.simple_test import SpannerSimpleTestClass @@ -114,81 +114,130 @@ def test_convert_uuidfield_value_none(self): def test_date_extract_sql(self): self.assertEqual( self.db_operations.date_extract_sql("week", "dummy_field"), - ("EXTRACT(isoweek FROM dummy_field)", None), + ("EXTRACT(isoweek FROM dummy_field)", ()), ) def test_date_extract_sql_lookup_type_dayofweek(self): self.assertEqual( self.db_operations.date_extract_sql("dayofweek", "dummy_field"), - ("EXTRACT(dayofweek FROM dummy_field)", None), + ("EXTRACT(dayofweek FROM dummy_field)", ()), ) + @override_settings(USE_TZ=True) def test_datetime_extract_sql(self): - settings.USE_TZ = True self.assertEqual( self.db_operations.datetime_extract_sql( "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",), + ), + ) + self.assertEqual( + self.db_operations.datetime_extract_sql( + "dayofweek", "dummy_field", None, None + ), + ( + "EXTRACT(dayofweek FROM dummy_field)", + (), ), ) + @override_settings(USE_TZ=False) def test_datetime_extract_sql_use_tz_false(self): - settings.USE_TZ = False self.assertEqual( self.db_operations.datetime_extract_sql( "dayofweek", "dummy_field", None, "IST" ), ( - 'EXTRACT(dayofweek FROM dummy_field AT TIME ZONE "UTC")', - None, + "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)", + ("IST",), + ), + ) + self.assertEqual( + self.db_operations.datetime_extract_sql( + "dayofweek", "dummy_field", None, None + ), + ( + "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)", + ("UTC",), ), ) - settings.USE_TZ = True # reset changes. 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): + @override_settings(USE_TZ=True) + def test_time_trunc_sql_use_tz_true(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",)), + ) + self.assertEqual( + self.db_operations.time_trunc_sql( + "dayofweek", "dummy_field", None, tzname="UTC" + ), + ("TIMESTAMP_TRUNC(dummy_field, dayofweek, %s)", ("UTC",)), + ) + self.assertEqual( + self.db_operations.time_trunc_sql( + "dayofweek", "dummy_field", None, tzname="America/New_York" + ), + ("TIMESTAMP_TRUNC(dummy_field, dayofweek, %s)", ("America/New_York",)), + ) + + @override_settings(USE_TZ=False) + def test_time_trunc_sql_use_tz_false(self): + self.assertEqual( + self.db_operations.time_trunc_sql("dayofweek", "dummy_field", None), + ("TIMESTAMP_TRUNC(dummy_field, dayofweek, %s)", ("UTC",)), + ) + self.assertEqual( + self.db_operations.time_trunc_sql( + "dayofweek", "dummy_field", None, tzname="America/New_York" + ), + ("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",)), ) + @override_settings(USE_TZ=True) def test_datetime_cast_time_sql(self): - settings.USE_TZ = True 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('1970-01-01 %H:%M:%E9S', dummy_field, %s), 'UTC')", + ("IST",), ), ) + @override_settings(USE_TZ=False) def test_datetime_cast_time_sql_use_tz_false(self): - settings.USE_TZ = False 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('1970-01-01 %H:%M:%E9S', dummy_field, %s), 'UTC')", + ("IST",), + ), + ) + self.assertEqual( + self.db_operations.datetime_cast_time_sql("dummy_field", None, None), + ( + "TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', dummy_field, %s), 'UTC')", + ("UTC",), ), ) - settings.USE_TZ = True # reset changes. def test_date_interval_sql(self): self.assertEqual( @@ -324,26 +373,343 @@ def test_bulk_insert_sql(self): def test_date_and_time_trunc_sql(self): sql, params = self.db_operations.date_trunc_sql("year", "field", None) self.assertEqual(sql, "DATE_TRUNC(CAST(field AS DATE), year)") - self.assertIsNone(params) + self.assertEqual(params, ()) + + sql, params = self.db_operations.date_extract_sql("year", "field", None) + self.assertEqual(sql, "EXTRACT(year FROM field)") + self.assertEqual(params, ()) + + sql, params = self.db_operations.time_extract_sql("hour", "field", None) + self.assertEqual(sql, "EXTRACT(hour FROM field AT TIME ZONE %s)") + self.assertEqual(params, ("UTC",)) 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.time_trunc_sql( + "hour", "field", None, tzname="America/New_York" + ) + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, hour, %s)") + self.assertEqual(params, ("America/New_York",)) + + sql, params = self.db_operations.datetime_trunc_sql( + "day", "field", None, tzname=None + ) + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, day)") + self.assertEqual(params, ()) sql, params = self.db_operations.datetime_trunc_sql( "day", "field", None, tzname="UTC" ) - self.assertEqual(sql, 'TIMESTAMP_TRUNC(field, day, "UTC")') - self.assertIsNone(params) + 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.assertEqual(params, ()) + + sql, params = self.db_operations.datetime_trunc_sql( + "week", "field", None, tzname=None + ) + self.assertEqual( + sql, + "TIMESTAMP_ADD(TIMESTAMP_TRUNC(TIMESTAMP_SUB(field, INTERVAL 1 DAY), week), INTERVAL 1 DAY)", + ) + self.assertEqual(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",)) + @override_settings(USE_TZ=True) def test_datetime_cast_sql(self): + sql, params = self.db_operations.datetime_cast_date_sql("field", None, None) + self.assertEqual(sql, "DATE(field)") + self.assertEqual(params, ()) + 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, None) + self.assertEqual( + sql, + "TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', field), 'UTC')", + ) + self.assertEqual(params, ()) 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('1970-01-01 %H:%M:%E9S', field, %s), 'UTC')", + ) + self.assertEqual(params, ("UTC",)) + + @override_settings(USE_TZ=False) + def test_datetime_trunc_and_cast_sql_use_tz_false(self): + sql, params = self.db_operations.datetime_trunc_sql( + "day", "field", None, tzname="America/New_York" + ) + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, day, %s)") + self.assertEqual(params, ("America/New_York",)) + + sql, params = self.db_operations.datetime_trunc_sql( + "day", "field", None, tzname=None + ) + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, day, %s)") + self.assertEqual(params, ("UTC",)) + + sql, params = self.db_operations.datetime_cast_date_sql( + "field", None, tzname="America/New_York" + ) + self.assertEqual(sql, "DATE(field, %s)") + self.assertEqual(params, ("America/New_York",)) + + sql, params = self.db_operations.datetime_cast_date_sql( + "field", None, tzname=None + ) + self.assertEqual(sql, "DATE(field, %s)") + self.assertEqual(params, ("UTC",)) + + sql, params = self.db_operations.datetime_cast_time_sql( + "field", None, tzname="America/New_York" + ) + self.assertEqual( + sql, + "TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', field, %s), 'UTC')", + ) + self.assertEqual(params, ("America/New_York",)) + + sql, params = self.db_operations.datetime_cast_time_sql( + "field", None, tzname=None + ) + self.assertEqual( + sql, + "TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', field, %s), 'UTC')", + ) + self.assertEqual(params, ("UTC",)) + + @override_settings(USE_TZ=True) + def test_get_timezone_param_use_tz_true(self): + self.assertEqual( + self.db_operations._get_timezone_param("America/New_York"), + "America/New_York", + ) + self.assertEqual( + self.db_operations._get_timezone_param("+05:30"), + "+05:30", + ) + self.assertIsNone(self.db_operations._get_timezone_param(None)) + + @override_settings(USE_TZ=False) + def test_get_timezone_param_use_tz_false(self): + self.assertEqual( + self.db_operations._get_timezone_param("America/New_York"), + "America/New_York", + ) + self.assertEqual( + self.db_operations._get_timezone_param("-08:00"), + "-08:00", + ) + self.assertEqual( + self.db_operations._get_timezone_param(None), + "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"), "-0443") + self.assertEqual(self.db_operations._prepare_tzname_delta("+0500"), "+0500") + 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"), "-0443") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC+0500"), "+0500") + self.assertEqual(self.db_operations._prepare_tzname_delta("+05"), "+05") + self.assertEqual(self.db_operations._prepare_tzname_delta("-08"), "-08") + self.assertEqual(self.db_operations._prepare_tzname_delta("+5"), "+5") + self.assertEqual(self.db_operations._prepare_tzname_delta("-8"), "-8") + self.assertEqual(self.db_operations._prepare_tzname_delta("+5:30"), "+5:30") + self.assertEqual(self.db_operations._prepare_tzname_delta("-8:00"), "-8:00") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC+5"), "+5") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC-8"), "-8") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC+5:30"), "+5:30") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC-8:00"), "-8:00") + self.assertEqual( + self.db_operations._prepare_tzname_delta("America/New_York"), + "America/New_York", + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("Etc/GMT+5"), + "Etc/GMT+5", + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("Etc/GMT-5"), + "Etc/GMT-5", + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("Etc/GMT+10"), + "Etc/GMT+10", + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("Etc/GMT-10"), + "Etc/GMT-10", + ) + 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") + self.assertEqual( + self.db_operations._prepare_tzname_delta("+invalid"), "+invalid" + ) + + 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) + + @override_settings(USE_TZ=True) + 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")) + + @override_settings(USE_TZ=True) + def test_tzname_sql_injection_defense(self): + 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.assertIsNone(params) + self.assertNotIn(malicious_tzname, sql) + self.assertIn("%s", sql) + self.assertEqual(params, (malicious_tzname,))