CVE-2024-26134 (Python, published 2024-02-19T23:15Z)
Description: value: cbor2 provides encoding and decoding for the Concise Binary Object Representation (CBOR) (RFC 8949) serialization format. Starting in version 5.5.1 and prior to version 5.6.2, an attacker can crash a service using cbor2 to parse a CBOR binary by sending a long enough object. Version 5.6.2 contains a patch for this issue.
Severity: nan (CVSS v3 base score: )
Weakness: CWE-120 \u2014 Buffer Copy without Checking Size of Input ('Classic Buffer Overflow')
The product copies an input buffer to an output buffer without verifying that the size of the input buffer is less than the size of the output buffer, leading to a buffer overflow.
Repository: https://github.com/agronholm/cbor2
Files changed:
cbor2/_decoder.py
docs/versionhistory.rst
setup.py
source/decoder.c
source/tags.c
tests/test_decoder.py
Commit message: Fixed MemoryError when decoding large definite strings (#204)

Also fixed a return value check in `CBORTag_hash()`.

Vulnerable code:
            result = self.read(length)
            result = self.read(length).decode("utf-8", self._str_errors)
- Added support for native date encoding (bschoenmaeckers)
        optional=True,
static int
fp_read(CBORDecoderObject *self, char *buf, const Py_ssize_t size)
    char *data;
    int ret = -1;

                data = PyBytes_AS_STRING(obj);
                memcpy(buf, data, size);
                ret = 0;
            Py_DECREF(obj);
        Py_DECREF(size_obj);
decode_definite_bytestring(CBORDecoderObject *self, Py_ssize_t length)
    PyObject *ret = NULL;

    ret = PyBytes_FromStringAndSize(NULL, length);
    if (fp_read(self, PyBytes_AS_STRING(ret), length) == -1) {
        Py_DECREF(ret);
        return NULL;
    }
        ret = decode_definite_bytestring(self, (Py_ssize_t)length);
    set_shareable(self, ret);
decode_definite_string(CBORDecoderObject *self, Py_ssize_t length)
    PyObject *ret = NULL;
    char *buf;

    buf = PyMem_Malloc(length);
    if (!buf)
        return PyErr_NoMemory();

    if (fp_read(self, buf, length) == 0)
        ret = PyUnicode_DecodeUTF8(
                buf, length, PyBytes_AS_STRING(self->str_errors));
    PyMem_Free(buf);

    if (!ret)
    if (string_namespace_add(self, ret, length) == -1) {
        ret = decode_definite_string(self, (Py_ssize_t)length);
    set_shareable(self, ret);
    if (length == 1) {
            b"\xd8\x1e\x84\xff\xff\xff\xff",
            b"\xae\xae\xae\xae\xae\xae\xae\xae\xae\x01\x08\xc2\x98C\xd9\x01\x00\xd8$"
            b"\x9f\x00\x00\xae\xae\xff\xc2l\xa7\x99",
            Exception,
def test_invalid_data(impl, payload, exception, pattern) -> None:

Fixed code:
from codecs import getincrementaldecoder
incremental_utf8_decoder = getincrementaldecoder("utf-8")
            elif length <= 65536:
                result = self.read(length)
            else:
                # Read large bytestrings 65536 (2 ** 16) bytes at a time
                left = length
                buffer = bytearray()
                while left:
                    chunk_size = min(left, 65536)
                    buffer.extend(self.read(chunk_size))
                    left -= chunk_size

                result = bytes(buffer)
            if length <= 65536:
                result = self.read(length).decode("utf-8", self._str_errors)
            else:
                # Read and decode large text strings 65536 (2 ** 16) bytes at a time
                codec = incremental_utf8_decoder(self._str_errors)
                left = length
                result = ""
                while left:
                    chunk_size = min(left, 65536)
                    final = left <= chunk_size
                    result += codec.decode(self.read(chunk_size), final)
                    left -= chunk_size

- Added support for native date encoding (bschoenmaeckers)
- Fixed ``MemoryError`` when maliciously constructed bytestrings or string (declared to be absurdly
  large) are being decoded
        # optional=True,
static PyObject *
fp_read_object(CBORDecoderObject *self, const Py_ssize_t size)
    PyObject *ret = NULL;
        Py_DECREF(size_obj);
                ret = obj;
                Py_DECREF(obj);
    }
    return ret;
}


static int
fp_read(CBORDecoderObject *self, char *buf, const Py_ssize_t size)
{
    int ret = -1;
    PyObject *obj = fp_read_object(self, size);
    if (obj) {
        char *data = PyBytes_AS_STRING(obj);
        if (data) {
            memcpy(buf, data, size);
            ret = 0;
        }
        Py_DECREF(obj);
decode_definite_short_bytestring(CBORDecoderObject *self, Py_ssize_t length)
    PyObject *ret = fp_read_object(self, length);

static PyObject *
decode_definite_long_bytestring(CBORDecoderObject *self, Py_ssize_t length)
{
    PyObject *buffer = NULL;
    Py_ssize_t left = length;
    while (left) {
        Py_ssize_t chunk_length = length <= 65536 ? length : 65536;
        PyObject *chunk = fp_read_object(self, chunk_length);
        if (!chunk) {
            break;
        }

        if (!PyBytes_CheckExact(chunk)) {
            Py_DECREF(chunk);
            break;
        }

        if (buffer) {
            PyObject *new_buffer = PyByteArray_Concat(buffer, chunk);
            Py_DECREF(chunk);
            if (!new_buffer)
                break;

            if (new_buffer != buffer) {
                Py_DECREF(buffer);
                buffer = new_buffer;
            }
        } else {
            buffer = PyByteArray_FromObject(chunk);
            Py_DECREF(chunk);
            if (!buffer)
                break;
        }
        left -= chunk_length;
    }

    PyObject *ret = NULL;
    if (buffer) {
        ret = PyBytes_FromObject(buffer);
        Py_DECREF(buffer);

        if (ret && string_namespace_add(self, ret, length) == -1) {
            Py_DECREF(ret);
            ret = NULL;
        }
    }
    return ret;
}


    else if (length <= 65536)
        ret = decode_definite_short_bytestring(self, (Py_ssize_t)length);
        ret = decode_definite_long_bytestring(self, (Py_ssize_t)length);

    if (ret)
        set_shareable(self, ret);

decode_definite_short_string(CBORDecoderObject *self, Py_ssize_t length)
    PyObject *bytes_obj = fp_read_object(self, length);
    if (!bytes_obj)
    const char *bytes = PyBytes_AS_STRING(bytes_obj);
    PyObject *ret = PyUnicode_FromStringAndSize(bytes, length);
    Py_DECREF(bytes_obj);
    if (ret && string_namespace_add(self, ret, length) == -1) {
static PyObject *
decode_definite_long_string(CBORDecoderObject *self, Py_ssize_t length)
{
    PyObject *ret = NULL, *chunk = NULL, *string = NULL;
    Py_ssize_t left = length;
    Py_ssize_t consumed;
    Py_ssize_t buffer_size = 0;  // how many bytes are allocated for the buffer
    Py_ssize_t buffer_length = 0;  // how many bytes are actually stored in the buffer
    char *buffer = NULL;
    while (left) {
        // Read up to 65536 bytes of data from the stream
        Py_ssize_t chunk_length = 65536 - buffer_size;
        if (left < chunk_length)
            chunk_length = left;

        PyObject *chunk = fp_read_object(self, chunk_length);
        left -= chunk_length;
        if (!chunk)
            goto error;

        // Get the internal buffer of the bytes object
        char *bytes_buffer = PyBytes_AsString(chunk);
        if (!bytes_buffer)
            goto error;

        char *source_buffer;
        if (buffer) {
            // Grow the buffer to accommodate the previous data plus the new chunk
            if (buffer_length + chunk_length > buffer_size) {
                buffer_size = buffer_length + chunk_length;
                char *new_buffer = PyMem_Realloc(buffer, buffer_size);
                if (!new_buffer)
                    goto error;

                buffer = new_buffer;
            }

            // Concatenate the chunk into the buffer
            memcpy(buffer + buffer_length, bytes_buffer, chunk_length);
            buffer_length += chunk_length;

            source_buffer = buffer;
            chunk_length = buffer_length;
        } else {
            // Use the chunk's internal buffer directly to decode as many characters as possible
            source_buffer = bytes_buffer;
        }

        string = PyUnicode_DecodeUTF8Stateful(source_buffer, chunk_length, NULL, &consumed);
        if (!string)
            goto error;

        if (ret) {
            // Concatenate the result to the existing result
            PyObject *joined = PyUnicode_Concat(ret, string);
            if (!joined)
                goto error;

            Py_DECREF(string);
            string = NULL;
            ret = joined;
        } else {
            // Set the result to the decoded string
            ret = string;
        }

        Py_ssize_t unconsumed = chunk_length - consumed;
        if (consumed != chunk_length) {
            if (buffer) {
                // Move the unconsumed bytes to the start of the buffer
                memmove(buffer, buffer + consumed, unconsumed);
            } else {
                // Create a new buffer
                buffer = PyMem_Malloc(unconsumed);
                if (!buffer)
                    goto error;

                memcpy(buffer, bytes_buffer + consumed, unconsumed);
            }
            buffer_length = unconsumed;
        }
    }

    if (ret && string_namespace_add(self, ret, length) == -1)
        goto error;

    return ret;
error:
    Py_XDECREF(ret);
    Py_XDECREF(chunk);
    Py_XDECREF(string);
    if (buffer)
        PyMem_Free(buffer);

    return NULL;
}


    else if (length <= 65536)
        ret = decode_definite_short_string(self, (Py_ssize_t)length);
        ret = decode_definite_long_string(self, (Py_ssize_t)length);

    if (ret)
        set_shareable(self, ret);

    if (length == -1) {
from __future__ import annotations

from pathlib import Path
from typing import Type, cast
        pytest.param("7a00010001" + "61" * 65535 + "c3b6", "a" * 65535 + "ö", id="split_unicode"),
            unhexlify("d81e84ffffffff"),
            unhexlify("aeaeaeaeaeaeaeaeae0108c29843d90100d8249f0000aeaeffc26ca799"),
            "CBORDecodeEOF",
def test_invalid_data(
    impl, payload: bytes, exception: type[Exception] | str, pattern: str
) -> None:
    if isinstance(exception, str):
        exception = getattr(impl, exception)



@pytest.mark.parametrize(
    "payload",
    [
        pytest.param(
            unhexlify("5b7fffffffffffff00"),
            id="bytestring",
        ),
        pytest.param(
            unhexlify("7b7fffffffffffff00"),
            id="unicode",
        ),
    ],
)
def test_oversized_read(impl, payload: bytes, tmp_path: Path) -> None:
    CBORDecodeEOF = cast(Type[Exception], getattr(impl, "CBORDecodeEOF"))
    with pytest.raises(CBORDecodeEOF, match="premature end of stream"):
        dummy_path = tmp_path / "testdata"
        dummy_path.write_bytes(payload)
        with dummy_path.open("rb") as f:
            impl.load(f)