fix(django-spanner): prevent SQL injection in date and time functions

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",

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.

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()