fix(spanner): catch recursion and decode errors in proto parsing to p… (#16561)

This PR fixes a Persistent Stored Denial of Service (DoS) vulnerability
in the google-cloud-spanner Python SDK (Issue 479858035).

**The Problem**
When the SDK attempts to deserialize a Protobuf-encoded row (via
_parse_proto() in _helpers.py) that contains a maliciously crafted
"recursion bomb" (e.g., a ListValue nested 1,000+ times), it triggers a
DecodeError or RecursionError. This unhandled exception crashes the
consumer thread and blocks the entire result set stream ("pipeline
blackhole").

**The Solution**
We modify _parse_proto to wrap the ParseFromString() call in a defensive
try...except block:
Catch RecursionError (triggered if Python hits its stack limit first in
pure Python implementations).
Catch google.protobuf.message.DecodeError (triggered by the C++
extension's internal limits).

If an error is caught: A warning is logged. The original raw bytes_value
is returned as a fallback (consistent with existing behavior when no
prototype is found). This allows the stream iterator to continue
processing subsequent rows.

Author: sinhasubham

packages/google-cloud-spanner/google/cloud/spanner_v1/_helpers.py

@@ -28,7 +28,7 @@
 from google.api_core.exceptions import Aborted
 from google.cloud._helpers import _date_from_iso8601_date
 from google.protobuf.internal.enum_type_wrapper import EnumTypeWrapper
-from google.protobuf.message import Message
+from google.protobuf.message import DecodeError, Message
 from google.protobuf.struct_pb2 import ListValue, Value
 from google.rpc.error_details_pb2 import RetryInfo
 
@@ -76,7 +76,7 @@
 
 GOOGLE_CLOUD_REGION_GLOBAL = "global"
 
-log = logging.getLogger(__name__)
+_LOGGER = logging.getLogger(__name__)
 
 _cloud_region: str = None
 
@@ -122,7 +122,7 @@ def _get_cloud_region() -> str:
         else:
             _cloud_region = GOOGLE_CLOUD_REGION_GLOBAL
     except Exception as e:
-        log.warning(
+        _LOGGER.warning(
             "Failed to detect GCP resource location for Spanner metrics, defaulting to 'global'. Error: %s",
             e,
         )
@@ -603,8 +603,14 @@ def _parse_proto(value_pb, column_info, field_name):
         default_proto_message = column_info.get(field_name)
         if isinstance(default_proto_message, Message):
             proto_message = type(default_proto_message)()
-            proto_message.ParseFromString(bytes_value)
-            return proto_message
+            try:
+                proto_message.ParseFromString(bytes_value)
+                return proto_message
+            except (DecodeError, RecursionError):
+                _LOGGER.warning(
+                    "Field could not be parsed as Proto due to excessive nesting/corruption. Returning raw bytes."
+                )
+                return bytes_value
     return bytes_value
 
 

packages/google-cloud-spanner/tests/unit/test__helpers.py

@@ -771,6 +771,59 @@ def test_w_proto_message(self):
             self._callFUT(value_pb, field_type, field_name, column_info), VALUE
         )
 
+    def test_w_proto_message_decode_error(self):
+        import base64
+        from unittest import mock
+
+        from google.protobuf.message import DecodeError
+        from google.protobuf.struct_pb2 import Value
+
+        from google.cloud.spanner_v1 import Type, TypeCode
+
+        from .testdata import singer_pb2
+
+        VALUE = singer_pb2.SingerInfo(singer_id=1, nationality="Canadian")
+        field_type = Type(code=TypeCode.PROTO)
+        field_name = "proto_message_column"
+        raw_bytes = VALUE.SerializeToString()
+        value_pb = Value(string_value=base64.b64encode(raw_bytes).decode("utf-8"))
+        column_info = {"proto_message_column": singer_pb2.SingerInfo()}
+
+        # Mock ParseFromString to raise DecodeError
+        with mock.patch.object(
+            singer_pb2.SingerInfo,
+            "ParseFromString",
+            side_effect=DecodeError("Mock Decode Error"),
+        ):
+            result = self._callFUT(value_pb, field_type, field_name, column_info)
+            # Should return raw bytes
+            self.assertEqual(result, raw_bytes)
+
+    def test_w_proto_message_recursion_error(self):
+        import base64
+        from unittest import mock
+
+        from google.protobuf.struct_pb2 import Value
+
+        from google.cloud.spanner_v1 import Type, TypeCode
+
+        from .testdata import singer_pb2
+
+        VALUE = singer_pb2.SingerInfo(singer_id=1, nationality="Canadian")
+        field_type = Type(code=TypeCode.PROTO)
+        field_name = "proto_message_column"
+        raw_bytes = VALUE.SerializeToString()
+        value_pb = Value(string_value=base64.b64encode(raw_bytes).decode("utf-8"))
+        column_info = {"proto_message_column": singer_pb2.SingerInfo()}
+
+        with mock.patch.object(
+            singer_pb2.SingerInfo,
+            "ParseFromString",
+            side_effect=RecursionError("Mock Recursion Error"),
+        ):
+            result = self._callFUT(value_pb, field_type, field_name, column_info)
+            self.assertEqual(result, raw_bytes)
+
     def test_w_proto_enum(self):
         from google.protobuf.struct_pb2 import Value