在 Python 中创建一个临时 FIFO（命名管道）？

Posted 2023-02-19

【中文标题】在 Python 中创建一个临时 FIFO（命名管道）？

【英文标题】：Create a temporary FIFO (named pipe) in Python?

【发布时间】：2010-11-28 15:54:42

【问题描述】：

如何在 Python 中创建临时 FIFO（命名管道）？这应该有效：

import tempfile

temp_file_name = mktemp()
os.mkfifo(temp_file_name)
open(temp_file_name, os.O_WRONLY)
# ... some process, somewhere, will read it ...

但是，我犹豫不决，因为 Python Docs 11.6 中的大警告和可能会被删除，因为它已被弃用。

编辑：值得注意的是，我尝试过tempfile.NamedTemporaryFile（以及扩展名tempfile.mkstemp），但os.mkfifo 抛出：

OSError -17: 文件已存在

当您在 mkstemp/NamedTemporaryFile 创建的文件上运行它时。

【问题讨论】：

正如您所提到的，由于 os.mkfifo 如果文件存在会引发错误，因此使用 mkstemp 甚至完全确定的文件名没有安全漏洞（除了不好的感觉和有人天真地复制您的文件的风险之外）代码）。

问题：您打算如何将管道名称传达给其他进程？这个频道是如何保护的？

@Joe：你是对的。我没有想到这一点。 @Steven Huwig：这两个进程都是由同一个父进程产生的，尽管我可能会在某个时候为此移动到一个新的 fifo。为什么要问？

【参考方案1】：

如果文件已经存在，os.mkfifo() 将失败并出现异常OSError: [Errno 17] File exists，因此这里不存在安全问题。使用tempfile.mktemp() 的安全问题是竞争条件，攻击者可以在您自己打开文件之前创建具有相同名称的文件，但由于os.mkfifo() 失败，如果文件已经存在，这不是问题。

但是，由于 mktemp() 已被弃用，因此您不应使用它。你可以改用tempfile.mkdtemp()：

import os, tempfile

tmpdir = tempfile.mkdtemp()
filename = os.path.join(tmpdir, 'myfifo')
print filename
try:
    os.mkfifo(filename)
except OSError, e:
    print "Failed to create FIFO: %s" % e
else:
    fifo = open(filename, 'w')
    # write stuff to fifo
    print >> fifo, "hello"
    fifo.close()
    os.remove(filename)
    os.rmdir(tmpdir)

编辑：我应该明确指出，仅仅因为 mktemp() 漏洞被避免，仍然需要考虑其他常见的安全问题；例如攻击者可以在您的程序执行之前创建先进先出（如果他们有适当的权限），如果错误/异常处理不当，可能会导致您的程序崩溃。

【讨论】：

os.rmdir(tmpdir) 不应该在 try-else 块之外吗？

你的意思是except OSError as e。

@TimDierks: 的确，新语法更具可读性并且是 Python 3 中唯一支持的语法，但旧语法适用于所有 Python 2.x 版本，而新语法需要在至少 Python 2.6。

【参考方案2】：

您可能会发现使用以下上下文管理器很方便，它会为您创建和删除临时文件：

import os
import tempfile
from contextlib import contextmanager


@contextmanager
def temp_fifo():
    """Context Manager for creating named pipes with temporary names."""
    tmpdir = tempfile.mkdtemp()
    filename = os.path.join(tmpdir, 'fifo')  # Temporary filename
    os.mkfifo(filename)  # Create FIFO
    try:
        yield filename
    finally:
        os.unlink(filename)  # Remove file
        os.rmdir(tmpdir)  # Remove directory

例如，您可以像这样使用它：

with temp_fifo() as fifo_file:
    # Pass the fifo_file filename e.g. to some other process to read from.
    # Write something to the pipe 
    with open(fifo_file, 'w') as f:
        f.write("Hello\n")

【参考方案3】：

如何使用

d = mkdtemp()
t = os.path.join(d, 'fifo')

【讨论】：

这与 mktemp 存在相同的安全问题。不过谢谢，布伦丹。

只有当用户已经攻破了运行脚本的账号（目录为0700）时，才会有同样的安全问题。在这种情况下，他们可能会做得更糟。

糟糕——我错了。这与 mktemp 没有相同的安全问题，因为如果文件存在，mkfifo() 会拒绝（即没有机会进行 MiM 攻击）。这会奏效。

【参考方案4】：

如果它用于您的程序中，而不是用于任何外部，请查看Queue module。另外一个好处是，python 队列是线程安全的。

【讨论】：

谢谢尼拉莫。不幸的是，它不适用于内部交流。不过，我在其他地方使用队列。感谢您的建议。

没问题，只是抛出“使用最简单可行的东西”标志。如果你真的需要先进先出，那么这个答案根本没有帮助。

【参考方案5】：

实际上，mkstemp 所做的一切都是在循环中运行mktemp 并不断尝试以独占方式创建，直到成功（参见 stdlib 源代码 here）。你可以用os.mkfifo做同样的事情：

import os, errno, tempfile

def mkftemp(*args, **kwargs):
    for attempt in xrange(1024):
        tpath = tempfile.mktemp(*args, **kwargs)

        try:
            os.mkfifo(tpath, 0600)
        except OSError as e:
            if e.errno == errno.EEXIST:
                # lets try again
                continue
            else:
                raise
        else:
           # NOTE: we only return the path because opening with
           # os.open here would block indefinitely since there 
           # isn't anyone on the other end of the fifo.
           return tpath
    else:
        raise IOError(errno.EEXIST, "No usable temporary file name found")

【参考方案6】：

为什么不直接使用mkstemp()？

例如：

import tempfile
import os

handle, filename = tempfile.mkstemp()
os.mkfifo(filename)
writer = open(filename, os.O_WRONLY)
reader = open(filename, os.O_RDONLY)
os.close(handle)

【讨论】：

os.mkfifo 抛出 OSError 17: File Exists。不过还是谢谢。