Function _gpg_run

apps/plugins/api_views.py:620–712

Run a GPG command using os.posix_spawn. os.posix_spawn skips pthread_atfork handlers, avoiding the indefinite hang that fork()-based approaches suffer under gevent+uWSGI. select.select() and time.sleep() are gevent-patched so reads and the waitpid poll yield to the hub coopera

(cmd, input_data=None, timeout=30)

def _gpg_run(cmd, input_data=None, timeout=30):
    """
    Run a GPG command using os.posix_spawn.

    os.posix_spawn skips pthread_atfork handlers, avoiding the indefinite hang
    that fork()-based approaches suffer under gevent+uWSGI.  select.select()
    and time.sleep() are gevent-patched so reads and the waitpid poll yield to
    the hub cooperatively.

    Returns (returncode, stdout_bytes, stderr_bytes).
    """
    import select as _select
    import signal as _signal
    import time as _time

    stdin_r,  stdin_w  = os.pipe()
    stdout_r, stdout_w = os.pipe()
    stderr_r, stderr_w = os.pipe()

    try:
        executable = shutil.which(cmd[0]) or cmd[0]
        pid = os.posix_spawn(
            executable, cmd, os.environ,
            file_actions=[
                (os.POSIX_SPAWN_DUP2,  stdin_r,  0),
                (os.POSIX_SPAWN_DUP2,  stdout_w, 1),
                (os.POSIX_SPAWN_DUP2,  stderr_w, 2),
                (os.POSIX_SPAWN_CLOSE, stdin_r),
                (os.POSIX_SPAWN_CLOSE, stdin_w),
                (os.POSIX_SPAWN_CLOSE, stdout_w),
                (os.POSIX_SPAWN_CLOSE, stderr_w),
                (os.POSIX_SPAWN_CLOSE, stdout_r),
                (os.POSIX_SPAWN_CLOSE, stderr_r),
            ],
        )
    except Exception:
        for fd in (stdin_r, stdin_w, stdout_r, stdout_w, stderr_r, stderr_w):
            try:
                os.close(fd)
            except OSError:
                pass
        raise

    for fd in (stdin_r, stdout_w, stderr_w):
        os.close(fd)

    try:
        if input_data:
            os.write(stdin_w, input_data)
    finally:
        os.close(stdin_w)

    out, err = [], []
    done = set()
    deadline = _time.monotonic() + timeout
    try:
        while len(done) < 2:
            remaining = deadline - _time.monotonic()