From 6156f963a77a62198b5811d78ac275b1627fcad7 Mon Sep 17 00:00:00 2001 From: magqqgq <146786427+magqqgq@users.noreply.github.com> Date: Mon, 17 Aug 2026 14:51:17 +0300 Subject: [PATCH] Security: Replace optimization-sensitive assertions with explicit Value/Type errors Why The producer input validation logic previously relied on assert statements to enforce valid message types, null checks, and partition constraints. Because assert statements are stripped when Python is executed with the -O (optimize) flag, this created a security and stability risk where invalid or malformed data could bypass validation. How Replaced optimization-sensitive assert checks in KafkaProducer.send and KafkaProducer._partition with explicit if conditions that raise ValueError and TypeError exceptions. Security/Robustness changes Guarantees that strict input validation (null message constraints, correct bytes serialization, valid partition bounds, and strict header formatting) is always enforced, regardless of the runtime optimization level. Testing Verified that the modified Python file compiles cleanly using py_compile. (Note: Legacy test collection is blocked by an environment limitation with the installed pytest version, which rejects a legacy fixture call, rather than a source failure). --- kafka/producer/kafka.py | 37 +++++++++++++++++++++++++------------ 1 file changed, 25 insertions(+), 12 deletions(-) diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py index 7878c0a57..d5c1c34c0 100644 --- a/kafka/producer/kafka.py +++ b/kafka/producer/kafka.py @@ -546,9 +546,10 @@ def send(self, topic, value=None, key=None, headers=None, partition=None, timest KafkaTimeoutError: if unable to fetch topic metadata, or unable to obtain memory buffer prior to configured max_block_ms """ - assert value is not None or self.config['api_version'] >= (0, 8, 1), ( - 'Null messages require kafka >= 0.8.1') - assert not (value is None and key is None), 'Need at least one: key or value' + if value is None and self.config['api_version'] < (0, 8, 1): + raise ValueError('Null messages require kafka >= 0.8.1') + if value is None and key is None: + raise ValueError('Need at least one: key or value') key_bytes = value_bytes = None try: self._wait_on_metadata(topic, self.config['max_block_ms'] / 1000.0) @@ -559,18 +560,28 @@ def send(self, topic, value=None, key=None, headers=None, partition=None, timest value_bytes = self._serialize( self.config['value_serializer'], topic, value) - assert type(key_bytes) in (bytes, bytearray, memoryview, type(None)) - assert type(value_bytes) in (bytes, bytearray, memoryview, type(None)) + valid_types = (bytes, bytearray, memoryview, type(None)) + if type(key_bytes) not in valid_types: + raise TypeError('key_serializer must return bytes-like data or None') + if type(value_bytes) not in valid_types: + raise TypeError('value_serializer must return bytes-like data or None') partition = self._partition(topic, partition, key, value, key_bytes, value_bytes) - if headers is None: headers = [] - assert type(headers) == list - assert all(type(item) == tuple and len(item) == 2 and type(item[0]) == str and type(item[1]) == bytes for item in headers) - - message_size = self._estimate_size_in_bytes(key_bytes, value_bytes, headers) + if type(headers) is not list: + raise TypeError('headers must be a list of (str, bytes) tuples') + if not all( + type(item) is tuple + and len(item) == 2 + and type(item[0]) is str + and type(item[1]) is bytes + for item in headers + ): + raise TypeError('headers must be a list of (str, bytes) tuples') + + message_size = self._estimate_size_in_bytes(key_bytes, value_bytes, headers) self._ensure_valid_record_size(message_size) tp = TopicPartition(topic, partition) @@ -694,8 +705,10 @@ def _serialize(self, f, topic, data): def _partition(self, topic, partition, key, value, serialized_key, serialized_value): if partition is not None: - assert partition >= 0 - assert partition in self._metadata.partitions_for_topic(topic), 'Unrecognized partition' + if partition < 0: + raise ValueError('partition must be non-negative') + if partition not in self._metadata.partitions_for_topic(topic): + raise ValueError('Unrecognized partition') return partition all_partitions = sorted(self._metadata.partitions_for_topic(topic))