CVE-2020-26759 (Python, published 2021-01-06T13:15Z)
Description: value: clickhouse-driver before 0.1.5 allows a malicious clickhouse server to trigger a crash or execute arbitrary code (on a database client) via a crafted server response, due to a buffer overflow.
Severity: HIGH (CVSS v3 base score: 9.8)
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/mymarilyn/clickhouse-driver
Files changed:
clickhouse_driver/varint.c
clickhouse_driver/varint.pyx
tests/test_varint.py
Commit message: Fix read_varint overflow


Vulnerable code:
static CYTHON_INLINE unsigned char __Pyx_PyInt_As_unsigned_char(PyObject *);
static const char __pyx_k_towrite[] = "towrite";
static PyObject *__pyx_n_s_towrite;
static PyObject *__pyx_pf_17clickhouse_driver_6varint_write_varint(CYTHON_UNUSED PyObject *__pyx_self, Py_ssize_t __pyx_v_number, PyObject *__pyx_v_buf); /* proto */
 * def write_varint(Py_ssize_t number, buf):             # <<<<<<<<<<<<<<
  Py_ssize_t __pyx_v_number;
    __pyx_v_number = __Pyx_PyIndex_AsSsize_t(values[0]); if (unlikely((__pyx_v_number == (Py_ssize_t)-1) && PyErr_Occurred())) __PYX_ERR(0, 4, __pyx_L3_error)
static PyObject *__pyx_pf_17clickhouse_driver_6varint_write_varint(CYTHON_UNUSED PyObject *__pyx_self, Py_ssize_t __pyx_v_number, PyObject *__pyx_v_buf) {
  Py_ssize_t __pyx_v_i;
  unsigned char __pyx_v_towrite;
 *     cdef Py_ssize_t i = 0             # <<<<<<<<<<<<<<
 *     cdef unsigned char towrite
 *     # Py_ssize_t checks integer on function call and
  /* "clickhouse_driver/varint.pyx":15
 *         towrite = number & 0x7f
    /* "clickhouse_driver/varint.pyx":16
 *         towrite = number & 0x7f             # <<<<<<<<<<<<<<
    __pyx_v_towrite = (__pyx_v_number & 0x7f);
    /* "clickhouse_driver/varint.pyx":17
 *         towrite = number & 0x7f
 *             num_buf[i] = towrite | 0x80
    /* "clickhouse_driver/varint.pyx":18
 *         towrite = number & 0x7f
 *             num_buf[i] = towrite | 0x80
      /* "clickhouse_driver/varint.pyx":19
 *             num_buf[i] = towrite | 0x80             # <<<<<<<<<<<<<<
      (__pyx_v_num_buf[__pyx_v_i]) = (__pyx_v_towrite | 0x80);
      /* "clickhouse_driver/varint.pyx":20
 *             num_buf[i] = towrite | 0x80
 *             num_buf[i] = towrite
      /* "clickhouse_driver/varint.pyx":18
 *         towrite = number & 0x7f
 *             num_buf[i] = towrite | 0x80
    /* "clickhouse_driver/varint.pyx":22
 *             num_buf[i] = towrite             # <<<<<<<<<<<<<<
      (__pyx_v_num_buf[__pyx_v_i]) = __pyx_v_towrite;
      /* "clickhouse_driver/varint.pyx":23
 *             num_buf[i] = towrite
      /* "clickhouse_driver/varint.pyx":24
 *             num_buf[i] = towrite
  /* "clickhouse_driver/varint.pyx":26
  __pyx_t_3 = __Pyx_PyObject_GetAttrStr(__pyx_v_buf, __pyx_n_s_write); if (unlikely(!__pyx_t_3)) __PYX_ERR(0, 26, __pyx_L1_error)
  __pyx_t_4 = PyBytes_FromStringAndSize(((char *)__pyx_v_num_buf), __pyx_v_i); if (unlikely(!__pyx_t_4)) __PYX_ERR(0, 26, __pyx_L1_error)
  if (unlikely(!__pyx_t_2)) __PYX_ERR(0, 26, __pyx_L1_error)
 * def write_varint(Py_ssize_t number, buf):             # <<<<<<<<<<<<<<
/* "clickhouse_driver/varint.pyx":29
  Py_ssize_t __pyx_v_shift;
  Py_ssize_t __pyx_v_result;
  unsigned char __pyx_v_i;
  unsigned char __pyx_t_4;
  /* "clickhouse_driver/varint.pyx":33
 *     cdef Py_ssize_t shift = 0             # <<<<<<<<<<<<<<
 *     cdef Py_ssize_t result = 0
 *     cdef unsigned char i
  /* "clickhouse_driver/varint.pyx":34
 *     cdef Py_ssize_t shift = 0
 *     cdef Py_ssize_t result = 0             # <<<<<<<<<<<<<<
 *     cdef unsigned char i
  /* "clickhouse_driver/varint.pyx":37
 *     cdef unsigned char i
  __pyx_t_1 = __Pyx_PyObject_GetAttrStr(__pyx_v_f, __pyx_n_s_read_one); if (unlikely(!__pyx_t_1)) __PYX_ERR(0, 37, __pyx_L1_error)
  /* "clickhouse_driver/varint.pyx":39
    /* "clickhouse_driver/varint.pyx":40
    if (unlikely(!__pyx_t_1)) __PYX_ERR(0, 40, __pyx_L1_error)
    __pyx_t_4 = __Pyx_PyInt_As_unsigned_char(__pyx_t_1); if (unlikely((__pyx_t_4 == (unsigned char)-1) && PyErr_Occurred())) __PYX_ERR(0, 40, __pyx_L1_error)
    /* "clickhouse_driver/varint.pyx":41
    /* "clickhouse_driver/varint.pyx":42
    /* "clickhouse_driver/varint.pyx":43
      /* "clickhouse_driver/varint.pyx":44
      /* "clickhouse_driver/varint.pyx":43
  /* "clickhouse_driver/varint.pyx":46
  __pyx_t_1 = PyInt_FromSsize_t(__pyx_v_result); if (unlikely(!__pyx_t_1)) __PYX_ERR(0, 46, __pyx_L1_error)
  /* "clickhouse_driver/varint.pyx":29
  {&__pyx_n_s_towrite, __pyx_k_towrite, sizeof(__pyx_k_towrite), 0, 0, 1, 1},
 * def write_varint(Py_ssize_t number, buf):             # <<<<<<<<<<<<<<
  __pyx_tuple_ = PyTuple_Pack(5, __pyx_n_s_number, __pyx_n_s_buf, __pyx_n_s_i, __pyx_n_s_towrite, __pyx_n_s_num_buf); if (unlikely(!__pyx_tuple_)) __PYX_ERR(0, 4, __pyx_L1_error)
  /* "clickhouse_driver/varint.pyx":29
  __pyx_tuple__3 = PyTuple_Pack(5, __pyx_n_s_f, __pyx_n_s_shift, __pyx_n_s_result, __pyx_n_s_i, __pyx_n_s_read_one); if (unlikely(!__pyx_tuple__3)) __PYX_ERR(0, 29, __pyx_L1_error)
  __pyx_codeobj__4 = (PyObject*)__Pyx_PyCode_New(1, 0, 5, 0, CO_OPTIMIZED|CO_NEWLOCALS, __pyx_empty_bytes, __pyx_empty_tuple, __pyx_empty_tuple, __pyx_tuple__3, __pyx_empty_tuple, __pyx_empty_tuple, __pyx_kp_s_clickhouse_driver_varint_pyx, __pyx_n_s_read_varint, 29, __pyx_empty_bytes); if (unlikely(!__pyx_codeobj__4)) __PYX_ERR(0, 29, __pyx_L1_error)
 * def write_varint(Py_ssize_t number, buf):             # <<<<<<<<<<<<<<
  /* "clickhouse_driver/varint.pyx":29
  __pyx_t_1 = PyCFunction_NewEx(&__pyx_mdef_17clickhouse_driver_6varint_3read_varint, NULL, __pyx_n_s_clickhouse_driver_varint); if (unlikely(!__pyx_t_1)) __PYX_ERR(0, 29, __pyx_L1_error)
  if (PyDict_SetItem(__pyx_d, __pyx_n_s_read_varint, __pyx_t_1) < 0) __PYX_ERR(0, 29, __pyx_L1_error)
static CYTHON_INLINE unsigned char __Pyx_PyInt_As_unsigned_char(PyObject *x) {
    const unsigned char neg_one = (unsigned char) ((unsigned char) 0 - (unsigned char) 1), const_zero = (unsigned char) 0;
        if (sizeof(unsigned char) < sizeof(long)) {
            __PYX_VERIFY_RETURN_INT(unsigned char, long, PyInt_AS_LONG(x))
            return (unsigned char) val;
                case  0: return (unsigned char) 0;
                case  1: __PYX_VERIFY_RETURN_INT(unsigned char, digit, digits[0])
                    if (8 * sizeof(unsigned char) > 1 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned char, unsigned long, (((((unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned char) >= 2 * PyLong_SHIFT) {
                            return (unsigned char) (((((unsigned char)digits[1]) << PyLong_SHIFT) | (unsigned char)digits[0]));
                    if (8 * sizeof(unsigned char) > 2 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned char, unsigned long, (((((((unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned char) >= 3 * PyLong_SHIFT) {
                            return (unsigned char) (((((((unsigned char)digits[2]) << PyLong_SHIFT) | (unsigned char)digits[1]) << PyLong_SHIFT) | (unsigned char)digits[0]));
                    if (8 * sizeof(unsigned char) > 3 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned char, unsigned long, (((((((((unsigned long)digits[3]) << PyLong_SHIFT) | (unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned char) >= 4 * PyLong_SHIFT) {
                            return (unsigned char) (((((((((unsigned char)digits[3]) << PyLong_SHIFT) | (unsigned char)digits[2]) << PyLong_SHIFT) | (unsigned char)digits[1]) << PyLong_SHIFT) | (unsigned char)digits[0]));
                    return (unsigned char) -1;
            if (sizeof(unsigned char) <= sizeof(unsigned long)) {
                __PYX_VERIFY_RETURN_INT_EXC(unsigned char, unsigned long, PyLong_AsUnsignedLong(x))
            } else if (sizeof(unsigned char) <= sizeof(unsigned PY_LONG_LONG)) {
                __PYX_VERIFY_RETURN_INT_EXC(unsigned char, unsigned PY_LONG_LONG, PyLong_AsUnsignedLongLong(x))
                case  0: return (unsigned char) 0;
                case -1: __PYX_VERIFY_RETURN_INT(unsigned char, sdigit, (sdigit) (-(sdigit)digits[0]))
                case  1: __PYX_VERIFY_RETURN_INT(unsigned char,  digit, +digits[0])
                    if (8 * sizeof(unsigned char) - 1 > 1 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned char, long, -(long) (((((unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned char) - 1 > 2 * PyLong_SHIFT) {
                            return (unsigned char) (((unsigned char)-1)*(((((unsigned char)digits[1]) << PyLong_SHIFT) | (unsigned char)digits[0])));
                    if (8 * sizeof(unsigned char) > 1 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned char, unsigned long, (((((unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned char) - 1 > 2 * PyLong_SHIFT) {
                            return (unsigned char) ((((((unsigned char)digits[1]) << PyLong_SHIFT) | (unsigned char)digits[0])));
                    if (8 * sizeof(unsigned char) - 1 > 2 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned char, long, -(long) (((((((unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned char) - 1 > 3 * PyLong_SHIFT) {
                            return (unsigned char) (((unsigned char)-1)*(((((((unsigned char)digits[2]) << PyLong_SHIFT) | (unsigned char)digits[1]) << PyLong_SHIFT) | (unsigned char)digits[0])));
                    if (8 * sizeof(unsigned char) > 2 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned char, unsigned long, (((((((unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned char) - 1 > 3 * PyLong_SHIFT) {
                            return (unsigned char) ((((((((unsigned char)digits[2]) << PyLong_SHIFT) | (unsigned char)digits[1]) << PyLong_SHIFT) | (unsigned char)digits[0])));
                    if (8 * sizeof(unsigned char) - 1 > 3 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned char, long, -(long) (((((((((unsigned long)digits[3]) << PyLong_SHIFT) | (unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned char) - 1 > 4 * PyLong_SHIFT) {
                            return (unsigned char) (((unsigned char)-1)*(((((((((unsigned char)digits[3]) << PyLong_SHIFT) | (unsigned char)digits[2]) << PyLong_SHIFT) | (unsigned char)digits[1]) << PyLong_SHIFT) | (unsigned char)digits[0])));
                    if (8 * sizeof(unsigned char) > 3 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned char, unsigned long, (((((((((unsigned long)digits[3]) << PyLong_SHIFT) | (unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned char) - 1 > 4 * PyLong_SHIFT) {
                            return (unsigned char) ((((((((((unsigned char)digits[3]) << PyLong_SHIFT) | (unsigned char)digits[2]) << PyLong_SHIFT) | (unsigned char)digits[1]) << PyLong_SHIFT) | (unsigned char)digits[0])));
            if (sizeof(unsigned char) <= sizeof(long)) {
                __PYX_VERIFY_RETURN_INT_EXC(unsigned char, long, PyLong_AsLong(x))
            } else if (sizeof(unsigned char) <= sizeof(PY_LONG_LONG)) {
                __PYX_VERIFY_RETURN_INT_EXC(unsigned char, PY_LONG_LONG, PyLong_AsLongLong(x))
            unsigned char val;
            return (unsigned char) -1;
        unsigned char val;
        if (!tmp) return (unsigned char) -1;
        val = __Pyx_PyInt_As_unsigned_char(tmp);
        "value too large to convert to unsigned char");
    return (unsigned char) -1;
        "can't convert negative value to unsigned char");
    return (unsigned char) -1;
def write_varint(Py_ssize_t number, buf):
    cdef Py_ssize_t i = 0
    cdef unsigned char towrite
    # Py_ssize_t checks integer on function call and
    # raises OverflowError if integer overflows Py_ssize_t.
    # Long enough for handling Py_ssize_t.
        towrite = number & 0x7f
            num_buf[i] = towrite | 0x80
            num_buf[i] = towrite
    cdef Py_ssize_t shift = 0
    cdef Py_ssize_t result = 0
    cdef unsigned char i

Fixed code:
/* CIntToPy.proto */
static CYTHON_INLINE PyObject* __Pyx_PyInt_From_unsigned_PY_LONG_LONG(unsigned PY_LONG_LONG value);

static CYTHON_INLINE unsigned PY_LONG_LONG __Pyx_PyInt_As_unsigned_PY_LONG_LONG(PyObject *);
static const char __pyx_k_to_write[] = "to_write";
static PyObject *__pyx_n_s_to_write;
static PyObject *__pyx_pf_17clickhouse_driver_6varint_write_varint(CYTHON_UNUSED PyObject *__pyx_self, unsigned PY_LONG_LONG __pyx_v_number, PyObject *__pyx_v_buf); /* proto */
 * def write_varint(unsigned long long number, buf):             # <<<<<<<<<<<<<<
  unsigned PY_LONG_LONG __pyx_v_number;
    __pyx_v_number = __Pyx_PyInt_As_unsigned_PY_LONG_LONG(values[0]); if (unlikely((__pyx_v_number == (unsigned PY_LONG_LONG)-1) && PyErr_Occurred())) __PYX_ERR(0, 4, __pyx_L3_error)
static PyObject *__pyx_pf_17clickhouse_driver_6varint_write_varint(CYTHON_UNUSED PyObject *__pyx_self, unsigned PY_LONG_LONG __pyx_v_number, PyObject *__pyx_v_buf) {
  unsigned char __pyx_v_to_write;
  unsigned char __pyx_v_i;
 *     cdef unsigned char to_write, i = 0             # <<<<<<<<<<<<<<
 *     # unsigned PY_LONG_LONG checks integer on function call and
 *     # raises OverflowError if integer overflows unsigned PY_LONG_LONG.
  /* "clickhouse_driver/varint.pyx":14
 *         to_write = number & 0x7f
    /* "clickhouse_driver/varint.pyx":15
 *         to_write = number & 0x7f             # <<<<<<<<<<<<<<
    __pyx_v_to_write = (__pyx_v_number & 0x7f);
    /* "clickhouse_driver/varint.pyx":16
 *         to_write = number & 0x7f
 *             num_buf[i] = to_write | 0x80
    /* "clickhouse_driver/varint.pyx":17
 *         to_write = number & 0x7f
 *             num_buf[i] = to_write | 0x80
      /* "clickhouse_driver/varint.pyx":18
 *             num_buf[i] = to_write | 0x80             # <<<<<<<<<<<<<<
      (__pyx_v_num_buf[__pyx_v_i]) = (__pyx_v_to_write | 0x80);
      /* "clickhouse_driver/varint.pyx":19
 *             num_buf[i] = to_write | 0x80
 *             num_buf[i] = to_write
      /* "clickhouse_driver/varint.pyx":17
 *         to_write = number & 0x7f
 *             num_buf[i] = to_write | 0x80
    /* "clickhouse_driver/varint.pyx":21
 *             num_buf[i] = to_write             # <<<<<<<<<<<<<<
      (__pyx_v_num_buf[__pyx_v_i]) = __pyx_v_to_write;
      /* "clickhouse_driver/varint.pyx":22
 *             num_buf[i] = to_write
      /* "clickhouse_driver/varint.pyx":23
 *             num_buf[i] = to_write
  /* "clickhouse_driver/varint.pyx":25
  __pyx_t_3 = __Pyx_PyObject_GetAttrStr(__pyx_v_buf, __pyx_n_s_write); if (unlikely(!__pyx_t_3)) __PYX_ERR(0, 25, __pyx_L1_error)
  __pyx_t_4 = PyBytes_FromStringAndSize(((char *)__pyx_v_num_buf), __pyx_v_i); if (unlikely(!__pyx_t_4)) __PYX_ERR(0, 25, __pyx_L1_error)
  if (unlikely(!__pyx_t_2)) __PYX_ERR(0, 25, __pyx_L1_error)
 * def write_varint(unsigned long long number, buf):             # <<<<<<<<<<<<<<
/* "clickhouse_driver/varint.pyx":28
  unsigned char __pyx_v_shift;
  unsigned PY_LONG_LONG __pyx_v_i;
  unsigned PY_LONG_LONG __pyx_v_result;
  unsigned PY_LONG_LONG __pyx_t_4;
  /* "clickhouse_driver/varint.pyx":32
 *     cdef unsigned char shift = 0             # <<<<<<<<<<<<<<
 *     cdef unsigned long long i, result = 0
 * 
  /* "clickhouse_driver/varint.pyx":33
 *     cdef unsigned char shift = 0
 *     cdef unsigned long long i, result = 0             # <<<<<<<<<<<<<<
 *     read_one = f.read_one
  /* "clickhouse_driver/varint.pyx":35
 *     cdef unsigned long long i, result = 0
  __pyx_t_1 = __Pyx_PyObject_GetAttrStr(__pyx_v_f, __pyx_n_s_read_one); if (unlikely(!__pyx_t_1)) __PYX_ERR(0, 35, __pyx_L1_error)
  /* "clickhouse_driver/varint.pyx":37
    /* "clickhouse_driver/varint.pyx":38
    if (unlikely(!__pyx_t_1)) __PYX_ERR(0, 38, __pyx_L1_error)
    __pyx_t_4 = __Pyx_PyInt_As_unsigned_PY_LONG_LONG(__pyx_t_1); if (unlikely((__pyx_t_4 == (unsigned PY_LONG_LONG)-1) && PyErr_Occurred())) __PYX_ERR(0, 38, __pyx_L1_error)
    /* "clickhouse_driver/varint.pyx":39
    /* "clickhouse_driver/varint.pyx":40
    /* "clickhouse_driver/varint.pyx":41
      /* "clickhouse_driver/varint.pyx":42
      /* "clickhouse_driver/varint.pyx":41
  /* "clickhouse_driver/varint.pyx":44
  __pyx_t_1 = __Pyx_PyInt_From_unsigned_PY_LONG_LONG(__pyx_v_result); if (unlikely(!__pyx_t_1)) __PYX_ERR(0, 44, __pyx_L1_error)
  /* "clickhouse_driver/varint.pyx":28
  {&__pyx_n_s_to_write, __pyx_k_to_write, sizeof(__pyx_k_to_write), 0, 0, 1, 1},
 * def write_varint(unsigned long long number, buf):             # <<<<<<<<<<<<<<
  __pyx_tuple_ = PyTuple_Pack(5, __pyx_n_s_number, __pyx_n_s_buf, __pyx_n_s_to_write, __pyx_n_s_i, __pyx_n_s_num_buf); if (unlikely(!__pyx_tuple_)) __PYX_ERR(0, 4, __pyx_L1_error)
  /* "clickhouse_driver/varint.pyx":28
  __pyx_tuple__3 = PyTuple_Pack(5, __pyx_n_s_f, __pyx_n_s_shift, __pyx_n_s_i, __pyx_n_s_result, __pyx_n_s_read_one); if (unlikely(!__pyx_tuple__3)) __PYX_ERR(0, 28, __pyx_L1_error)
  __pyx_codeobj__4 = (PyObject*)__Pyx_PyCode_New(1, 0, 5, 0, CO_OPTIMIZED|CO_NEWLOCALS, __pyx_empty_bytes, __pyx_empty_tuple, __pyx_empty_tuple, __pyx_tuple__3, __pyx_empty_tuple, __pyx_empty_tuple, __pyx_kp_s_clickhouse_driver_varint_pyx, __pyx_n_s_read_varint, 28, __pyx_empty_bytes); if (unlikely(!__pyx_codeobj__4)) __PYX_ERR(0, 28, __pyx_L1_error)
 * def write_varint(unsigned long long number, buf):             # <<<<<<<<<<<<<<
  /* "clickhouse_driver/varint.pyx":28
  __pyx_t_1 = PyCFunction_NewEx(&__pyx_mdef_17clickhouse_driver_6varint_3read_varint, NULL, __pyx_n_s_clickhouse_driver_varint); if (unlikely(!__pyx_t_1)) __PYX_ERR(0, 28, __pyx_L1_error)
  if (PyDict_SetItem(__pyx_d, __pyx_n_s_read_varint, __pyx_t_1) < 0) __PYX_ERR(0, 28, __pyx_L1_error)
/* CIntToPy */
static CYTHON_INLINE PyObject* __Pyx_PyInt_From_unsigned_PY_LONG_LONG(unsigned PY_LONG_LONG value) {
    const unsigned PY_LONG_LONG neg_one = (unsigned PY_LONG_LONG) ((unsigned PY_LONG_LONG) 0 - (unsigned PY_LONG_LONG) 1), const_zero = (unsigned PY_LONG_LONG) 0;
    const int is_unsigned = neg_one > const_zero;
    if (is_unsigned) {
        if (sizeof(unsigned PY_LONG_LONG) < sizeof(long)) {
            return PyInt_FromLong((long) value);
        } else if (sizeof(unsigned PY_LONG_LONG) <= sizeof(unsigned long)) {
            return PyLong_FromUnsignedLong((unsigned long) value);
#ifdef HAVE_LONG_LONG
        } else if (sizeof(unsigned PY_LONG_LONG) <= sizeof(unsigned PY_LONG_LONG)) {
            return PyLong_FromUnsignedLongLong((unsigned PY_LONG_LONG) value);
#endif
        }
    } else {
        if (sizeof(unsigned PY_LONG_LONG) <= sizeof(long)) {
            return PyInt_FromLong((long) value);
#ifdef HAVE_LONG_LONG
        } else if (sizeof(unsigned PY_LONG_LONG) <= sizeof(PY_LONG_LONG)) {
            return PyLong_FromLongLong((PY_LONG_LONG) value);
#endif
        }
    }
    {
        int one = 1; int little = (int)*(unsigned char *)&one;
        unsigned char *bytes = (unsigned char *)&value;
        return _PyLong_FromByteArray(bytes, sizeof(unsigned PY_LONG_LONG),
                                     little, !is_unsigned);
    }
}

static CYTHON_INLINE unsigned PY_LONG_LONG __Pyx_PyInt_As_unsigned_PY_LONG_LONG(PyObject *x) {
    const unsigned PY_LONG_LONG neg_one = (unsigned PY_LONG_LONG) ((unsigned PY_LONG_LONG) 0 - (unsigned PY_LONG_LONG) 1), const_zero = (unsigned PY_LONG_LONG) 0;
        if (sizeof(unsigned PY_LONG_LONG) < sizeof(long)) {
            __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, long, PyInt_AS_LONG(x))
            return (unsigned PY_LONG_LONG) val;
                case  0: return (unsigned PY_LONG_LONG) 0;
                case  1: __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, digit, digits[0])
                    if (8 * sizeof(unsigned PY_LONG_LONG) > 1 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, unsigned long, (((((unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned PY_LONG_LONG) >= 2 * PyLong_SHIFT) {
                            return (unsigned PY_LONG_LONG) (((((unsigned PY_LONG_LONG)digits[1]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[0]));
                    if (8 * sizeof(unsigned PY_LONG_LONG) > 2 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, unsigned long, (((((((unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned PY_LONG_LONG) >= 3 * PyLong_SHIFT) {
                            return (unsigned PY_LONG_LONG) (((((((unsigned PY_LONG_LONG)digits[2]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[1]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[0]));
                    if (8 * sizeof(unsigned PY_LONG_LONG) > 3 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, unsigned long, (((((((((unsigned long)digits[3]) << PyLong_SHIFT) | (unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned PY_LONG_LONG) >= 4 * PyLong_SHIFT) {
                            return (unsigned PY_LONG_LONG) (((((((((unsigned PY_LONG_LONG)digits[3]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[2]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[1]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[0]));
                    return (unsigned PY_LONG_LONG) -1;
            if (sizeof(unsigned PY_LONG_LONG) <= sizeof(unsigned long)) {
                __PYX_VERIFY_RETURN_INT_EXC(unsigned PY_LONG_LONG, unsigned long, PyLong_AsUnsignedLong(x))
            } else if (sizeof(unsigned PY_LONG_LONG) <= sizeof(unsigned PY_LONG_LONG)) {
                __PYX_VERIFY_RETURN_INT_EXC(unsigned PY_LONG_LONG, unsigned PY_LONG_LONG, PyLong_AsUnsignedLongLong(x))
                case  0: return (unsigned PY_LONG_LONG) 0;
                case -1: __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, sdigit, (sdigit) (-(sdigit)digits[0]))
                case  1: __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG,  digit, +digits[0])
                    if (8 * sizeof(unsigned PY_LONG_LONG) - 1 > 1 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, long, -(long) (((((unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned PY_LONG_LONG) - 1 > 2 * PyLong_SHIFT) {
                            return (unsigned PY_LONG_LONG) (((unsigned PY_LONG_LONG)-1)*(((((unsigned PY_LONG_LONG)digits[1]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[0])));
                    if (8 * sizeof(unsigned PY_LONG_LONG) > 1 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, unsigned long, (((((unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned PY_LONG_LONG) - 1 > 2 * PyLong_SHIFT) {
                            return (unsigned PY_LONG_LONG) ((((((unsigned PY_LONG_LONG)digits[1]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[0])));
                    if (8 * sizeof(unsigned PY_LONG_LONG) - 1 > 2 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, long, -(long) (((((((unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned PY_LONG_LONG) - 1 > 3 * PyLong_SHIFT) {
                            return (unsigned PY_LONG_LONG) (((unsigned PY_LONG_LONG)-1)*(((((((unsigned PY_LONG_LONG)digits[2]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[1]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[0])));
                    if (8 * sizeof(unsigned PY_LONG_LONG) > 2 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, unsigned long, (((((((unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned PY_LONG_LONG) - 1 > 3 * PyLong_SHIFT) {
                            return (unsigned PY_LONG_LONG) ((((((((unsigned PY_LONG_LONG)digits[2]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[1]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[0])));
                    if (8 * sizeof(unsigned PY_LONG_LONG) - 1 > 3 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, long, -(long) (((((((((unsigned long)digits[3]) << PyLong_SHIFT) | (unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned PY_LONG_LONG) - 1 > 4 * PyLong_SHIFT) {
                            return (unsigned PY_LONG_LONG) (((unsigned PY_LONG_LONG)-1)*(((((((((unsigned PY_LONG_LONG)digits[3]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[2]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[1]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[0])));
                    if (8 * sizeof(unsigned PY_LONG_LONG) > 3 * PyLong_SHIFT) {
                            __PYX_VERIFY_RETURN_INT(unsigned PY_LONG_LONG, unsigned long, (((((((((unsigned long)digits[3]) << PyLong_SHIFT) | (unsigned long)digits[2]) << PyLong_SHIFT) | (unsigned long)digits[1]) << PyLong_SHIFT) | (unsigned long)digits[0])))
                        } else if (8 * sizeof(unsigned PY_LONG_LONG) - 1 > 4 * PyLong_SHIFT) {
                            return (unsigned PY_LONG_LONG) ((((((((((unsigned PY_LONG_LONG)digits[3]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[2]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[1]) << PyLong_SHIFT) | (unsigned PY_LONG_LONG)digits[0])));
            if (sizeof(unsigned PY_LONG_LONG) <= sizeof(long)) {
                __PYX_VERIFY_RETURN_INT_EXC(unsigned PY_LONG_LONG, long, PyLong_AsLong(x))
            } else if (sizeof(unsigned PY_LONG_LONG) <= sizeof(PY_LONG_LONG)) {
                __PYX_VERIFY_RETURN_INT_EXC(unsigned PY_LONG_LONG, PY_LONG_LONG, PyLong_AsLongLong(x))
            unsigned PY_LONG_LONG val;
            return (unsigned PY_LONG_LONG) -1;
        unsigned PY_LONG_LONG val;
        if (!tmp) return (unsigned PY_LONG_LONG) -1;
        val = __Pyx_PyInt_As_unsigned_PY_LONG_LONG(tmp);
        "value too large to convert to unsigned PY_LONG_LONG");
    return (unsigned PY_LONG_LONG) -1;
        "can't convert negative value to unsigned PY_LONG_LONG");
    return (unsigned PY_LONG_LONG) -1;
def write_varint(unsigned long long number, buf):
    cdef unsigned char to_write, i = 0
    # unsigned PY_LONG_LONG checks integer on function call and
    # raises OverflowError if integer overflows unsigned PY_LONG_LONG.
    # Long enough for handling unsigned PY_LONG_LONG.
        to_write = number & 0x7f
            num_buf[i] = to_write | 0x80
            num_buf[i] = to_write
    cdef unsigned char shift = 0
    cdef unsigned long long i, result = 0
from io import BytesIO
from unittest import TestCase

from clickhouse_driver.varint import read_varint, write_varint


class VarIntTestCase(TestCase):
    def test_check_not_negative(self):
        n = 0x9FFFFFFF

        buf = BytesIO()
        write_varint(n, buf)
        val = buf.getvalue()
        self.assertEqual(b'\xFF\xFF\xFF\xFF\t', val)

        buf = BytesIO(val)
        buf.read_one = lambda: ord(buf.read(1))
        m = read_varint(buf)
        self.assertEqual(m, n)