#include "Python.h"

#include "pycore_audit.h" // _Py_AuditHookEntry

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

#include "pycore_ceval.h" // _PyEval_SetAsyncGenFinalizer()

#include "pycore_frame.h" // _PyInterpreterFrame

#include "pycore_import.h" // _PyImport_SetDLOpenFlags()

#include "pycore_initconfig.h" // _PyStatus_EXCEPTION()

#include "pycore_interpframe.h" // _PyFrame_GetFirstComplete()

#include "pycore_long.h" // _PY_LONG_MAX_STR_DIGITS_THRESHOLD

#include "pycore_modsupport.h" // _PyModule_CreateInitialized()

#include "pycore_namespace.h" // _PyNamespace_New()

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

#include "pycore_optimizer.h" // _PyDumpExecutors()

#include "pycore_pathconfig.h" // _PyPathConfig_ComputeSysPath0()

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

#include "pycore_pylifecycle.h" // _PyErr_WriteUnraisableDefaultHook()

#include "pycore_pymath.h" // _PY_SHORT_FLOAT_REPR

#include "pycore_pymem.h" // _PyMem_DefaultRawFree()

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

#include "pycore_pystats.h" // _Py_PrintSpecializationStats()

#include "pycore_runtime.h" // _PyRuntimeState_Get*()

#include "pycore_structseq.h" // _PyStructSequence_InitBuiltinWithFlags()

#include "pycore_sysmodule.h" // export _PySys_GetSizeOf()

#include "pycore_unicodeobject.h" // _PyUnicode_InternImmortal()

#include "pydtrace.h" // PyDTrace_AUDIT()

#include "osdefs.h" // DELIM

#include "stdlib_module_names.h" // _Py_stdlib_module_names

#ifdef HAVE_UNISTD_H

# include <unistd.h> // getpid()

#ifdef MS_WINDOWS

# ifndef WIN32_LEAN_AND_MEAN

# define WIN32_LEAN_AND_MEAN

# include <windows.h>

#endif /* MS_WINDOWS */

#ifdef MS_COREDLL

extern void *PyWin_DLLhModule;

extern const char *PyWin_DLLVersionString;

#ifdef __EMSCRIPTEN__

# include <emscripten.h>

#ifdef __wasi__

# include <wasi/version.h>

#ifdef HAVE_FCNTL_H

# include <fcntl.h>

#include "clinic/sysmodule.c.h"

PyObject *

PySys_GetAttr(PyObject *name)

{

if (!PyUnicode_Check(name)) {

PyErr_Format(PyExc_TypeError,

"attribute name must be string, not '%T'",

name);

return NULL;

}

PyThreadState *tstate = _PyThreadState_GET();

PyObject *sysdict = tstate->interp->sysdict;

if (sysdict == NULL) {

PyErr_SetString(PyExc_RuntimeError, "no sys module");

return NULL;

}

PyObject *value;

if (PyDict_GetItemRef(sysdict, name, &value) == 0) {

PyErr_Format(PyExc_RuntimeError, "lost sys.%U", name);

}

return value;

}

PyObject *

PySys_GetAttrString(const char *name)

{

PyThreadState *tstate = _PyThreadState_GET();

PyObject *sysdict = tstate->interp->sysdict;

if (sysdict == NULL) {

PyErr_SetString(PyExc_RuntimeError, "no sys module");

return NULL;

}

PyObject *value;

if (PyDict_GetItemStringRef(sysdict, name, &value) == 0) {

PyErr_Format(PyExc_RuntimeError, "lost sys.%s", name);

}

return value;

}

PySys_GetOptionalAttr(PyObject *name, PyObject **value)

{

if (!PyUnicode_Check(name)) {

PyErr_Format(PyExc_TypeError,

"attribute name must be string, not '%T'",

name);

*value = NULL;

return -1;

}

PyThreadState *tstate = _PyThreadState_GET();

PyObject *sysdict = tstate->interp->sysdict;

if (sysdict == NULL) {

*value = NULL;

return 0;

}

return PyDict_GetItemRef(sysdict, name, value);

}

PySys_GetOptionalAttrString(const char *name, PyObject **value)

{

PyThreadState *tstate = _PyThreadState_GET();

PyObject *sysdict = tstate->interp->sysdict;

if (sysdict == NULL) {

*value = NULL;

return 0;

}

return PyDict_GetItemStringRef(sysdict, name, value);

}

PyObject *

PySys_GetObject(const char *name)

{

PyThreadState *tstate = _PyThreadState_GET();

PyObject *sysdict = tstate->interp->sysdict;

if (sysdict == NULL) {

return NULL;

}

PyObject *exc = _PyErr_GetRaisedException(tstate);

PyObject *value;

(void) PyDict_GetItemStringRef(sysdict, name, &value);

/* XXX Suppress a new exception if it was raised and restore

if (_PyErr_Occurred(tstate)) {

PyErr_FormatUnraisable("Exception ignored in PySys_GetObject()");

}

_PyErr_SetRaisedException(tstate, exc);

Py_XDECREF(value); // return a borrowed reference

return value;

}

static int

sys_set_object(PyInterpreterState *interp, PyObject *key, PyObject *v)

{

if (key == NULL) {

return -1;

}

PyObject *sd = interp->sysdict;

if (sd == NULL) {

PyErr_SetString(PyExc_RuntimeError, "no sys module");

return -1;

}

if (v == NULL) {

if (PyDict_Pop(sd, key, NULL) < 0) {

return -1;

}

return 0;

}

else {

return PyDict_SetItem(sd, key, v);

}

}

_PySys_SetAttr(PyObject *key, PyObject *v)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

return sys_set_object(interp, key, v);

}

static int

sys_set_object_str(PyInterpreterState *interp, const char *name, PyObject *v)

{

PyObject *key = v ? PyUnicode_InternFromString(name)

: PyUnicode_FromString(name);

int r = sys_set_object(interp, key, v);

Py_XDECREF(key);

return r;

}

PySys_SetObject(const char *name, PyObject *v)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

return sys_set_object_str(interp, name, v);

}

_PySys_ClearAttrString(PyInterpreterState *interp,

const char *name, int verbose)

{

if (verbose) {

PySys_WriteStderr("# clear sys.%s\n", name);

}

/* To play it safe, we set the attr to None instead of deleting it. */

if (PyDict_SetItemString(interp->sysdict, name, Py_None) < 0) {

return -1;

}

return 0;

}

static int

should_audit(PyInterpreterState *interp)

{

/* interp must not be NULL, but test it just in case for extra safety */

assert(interp != NULL);

if (!interp) {

return 0;

}

return (interp->runtime->audit_hooks.head

|| interp->audit_hooks

|| PyDTrace_AUDIT_ENABLED());

}

static int

sys_audit_tstate(PyThreadState *ts, const char *event,

const char *argFormat, va_list vargs)

{

assert(event != NULL);

assert(!argFormat || !strchr(argFormat, 'N'));

if (!ts) {

/* Audit hooks cannot be called with a NULL thread state */

return 0;

}

/* The current implementation cannot be called if tstate is not

assert(ts == _PyThreadState_GET());

/* Early exit when no hooks are registered */

PyInterpreterState *is = ts->interp;

if (!should_audit(is)) {

return 0;

}

PyObject *eventName = NULL;

PyObject *eventArgs = NULL;

PyObject *hooks = NULL;

PyObject *hook = NULL;

int res = -1;

int dtrace = PyDTrace_AUDIT_ENABLED();

PyObject *exc = _PyErr_GetRaisedException(ts);

/* Initialize event args now */

if (argFormat && argFormat[0]) {

eventArgs = Py_VaBuildValue(argFormat, vargs);

if (eventArgs && !PyTuple_Check(eventArgs)) {

PyObject *argTuple = PyTuple_Pack(1, eventArgs);

Py_SETREF(eventArgs, argTuple);

}

}

else {

eventArgs = PyTuple_New(0);

}

if (!eventArgs) {

goto exit;

}

/* Call global hooks

_Py_AuditHookEntry *e = is->runtime->audit_hooks.head;

for (; e; e = e->next) {

if (e->hookCFunction(event, eventArgs, e->userData) < 0) {

goto exit;

}

}

/* Dtrace USDT point */

if (dtrace) {

PyDTrace_AUDIT(event, (void *)eventArgs);

}

/* Call interpreter hooks */

if (is->audit_hooks) {

eventName = PyUnicode_FromString(event);

if (!eventName) {

goto exit;

}

hooks = PyObject_GetIter(is->audit_hooks);

if (!hooks) {

goto exit;

}

/* Disallow tracing in hooks unless explicitly enabled */

PyThreadState_EnterTracing(ts);

while ((hook = PyIter_Next(hooks)) != NULL) {

PyObject *o;

int canTrace = PyObject_GetOptionalAttr(hook, &_Py_ID(__cantrace__), &o);

if (o) {

canTrace = PyObject_IsTrue(o);

Py_DECREF(o);

}

if (canTrace < 0) {

break;

}

if (canTrace) {

PyThreadState_LeaveTracing(ts);

}

PyObject* args[2] = {eventName, eventArgs};

o = _PyObject_VectorcallTstate(ts, hook, args, 2, NULL);

if (canTrace) {

PyThreadState_EnterTracing(ts);

}

if (!o) {

break;

}

Py_DECREF(o);

Py_CLEAR(hook);

}

PyThreadState_LeaveTracing(ts);

if (_PyErr_Occurred(ts)) {

goto exit;

}

}

res = 0;

exit:

Py_XDECREF(hook);

Py_XDECREF(hooks);

Py_XDECREF(eventName);

Py_XDECREF(eventArgs);

if (!res) {

_PyErr_SetRaisedException(ts, exc);

}

else {

assert(_PyErr_Occurred(ts));

Py_XDECREF(exc);

}

return res;

}

_PySys_Audit(PyThreadState *tstate, const char *event,

const char *argFormat, ...)

{

va_list vargs;

va_start(vargs, argFormat);

int res = sys_audit_tstate(tstate, event, argFormat, vargs);

va_end(vargs);

return res;

}

PySys_Audit(const char *event, const char *argFormat, ...)

{

PyThreadState *tstate = _PyThreadState_GET();

va_list vargs;

va_start(vargs, argFormat);

int res = sys_audit_tstate(tstate, event, argFormat, vargs);

va_end(vargs);

return res;

}

PySys_AuditTuple(const char *event, PyObject *args)

{

if (args == NULL) {

return PySys_Audit(event, NULL);

}

if (!PyTuple_Check(args)) {

PyErr_Format(PyExc_TypeError, "args must be tuple, got %s",

Py_TYPE(args)->tp_name);

return -1;

}

return PySys_Audit(event, "O", args);

}

_PySys_ClearAuditHooks(PyThreadState *ts)

{

assert(ts != NULL);

if (!ts) {

return;

}

_PyRuntimeState *runtime = ts->interp->runtime;

/* The hooks are global so we have to check for runtime finalization. */

PyThreadState *finalizing = _PyRuntimeState_GetFinalizing(runtime);

assert(finalizing == ts);

if (finalizing != ts) {

return;

}

const PyConfig *config = _PyInterpreterState_GetConfig(ts->interp);

if (config->verbose) {

PySys_WriteStderr("# clear sys.audit hooks\n");

}

/* Hooks can abort later hooks for this event, but cannot

_PySys_Audit(ts, "cpython._PySys_ClearAuditHooks", NULL);

_PyErr_Clear(ts);

/* We don't worry about the very unlikely race right here,

_Py_AuditHookEntry *e = runtime->audit_hooks.head, *n;

runtime->audit_hooks.head = NULL;

while (e) {

n = e->next;

PyMem_RawFree(e);

e = n;

}

}

static void

add_audit_hook_entry_unlocked(_PyRuntimeState *runtime,

_Py_AuditHookEntry *entry)

{

if (runtime->audit_hooks.head == NULL) {

runtime->audit_hooks.head = entry;

}

else {

_Py_AuditHookEntry *last = runtime->audit_hooks.head;

while (last->next) {

last = last->next;

}

last->next = entry;

}

}

PySys_AddAuditHook(Py_AuditHookFunction hook, void *userData)

{

/* tstate can be NULL, so access directly _PyRuntime:

_PyRuntimeState *runtime = &_PyRuntime;

PyThreadState *tstate;

if (_PyRuntimeState_GetInitialized(runtime)) {

tstate = _PyThreadState_GET();

}

else {

tstate = NULL;

}

/* Invoke existing audit hooks to allow them an opportunity to abort. */

/* Cannot invoke hooks until we are initialized */

if (tstate != NULL) {

if (_PySys_Audit(tstate, "sys.addaudithook", NULL) < 0) {

if (_PyErr_ExceptionMatches(tstate, PyExc_RuntimeError)) {

/* We do not report errors derived from RuntimeError */

_PyErr_Clear(tstate);

return 0;

}

return -1;

}

}

_Py_AuditHookEntry *e = (_Py_AuditHookEntry*)PyMem_RawMalloc(

sizeof(_Py_AuditHookEntry));

if (!e) {

if (tstate != NULL) {

_PyErr_NoMemory(tstate);

}

return -1;

}

e->next = NULL;

e->hookCFunction = (Py_AuditHookFunction)hook;

e->userData = userData;

PyMutex_Lock(&runtime->audit_hooks.mutex);

add_audit_hook_entry_unlocked(runtime, e);

PyMutex_Unlock(&runtime->audit_hooks.mutex);

return 0;

}

static PyObject *

sys_addaudithook_impl(PyObject *module, PyObject *hook)

{

PyThreadState *tstate = _PyThreadState_GET();

/* Invoke existing audit hooks to allow them an opportunity to abort. */

if (_PySys_Audit(tstate, "sys.addaudithook", NULL) < 0) {

if (_PyErr_ExceptionMatches(tstate, PyExc_RuntimeError)) {

/* We do not report errors derived from RuntimeError */

_PyErr_Clear(tstate);

Py_RETURN_NONE;

}

return NULL;

}

PyInterpreterState *interp = tstate->interp;

if (interp->audit_hooks == NULL) {

interp->audit_hooks = PyList_New(0);

if (interp->audit_hooks == NULL) {

return NULL;

}

/* Avoid having our list of hooks show up in the GC module */

PyObject_GC_UnTrack(interp->audit_hooks);

}

if (PyList_Append(interp->audit_hooks, hook) < 0) {

return NULL;

}

Py_RETURN_NONE;

}

static PyObject *

sys_audit_impl(PyObject *module, const char *event, PyObject *args)

{

PyThreadState *tstate = _PyThreadState_GET();

_Py_EnsureTstateNotNULL(tstate);

if (!should_audit(tstate->interp)) {

Py_RETURN_NONE;

}

int res = _PySys_Audit(tstate, event, "O", args);

if (res < 0) {

return NULL;

}

Py_RETURN_NONE;

}

static PyObject *

sys_breakpointhook(PyObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *keywords)

{

PyThreadState *tstate = _PyThreadState_GET();

assert(!_PyErr_Occurred(tstate));

char *envar = Py_GETENV("PYTHONBREAKPOINT");

if (envar == NULL || strlen(envar) == 0) {

envar = "pdb.set_trace";

}

else if (!strcmp(envar, "0")) {

/* The breakpoint is explicitly no-op'd. */

Py_RETURN_NONE;

}

/* According to POSIX the string returned by getenv() might be invalidated

envar = _PyMem_RawStrdup(envar);

if (envar == NULL) {

_PyErr_NoMemory(tstate);

return NULL;

}

const char *last_dot = strrchr(envar, '.');

const char *attrname = NULL;

PyObject *modulepath = NULL;

if (last_dot == NULL) {

/* The breakpoint is a built-in, e.g. PYTHONBREAKPOINT=int */

modulepath = &_Py_ID(builtins);

attrname = envar;

}

else if (last_dot != envar) {

/* Split on the last dot; */

modulepath = PyUnicode_FromStringAndSize(envar, last_dot - envar);

attrname = last_dot + 1;

}

else {

goto warn;

}

if (modulepath == NULL) {

PyMem_RawFree(envar);

return NULL;

}

PyObject *module = PyImport_Import(modulepath);

Py_DECREF(modulepath);

if (module == NULL) {

if (_PyErr_ExceptionMatches(tstate, PyExc_ImportError)) {

goto warn;

}

PyMem_RawFree(envar);

return NULL;

}

PyObject *hook = PyObject_GetAttrString(module, attrname);

Py_DECREF(module);

if (hook == NULL) {

if (_PyErr_ExceptionMatches(tstate, PyExc_AttributeError)) {

goto warn;

}

PyMem_RawFree(envar);

return NULL;

}

PyMem_RawFree(envar);

PyObject *retval = PyObject_Vectorcall(hook, args, nargs, keywords);

Py_DECREF(hook);

return retval;

warn:

/* If any of the imports went wrong, then warn and ignore. */

_PyErr_Clear(tstate);

int status = PyErr_WarnFormat(

PyExc_RuntimeWarning, 0,

"Ignoring unimportable $PYTHONBREAKPOINT: \"%s\"", envar);

PyMem_RawFree(envar);

if (status < 0) {

/* Printing the warning raised an exception. */

return NULL;

}

/* The warning was (probably) issued. */

Py_RETURN_NONE;

}

PyDoc_STRVAR(breakpointhook_doc,

);

static int

sys_displayhook_unencodable(PyObject *outf, PyObject *o)

{

PyObject *stdout_encoding = NULL;

PyObject *encoded, *escaped_str, *repr_str, *buffer, *result;

const char *stdout_encoding_str;

int ret;

stdout_encoding = PyObject_GetAttr(outf, &_Py_ID(encoding));

if (stdout_encoding == NULL)

goto error;

stdout_encoding_str = PyUnicode_AsUTF8(stdout_encoding);

if (stdout_encoding_str == NULL)

goto error;

repr_str = PyObject_Repr(o);

if (repr_str == NULL)

goto error;

encoded = PyUnicode_AsEncodedString(repr_str,

stdout_encoding_str,

"backslashreplace");

Py_DECREF(repr_str);

if (encoded == NULL)

goto error;

if (PyObject_GetOptionalAttr(outf, &_Py_ID(buffer), &buffer) < 0) {

Py_DECREF(encoded);

goto error;

}

if (buffer) {

result = PyObject_CallMethodOneArg(buffer, &_Py_ID(write), encoded);

Py_DECREF(buffer);

Py_DECREF(encoded);

if (result == NULL)

goto error;

Py_DECREF(result);

}

else {

escaped_str = PyUnicode_FromEncodedObject(encoded,

stdout_encoding_str,

"strict");

Py_DECREF(encoded);

if (PyFile_WriteObject(escaped_str, outf, Py_PRINT_RAW) != 0) {

Py_DECREF(escaped_str);

goto error;

}

Py_DECREF(escaped_str);

}

ret = 0;

goto finally;

error:

ret = -1;

finally:

Py_XDECREF(stdout_encoding);

return ret;

}

static PyObject *

sys_displayhook(PyObject *module, PyObject *o)

{

PyObject *outf;

PyObject *builtins;

PyThreadState *tstate = _PyThreadState_GET();

builtins = PyImport_GetModule(&_Py_ID(builtins));

if (builtins == NULL) {

if (!_PyErr_Occurred(tstate)) {

_PyErr_SetString(tstate, PyExc_RuntimeError,

"lost builtins module");

}

return NULL;

}

Py_DECREF(builtins);

/* Print value except if None */

/* After printing, also assign to '_' */

/* Before, set '_' to None to avoid recursion */

if (o == Py_None) {

Py_RETURN_NONE;

}

if (PyObject_SetAttr(builtins, _Py_LATIN1_CHR('_'), Py_None) != 0)

return NULL;

outf = PySys_GetAttr(&_Py_ID(stdout));

if (outf == NULL) {

return NULL;

}

if (outf == Py_None) {

_PyErr_SetString(tstate, PyExc_RuntimeError, "lost sys.stdout");

Py_DECREF(outf);

return NULL;

}

if (PyFile_WriteObject(o, outf, 0) != 0) {

if (_PyErr_ExceptionMatches(tstate, PyExc_UnicodeEncodeError)) {

int err;

/* repr(o) is not encodable to sys.stdout.encoding with

_PyErr_Clear(tstate);

err = sys_displayhook_unencodable(outf, o);

if (err) {

Py_DECREF(outf);

return NULL;

}

}

else {

Py_DECREF(outf);

return NULL;

}

}

if (PyFile_WriteObject(_Py_LATIN1_CHR('\n'), outf, Py_PRINT_RAW) != 0) {

Py_DECREF(outf);

return NULL;

}

Py_DECREF(outf);

if (PyObject_SetAttr(builtins, _Py_LATIN1_CHR('_'), o) != 0) {

return NULL;

}

Py_RETURN_NONE;

}

static PyObject *

sys_excepthook_impl(PyObject *module, PyObject *exctype, PyObject *value,

PyObject *traceback)

{

PyErr_Display(NULL, value, traceback);

Py_RETURN_NONE;

}

static PyObject *

sys_exception_impl(PyObject *module)

{

_PyErr_StackItem *err_info = _PyErr_GetTopmostException(_PyThreadState_GET());

if (err_info->exc_value != NULL) {

return Py_NewRef(err_info->exc_value);

}

Py_RETURN_NONE;

}

static PyObject *

sys_exc_info_impl(PyObject *module)

{

_PyErr_StackItem *err_info = _PyErr_GetTopmostException(_PyThreadState_GET());

return _PyErr_StackItemToExcInfoTuple(err_info);

}

static PyObject *

sys_unraisablehook(PyObject *module, PyObject *unraisable)

{

return _PyErr_WriteUnraisableDefaultHook(unraisable);

}

static PyObject *

sys_exit_impl(PyObject *module, PyObject *status)

{

/* Raise SystemExit so callers may catch it or clean up. */

PyErr_SetObject(PyExc_SystemExit, status);

return NULL;

}

static PyObject *

get_utf8_unicode(void)

{

_Py_DECLARE_STR(utf_8, "utf-8");

PyObject *ret = &_Py_STR(utf_8);

return Py_NewRef(ret);

}

static PyObject *

sys_getdefaultencoding_impl(PyObject *module)

{

return get_utf8_unicode();

}

static PyObject *

sys_getfilesystemencoding_impl(PyObject *module)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

const PyConfig *config = _PyInterpreterState_GetConfig(interp);

if (wcscmp(config->filesystem_encoding, L"utf-8") == 0) {

return get_utf8_unicode();

}

PyObject *u = PyUnicode_FromWideChar(config->filesystem_encoding, -1);

if (u == NULL) {

return NULL;

}

_PyUnicode_InternImmortal(interp, &u);

return u;

}

static PyObject *

sys_getfilesystemencodeerrors_impl(PyObject *module)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

const PyConfig *config = _PyInterpreterState_GetConfig(interp);

PyObject *u = PyUnicode_FromWideChar(config->filesystem_errors, -1);

if (u == NULL) {

return NULL;

}

_PyUnicode_InternImmortal(interp, &u);

return u;

}