GH-49155: [C++][IPC] Allow disabling extension type deserialization

Add extension_types_blocked option to IpcReadOptions to allow users to disable extension type deserialization for security/robustness.

When enabled, extension types are returned as their storage types instead of calling custom deserialization code.

Author: Alirana2829

cpp/src/arrow/ipc/metadata_internal.cc

@@ -851,7 +851,9 @@ class FieldToFlatbufferVisitor {
 };
 
 Status FieldFromFlatbuffer(const flatbuf::Field* field, FieldPosition field_pos,
-                           DictionaryMemo* dictionary_memo, std::shared_ptr<Field>* out) {
+                           DictionaryMemo* dictionary_memo,
+                           const IpcReadOptions* options,
+                           std::shared_ptr<Field>* out) {
   std::shared_ptr<DataType> type;
 
   std::shared_ptr<KeyValueMetadata> metadata;
@@ -866,7 +868,7 @@ Status FieldFromFlatbuffer(const flatbuf::Field* field, FieldPosition field_pos,
     child_fields.resize(children->size());
     for (int i = 0; i < static_cast<int>(children->size()); ++i) {
       RETURN_NOT_OK(FieldFromFlatbuffer(children->Get(i), field_pos.child(i),
-                                        dictionary_memo, &child_fields[i]));
+                                        dictionary_memo, options, &child_fields[i]));
     }
   }
 
@@ -899,21 +901,26 @@ Status FieldFromFlatbuffer(const flatbuf::Field* field, FieldPosition field_pos,
     // Look for extension metadata in custom_metadata field
     int name_index = metadata->FindKey(kExtensionTypeKeyName);
     if (name_index != -1) {
-      std::shared_ptr<ExtensionType> ext_type =
-          GetExtensionType(metadata->value(name_index));
-      if (ext_type != nullptr) {
-        int data_index = metadata->FindKey(kExtensionMetadataKeyName);
-        std::string type_data = data_index == -1 ? "" : metadata->value(data_index);
-
-        ARROW_ASSIGN_OR_RAISE(type, ext_type->Deserialize(type, type_data));
-        // Remove the metadata, for faithful roundtripping
-        if (data_index != -1) {
-          RETURN_NOT_OK(metadata->DeleteMany({name_index, data_index}));
-        } else {
-          RETURN_NOT_OK(metadata->Delete(name_index));
+      // Check if extension types are blocked
+      bool should_deserialize = (options == nullptr || !options->extension_types_blocked);
+      
+      if (should_deserialize) {
+        std::shared_ptr<ExtensionType> ext_type =
+            GetExtensionType(metadata->value(name_index));
+        if (ext_type != nullptr) {
+          int data_index = metadata->FindKey(kExtensionMetadataKeyName);
+          std::string type_data = data_index == -1 ? "" : metadata->value(data_index);
+
+          ARROW_ASSIGN_OR_RAISE(type, ext_type->Deserialize(type, type_data));
+          // Remove the metadata, for faithful roundtripping
+          if (data_index != -1) {
+            RETURN_NOT_OK(metadata->DeleteMany({name_index, data_index}));
+          } else {
+            RETURN_NOT_OK(metadata->Delete(name_index));
+          }
         }
       }
-      // NOTE: if extension type is unknown, we do not raise here and
+      // NOTE: if extension type is unknown or blocked, we do not raise here and
       // simply return the storage type.
     }
   }
@@ -933,6 +940,13 @@ Status FieldFromFlatbuffer(const flatbuf::Field* field, FieldPosition field_pos,
   return Status::OK();
 }
 
+// Backward-compatible overload without options
+Status FieldFromFlatbuffer(const flatbuf::Field* field, FieldPosition field_pos,
+                           DictionaryMemo* dictionary_memo, std::shared_ptr<Field>* out) {
+  return FieldFromFlatbuffer(field, field_pos, dictionary_memo, nullptr, out);
+}
+
+
 flatbuffers::Offset<KVVector> SerializeCustomMetadata(
     FBB& fbb, const std::shared_ptr<const KeyValueMetadata>& metadata) {
   std::vector<KeyValueOffset> key_values;
@@ -1433,7 +1447,7 @@ Status WriteFileFooter(const Schema& schema, const std::vector<FileBlock>& dicti
 // ----------------------------------------------------------------------
 
 Status GetSchema(const void* opaque_schema, DictionaryMemo* dictionary_memo,
-                 std::shared_ptr<Schema>* out) {
+                 const IpcReadOptions* options, std::shared_ptr<Schema>* out) {
   auto schema = static_cast<const flatbuf::Schema*>(opaque_schema);
   CHECK_FLATBUFFERS_NOT_NULL(schema, "schema");
   CHECK_FLATBUFFERS_NOT_NULL(schema->fields(), "Schema.fields");
@@ -1447,7 +1461,7 @@ Status GetSchema(const void* opaque_schema, DictionaryMemo* dictionary_memo,
     // XXX I don't think this check is necessary (AP)
     CHECK_FLATBUFFERS_NOT_NULL(field, "DictionaryEncoding.indexType");
     RETURN_NOT_OK(
-        FieldFromFlatbuffer(field, field_pos.child(i), dictionary_memo, &fields[i]));
+        FieldFromFlatbuffer(field, field_pos.child(i), dictionary_memo, options, &fields[i]));
   }
 
   std::shared_ptr<KeyValueMetadata> metadata;
@@ -1460,6 +1474,12 @@ Status GetSchema(const void* opaque_schema, DictionaryMemo* dictionary_memo,
   return Status::OK();
 }
 
+// Backward-compatible overload
+Status GetSchema(const void* opaque_schema, DictionaryMemo* dictionary_memo,
+                 std::shared_ptr<Schema>* out) {
+  return GetSchema(opaque_schema, dictionary_memo, nullptr, out);
+}
+
 Status GetTensorMetadata(const Buffer& metadata, std::shared_ptr<DataType>* type,
                          std::vector<int64_t>* shape, std::vector<int64_t>* strides,
                          std::vector<std::string>* dim_names) {

cpp/src/arrow/ipc/options.h

@@ -189,6 +189,16 @@ struct ARROW_EXPORT IpcReadOptions {
   /// The lazy property will always be reset to true to deliver the expected behavior
   io::CacheOptions pre_buffer_cache_options = io::CacheOptions::LazyDefaults();
 
+  /// \brief Whether to disable deserialization of extension types
+  ///
+  /// If true, extension types will be deserialized as their storage types instead
+  /// of calling custom deserialization code. This can be useful for security-sensitive
+  /// applications that want to avoid potentially buggy third-party extension type
+  /// deserialization code.
+  ///
+  /// Default is false (extension types are deserialized normally).
+  bool extension_types_blocked = false;
+
   static IpcReadOptions Defaults();
 };
 

cpp/src/arrow/ipc/reader.cc

@@ -810,7 +810,7 @@ Status UnpackSchemaMessage(const void* opaque_schema, const IpcReadOptions& opti
                            std::shared_ptr<Schema>* schema,
                            std::shared_ptr<Schema>* out_schema,
                            std::vector<bool>* field_inclusion_mask, bool* swap_endian) {
-  RETURN_NOT_OK(internal::GetSchema(opaque_schema, dictionary_memo, schema));
+  RETURN_NOT_OK(internal::GetSchema(opaque_schema, dictionary_memo, &options, schema));
 
   // If we are selecting only certain fields, populate the inclusion mask now
   // for fast lookups