如何动态导入超时的不安全 Python 模块？
How to dynamically import an unsafe Python module with a timeout?
出于测试目的，我需要动态加载几个可能不安全的模块。
关于安全性，我的脚本是由低访问权限的用户执行的。
尽管如此，我仍然需要一种优雅地使导入过程超时的方法，因为我无法保证模块脚本会终止。例如，它可以包含对 input 或无限循环的调用。
我目前正在使用 Thread.join 和 timeout，但这并不能完全解决问题，因为脚本仍然在后台运行并且无法终止线程。
from threading import Thread
import importlib.util
class ReturnThread(Thread):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._return = None
def run(self):
if self._target is not None:
self._return = self._target(*self._args, **self._kwargs)
def join(self, *args, **kwargs):
super().join(*args, **kwargs)
return self._return
def loader(name, path):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module) # This may run into an infinite loop
return module
module_loader = ReturnThread(loader, ('module_name', 'module/path'))
module_loader.start()
module = module_loader.join(timeout=0.1)
# The thread might still be alive here
if module is None:
...
else:
...
我如何导入一个模块，但是 return None 如果脚本超时？
您无法可靠地终止导入模块。您实质上是在自己的解释器中执行实时代码，所以一切都没有了。
永远不要导入不受信任的代码
首先，没有办法从不受信任的来源安全地导入不安全的模块。如果您使用的是低访问权限用户，这无关紧要。 切勿导入不受信任的代码。导入代码的那一刻，它可能已经利用了系统中的安全漏洞，远远超出 Python 进程本身。 Python 是一种通用编程语言，不是沙盒环境，您导入的任何代码都具有您系统的完整 运行
而不是使用低访问权限的用户，至少 运行这是一个虚拟机。虚拟机环境可以从已知良好的快照设置，无需网络访问，并在达到时间限制时关闭。然后您可以比较快照以查看代码试图做什么（如果有的话）。该级别的任何安全漏洞都是短暂的，没有价值。另请参阅 Best practices for execution of untrusted code Software Engineering Stack Exchange。
您无法阻止代码撤消您的工作
接下来，由于您无法控制导入代码的作用，它可能会干扰任何使代码超时的尝试。导入的代码可以做的第一件事就是撤销您设置的保护！导入的代码可以访问 Python 的所有全局状态，包括触发导入的代码。代码可以将 thread switch interval 设置为最大值（在内部，unsigned long 建模毫秒，因此最大值为 ((2 ** 32) - 1) 毫秒，仅略低于 71 分 35 秒）以扰乱调度。
如果线程不想停止，则无法可靠地停止线程
在 Python 中退出线程是 handled by raising a exception:
Raise the SystemExit exception. When not caught, this will cause the thread to exit silently.
（大胆强调我的。）
在纯 Python 代码中，您只能从线程 运行ning 中的代码退出线程 ，但有一种解决方法，见下文。
但是你不能保证你导入的代码不只是捕获和处理所有的异常；如果是这样，代码将继续 运行ning。到那时它就变成了一场武器竞赛；您的线程能否设法在另一个线程位于异常处理程序中时插入异常？然后你可以退出那个线程，否则，你就输了。你必须不断尝试直到成功。
等待阻塞 I/O 或在本机扩展中启动阻塞操作的线程不能（轻易）被杀死
如果您导入的代码等待阻塞 I/O（例如 input() 调用），那么您无法中断该调用。引发异常不会执行任何操作，并且您不能使用信号（因为 Python 处理主线程上的信号 仅 ）。您必须找到并关闭他们可能被阻止的每个打开的 I/O 频道。这超出了我在这里回答的范围，启动 I/O 操作的方法太多了。
如果代码启动了一些用本机代码实现的东西（Python 扩展）并且 that 块，所有赌注都完全关闭。
您的解释器状态可以在您停止它们时得到处理
在您设法阻止它们时，您导入的代码可能已经完成了任何操作。导入的模块可能已被替换。磁盘上的源代码可能已被更改。您不能确定没有其他线程已启动。 Python 一切皆有可能，所以假设它已经发生了。
如果你想这样做，无论如何
考虑到这些注意事项，所以您接受
- 您导入的代码可以对 OS 它们 运行 正在进入的 OS 进行恶意操作，而您无法从同一进程中甚至 OS[ 阻止它们=86=]
- 您导入的代码可能会使您的代码无法工作。
- 您导入的代码可能已经导入并启动了您不想导入或启动的内容。
- 代码可能会启动阻止您完全停止线程的操作
然后您可以通过 运行 在单独的线程中将导入超时，然后在线程中引发 SystemExit 异常。您可以通过调用 PyThreadState_SetAsyncExc C-API function via the ctypes.pythonapi object. The Python test suite actually uses this path in a test 在另一个线程中引发异常，我将其用作下面我的解决方案的模板。
所以这是一个完整的实现，它会在导入无法中断的情况下引发自定义 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 秒以查看导入是否真的无法终止。
出于测试目的，我需要动态加载几个可能不安全的模块。
关于安全性，我的脚本是由低访问权限的用户执行的。
尽管如此，我仍然需要一种优雅地使导入过程超时的方法，因为我无法保证模块脚本会终止。例如，它可以包含对 input 或无限循环的调用。
我目前正在使用 Thread.join 和 timeout，但这并不能完全解决问题，因为脚本仍然在后台运行并且无法终止线程。
from threading import Thread
import importlib.util
class ReturnThread(Thread):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._return = None
def run(self):
if self._target is not None:
self._return = self._target(*self._args, **self._kwargs)
def join(self, *args, **kwargs):
super().join(*args, **kwargs)
return self._return
def loader(name, path):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module) # This may run into an infinite loop
return module
module_loader = ReturnThread(loader, ('module_name', 'module/path'))
module_loader.start()
module = module_loader.join(timeout=0.1)
# The thread might still be alive here
if module is None:
...
else:
...
我如何导入一个模块，但是 return None 如果脚本超时？
您无法可靠地终止导入模块。您实质上是在自己的解释器中执行实时代码，所以一切都没有了。
永远不要导入不受信任的代码
首先，没有办法从不受信任的来源安全地导入不安全的模块。如果您使用的是低访问权限用户，这无关紧要。 切勿导入不受信任的代码。导入代码的那一刻，它可能已经利用了系统中的安全漏洞，远远超出 Python 进程本身。 Python 是一种通用编程语言，不是沙盒环境，您导入的任何代码都具有您系统的完整 运行
而不是使用低访问权限的用户，至少 运行这是一个虚拟机。虚拟机环境可以从已知良好的快照设置，无需网络访问，并在达到时间限制时关闭。然后您可以比较快照以查看代码试图做什么（如果有的话）。该级别的任何安全漏洞都是短暂的，没有价值。另请参阅 Best practices for execution of untrusted code Software Engineering Stack Exchange。
您无法阻止代码撤消您的工作
接下来，由于您无法控制导入代码的作用，它可能会干扰任何使代码超时的尝试。导入的代码可以做的第一件事就是撤销您设置的保护！导入的代码可以访问 Python 的所有全局状态，包括触发导入的代码。代码可以将 thread switch interval 设置为最大值（在内部，unsigned long 建模毫秒，因此最大值为 ((2 ** 32) - 1) 毫秒，仅略低于 71 分 35 秒）以扰乱调度。
如果线程不想停止，则无法可靠地停止线程
在 Python 中退出线程是 handled by raising a exception:
Raise the
SystemExitexception. When not caught, this will cause the thread to exit silently.
（大胆强调我的。）
在纯 Python 代码中，您只能从线程 运行ning 中的代码退出线程 ，但有一种解决方法，见下文。
但是你不能保证你导入的代码不只是捕获和处理所有的异常；如果是这样，代码将继续 运行ning。到那时它就变成了一场武器竞赛；您的线程能否设法在另一个线程位于异常处理程序中时插入异常？然后你可以退出那个线程，否则，你就输了。你必须不断尝试直到成功。
等待阻塞 I/O 或在本机扩展中启动阻塞操作的线程不能（轻易）被杀死
如果您导入的代码等待阻塞 I/O（例如 input() 调用），那么您无法中断该调用。引发异常不会执行任何操作，并且您不能使用信号（因为 Python 处理主线程上的信号 仅 ）。您必须找到并关闭他们可能被阻止的每个打开的 I/O 频道。这超出了我在这里回答的范围，启动 I/O 操作的方法太多了。
如果代码启动了一些用本机代码实现的东西（Python 扩展）并且 that 块，所有赌注都完全关闭。
您的解释器状态可以在您停止它们时得到处理
在您设法阻止它们时，您导入的代码可能已经完成了任何操作。导入的模块可能已被替换。磁盘上的源代码可能已被更改。您不能确定没有其他线程已启动。 Python 一切皆有可能，所以假设它已经发生了。
如果你想这样做，无论如何
考虑到这些注意事项，所以您接受
- 您导入的代码可以对 OS 它们 运行 正在进入的 OS 进行恶意操作，而您无法从同一进程中甚至 OS[ 阻止它们=86=]
- 您导入的代码可能会使您的代码无法工作。
- 您导入的代码可能已经导入并启动了您不想导入或启动的内容。
- 代码可能会启动阻止您完全停止线程的操作
然后您可以通过 运行 在单独的线程中将导入超时，然后在线程中引发 SystemExit 异常。您可以通过调用 PyThreadState_SetAsyncExc C-API function via the ctypes.pythonapi object. The Python test suite actually uses this path in a test 在另一个线程中引发异常，我将其用作下面我的解决方案的模板。
所以这是一个完整的实现，它会在导入无法中断的情况下引发自定义 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 秒以查看导入是否真的无法终止。