Send bytes-like data verbatim in SSLSocket.write (F-5622)

write() converted data with t2b(), which str()-encodes anything that is
not already bytes. Valid bytes-like inputs such as bytearray and
memoryview were transmitted as their Python repr ("bytearray(b'...')",
"<memory at ...>") instead of their contents, corrupting the stream.
Convert via the buffer protocol (bytes(memoryview(data))) and raise
TypeError for objects that are not bytes-like, matching the stdlib ssl
module.
pull/70/head
Juliusz Sosinowicz 2026-06-23 12:49:01 +00:00
parent 9c26572a41
commit 3dd1b902e1
2 changed files with 92 additions and 2 deletions

View File

@ -0,0 +1,84 @@
# -*- coding: utf-8 -*-
#
# test_write_bytes.py
#
# Copyright (C) 2006-2020 wolfSSL Inc.
#
# This file is part of wolfSSL. (formerly known as CyaSSL)
#
# wolfSSL is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# wolfSSL is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
# pylint: disable=missing-docstring, invalid-name, import-error
# pylint: disable=protected-access
"""
F-5622: SSLSocket.write() ran data through t2b(), which str()-encodes anything
that is not already bytes. Valid bytes-like inputs (bytearray, memoryview)
were therefore serialized as their Python repr ("bytearray(b'...')",
"<memory at 0x...>") instead of their actual contents.
"""
from types import SimpleNamespace
import wolfssl
class _CaptureLib:
def __init__(self):
self.written = None
def wolfSSL_write(self, ssl, data, length):
self.written = bytes(data[:length])
return length
def wolfSSL_get_error(self, ssl, ret): # pragma: no cover
return 0
def _make_socket(monkeypatch):
lib = _CaptureLib()
monkeypatch.setattr(wolfssl, "_lib", lib)
sock = wolfssl.SSLSocket.__new__(wolfssl.SSLSocket)
sock.native_object = object()
sock._connected = True
sock._context = SimpleNamespace(protocol=wolfssl.PROTOCOL_TLS)
sock._release_native_object = lambda: None
return sock, lib
def test_write_bytes_unchanged(monkeypatch):
sock, lib = _make_socket(monkeypatch)
sock.write(b"hello")
assert lib.written == b"hello"
def test_write_bytearray_sends_contents(monkeypatch):
sock, lib = _make_socket(monkeypatch)
sock.write(bytearray(b"hello"))
assert lib.written == b"hello"
def test_write_memoryview_sends_contents(monkeypatch):
sock, lib = _make_socket(monkeypatch)
sock.write(memoryview(b"hello"))
assert lib.written == b"hello"
def test_write_str_is_utf8_encoded(monkeypatch):
# Backward compatibility: str is UTF-8 encoded (historical t2b()
# behavior), not rejected.
sock, lib = _make_socket(monkeypatch)
sock.write("héllo")
assert lib.written == "héllo".encode("utf-8")

View File

@ -577,14 +577,20 @@ class SSLSocket(object):
Returns number of bytes of DATA actually transmitted.
"""
self._check_closed("write")
# Check connected if not DTLS
# Check connected if not DTLS
if self._context.protocol < PROTOCOL_DTLSv1:
self._check_connected()
# Drive the DTLS handshake only until it has completed.
elif not self._handshake_complete:
self.do_handshake()
data = t2b(data)
# Send bytes-like objects verbatim; fall back to t2b() for other
# types (e.g. str) to preserve backward compatibility.
if not isinstance(data, bytes):
try:
data = bytes(memoryview(data))
except TypeError:
data = t2b(data)
ret = _lib.wolfSSL_write(
self.native_object, data, len(data))