#include "Python.h"

#include "pycore_call.h" // _PyObject_CallMethod()

#include "pycore_codecs.h" // _PyCodecInfo_GetIncrementalDecoder()

#include "pycore_fileutils.h" // _Py_GetLocaleEncoding()

#include "pycore_interp.h" // PyInterpreterState.fs_codec

#include "pycore_long.h" // _PyLong_GetZero()

#include "pycore_object.h" // _PyObject_GC_UNTRACK()

#include "pycore_pyerrors.h" // _PyErr_ChainExceptions1()

#include "pycore_pystate.h" // _PyInterpreterState_GET()

#include "_iomodule.h"

typedef struct nldecoder_object nldecoder_object;

typedef struct textio textio;

#define clinic_state() (find_io_state_by_def(Py_TYPE(self)))

#include "clinic/textio.c.h"

#undef clinic_state

PyDoc_STRVAR(textiobase_doc,

"Base class for text I/O.\n"

"\n"

"This class provides a character and line based interface to stream\n"

"I/O. There is no readinto method because Python's character strings\n"

"are immutable.\n"

);

static PyObject *

_unsupported(_PyIO_State *state, const char *message)

{

PyErr_SetString(state->unsupported_operation, message);

return NULL;

}

static PyObject *

_io__TextIOBase_detach_impl(PyObject *self, PyTypeObject *cls)

{

_PyIO_State *state = get_io_state_by_cls(cls);

return _unsupported(state, "detach");

}

static PyObject *

_io__TextIOBase_read_impl(PyObject *self, PyTypeObject *cls,

int Py_UNUSED(size))

{

_PyIO_State *state = get_io_state_by_cls(cls);

return _unsupported(state, "read");

}

static PyObject *

_io__TextIOBase_readline_impl(PyObject *self, PyTypeObject *cls,

int Py_UNUSED(size))

{

_PyIO_State *state = get_io_state_by_cls(cls);

return _unsupported(state, "readline");

}

static PyObject *

_io__TextIOBase_write_impl(PyObject *self, PyTypeObject *cls,

const char *Py_UNUSED(s))

{

_PyIO_State *state = get_io_state_by_cls(cls);

return _unsupported(state, "write");

}

static PyObject *

_io__TextIOBase_encoding_get_impl(PyObject *self)

{

Py_RETURN_NONE;

}

static PyObject *

_io__TextIOBase_newlines_get_impl(PyObject *self)

{

Py_RETURN_NONE;

}

static PyObject *

_io__TextIOBase_errors_get_impl(PyObject *self)

{

Py_RETURN_NONE;

}

static PyMethodDef textiobase_methods[] = {

_IO__TEXTIOBASE_DETACH_METHODDEF

_IO__TEXTIOBASE_READ_METHODDEF

_IO__TEXTIOBASE_READLINE_METHODDEF

_IO__TEXTIOBASE_WRITE_METHODDEF

{NULL, NULL}

};

static PyGetSetDef textiobase_getset[] = {

_IO__TEXTIOBASE_ENCODING_GETSETDEF

_IO__TEXTIOBASE_NEWLINES_GETSETDEF

_IO__TEXTIOBASE_ERRORS_GETSETDEF

{NULL}

};

static PyType_Slot textiobase_slots[] = {

{Py_tp_doc, (void *)textiobase_doc},

{Py_tp_methods, textiobase_methods},

{Py_tp_getset, textiobase_getset},

{0, NULL},

};

PyType_Spec textiobase_spec = {

.name = "_io._TextIOBase",

.flags = (Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE |

Py_TPFLAGS_IMMUTABLETYPE),

.slots = textiobase_slots,

};

struct nldecoder_object {

};

#define nldecoder_object_CAST(op) ((nldecoder_object *)(op))

static int

_io_IncrementalNewlineDecoder___init___impl(nldecoder_object *self,

PyObject *decoder, int translate,

PyObject *errors)

{

if (errors == NULL) {

errors = &_Py_ID(strict);

}

else {

errors = Py_NewRef(errors);

}

Py_XSETREF(self->errors, errors);

Py_XSETREF(self->decoder, Py_NewRef(decoder));

self->translate = translate ? 1 : 0;

self->seennl = 0;

self->pendingcr = 0;

return 0;

}

static int

incrementalnewlinedecoder_traverse(PyObject *op, visitproc visit, void *arg)

{

nldecoder_object *self = nldecoder_object_CAST(op);

Py_VISIT(Py_TYPE(self));

Py_VISIT(self->decoder);

Py_VISIT(self->errors);

return 0;

}

static int

incrementalnewlinedecoder_clear(PyObject *op)

{

nldecoder_object *self = nldecoder_object_CAST(op);

Py_CLEAR(self->decoder);

Py_CLEAR(self->errors);

return 0;

}

static void

incrementalnewlinedecoder_dealloc(PyObject *op)

{

nldecoder_object *self = nldecoder_object_CAST(op);

PyTypeObject *tp = Py_TYPE(self);

_PyObject_GC_UNTRACK(self);

(void)incrementalnewlinedecoder_clear(op);

tp->tp_free(self);

Py_DECREF(tp);

}

static int

check_decoded(PyObject *decoded)

{

if (decoded == NULL)

return -1;

if (!PyUnicode_Check(decoded)) {

PyErr_Format(PyExc_TypeError,

"decoder should return a string result, not '%.200s'",

Py_TYPE(decoded)->tp_name);

Py_DECREF(decoded);

return -1;

}

return 0;

}

#define CHECK_INITIALIZED_DECODER(self) \

if (self->errors == NULL) { \

PyErr_SetString(PyExc_ValueError, \

"IncrementalNewlineDecoder.__init__() not called"); \

return NULL; \

}

#define SEEN_CR 1

#define SEEN_LF 2

#define SEEN_CRLF 4

#define SEEN_ALL (SEEN_CR | SEEN_LF | SEEN_CRLF)

PyObject *

_PyIncrementalNewlineDecoder_decode(PyObject *myself,

PyObject *input, int final)

{

PyObject *output;

Py_ssize_t output_len;

nldecoder_object *self = nldecoder_object_CAST(myself);

CHECK_INITIALIZED_DECODER(self);

/* decode input (with the eventual \r from a previous pass) */

if (self->decoder != Py_None) {

output = PyObject_CallMethodObjArgs(self->decoder,

&_Py_ID(decode), input, final ? Py_True : Py_False, NULL);

}

else {

output = Py_NewRef(input);

}

if (check_decoded(output) < 0)

return NULL;

output_len = PyUnicode_GET_LENGTH(output);

if (self->pendingcr && (final || output_len > 0)) {

/* Prefix output with CR */

int kind;

PyObject *modified;

char *out;

modified = PyUnicode_New(output_len + 1,

PyUnicode_MAX_CHAR_VALUE(output));

if (modified == NULL)

goto error;

kind = PyUnicode_KIND(modified);

out = PyUnicode_DATA(modified);

PyUnicode_WRITE(kind, out, 0, '\r');

memcpy(out + kind, PyUnicode_DATA(output), kind * output_len);

Py_SETREF(output, modified);

self->pendingcr = 0;

output_len++;

}

/* retain last \r even when not translating data:

if (!final) {

if (output_len > 0

&& PyUnicode_READ_CHAR(output, output_len - 1) == '\r')

{

PyObject *modified = PyUnicode_Substring(output, 0, output_len -1);

if (modified == NULL)

goto error;

Py_SETREF(output, modified);

self->pendingcr = 1;

}

}

/* Record which newlines are read and do newline translation if desired,

{

const void *in_str;

Py_ssize_t len;

int seennl = self->seennl;

int only_lf = 0;

int kind;

in_str = PyUnicode_DATA(output);

len = PyUnicode_GET_LENGTH(output);

kind = PyUnicode_KIND(output);

if (len == 0)

return output;

/* If, up to now, newlines are consistently \n, do a quick check

if (seennl == SEEN_LF || seennl == 0) {

only_lf = (memchr(in_str, '\r', kind * len) == NULL);

}

if (only_lf) {

/* If not already seen, quick scan for a possible "\n" character.

if (seennl == 0 &&

memchr(in_str, '\n', kind * len) != NULL) {

if (kind == PyUnicode_1BYTE_KIND)

seennl |= SEEN_LF;

else {

Py_ssize_t i = 0;

for (;;) {

Py_UCS4 c;

/* Fast loop for non-control characters */

while (PyUnicode_READ(kind, in_str, i) > '\n')

i++;

c = PyUnicode_READ(kind, in_str, i++);

if (c == '\n') {

seennl |= SEEN_LF;

break;

}

if (i >= len)

break;

}

}

}

/* Finished: we have scanned for newlines, and none of them

}

else if (!self->translate) {

Py_ssize_t i = 0;

/* We have already seen all newline types, no need to scan again */

if (seennl == SEEN_ALL)

goto endscan;

for (;;) {

Py_UCS4 c;

/* Fast loop for non-control characters */

while (PyUnicode_READ(kind, in_str, i) > '\r')

i++;

c = PyUnicode_READ(kind, in_str, i++);

if (c == '\n')

seennl |= SEEN_LF;

else if (c == '\r') {

if (PyUnicode_READ(kind, in_str, i) == '\n') {

seennl |= SEEN_CRLF;

i++;

}

else

seennl |= SEEN_CR;

}

if (i >= len)

break;

if (seennl == SEEN_ALL)

break;

}

endscan:

;

}

else {

void *translated;

int kind = PyUnicode_KIND(output);

const void *in_str = PyUnicode_DATA(output);

Py_ssize_t in, out;

/* XXX: Previous in-place translation here is disabled as

/* We could try to optimize this so that we only do a copy

translated = PyMem_Malloc(kind * len);

if (translated == NULL) {

PyErr_NoMemory();

goto error;

}

in = out = 0;

for (;;) {

Py_UCS4 c;

/* Fast loop for non-control characters */

while ((c = PyUnicode_READ(kind, in_str, in++)) > '\r')

PyUnicode_WRITE(kind, translated, out++, c);

if (c == '\n') {

PyUnicode_WRITE(kind, translated, out++, c);

seennl |= SEEN_LF;

continue;

}

if (c == '\r') {

if (PyUnicode_READ(kind, in_str, in) == '\n') {

in++;

seennl |= SEEN_CRLF;

}

else

seennl |= SEEN_CR;

PyUnicode_WRITE(kind, translated, out++, '\n');

continue;

}

if (in > len)

break;

PyUnicode_WRITE(kind, translated, out++, c);

}

Py_DECREF(output);

output = PyUnicode_FromKindAndData(kind, translated, out);

PyMem_Free(translated);

if (!output)

return NULL;

}

self->seennl |= seennl;

}

return output;

error:

Py_DECREF(output);

return NULL;

}

static PyObject *

_io_IncrementalNewlineDecoder_decode_impl(nldecoder_object *self,

PyObject *input, int final)

{

return _PyIncrementalNewlineDecoder_decode((PyObject *) self, input, final);

}

static PyObject *

_io_IncrementalNewlineDecoder_getstate_impl(nldecoder_object *self)

{

PyObject *buffer;

unsigned long long flag;

CHECK_INITIALIZED_DECODER(self);

if (self->decoder != Py_None) {

PyObject *state = PyObject_CallMethodNoArgs(self->decoder,

&_Py_ID(getstate));

if (state == NULL)

return NULL;

if (!PyTuple_Check(state)) {

PyErr_SetString(PyExc_TypeError,

"illegal decoder state");

Py_DECREF(state);

return NULL;

}

if (!PyArg_ParseTuple(state, "OK;illegal decoder state",

&buffer, &flag))

{

Py_DECREF(state);

return NULL;

}

Py_INCREF(buffer);

Py_DECREF(state);

}

else {

buffer = Py_GetConstant(Py_CONSTANT_EMPTY_BYTES);

flag = 0;

}

flag <<= 1;

if (self->pendingcr)

flag |= 1;

return Py_BuildValue("NK", buffer, flag);

}

static PyObject *

_io_IncrementalNewlineDecoder_setstate(nldecoder_object *self,

PyObject *state)

{

PyObject *buffer;

unsigned long long flag;

CHECK_INITIALIZED_DECODER(self);

if (!PyTuple_Check(state)) {

PyErr_SetString(PyExc_TypeError, "state argument must be a tuple");

return NULL;

}

if (!PyArg_ParseTuple(state, "OK;setstate(): illegal state argument",

&buffer, &flag))

{

return NULL;

}

self->pendingcr = (int) (flag & 1);

flag >>= 1;

if (self->decoder != Py_None) {

return _PyObject_CallMethod(self->decoder, &_Py_ID(setstate),

"((OK))", buffer, flag);

}

else {

Py_RETURN_NONE;

}

}

static PyObject *

_io_IncrementalNewlineDecoder_reset_impl(nldecoder_object *self)

{

CHECK_INITIALIZED_DECODER(self);

self->seennl = 0;

self->pendingcr = 0;

if (self->decoder != Py_None)

return PyObject_CallMethodNoArgs(self->decoder, &_Py_ID(reset));

else

Py_RETURN_NONE;

}

static PyObject *

incrementalnewlinedecoder_newlines_get(PyObject *op, void *Py_UNUSED(context))

{

nldecoder_object *self = nldecoder_object_CAST(op);

CHECK_INITIALIZED_DECODER(self);

switch (self->seennl) {

case SEEN_CR:

return PyUnicode_FromString("\r");

case SEEN_LF:

return PyUnicode_FromString("\n");

case SEEN_CRLF:

return PyUnicode_FromString("\r\n");

case SEEN_CR | SEEN_LF:

return Py_BuildValue("ss", "\r", "\n");

case SEEN_CR | SEEN_CRLF:

return Py_BuildValue("ss", "\r", "\r\n");

case SEEN_LF | SEEN_CRLF:

return Py_BuildValue("ss", "\n", "\r\n");

case SEEN_CR | SEEN_LF | SEEN_CRLF:

return Py_BuildValue("sss", "\r", "\n", "\r\n");

default:

Py_RETURN_NONE;

}

}

typedef PyObject *(*encodefunc_t)(PyObject *, PyObject *);

struct textio

};

#define textio_CAST(op) ((textio *)(op))

static void

textiowrapper_set_decoded_chars(textio *self, PyObject *chars);

static PyObject *

ascii_encode(PyObject *op, PyObject *text)

{

textio *self = textio_CAST(op);

return _PyUnicode_AsASCIIString(text, PyUnicode_AsUTF8(self->errors));

}

static PyObject *

utf16be_encode(PyObject *op, PyObject *text)

{

textio *self = textio_CAST(op);

return _PyUnicode_EncodeUTF16(text, PyUnicode_AsUTF8(self->errors), 1);

}

static PyObject *

utf16le_encode(PyObject *op, PyObject *text)

{

textio *self = textio_CAST(op);

return _PyUnicode_EncodeUTF16(text, PyUnicode_AsUTF8(self->errors), -1);

}

static PyObject *

utf16_encode(PyObject *op, PyObject *text)

{

textio *self = textio_CAST(op);

if (!self->encoding_start_of_stream) {

/* Skip the BOM and use native byte ordering */

#if PY_BIG_ENDIAN

return utf16be_encode(op, text);

return utf16le_encode(op, text);

}

return _PyUnicode_EncodeUTF16(text, PyUnicode_AsUTF8(self->errors), 0);

}

static PyObject *

utf32be_encode(PyObject *op, PyObject *text)

{

textio *self = textio_CAST(op);

return _PyUnicode_EncodeUTF32(text, PyUnicode_AsUTF8(self->errors), 1);

}

static PyObject *

utf32le_encode(PyObject *op, PyObject *text)

{

textio *self = textio_CAST(op);

return _PyUnicode_EncodeUTF32(text, PyUnicode_AsUTF8(self->errors), -1);

}

static PyObject *

utf32_encode(PyObject *op, PyObject *text)

{

textio *self = textio_CAST(op);

if (!self->encoding_start_of_stream) {

/* Skip the BOM and use native byte ordering */

#if PY_BIG_ENDIAN

return utf32be_encode(op, text);

return utf32le_encode(op, text);

}

return _PyUnicode_EncodeUTF32(text, PyUnicode_AsUTF8(self->errors), 0);

}

static PyObject *

utf8_encode(PyObject *op, PyObject *text)

{

textio *self = textio_CAST(op);

return _PyUnicode_AsUTF8String(text, PyUnicode_AsUTF8(self->errors));

}

static PyObject *

latin1_encode(PyObject *op, PyObject *text)

{

textio *self = textio_CAST(op);

return _PyUnicode_AsLatin1String(text, PyUnicode_AsUTF8(self->errors));

}

static inline int

is_asciicompat_encoding(encodefunc_t f)

{

return f == ascii_encode || f == latin1_encode || f == utf8_encode;

}

typedef struct {

const char *name;

encodefunc_t encodefunc;

} encodefuncentry;

static const encodefuncentry encodefuncs[] = {

{"ascii", ascii_encode},

{"iso8859-1", latin1_encode},

{"utf-8", utf8_encode},

{"utf-16-be", utf16be_encode},

{"utf-16-le", utf16le_encode},

{"utf-16", utf16_encode},

{"utf-32-be", utf32be_encode},

{"utf-32-le", utf32le_encode},

{"utf-32", utf32_encode},

{NULL, NULL}

};

static int

validate_newline(const char *newline)

{

if (newline && newline[0] != '\0'

&& !(newline[0] == '\n' && newline[1] == '\0')

&& !(newline[0] == '\r' && newline[1] == '\0')

&& !(newline[0] == '\r' && newline[1] == '\n' && newline[2] == '\0')) {

PyErr_Format(PyExc_ValueError,

"illegal newline value: %s", newline);

return -1;

}

return 0;

}

static int

set_newline(textio *self, const char *newline)

{

PyObject *old = self->readnl;

if (newline == NULL) {

self->readnl = NULL;

}

else {

self->readnl = PyUnicode_FromString(newline);

if (self->readnl == NULL) {

self->readnl = old;

return -1;

}

}

self->readuniversal = (newline == NULL || newline[0] == '\0');

self->readtranslate = (newline == NULL);

self->writetranslate = (newline == NULL || newline[0] != '\0');

if (!self->readuniversal && self->readnl != NULL) {

// validate_newline() accepts only ASCII newlines.

assert(PyUnicode_KIND(self->readnl) == PyUnicode_1BYTE_KIND);

self->writenl = (const char *)PyUnicode_1BYTE_DATA(self->readnl);

if (strcmp(self->writenl, "\n") == 0) {

self->writenl = NULL;

}

}

else {

#ifdef MS_WINDOWS

self->writenl = "\r\n";

self->writenl = NULL;

}

Py_XDECREF(old);

return 0;

}

static int

_textiowrapper_set_decoder(textio *self, PyObject *codec_info,

const char *errors)

{

PyObject *res;

int r;

res = PyObject_CallMethodNoArgs(self->buffer, &_Py_ID(readable));

if (res == NULL)

return -1;

r = PyObject_IsTrue(res);

Py_DECREF(res);

if (r == -1)

return -1;

if (r != 1)

return 0;

Py_CLEAR(self->decoder);

self->decoder = _PyCodecInfo_GetIncrementalDecoder(codec_info, errors);

if (self->decoder == NULL)

return -1;

if (self->readuniversal) {

_PyIO_State *state = self->state;

PyObject *incrementalDecoder = PyObject_CallFunctionObjArgs(

(PyObject *)state->PyIncrementalNewlineDecoder_Type,

self->decoder, self->readtranslate ? Py_True : Py_False, NULL);

if (incrementalDecoder == NULL)

return -1;

Py_XSETREF(self->decoder, incrementalDecoder);

}

return 0;

}

static PyObject*

_textiowrapper_decode(_PyIO_State *state, PyObject *decoder, PyObject *bytes,

int eof)

{

PyObject *chars;

if (Py_IS_TYPE(decoder, state->PyIncrementalNewlineDecoder_Type))

chars = _PyIncrementalNewlineDecoder_decode(decoder, bytes, eof);

else

chars = PyObject_CallMethodObjArgs(decoder, &_Py_ID(decode), bytes,

eof ? Py_True : Py_False, NULL);

if (check_decoded(chars) < 0)

// check_decoded already decreases refcount

return NULL;

return chars;

}

static int

_textiowrapper_set_encoder(textio *self, PyObject *codec_info,

const char *errors)

{

PyObject *res;

int r;

res = PyObject_CallMethodNoArgs(self->buffer, &_Py_ID(writable));

if (res == NULL)

return -1;

r = PyObject_IsTrue(res);

Py_DECREF(res);

if (r == -1)

return -1;

if (r != 1)

return 0;

Py_CLEAR(self->encoder);

self->encodefunc = NULL;

self->encoder = _PyCodecInfo_GetIncrementalEncoder(codec_info, errors);

if (self->encoder == NULL)

return -1;

/* Get the normalized named of the codec */

if (PyObject_GetOptionalAttr(codec_info, &_Py_ID(name), &res) < 0) {

return -1;

}

if (res != NULL && PyUnicode_Check(res)) {

const encodefuncentry *e = encodefuncs;

while (e->name != NULL) {

if (_PyUnicode_EqualToASCIIString(res, e->name)) {

self->encodefunc = e->encodefunc;

break;

}

e++;

}

}

Py_XDECREF(res);

return 0;

}

static int

_textiowrapper_fix_encoder_state(textio *self)

{

if (!self->seekable || !self->encoder) {

return 0;

}

self->encoding_start_of_stream = 1;

PyObject *cookieObj = PyObject_CallMethodNoArgs(

self->buffer, &_Py_ID(tell));

if (cookieObj == NULL) {

return -1;

}