tcsetpgrp.py

import pty
import time
import os
import termios
import signal
import sys
import traceback
import threading

if len(sys.argv) > 1:
    if sys.argv[1] == "":
        sigmask = []
    else:
        sigmask = [getattr(signal, sig) for sig in sys.argv[1:]]
else:
    sigmask = [signal.SIGTTOU, signal.SIGTTIN]


def child(fd):
    try:
        ctty = 1
        pid = os.fork()
        if pid == 0:
            pid = os.getpid()
            os.setpgrp()

            signal.pthread_sigmask(signal.SIG_BLOCK, sigmask)
            os.tcsetpgrp(ctty, pid)
            print("[+] did tcsetpgrp")
            signal.pthread_sigmask(signal.SIG_UNBLOCK, sigmask)
            print("[+] did unblock")
        else:
            _, st = os.waitpid(pid, os.WUNTRACED)
            if st == 0:
                print(f"[+] child exited successfully")
            else:
                if os.WIFSTOPPED(st):
                    print(f"[-] child signalled: {os.WSTOPSIG(st)}")
                os.kill(pid, 9)
            if os.tcgetpgrp(1) == pid:
                print("[+] child DID succeed in tcsetpgrp")
            elif os.tcgetpgrp(1) == os.getpid():
                print("[-] child failed, we are still the foreground group")
        os._exit(0)
    except Exception:
        traceback.print_exc()
    finally:
        os._exit(1)


def copy_out(fh):
    while True:
        try:
            data = os.read(fh, 1024)
        except OSError:
            break
        if len(data) == 0:
            break
        sys.stdout.buffer.write(data)


if __name__ == "__main__":
    (pid, fd) = pty.fork()
    if pid == 0:
        child(fd)

    threading.Thread(target=copy_out, args=(fd,), daemon=True).start()

    # print(os.read(fd, 1024).decode("utf-8"))
    # time.sleep(5)
    print("child exit: ", os.waitpid(pid, 0))
    os.close(fd)