如何在超时的情况下动态导入不安全的 Python 模块？

Question

我需要动态加载几个可能不安全的模块以进行测试。

关于安全性，我的脚本由低访问权限的用户执行。

尽管如此，我仍然需要一种方法来优雅地使导入过程超时，因为我不能保证模块脚本会终止。例如，它可以包含对input或无限循环的调用。

我目前正在使用Thread.joina 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:
    ...

如何导入模块，但None在脚本超时时返回？

Answer 1

您无法可靠地终止导入模块。您本质上是在自己的解释器中执行实时代码，因此所有的赌注都没有。

切勿导入不受信任的代码

首先，没有办法从不受信任的来源安全地导入不安全的模块。如果您使用的是低访问权限的用户，这并不重要。切勿导入不受信任的代码。导入代码的那一刻，它可能会利用系统中远远超出 Python 进程本身的安全漏洞。Python 是一种通用编程语言，不是沙盒环境，您导入的任何代码都可以在您的系统中完整运行

与其使用低访问权限的用户，至少运行这是一个虚拟机。虚拟机环境可以从已知良好的快照进行设置，无需网络访问，并在达到时间限制时关闭。然后，您可以比较快照以查看代码尝试执行的操作（如果有）。该级别的任何安全漏洞都是短暂的且没有价值。另请参阅Software Engineering Stack Exchange 上执行不受信任代码的最佳实践。

您无法阻止代码撤消您的工作

其次，由于您无法控制导入的代码执行的操作，因此它可能会轻微干扰任何使代码超时的尝试。导入的代码可以做的第一件事就是撤销您设置的保护！导入的代码可以访问Python的所有全局状态，包括触发导入的代码。该代码可以将线程切换间隔设置为最大值（内部是一个无符号长建模毫秒，因此最大值为((2 ** 32) - 1)毫秒，仅比 71 分 35 秒少一点），以扰乱调度。

如果线程不想被停止，你就无法可靠地停止它们

Python 中退出线程是通过引发异常来处理的：

引发SystemExit异常。如果没有被捕获，这将导致线程静默退出。

（粗体强调我的。）

在纯 Python 代码中，您只能从该线程中运行的代码中退出该线程，但有一种方法可以解决此问题，请参见下文。

但是您不能保证您导入的代码不仅仅是捕获和处理所有异常；而是捕获和处理所有异常。如果是这种情况，代码将继续运行。到那时，它就变成了一场武器竞赛；您的线程能否设法在另一个线程位于异常处理程序内时插入异常？然后你就可以退出该线程，否则你就输了。你必须继续努力，直到成功。

等待阻塞 I/O 或在本机扩展中启动阻塞操作的线程无法（轻易）被终止

如果您导入的代码等待阻塞 I/O（例如调用input()），那么您无法中断该调用。引发异常不会执行任何操作，并且您不能使用信号（因为 Python仅处理主线程上的信号）。您必须找到并关闭每个可能被阻塞的开放 I/O 通道。这超出了我的回答范围，启动 I/O 操作的方法太多了。

如果代码启动了以本机代码（Python 扩展）实现的某些内容并且该代码被阻止，那么所有的赌注都将完全失败。

当您停止解释器时，您的解释器状态可能会被破坏

当您设法阻止它们时，您导入的代码可能已经做了任何事情。进口模块可能已被替换。磁盘上的源代码可能已被更改。您无法确定没有其他线程已启动。在 Python 中一切皆有可能，所以假设它已经发生了。

无论如何，如果你想这样做

考虑到这些注意事项，所以您接受这一点

• 您导入的代码可以对它们运行的​​操作系统执行恶意操作，而您无法在同一进程甚至操作系统中阻止它们
• 您导入的代码可能会阻止您的代码运行。
• 您导入的代码可能已导入并启动了您不想导入或启动的内容。
• 该代码可能会启动阻止您完全停止线程的操作

那么您可以通过在单独的线程中运行导入来使导入超时，然后SystemExit在线程中引发异常。您可以通过对象调用PyThreadState_SetAsyncExcC-API 函数来在另一个线程中引发异常。Python 测试套件实际上在测试中使用了此路径，我将其用作下面的解决方案的模板。ctypes.pythonapi

因此，这里是一个完整的实现，它就是这样做的，并且如果导入无法中断，则会引发自定义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 秒以查看导入是否确实无法终止。