Skip to content

Security: Replace optimization-sensitive assertions with explicit Value/Type errors #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
magqqgq wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Security: Replace optimization-sensitive assertions with explicit Value/Type errors #3

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 25 additions & 12 deletions kafka/producer/kafka.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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)
Expand Down Expand Up @@ -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))
Expand Down