Capturing print output from shared library called from python with ctypes module

Question

I am working with a shared library that is being called through the ctypes module. I would like to redirect the stdout associated with this module to a variable or a file that I can access in my program. However ctypes uses a separate stdout from sys.stdout.

I'll demonstrate the problem I am having with libc. If anyone is copying and pasting the code they might have to change the filename on line 2.

import ctypes
libc = ctypes.CDLL('libc.so.6')

from cStringIO import StringIO
import sys
oldStdOut = sys.stdout
sys.stdout = myStdOut = StringIO()

print 'This text gets captured by myStdOut'
libc.printf('This text fails to be captured by myStdOut\n')

sys.stdout = oldStdOut
myStdOut.getvalue()

Is there any way I can capture the stdout that is associated with the ctypes loaded shared library?

Answer 1

Late to the party but here is my approach, basically using the flag --capture=fd will use a different file descriptor value than 1 which is the value for c-level stdout, that being said I came up with this code, the commented section is how I came up with the answer

import os 
import sys
import pytest 
import ctypes

libc = ctypes.CDLL(None)
c_stdout = ctypes.c_void_p.in_dll(libc, 'stdout')
c_stdout_fd = libc.fileno(c_stdout)

def test_one():
    c_fd = os.memfd_create('c_fd')
    # py_fd = sys.stdout.fileno()
    # print (c_fd, py_fd, c_stdout_fd)
    # assert py_fd == c_stdout_fd
    try:
        os.dup2(c_fd, c_stdout_fd)
        libc.printf(b"Hello world")
        libc.fflush(None)
        print("hello world")

        os.lseek(c_fd, 0, os.SEEK_SET)

        output = os.read(c_fd, 1024)
        assert output == "Hello world".encode("utf-8")
    
    finally:
        os.dup2(c_stdout_fd, 1)
        os.close(c_fd)
        

I don't recommend using this code as it does not safely handle errors and it uses platform specifc approach os.memfd_create.

Note you could replace os.memfd_create for tempfile

Answer 2

There is a Python project called Wurlitzer that very elegantly solves this problem. It's a work of art and deserves to be one of the top answers to this question.

https://github.com/minrk/wurlitzer

https://pypi.org/project/wurlitzer/

pip install wurlitzer

from wurlitzer import pipes

with pipes() as (out, err):
    call_some_c_function()

stdout = out.read()

from io import StringIO
from wurlitzer import pipes, STDOUT

out = StringIO()
with pipes(stdout=out, stderr=STDOUT):
    call_some_c_function()

stdout = out.getvalue()

from wurlitzer import sys_pipes

with sys_pipes():
    call_some_c_function()

And the most magical part: it supports Jupyter:

%load_ext wurlitzer

Answer 3

If the data the native process writes are large (larger than pipe buffer), the native program would block until you make some space in the pipe by reading it.

The solution from lunixbochs, however, needs the native process to finish before it starts reading the pipe. I improved the solution so that it reads the pipe in parallel from a separate thread. This way you can capture output of any size.

This solution is also inspired by https://stackoverflow.com/a/16571630/1076564 and captures both stdout and stderr:

class CtypesStdoutCapture(object):
    def __enter__(self):
        self._pipe_out, self._pipe_in = os.pipe()
        self._err_pipe_out, self._err_pipe_in = os.pipe()
        self._stdout = os.dup(1)
        self._stderr = os.dup(2)
        self.text = ""
        self.err = ""
        # replace stdout with our write pipe
        os.dup2(self._pipe_in, 1)
        os.dup2(self._err_pipe_in, 2)
        self._stop = False
        self._read_thread = threading.Thread(target=self._read, args=["text", self._pipe_out])
        self._read_err_thread = threading.Thread(target=self._read, args=["err", self._err_pipe_out])
        self._read_thread.start()
        self._read_err_thread.start()
        return self

    def __exit__(self, *args):
        self._stop = True
        self._read_thread.join()
        self._read_err_thread.join()
        # put stdout back in place
        os.dup2(self._stdout, 1)
        os.dup2(self._stderr, 2)
        self.text += self.read_pipe(self._pipe_out)
        self.err += self.read_pipe(self._err_pipe_out)

    # check if we have more to read from the pipe
    def more_data(self, pipe):
        r, _, _ = select.select([pipe], [], [], 0)
        return bool(r)

    # read the whole pipe
    def read_pipe(self, pipe):
        out = ''
        while self.more_data(pipe):
            out += os.read(pipe, 1024)

        return out

    def _read(self, type, pipe):
        while not self._stop:
            setattr(self, type, getattr(self, type) + self.read_pipe(pipe))
            sleep(0.001)

    def __str__(self):
        return self.text

# Usage:

with CtypesStdoutCapture as capture:
  lib.native_fn()

print(capture.text)
print(capture.err)

Answer 4

Simplest example, because this question in google top.

import os
from ctypes import CDLL

libc = CDLL(None)
stdout = os.dup(1)
silent = os.open(os.devnull, os.O_WRONLY)
os.dup2(silent, 1)
libc.printf(b"Hate this text")
os.dup2(stdout, 1)

Answer 5

We can use os.dup2() and os.pipe() to replace the entire stdout file descriptor (fd 1) with a pipe we can read from ourselves. You can do the same thing with stderr (fd 2).

This example uses select.select() to see if the pipe (our fake stdout) has data waiting to be written, so we can print it safely without blocking execution of our script.

As we are completely replacing the stdout file descriptor for this process and any subprocesses, this example can even capture output from child processes.

import os, sys, select

# the pipe would fail for some reason if I didn't write to stdout at some point
# so I write a space, then backspace (will show as empty in a normal terminal)
sys.stdout.write(' \b')
pipe_out, pipe_in = os.pipe()
# save a copy of stdout
stdout = os.dup(1)
# replace stdout with our write pipe
os.dup2(pipe_in, 1)

# check if we have more to read from the pipe
def more_data():
        r, _, _ = select.select([pipe_out], [], [], 0)
        return bool(r)

# read the whole pipe
def read_pipe():
        out = ''
        while more_data():
                out += os.read(pipe_out, 1024)

        return out

# testing print methods
import ctypes
libc = ctypes.CDLL('libc.so.6')

print 'This text gets captured by myStdOut'
libc.printf('This text fails to be captured by myStdOut\n')

# put stdout back in place 
os.dup2(stdout, 1)
print 'Contents of our stdout pipe:'
print read_pipe()