/* ------------------------------------------------------------------------

Python Codec Registry and support functions

Written by Marc-Andre Lemburg (mal@lemburg.com).

Copyright (c) Corporation for National Research Initiatives.

------------------------------------------------------------------------ */

#include "Python.h"

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

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

#include "pycore_lock.h" // PyMutex

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

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

#include "pycore_ucnhash.h" // _PyUnicode_Name_CAPI

static const char *codecs_builtin_error_handlers[] = {

"strict", "ignore", "replace",

"xmlcharrefreplace", "backslashreplace", "namereplace",

"surrogatepass", "surrogateescape",

};

const char *Py_hexdigits = "0123456789abcdef";

/* --- Codec Registry ----------------------------------------------------- */

int PyCodec_Register(PyObject *search_function)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

assert(interp->codecs.initialized);

if (search_function == NULL) {

PyErr_BadArgument();

goto onError;

}

if (!PyCallable_Check(search_function)) {

PyErr_SetString(PyExc_TypeError, "argument must be callable");

goto onError;

}

#ifdef Py_GIL_DISABLED

PyMutex_Lock(&interp->codecs.search_path_mutex);

#endif

int ret = PyList_Append(interp->codecs.search_path, search_function);

#ifdef Py_GIL_DISABLED

PyMutex_Unlock(&interp->codecs.search_path_mutex);

#endif

return ret;

onError:

return -1;

}

int

PyCodec_Unregister(PyObject *search_function)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

if (interp->codecs.initialized != 1) {

/* Do nothing if codecs state was cleared (only possible during

interpreter shutdown). */

return 0;

}

PyObject *codec_search_path = interp->codecs.search_path;

assert(PyList_CheckExact(codec_search_path));

for (Py_ssize_t i = 0; i < PyList_GET_SIZE(codec_search_path); i++) {

#ifdef Py_GIL_DISABLED

PyMutex_Lock(&interp->codecs.search_path_mutex);

#endif

PyObject *item = PyList_GetItemRef(codec_search_path, i);

int ret = 1;

if (item == search_function) {

// We hold a reference to the item, so its destructor can't run

// while we hold search_path_mutex.

ret = PyList_SetSlice(codec_search_path, i, i+1, NULL);

}

#ifdef Py_GIL_DISABLED

PyMutex_Unlock(&interp->codecs.search_path_mutex);

#endif

Py_DECREF(item);

if (ret != 1) {

assert(interp->codecs.search_cache != NULL);

assert(PyDict_CheckExact(interp->codecs.search_cache));

PyDict_Clear(interp->codecs.search_cache);

return ret;

}

}

return 0;

}

extern int _Py_normalize_encoding(const char *, char *, size_t);

/* Convert a string to a normalized Python string(decoded from UTF-8): all characters are

converted to lower case, spaces and hyphens are replaced with underscores. */

static

PyObject *normalizestring(const char *string)

{

size_t len = strlen(string);

char *encoding;

PyObject *v;

if (len > PY_SSIZE_T_MAX) {

PyErr_SetString(PyExc_OverflowError, "string is too large");

return NULL;

}

encoding = PyMem_Malloc(len + 1);

if (encoding == NULL)

return PyErr_NoMemory();

if (!_Py_normalize_encoding(string, encoding, len + 1))

{

PyErr_SetString(PyExc_RuntimeError, "_Py_normalize_encoding() failed");

PyMem_Free(encoding);

return NULL;

}

v = PyUnicode_FromString(encoding);

PyMem_Free(encoding);

return v;

}

/* Lookup the given encoding and return a tuple providing the codec

facilities.

The encoding string is looked up converted to all lower-case

characters. This makes encodings looked up through this mechanism

effectively case-insensitive.

If no codec is found, a LookupError is set and NULL returned.

As side effect, this tries to load the encodings package, if not

yet done. This is part of the lazy load strategy for the encodings

package.

*/

PyObject *_PyCodec_Lookup(const char *encoding)

{

if (encoding == NULL) {

PyErr_BadArgument();

return NULL;

}

PyInterpreterState *interp = _PyInterpreterState_GET();

assert(interp->codecs.initialized);

/* Convert the encoding to a normalized Python string: all

characters are converted to lower case, spaces and hyphens are

replaced with underscores. */

PyObject *v = normalizestring(encoding);

if (v == NULL) {

return NULL;

}

/* Intern the string. We'll make it immortal later if lookup succeeds. */

_PyUnicode_InternMortal(interp, &v);

/* First, try to lookup the name in the registry dictionary */

PyObject *result;

if (PyDict_GetItemRef(interp->codecs.search_cache, v, &result) < 0) {

goto onError;

}

if (result != NULL) {

Py_DECREF(v);

return result;

}

/* Next, scan the search functions in order of registration */

const Py_ssize_t len = PyList_Size(interp->codecs.search_path);

if (len < 0)

goto onError;

if (len == 0) {

PyErr_SetString(PyExc_LookupError,

"no codec search functions registered: "

"can't find encoding");

goto onError;

}

Py_ssize_t i;

for (i = 0; i < len; i++) {

PyObject *func;

func = PyList_GetItemRef(interp->codecs.search_path, i);

if (func == NULL)

goto onError;

result = PyObject_CallOneArg(func, v);

Py_DECREF(func);

if (result == NULL)

goto onError;

if (result == Py_None) {

Py_CLEAR(result);

continue;

}

if (!PyTuple_Check(result) || PyTuple_GET_SIZE(result) != 4) {

PyErr_SetString(PyExc_TypeError,

"codec search functions must return 4-tuples");

Py_DECREF(result);

goto onError;

}

break;

}

if (result == NULL) {

/* XXX Perhaps we should cache misses too ? */

PyErr_Format(PyExc_LookupError,

"unknown encoding: %s", encoding);

goto onError;

}

_PyUnicode_InternImmortal(interp, &v);

/* Cache and return the result */

if (PyDict_SetItem(interp->codecs.search_cache, v, result) < 0) {

Py_DECREF(result);

goto onError;

}

Py_DECREF(v);

return result;

onError:

Py_DECREF(v);

return NULL;

}

/* Codec registry encoding check API. */

int PyCodec_KnownEncoding(const char *encoding)

{

PyObject *codecs;

codecs = _PyCodec_Lookup(encoding);

if (!codecs) {

PyErr_Clear();

return 0;

}

else {

Py_DECREF(codecs);

return 1;

}

}

static

PyObject *args_tuple(PyObject *object,

const char *errors)

{

PyObject *args;

args = PyTuple_New(1 + (errors != NULL));

if (args == NULL)

return NULL;

PyTuple_SET_ITEM(args, 0, Py_NewRef(object));

if (errors) {

PyObject *v;

v = PyUnicode_FromString(errors);

if (v == NULL) {

Py_DECREF(args);

return NULL;

}

PyTuple_SET_ITEM(args, 1, v);

}

return args;

}

/* Helper function to get a codec item */

static

PyObject *codec_getitem(const char *encoding, int index)

{

PyObject *codecs;

PyObject *v;

codecs = _PyCodec_Lookup(encoding);

if (codecs == NULL)

return NULL;

v = PyTuple_GET_ITEM(codecs, index);

Py_DECREF(codecs);

return Py_NewRef(v);

}

/* Helper functions to create an incremental codec. */

static

PyObject *codec_makeincrementalcodec(PyObject *codec_info,

const char *errors,

const char *attrname)

{

PyObject *ret, *inccodec;

inccodec = PyObject_GetAttrString(codec_info, attrname);

if (inccodec == NULL)

return NULL;

if (errors)

ret = PyObject_CallFunction(inccodec, "s", errors);

else

ret = _PyObject_CallNoArgs(inccodec);

Py_DECREF(inccodec);

return ret;

}

static

PyObject *codec_getincrementalcodec(const char *encoding,

const char *errors,

const char *attrname)

{

PyObject *codec_info, *ret;

codec_info = _PyCodec_Lookup(encoding);

if (codec_info == NULL)

return NULL;

ret = codec_makeincrementalcodec(codec_info, errors, attrname);

Py_DECREF(codec_info);

return ret;

}

/* Helper function to create a stream codec. */

static

PyObject *codec_getstreamcodec(const char *encoding,

PyObject *stream,

const char *errors,

const int index)

{

PyObject *codecs, *streamcodec, *codeccls;

codecs = _PyCodec_Lookup(encoding);

if (codecs == NULL)

return NULL;

codeccls = PyTuple_GET_ITEM(codecs, index);

if (errors != NULL)

streamcodec = PyObject_CallFunction(codeccls, "Os", stream, errors);

else

streamcodec = PyObject_CallOneArg(codeccls, stream);

Py_DECREF(codecs);

return streamcodec;

}

/* Helpers to work with the result of _PyCodec_Lookup

*/

PyObject *_PyCodecInfo_GetIncrementalDecoder(PyObject *codec_info,

const char *errors)

{

return codec_makeincrementalcodec(codec_info, errors,

"incrementaldecoder");

}

PyObject *_PyCodecInfo_GetIncrementalEncoder(PyObject *codec_info,

const char *errors)

{

return codec_makeincrementalcodec(codec_info, errors,

"incrementalencoder");

}

/* Convenience APIs to query the Codec registry.

All APIs return a codec object with incremented refcount.

*/

PyObject *PyCodec_Encoder(const char *encoding)

{

return codec_getitem(encoding, 0);

}

PyObject *PyCodec_Decoder(const char *encoding)

{

return codec_getitem(encoding, 1);

}

PyObject *PyCodec_IncrementalEncoder(const char *encoding,

const char *errors)

{

return codec_getincrementalcodec(encoding, errors, "incrementalencoder");

}

PyObject *PyCodec_IncrementalDecoder(const char *encoding,

const char *errors)

{

return codec_getincrementalcodec(encoding, errors, "incrementaldecoder");

}

PyObject *PyCodec_StreamReader(const char *encoding,

PyObject *stream,

const char *errors)

{

return codec_getstreamcodec(encoding, stream, errors, 2);

}

PyObject *PyCodec_StreamWriter(const char *encoding,

PyObject *stream,

const char *errors)

{

return codec_getstreamcodec(encoding, stream, errors, 3);

}

/* Encode an object (e.g. a Unicode object) using the given encoding

and return the resulting encoded object (usually a Python string).

errors is passed to the encoder factory as argument if non-NULL. */

static PyObject *

_PyCodec_EncodeInternal(PyObject *object,

PyObject *encoder,

const char *encoding,

const char *errors)

{

PyObject *args = NULL, *result = NULL;

PyObject *v = NULL;

args = args_tuple(object, errors);

if (args == NULL)

goto onError;

result = PyObject_Call(encoder, args, NULL);

if (result == NULL) {

_PyErr_FormatNote("%s with '%s' codec failed", "encoding", encoding);

goto onError;

}

if (!PyTuple_Check(result) ||

PyTuple_GET_SIZE(result) != 2) {

PyErr_SetString(PyExc_TypeError,

"encoder must return a tuple (object, integer)");

goto onError;

}

v = Py_NewRef(PyTuple_GET_ITEM(result,0));

/* We don't check or use the second (integer) entry. */

Py_DECREF(args);

Py_DECREF(encoder);

Py_DECREF(result);

return v;

onError:

Py_XDECREF(result);

Py_XDECREF(args);

Py_XDECREF(encoder);

return NULL;

}

/* Decode an object (usually a Python string) using the given encoding

and return an equivalent object (e.g. a Unicode object).

errors is passed to the decoder factory as argument if non-NULL. */

static PyObject *

_PyCodec_DecodeInternal(PyObject *object,

PyObject *decoder,

const char *encoding,

const char *errors)

{

PyObject *args = NULL, *result = NULL;

PyObject *v;

args = args_tuple(object, errors);

if (args == NULL)

goto onError;

result = PyObject_Call(decoder, args, NULL);

if (result == NULL) {

_PyErr_FormatNote("%s with '%s' codec failed", "decoding", encoding);

goto onError;

}

if (!PyTuple_Check(result) ||

PyTuple_GET_SIZE(result) != 2) {

PyErr_SetString(PyExc_TypeError,

"decoder must return a tuple (object,integer)");

goto onError;

}

v = Py_NewRef(PyTuple_GET_ITEM(result,0));

/* We don't check or use the second (integer) entry. */

Py_DECREF(args);

Py_DECREF(decoder);

Py_DECREF(result);

return v;

onError:

Py_XDECREF(args);

Py_XDECREF(decoder);

Py_XDECREF(result);

return NULL;

}

/* Generic encoding/decoding API */

PyObject *PyCodec_Encode(PyObject *object,

const char *encoding,

const char *errors)

{

PyObject *encoder;

encoder = PyCodec_Encoder(encoding);

if (encoder == NULL)

return NULL;

return _PyCodec_EncodeInternal(object, encoder, encoding, errors);

}

PyObject *PyCodec_Decode(PyObject *object,

const char *encoding,

const char *errors)

{

PyObject *decoder;

decoder = PyCodec_Decoder(encoding);

if (decoder == NULL)

return NULL;

return _PyCodec_DecodeInternal(object, decoder, encoding, errors);

}

/* Text encoding/decoding API */

PyObject * _PyCodec_LookupTextEncoding(const char *encoding,

const char *alternate_command)

{

PyObject *codec;

PyObject *attr;

int is_text_codec;

codec = _PyCodec_Lookup(encoding);

if (codec == NULL)

return NULL;

/* Backwards compatibility: assume any raw tuple describes a text

* encoding, and the same for anything lacking the private

* attribute.

*/

if (!PyTuple_CheckExact(codec)) {

if (PyObject_GetOptionalAttr(codec, &_Py_ID(_is_text_encoding), &attr) < 0) {

Py_DECREF(codec);

return NULL;

}

if (attr != NULL) {

is_text_codec = PyObject_IsTrue(attr);

Py_DECREF(attr);

if (is_text_codec <= 0) {

Py_DECREF(codec);

if (!is_text_codec)

PyErr_Format(PyExc_LookupError,

"'%.400s' is not a text encoding; "

"use %s to handle arbitrary codecs",

encoding, alternate_command);

return NULL;

}

}

}

/* This appears to be a valid text encoding */

return codec;

}

static

PyObject *codec_getitem_checked(const char *encoding,

const char *alternate_command,

int index)

{

PyObject *codec;

PyObject *v;

codec = _PyCodec_LookupTextEncoding(encoding, alternate_command);

if (codec == NULL)

return NULL;

v = Py_NewRef(PyTuple_GET_ITEM(codec, index));

Py_DECREF(codec);

return v;

}

static PyObject * _PyCodec_TextEncoder(const char *encoding)

{

return codec_getitem_checked(encoding, "codecs.encode()", 0);

}

static PyObject * _PyCodec_TextDecoder(const char *encoding)

{

return codec_getitem_checked(encoding, "codecs.decode()", 1);

}

PyObject *_PyCodec_EncodeText(PyObject *object,

const char *encoding,

const char *errors)

{

PyObject *encoder;

encoder = _PyCodec_TextEncoder(encoding);

if (encoder == NULL)

return NULL;

return _PyCodec_EncodeInternal(object, encoder, encoding, errors);

}

PyObject *_PyCodec_DecodeText(PyObject *object,

const char *encoding,

const char *errors)

{

PyObject *decoder;

decoder = _PyCodec_TextDecoder(encoding);

if (decoder == NULL)

return NULL;

return _PyCodec_DecodeInternal(object, decoder, encoding, errors);

}

/* Register the error handling callback function error under the name

name. This function will be called by the codec when it encounters

an unencodable characters/undecodable bytes and doesn't know the

callback name, when name is specified as the error parameter

in the call to the encode/decode function.

Return 0 on success, -1 on error */

int PyCodec_RegisterError(const char *name, PyObject *error)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

assert(interp->codecs.initialized);

if (!PyCallable_Check(error)) {

PyErr_SetString(PyExc_TypeError, "handler must be callable");

return -1;

}

return PyDict_SetItemString(interp->codecs.error_registry,

name, error);

}

int _PyCodec_UnregisterError(const char *name)

{

for (size_t i = 0; i < Py_ARRAY_LENGTH(codecs_builtin_error_handlers); ++i) {

if (strcmp(name, codecs_builtin_error_handlers[i]) == 0) {

PyErr_Format(PyExc_ValueError,

"cannot un-register built-in error handler '%s'", name);

return -1;

}

}

PyInterpreterState *interp = _PyInterpreterState_GET();

assert(interp->codecs.initialized);

return PyDict_PopString(interp->codecs.error_registry, name, NULL);

}

/* Lookup the error handling callback function registered under the

name error. As a special case NULL can be passed, in which case

the error handling callback for strict encoding will be returned. */

PyObject *PyCodec_LookupError(const char *name)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

assert(interp->codecs.initialized);

if (name==NULL)

name = "strict";

PyObject *handler;

if (PyDict_GetItemStringRef(interp->codecs.error_registry, name, &handler) < 0) {

return NULL;

}

if (handler == NULL) {

PyErr_Format(PyExc_LookupError, "unknown error handler name '%.400s'", name);

return NULL;

}

return handler;

}

static inline void

wrong_exception_type(PyObject *exc)

{

PyErr_Format(PyExc_TypeError,

"don't know how to handle %T in error callback", exc);

}

#define _PyIsUnicodeEncodeError(EXC) \

PyObject_TypeCheck(EXC, (PyTypeObject *)PyExc_UnicodeEncodeError)

#define _PyIsUnicodeDecodeError(EXC) \

PyObject_TypeCheck(EXC, (PyTypeObject *)PyExc_UnicodeDecodeError)

#define _PyIsUnicodeTranslateError(EXC) \

PyObject_TypeCheck(EXC, (PyTypeObject *)PyExc_UnicodeTranslateError)

// --- codecs handlers: utilities ---------------------------------------------

/*

* Return the number of characters (including special prefixes)

* needed to represent 'ch' by codec_handler_write_unicode_hex().

*/

static inline Py_ssize_t

codec_handler_unicode_hex_width(Py_UCS4 ch)

{

if (ch >= 0x10000) {

// format: '\\' + 'U' + 8 hex digits

return 1 + 1 + 8;

}

else if (ch >= 0x100) {

// format: '\\' + 'u' + 4 hex digits

return 1 + 1 + 4;

}

else {

// format: '\\' + 'x' + 2 hex digits

return 1 + 1 + 2;

}

}

/*

* Write the hexadecimal representation of 'ch' to the buffer pointed by 'p'

* using 2, 4, or 8 characters prefixed by '\x', '\u', or '\U' respectively.

*/

static inline void

codec_handler_write_unicode_hex(Py_UCS1 **p, Py_UCS4 ch)

{

*(*p)++ = '\\';

if (ch >= 0x10000) {

*(*p)++ = 'U';

*(*p)++ = Py_hexdigits[(ch >> 28) & 0xf];

*(*p)++ = Py_hexdigits[(ch >> 24) & 0xf];

*(*p)++ = Py_hexdigits[(ch >> 20) & 0xf];

*(*p)++ = Py_hexdigits[(ch >> 16) & 0xf];

*(*p)++ = Py_hexdigits[(ch >> 12) & 0xf];

*(*p)++ = Py_hexdigits[(ch >> 8) & 0xf];

}

else if (ch >= 0x100) {

*(*p)++ = 'u';

*(*p)++ = Py_hexdigits[(ch >> 12) & 0xf];

*(*p)++ = Py_hexdigits[(ch >> 8) & 0xf];

}

else {

*(*p)++ = 'x';

}

*(*p)++ = Py_hexdigits[(ch >> 4) & 0xf];

*(*p)++ = Py_hexdigits[ch & 0xf];

}

// --- handler: 'strict' ------------------------------------------------------

PyObject *PyCodec_StrictErrors(PyObject *exc)

{

if (PyExceptionInstance_Check(exc)) {

PyErr_SetObject(PyExceptionInstance_Class(exc), exc);

}

else {

PyErr_SetString(PyExc_TypeError, "codec must pass exception instance");

}

return NULL;

}

// --- handler: 'ignore' ------------------------------------------------------

static PyObject *

_PyCodec_IgnoreError(PyObject *exc, int as_bytes)

{

Py_ssize_t end;

if (_PyUnicodeError_GetParams(exc, NULL, NULL, NULL,

&end, NULL, as_bytes) < 0)

{

return NULL;

}

return Py_BuildValue("(Nn)", Py_GetConstant(Py_CONSTANT_EMPTY_STR), end);

}

PyObject *PyCodec_IgnoreErrors(PyObject *exc)

{

if (_PyIsUnicodeEncodeError(exc) || _PyIsUnicodeTranslateError(exc)) {

return _PyCodec_IgnoreError(exc, false);

}

else if (_PyIsUnicodeDecodeError(exc)) {

return _PyCodec_IgnoreError(exc, true);

}

else {

wrong_exception_type(exc);

return NULL;

}

}

PyObject *PyCodec_ReplaceErrors(PyObject *exc)

{

Py_ssize_t start, end, slen;

if (PyObject_TypeCheck(exc, (PyTypeObject *)PyExc_UnicodeEncodeError)) {

if (_PyUnicodeError_GetParams(exc, NULL, NULL,

&start, &end, &slen, false) < 0) {

return NULL;

}

PyObject *res = PyUnicode_New(slen, '?');

if (res == NULL) {

return NULL;

}

assert(PyUnicode_KIND(res) == PyUnicode_1BYTE_KIND);

Py_UCS1 *outp = PyUnicode_1BYTE_DATA(res);

memset(outp, '?', sizeof(Py_UCS1) * slen);

assert(_PyUnicode_CheckConsistency(res, 1));

return Py_BuildValue("(Nn)", res, end);

}

else if (PyObject_TypeCheck(exc, (PyTypeObject *)PyExc_UnicodeDecodeError)) {

if (_PyUnicodeError_GetParams(exc, NULL, NULL,

NULL, &end, NULL, true) < 0) {

return NULL;

}

return Py_BuildValue("(Cn)",

(int)Py_UNICODE_REPLACEMENT_CHARACTER,

end);

}

else if (PyObject_TypeCheck(exc, (PyTypeObject *)PyExc_UnicodeTranslateError)) {

if (_PyUnicodeError_GetParams(exc, NULL, NULL,

&start, &end, &slen, false) < 0) {

return NULL;

}

PyObject *res = PyUnicode_New(slen, Py_UNICODE_REPLACEMENT_CHARACTER);

if (res == NULL) {

return NULL;

}

assert(slen == 0 || PyUnicode_KIND(res) == PyUnicode_2BYTE_KIND);

Py_UCS2 *outp = PyUnicode_2BYTE_DATA(res);

for (Py_ssize_t i = 0; i < slen; ++i) {

outp[i] = Py_UNICODE_REPLACEMENT_CHARACTER;

}

assert(_PyUnicode_CheckConsistency(res, 1));

return Py_BuildValue("(Nn)", res, end);

}

else {

wrong_exception_type(exc);

return NULL;

}

}

PyObject *PyCodec_XMLCharRefReplaceErrors(PyObject *exc)

{

if (!PyObject_TypeCheck(exc, (PyTypeObject *)PyExc_UnicodeEncodeError)) {

wrong_exception_type(exc);

return NULL;

}

PyObject *obj;

Py_ssize_t objlen, start, end, slen;

if (_PyUnicodeError_GetParams(exc,

&obj, &objlen,

&start, &end, &slen, false) < 0)

{

return NULL;

}

// The number of characters that each character 'ch' contributes

// in the result is 2 + k + 1, where k = min{t >= 1 | 10^t > ch}

// and will be formatted as "&#" + DIGITS + ";". Since the Unicode

// range is below 10^7, each "block" requires at most 2 + 7 + 1

// characters.

if (slen > PY_SSIZE_T_MAX / (2 + 7 + 1)) {

end = start + PY_SSIZE_T_MAX / (2 + 7 + 1);

end = Py_MIN(end, objlen);

slen = Py_MAX(0, end - start);

}

Py_ssize_t ressize = 0;

for (Py_ssize_t i = start; i < end; ++i) {

/* object is guaranteed to be "ready" */

Py_UCS4 ch = PyUnicode_READ_CHAR(obj, i);

if (ch < 10) {

ressize += 2 + 1 + 1;

}

else if (ch < 100) {

ressize += 2 + 2 + 1;

}

else if (ch < 1000) {

ressize += 2 + 3 + 1;

}

else if (ch < 10000) {

ressize += 2 + 4 + 1;

}

else if (ch < 100000) {

ressize += 2 + 5 + 1;

}

else if (ch < 1000000) {

ressize += 2 + 6 + 1;

}

else {

assert(ch < 10000000);

ressize += 2 + 7 + 1;

}

}

/* allocate replacement */

PyObject *res = PyUnicode_New(ressize, 127);

if (res == NULL) {

Py_DECREF(obj);

return NULL;

}

Py_UCS1 *outp = PyUnicode_1BYTE_DATA(res);

/* generate replacement */

for (Py_ssize_t i = start; i < end; ++i) {

int digits, base;

Py_UCS4 ch = PyUnicode_READ_CHAR(obj, i);

if (ch < 10) {

digits = 1;

base = 1;

}

else if (ch < 100) {

digits = 2;

base = 10;

}

else if (ch < 1000) {

digits = 3;

base = 100;

}

else if (ch < 10000) {

digits = 4;

base = 1000;

}

else if (ch < 100000) {

digits = 5;

base = 10000;

}

else if (ch < 1000000) {

digits = 6;

base = 100000;

}

else {

assert(ch < 10000000);

digits = 7;

base = 1000000;

}

*outp++ = '&';

*outp++ = '#';

while (digits-- > 0) {

assert(base >= 1);

*outp++ = '0' + ch / base;

ch %= base;

base /= 10;

}

*outp++ = ';';

}

assert(_PyUnicode_CheckConsistency(res, 1));

PyObject *restuple = Py_BuildValue("(Nn)", res, end);

Py_DECREF(obj);

return restuple;

}

PyObject *PyCodec_BackslashReplaceErrors(PyObject *exc)

{

PyObject *obj;

Py_ssize_t objlen, start, end, slen;

if (PyObject_TypeCheck(exc, (PyTypeObject *)PyExc_UnicodeDecodeError)) {

if (_PyUnicodeError_GetParams(exc,

&obj, &objlen,

&start, &end, &slen, true) < 0)

{

return NULL;

}

PyObject *res = PyUnicode_New(4 * slen, 127);

if (res == NULL) {

Py_DECREF(obj);

return NULL;

}

Py_UCS1 *outp = PyUnicode_1BYTE_DATA(res);

const unsigned char *p = (const unsigned char *)PyBytes_AS_STRING(obj);

for (Py_ssize_t i = start; i < end; i++, outp += 4) {

const unsigned char ch = p[i];

outp[0] = '\\';

outp[1] = 'x';

outp[2] = Py_hexdigits[(ch >> 4) & 0xf];

outp[3] = Py_hexdigits[ch & 0xf];

}

assert(_PyUnicode_CheckConsistency(res, 1));

Py_DECREF(obj);

return Py_BuildValue("(Nn)", res, end);

}

if (

PyObject_TypeCheck(exc, (PyTypeObject *)PyExc_UnicodeEncodeError)

|| PyObject_TypeCheck(exc, (PyTypeObject *)PyExc_UnicodeTranslateError)

) {

if (_PyUnicodeError_GetParams(exc,

&obj, &objlen,

&start, &end, &slen, false) < 0)

{

return NULL;

}

}

else {

wrong_exception_type(exc);

return NULL;

}

// The number of characters that each character 'ch' contributes

// in the result is 1 + 1 + k, where k >= min{t >= 1 | 16^t > ch}

// and will be formatted as "\\" + ('U'|'u'|'x') + HEXDIGITS,

// where the number of hexdigits is either 2, 4, or 8 (not 6).

// Since the Unicode range is below 10^7, we choose k = 8 whence

// each "block" requires at most 1 + 1 + 8 characters.

if (slen > PY_SSIZE_T_MAX / (1 + 1 + 8)) {

end = start + PY_SSIZE_T_MAX / (1 + 1 + 8);

end = Py_MIN(end, objlen);

slen = Py_MAX(0, end - start);

}

Py_ssize_t ressize = 0;

for (Py_ssize_t i = start; i < end; ++i) {

Py_UCS4 c = PyUnicode_READ_CHAR(obj, i);

ressize += codec_handler_unicode_hex_width(c);