5

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

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

虽然，我仍然需要一种优雅地使导入过程超时的方法，因为我不能保证模块脚本会终止。例如，它可能包含调用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如果脚本超时则返回？

4

1 回答 1

4

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

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

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

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

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

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

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

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

提出SystemExit例外。当不被捕获时，这将导致线程静默退出。

（我的粗体强调。）

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

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

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

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

如果代码启动了在本机代码（Python 扩展）中实现的东西并且阻塞了，那么所有的赌注都完全失败了。

当您阻止他们时，您的口译员状态可能会受到影响

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

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

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

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

然后您可以通过在单独的线程中运行它们来使导入超时，然后SystemExit在线程中引发异常。您可以通过object调用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 秒以查看导入是否真的无法杀死。

于 2018-11-28T15:15:05.170 回答