python - 创建一个用于在 Python3 中运行二进制程序的最小沙箱

Question

我正在尝试构建一个 Python 沙箱，以便在最小且安全的环境中运行学生的代码。我打算将其运行到容器中并限制其对该容器资源的访问。因此，我目前正在设计沙箱的一部分，该部分应该运行到容器中并处理对资源的访问。

目前，我的规范是限制进程使用的时间和内存量。我还需要能够通过 stdin 与进程通信并捕获 retcode、stdout 和 stderr > 在执行结束时。

此外，程序可能会进入无限循环并通过 stdout 或 stderr 填满内存(我有一个学生的程序导致我的容器崩溃，因为那)。因此，我还希望能够限制恢复的 stdout 和 stderr 的大小(达到一定限制后，我可以终止该进程并忽略其余的输出。我不关心这些额外的数据，因为它很可能是一个有缺陷的程序，应该被丢弃)。

目前，我的沙箱几乎捕获了所有内容，这意味着我可以:

• 根据需要设置超时；
• 设置进程使用的内存限制；
• 通过stdin(现在是给定的字符串)提供进程；
• 获取最终的retcode、stdout和stderr。

这是我当前的代码(我试图在示例中保持较小的代码):

MEMORY_LIMIT  = 64 * 1024 * 1024
TIMEOUT_LIMIT = 5 * 60

__NR_FILE_NOT_FOUND = -1
__NR_TIMEOUT        = -2
__NR_MEMORY_OUT     = -3

def limit_memory(memory):
    import resource
    return lambda :resource.setrlimit(resource.RLIMIT_AS, (memory, memory))

def run_program(cmd, sinput='', timeout=TIMEOUT_LIMIT, memory=MEMORY_LIMIT):
    """Run the command line and output (ret, sout, serr)."""
    from subprocess import Popen, PIPE
    try:
        proc =  Popen(cmd, stdin=PIPE, stdout=PIPE, stderr=PIPE,
                      preexec_fn=limit_memory(memory))
    except FileNotFoundError:
        return (__NR_FILE_NOT_FOUND, "", "")

    sout, serr = "".encode("utf-8"), "".encode("utf-8")
    try:
        sout, serr = proc.communicate(sinput.encode("utf-8"), timeout=timeout)
        ret = proc.wait()
    except subprocess.TimeoutExpired:
        ret = __NR_TIMEOUT
    except MemoryError:
        ret = __NR_MEMORY_OUT
    return (ret, sout.decode("utf-8"), serr.decode("utf-8"))

if __name__ == "__main__":
    ret, out, err = run_program(['./example.sh'], timeout=8)
    print("return code: %i\n" % ret)
    print("stdout:\n%s" % out)
    print("stderr:\n%s" % err)

缺少的功能是:

1. 对 stdout 和 stderr 的大小设置限制。我在网上看到了一些尝试，但没有一个真正有效。

2. 将函数附加到 stdin 比仅静态字符串更好。该函数应连接到管道 stdout 和 stderr 并将字节返回到 stdin。

有人对此有想法吗？

PS:我已经看过:

• Non blocking reading from a subprocess output stream in Python ;

• Python subprocess with timeout and large output (>64K)

Answer 1

正如我所说，您可以创建自己的缓冲区并将 STDOUT/STDERR 写入其中，同时检查大小。为了方便起见，您可以编写一个小的 io.BytesIO 包装器来为您进行检查，例如:

from io import BytesIO

# lets first create a size-controlled BytesIO buffer for convenience
class MeasuredStream(BytesIO):

    def __init__(self, maxsize=1024):  # lets use a 1 KB as a default
        super(MeasuredStream, self).__init__()
        self.maxsize = maxsize
        self.length = 0

    def write(self, b):
        if self.length + len(b) > self.maxsize:  # o-oh, max size exceeded
            # write only up to maxsize, truncate the rest
            super(MeasuredStream, self).write(b[:self.maxsize - self.length])
            raise ValueError("Max size reached, excess data is truncated")
        # plenty of space left, write the bytes and increase the length
        self.length += super(MeasuredStream, self).write(b)
        return len(b)  # convention: return the written number of bytes 

请注意，如果您打算进行截断/查找和替换，则必须考虑到您的长度中的内容，但这对于我们的目的来说已经足够了。

无论如何，现在您需要做的就是处理自己的流并考虑 MeasuredStream 中可能出现的 ValueError，而不是使用 Popen.communicate ()。不幸的是，这也意味着您必须自己处理超时。像这样的东西:

from subprocess import Popen, PIPE, STDOUT, TimeoutExpired
import sys
import time

MEMORY_LIMIT  = 64 * 1024 * 1024
TIMEOUT_LIMIT = 5 * 60
STDOUT_LIMIT  = 1024 * 1024  # let's use 1 MB as a STDOUT limit

__NR_FILE_NOT_FOUND      = -1
__NR_TIMEOUT             = -2
__NR_MEMORY_OUT          = -3
__NR_MAX_STDOUT_EXCEEDED = -4  # let's add a new return code

# a cross-platform precision clock
get_timer = time.clock if sys.platform == "win32" else time.time

def limit_memory(memory):
    import resource
    return lambda :resource.setrlimit(resource.RLIMIT_AS, (memory, memory))

def run_program(cmd, sinput='', timeout=TIMEOUT_LIMIT, memory=MEMORY_LIMIT):
    """Run the command line and output (ret, sout, serr)."""
    from subprocess import Popen, PIPE, STDOUT
    try:
        proc =  Popen(cmd, stdin=PIPE, stdout=PIPE, stderr=STDOUT,
                      preexec_fn=limit_memory(memory), timeout=timeout)
    except FileNotFoundError:
        return (__NR_FILE_NOT_FOUND, "", "")
    sout = MeasuredStream(STDOUT_LIMIT)  # store STDOUT in a measured stream
    start_time = get_timer()  # store a reference timer for our custom timeout
    try:
        proc.stdin.write(sinput.encode("utf-8"))  # write the input to STDIN
        proc.stdin.flush()  # flush the STDOUT buffer
        while True:  # our main listener loop
            line = proc.stdout.readline()  # read a line from the STDOUT
            # use proc.stdout.read(buf_size) instead to handle your own buffer
            if line != b"":  # content collected...
                sout.write(line)  # write it to our stream
            elif proc.poll() is not None:  # process finished, nothing to do
                break
            # finally, check the current time progress...
            if get_timer() >= start_time + TIMEOUT_LIMIT:
                raise TimeoutExpired(proc.args, TIMEOUT_LIMIT)
        ret = proc.poll()  # get the return code
    except TimeoutExpired:
        proc.kill()  # we're no longer interested in the process, kill it
        ret = __NR_TIMEOUT
    except MemoryError:
        ret = __NR_MEMORY_OUT
    except ValueError:  # max buffer reached
        proc.kill()  # we're no longer interested in the process, kill it
        ret = __NR_MAX_STDOUT_EXCEEDED
    sout.seek(0)  # rewind the buffer
    return ret, sout.read().decode("utf-8")  # send the results back

if __name__ == "__main__":
    ret, out, err = run_program(['./example.sh'], timeout=8)
    print("return code: %i\n" % ret)
    print("stdout:\n%s" % out)
    print("stderr:\n%s" % err)

这有两个“问题”，第一个非常明显 - 我将子进程 STDERR 传输到 STDOUT，因此结果将是混合的。由于从 STDOUT 和 STDERR 流读取是一个阻塞操作，如果您想分别读取它们，则必须生成两个线程(并在超出流大小时分别处理它们的 ValueError 异常)。第二个问题是子进程 STDOUT 可以锁定超时检查，因为它取决于 STDOUT 实际刷新一些数据。这也可以通过一个单独的计时器线程来解决，如果超过超时，该线程将强制终止进程。事实上，这正是 Popen.communicate() 所做的。

操作原理本质上是相同的，您只需将检查外包给单独的线程并最终将所有内容重新连接起来。这是我留给你的练习;)

至于您的第二个缺少的功能，您能详细说明一下您的想法吗？