pid-sock.py

pid-sock.py

#!/usr/bin/env python3

import os
import socket
import struct
import subprocess
import sys


def nested(fd):
    sock = socket.fromfd(int(fd), socket.AF_UNIX, socket.SOCK_STREAM)

    sock.sendmsg([b'hello'])

    real_pid, *_ = sock.recvmsg(1024)
    print('real PID from socket (transferred)', int(real_pid))


def main(prefix=''):
    a, b = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
    a.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
    b.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)

    with open(__file__) as fp:
        code = fp.read()

    command = ['python3', '-c', code, 'nested', str(b.fileno())]
    if prefix == 'flatpak':
        command = ['flatpak-spawn', '--directory=/', '--sandbox',
                   f'--forward-fd={b.fileno()}'] + command

    p = subprocess.Popen(command, pass_fds=(b.fileno(),), cwd='/')
    print('known PID from parent', p.pid)

    msg, anc, *_ = a.recvmsg(1024, socket.CMSG_LEN(1024))
    assert msg == b'hello'
    for level, ty, data in anc:
        if level == socket.SOL_SOCKET and ty == socket.SCM_CREDENTIALS:
            pid, uid, gid = struct.unpack('3i', data)
            print('real PID from socket (parent)', pid)

    a.sendmsg([bytes(str(pid), 'ascii')])

    p.wait()


if len(sys.argv) >= 2 and sys.argv[1] == 'nested':
    assert len(sys.argv) == 3
    nested(str(sys.argv[2]))
else:
    main(*sys.argv[1:])