fix null pointer reference segfault (#636)

Author: mensfeld

CHANGELOG.md

@@ -5,6 +5,7 @@
 - **[Feature]** Add precompiled `x86_64-linux-musl` setup.
 - **[Feature]** Add precompiled `macos_arm64` setup.
 - [Fix] Fix a case where using empty key on the `musl` architecture would cause a segfault.
+- [Fix] Fix for null pointer reference bypass on empty string being too wide causing segfault.
 - [Enhancement] Allow for producing to non-existing topics with `key` and `partition_key` present.
 - [Enhancement] Replace TTL-based partition count cache with a global cache that reuses `librdkafka` statistics data when possible.
 - [Enhancement] Support producing and consuming of headers with mulitple values (KIP-82).

lib/rdkafka/bindings.rb

@@ -384,18 +384,16 @@ class NativeErrorDesc < FFI::Struct
       hsh[name] = method_name
     end
 
-    def self.partitioner(str, partition_count, partitioner_name = "consistent_random")
+    def self.partitioner(topic_ptr, str, partition_count, partitioner = "consistent_random")
       # Return RD_KAFKA_PARTITION_UA(unassigned partition) when partition count is nil/zero.
       return -1 unless partition_count&.nonzero?
-      # musl architecture crashes with empty string
-      return 0 if str.empty?
 
-      str_ptr = FFI::MemoryPointer.from_string(str)
-      method_name = PARTITIONERS.fetch(partitioner_name) do
-        raise Rdkafka::Config::ConfigError.new("Unknown partitioner: #{partitioner_name}")
+      str_ptr = str.empty? ? FFI::MemoryPointer::NULL : FFI::MemoryPointer.from_string(str)
+      method_name = PARTITIONERS.fetch(partitioner) do
+        raise Rdkafka::Config::ConfigError.new("Unknown partitioner: #{partitioner}")
       end
 
-      public_send(method_name, nil, str_ptr, str.size, partition_count, nil, nil)
+      public_send(method_name, topic_ptr, str_ptr, str.size, partition_count, nil, nil)
     end
 
     # Create Topics

lib/rdkafka/producer.rb

@@ -51,13 +51,13 @@ class TopicHandleCreationError < RuntimeError; end
 
     # @private
     # @param native_kafka [NativeKafka]
-    # @param partitioner_name [String, nil] name of the partitioner we want to use or nil to use
+    # @param partitioner [String, nil] name of the partitioner we want to use or nil to use
     #   the "consistent_random" default
-    def initialize(native_kafka, partitioner_name)
+    def initialize(native_kafka, partitioner)
       @topics_refs_map = {}
       @topics_configs = {}
       @native_kafka = native_kafka
-      @partitioner_name = partitioner_name || "consistent_random"
+      @partitioner = partitioner || "consistent_random"
 
       # Makes sure, that native kafka gets closed before it gets GCed by Ruby
       ObjectSpace.define_finalizer(self, native_kafka.finalizer)
@@ -275,7 +275,8 @@ def produce(
       timestamp: nil,
       headers: nil,
       label: nil,
-      topic_config: EMPTY_HASH
+      topic_config: EMPTY_HASH,
+      partitioner: @partitioner
     )
       closed_producer_check(__method__)
 
@@ -307,10 +308,14 @@ def produce(
 
         # Check if there are no overrides for the partitioner and use the default one only when
         # no per-topic is present.
-        partitioner_name = @topics_configs.dig(topic, topic_config_hash, :partitioner) || @partitioner_name
+        selected_partitioner = @topics_configs.dig(topic, topic_config_hash, :partitioner) || partitioner
 
         # If the topic is not present, set to -1
-        partition = Rdkafka::Bindings.partitioner(partition_key, partition_count, partitioner_name) if partition_count.positive?
+        partition = Rdkafka::Bindings.partitioner(
+          topic_ref,
+          partition_key,
+          partition_count,
+          selected_partitioner) if partition_count.positive?
       end
 
       # If partition is nil, use -1 to let librdafka set the partition randomly or

spec/rdkafka/bindings_spec.rb

@@ -77,30 +77,6 @@
     end
   end
 
-  describe "partitioner" do
-    let(:partition_key) { ('a'..'z').to_a.shuffle.take(15).join('') }
-    let(:partition_count) { rand(50) + 1 }
-
-    it "should return the same partition for a similar string and the same partition count" do
-      result_1 = Rdkafka::Bindings.partitioner(partition_key, partition_count)
-      result_2 = Rdkafka::Bindings.partitioner(partition_key, partition_count)
-      expect(result_1).to eq(result_2)
-    end
-
-    it "should match the old partitioner" do
-      result_1 = Rdkafka::Bindings.partitioner(partition_key, partition_count)
-      result_2 = (Zlib.crc32(partition_key) % partition_count)
-      expect(result_1).to eq(result_2)
-    end
-
-    it "should return the partition calculated by the specified partitioner" do
-      result_1 = Rdkafka::Bindings.partitioner(partition_key, partition_count, "murmur2")
-      ptr = FFI::MemoryPointer.from_string(partition_key)
-      result_2 = Rdkafka::Bindings.rd_kafka_msg_partitioner_murmur2(nil, ptr, partition_key.size, partition_count, nil, nil)
-      expect(result_1).to eq(result_2)
-    end
-  end
-
   describe "stats callback" do
     context "without a stats callback" do
       it "should do nothing" do