file descriptor share across process

client.py

import socket
import os
import array


def send_fd(sock, fd):
    """Send a file descriptor to another process via a Unix domain socket."""
    fds = array.array("i", [fd])
    sock.sendmsg([b"x"], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fds)])


def recv_fd(sock):
    """Receive a file descriptor from another process via a Unix domain socket."""
    msg, ancdata, flags, addr = sock.recvmsg(
        1, socket.CMSG_SPACE(array.array("i").itemsize)
    )
    for cmsg_level, cmsg_type, cmsg_data in ancdata:
        if cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS:
            return array.array("i", cmsg_data)[0]


# Path to the Unix socket
socket_path = "/tmp/unix_socket_example"

# Make sure the socket does not already exist

# Create a Unix socket
client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

# Bind the socket to the path
client.connect(socket_path)

# Listen for incoming connections
fd_to_send = os.open("example.txt", os.O_RDONLY)

# Send the file descriptor to the parent process
send_fd(client, fd_to_send)

os.close(fd_to_send)
client.close()

server.py

import socket
import os
import array


def send_fd(sock, fd):
    """Send a file descriptor to another process via a Unix domain socket."""
    fds = array.array("i", [fd])
    sock.sendmsg([b"x"], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fds)])


def recv_fd(sock):
    """Receive a file descriptor from another process via a Unix domain socket."""
    msg, ancdata, flags, addr = sock.recvmsg(
        1, socket.CMSG_SPACE(array.array("i").itemsize)
    )
    for cmsg_level, cmsg_type, cmsg_data in ancdata:
        if cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS:
            return array.array("i", cmsg_data)[0]


# Path to the Unix socket
socket_path = "/tmp/unix_socket_example"

# Make sure the socket does not already exist
try:
    os.unlink(socket_path)
except OSError:
    if os.path.exists(socket_path):
        raise

# Create a Unix socket
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

# Bind the socket to the path
server.bind(socket_path)

# Listen for incoming connections
server.listen()

while 1:
    client, addr = server.accept()
    while 1:
        try:
            fd = recv_fd(client)
        except Exception as e:
            print(e)
            break
        if fd is None:
            break
        with os.fdopen(fd, "r") as f:
            print(f"Parent process read: {f.read()}")
    client.close()

https://stackoverflow.com/questions/1997622/can-i-open-a-socket-and-pass-it-to-another-process-in-linux