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

Question

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

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

尽管如此，我仍然需要一种方法来优雅地使导入过程超时，因为我不能保证模块脚本会终止。例如，它可能包含对 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:
    ...

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

Answer 1

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

永远不要导入不受信任的代码

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

而不是使用低访问用户，至少 运行这是一个虚拟机。虚拟机环境可以从已知良好的快照设置，无需网络访问，并在达到时间限制时关闭。然后，您可以比较快照以查看代码尝试执行的操作(如果有)。该级别的任何安全漏洞都是短暂的，没有值(value)。另见 Best practices for execution of untrusted code在软件工程堆栈交换上。

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

其次，由于您无法控制导入代码的作用，因此它可能会干扰任何使代码超时的尝试。导入的代码可以做的第一件事就是撤销您设置的保护!导入的代码可以访问 Python 的所有全局状态，包括触发导入的代码。代码可以设置thread switch interval到最大值(在内部，一个无符号长建模毫秒，所以最大值是 ((2 ** 32) - 1) 毫秒，只是 71 分 35 秒以下的一个smidgen)来干扰调度。

你不能可靠地停止线程，如果他们不想被停止

在 Python 中退出线程是 handled by raising a exception :

Raise the SystemExit exception. When not caught, this will cause the thread to exit silently.

(粗体强调我的。)

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

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

在本地扩展中等待阻塞 I/O 或启动阻塞操作的线程不能(容易)被杀死

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

如果代码开始以 native 代码(Python 扩展)实现的某些内容并且阻止了，那么所有赌注都将完全结束。

您的口译员状态可能会在您阻止他们时被冲洗掉

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

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

考虑到这些警告，所以你接受了

您导入的代码可以对它们运行的​​操作系统执行恶意操作，而您无法从同一进程甚至操作系统中阻止它们

您导入的代码可能会阻止您的代码工作。

您导入的代码可能已经导入并启动了您不想导入或启动的内容。

该代码可能会启动阻止您完全停止线程的操作

然后你可以通过在单独的线程中运行它们来超时导入，然后引发 SystemExit线程中的异常。您可以通过调用 PyThreadState_SetAsyncExc C-API function 在另一个线程中引发异常。通过 ctypes.pythonapi object . Python 测试套件 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 秒以查看导入是否真的无法终止。