Вы не можете надежно убить импорт модуля.По сути, вы выполняете живой код в своем собственном интерпретаторе, поэтому все ставки отключены.
Никогда не импортируйте ненадежный код
Прежде всего, нет способа безопасно импортировать небезопасные модули из ненадежного источника.,Не имеет значения, используете ли вы пользователя с низким доступом. НИКОГДА НЕ ИМПОРТИРУЙТЕ НЕПРАВИЛЬНЫЙ КОД .В тот момент, когда код импортируется, он может использовать дыры в безопасности вашей системы далеко за пределами самого процесса Python.Python - это язык программирования общего назначения, а не среда с песочницей, и любой импортируемый вами код имеет полный прогон вашей системы
Вместо использования пользователя с низким доступом, по крайней мере запустить это виртуальная машина.Среду виртуальной машины можно настроить на основе удачного снимка без сетевого доступа и закрыть по истечении определенного времени.Затем вы можете сравнить снимки, чтобы увидеть, что, если вообще, пытался сделать код.Любое нарушение безопасности на этом уровне является кратковременным и бесполезным.См. Также Рекомендации по выполнению ненадежного кода over в Software Engineering Stack Exchange.
Вы не можете помешать коду отменить вашу работу
Далее, поскольку вы не можете контролировать действия импортируемого кода, он может просто мешать любым попыткам тайм-аута кода.Первое, что может сделать импортированный код, это отменить защиту, которую вы установили!Импортированный код может получить доступ ко всему глобальному состоянию Python, включая код, вызвавший импорт.Код мог бы установить максимальное значение интервала переключения потоков 1020 * (внутреннее значение, длинное моделирование без знака, миллисекунды, так что максимальное значение составляет ((2 ** 32) - 1) миллисекунды, всего лишь smidgen менее 71 минуты 35 секунд), чтобы помешать планированию.
Вы не можете надежно останавливать потоки, если они не хотят останавливаться
Выход из потока в Python обрабатывается путем вызова исключения :
Поднимите исключение SystemExit. Если не поймано , это приведет к бесшумному завершению потока.
(выделение жирным шрифтом.)
Из чистого кода Python вы можете выйти только изпоток из кода , работающий в этом потоке , но есть способ обойти это, см. ниже.
Но вы не можете гарантировать, что импортируемый код не просто перехватывает и обрабатываетвсе исключения;если это так, код будет продолжать работать.В этот момент он становится гонкой вооружений;ваш поток может вставить исключение в тот момент, когда другой поток находится внутри обработчика исключения?Тогда вы можете выйти из этой ветки, иначе проиграете.Вам придется продолжать попытки, пока вы не добьетесь успеха.
Поток, который ожидает блокирования ввода-вывода или запускает операцию блокировки в собственном расширении, не может (легко) быть уничтожен
Еслиимпортируемый код ожидает блокировки ввода-вывода (например, input()), после чего вы не можете прервать этот вызов.Возбуждение исключения ничего не делает, и вы не можете использовать сигналы (поскольку Python обрабатывает их в основном потоке только ).Вам нужно будет найти и закрыть каждый открытый канал ввода-вывода, на котором они могут быть заблокированы.Это выходит за рамки моего ответа здесь, слишком много способов запустить операции ввода-вывода.
Если код начал что-то реализованное в собственном коде (расширение Python) и , что блоков, все ставки полностью отключены.
Состояние вашего интерпретатора может быть изменено к тому моменту, когда вы его остановите
Код, который вы импортировали, мог сделать что-нибудь к тому времени, когда вам удалось их остановить.,Импортированные модули могли быть заменены.Исходный код на диске может быть изменен.Вы не можете быть уверены, что никакие другие темы не были запущены.В Python все возможно, поэтому предположим, что это произошло.
Если вы хотите сделать это, в любом случае
С учетом этих предостережений, так что вы принимаете, что
- Код, который вы импортируете, может совершать вредоносные действия в ОС, в которой он работает, без возможности остановить их в рамках одного и того же процесса или даже в ОС
- Код, который вы импортируете, может помешать работе вашего кода.
- Код, который вы импортировали, мог импортировать и запускать вещи, которые вы не хотели импортировать или запускать.
- Код может запускать операции, которые не позволяют полностью остановить поток
затем вы можете установить время ожидания импорта, запустив их в отдельном потоке, а затем вызвать исключение SystemExit в потоке.Вы можете вызвать исключения в другом потоке, вызвав функцию PyThreadState_SetAsyncExc C-API через объект ctypes.pythonapi .Набор тестов Python фактически использует этот путь в тесте , я использовал его в качестве шаблона для своего решения ниже.
Итак, вот полная реализация, которая делает именно это и поднимает пользовательскийUninterruptableImport исключение (подкласс ImportError), если импорт не может быть прерван.Если при импорте возникло исключение, то это исключение повторно вызывается в потоке, который запустил процесс импорта:
"""Import a module within a timeframe
Uses the PyThreadState_SetAsyncExc C API and a signal handler to interrupt
the stack of calls triggered from an import within a timeframe
No guarantees are made as to the state of the interpreter after interrupting
"""
import ctypes
import importlib
import random
import sys
import threading
import time
_set_async_exc = ctypes.pythonapi.PyThreadState_SetAsyncExc
_set_async_exc.argtypes = (ctypes.c_ulong, ctypes.py_object)
_system_exit = ctypes.py_object(SystemExit)
class UninterruptableImport(ImportError):
pass
class TimeLimitedImporter():
def __init__(self, modulename, timeout=5):
self.modulename = modulename
self.module = None
self.exception = None
self.timeout = timeout
self._started = None
self._started_event = threading.Event()
self._importer = threading.Thread(target=self._import, daemon=True)
self._importer.start()
self._started_event.wait()
def _import(self):
self._started = time.time()
self._started_event.set()
timer = threading.Timer(self.timeout, self.exit)
timer.start()
try:
self.module = importlib.import_module(self.modulename)
except Exception as e:
self.exception = e
finally:
timer.cancel()
def result(self, timeout=None):
# give the importer a chance to finish first
if timeout is not None:
timeout += max(time.time() + self.timeout - self._started, 0)
self._importer.join(timeout)
if self._importer.is_alive():
raise UninterruptableImport(
f"Could not interrupt the import of {self.modulename}")
if self.module is not None:
return self.module
if self.exception is not None:
raise self.exception
def exit(self):
target_id = self._importer.ident
if target_id is None:
return
# set a very low switch interval to be able to interrupt an exception
# handler if SystemExit is being caught
old_interval = sys.getswitchinterval()
sys.setswitchinterval(1e-6)
try:
# repeatedly raise SystemExit until the import thread has exited.
# If the exception is being caught by a an exception handler,
# our only hope is to raise it again *while inside the handler*
while True:
_set_async_exc(target_id, _system_exit)
# short randomised wait times to 'surprise' an exception
# handler
self._importer.join(
timeout=random.uniform(1e-4, 1e-5)
)
if not self._importer.is_alive():
return
finally:
sys.setswitchinterval(old_interval)
def import_with_timeout(modulename, import_timeout=5, exit_timeout=1):
importer = TimeLimitedImporter(modulename, import_timeout)
return importer.result(exit_timeout)
Если код не может быть уничтожен, он будет запущен в потоке демонаЭто означает, что вы можете по крайней мере грациозно завершить работу с Python.
Используйте его следующим образом:
module = import_with_timeout(modulename)
для тайм-аута по умолчанию, равного 5 секундам, и 1 секунда ожидания, чтобы убедиться, что импорт действительно не удаляется.