A demo to generate coredump using a forked process

async_gcore.py

import os
import sys
import copy
import time
import syscall
import subprocess


def inject_syscall(
    pid: int,
    saved_regs: syscall.UserRegsStruct,
    syscall_no: int,
    arg1: int,
    arg2: int,
    arg3: int,
    arg4: int,
    arg5: int,
    arg6: int,
) -> int:
    regs = copy.copy(saved_regs)
    regs.rax = syscall_no
    regs.rdi = arg1
    regs.rsi = arg2
    regs.rdx = arg3
    regs.r10 = arg4
    regs.r8 = arg5
    regs.r9 = arg6
    regs.rip -= 2
    syscall.ptrace(syscall.PTRACE_SETREGS, pid, 0, regs.byref())
    syscall.ptrace(syscall.PTRACE_SINGLESTEP, pid, 0, 0)
    os.wait()
    rregs = syscall.UserRegsStruct()
    syscall.ptrace(syscall.PTRACE_GETREGS, pid, 0, rregs.byref())
    return rregs.rax


if __name__ == "__main__":
    elf, pid = sys.argv[1], int(sys.argv[2])

    tids = [int(tid) for tid in os.listdir(f"/proc/{pid}/task")]
    start = time.time()
    for tid in tids:
        syscall.ptrace(syscall.PTRACE_ATTACH, tid, 0, 0)
        os.waitpid(tid, 0)

    saved_regs = syscall.UserRegsStruct()
    syscall.ptrace(syscall.PTRACE_GETREGS, pid, 0, saved_regs.byref())
    offset = saved_regs.rip - 0x400000
    args = [
        "bpftrace",
        "--unsafe",
        "-e",
        f"u:{elf}:0x{offset:x} /pid != {pid}/ {{signal(19);}}",
    ]
    pipe_r, pipe_w = os.pipe()
    popen = subprocess.Popen(args, stdout=os.fdopen(pipe_w, "w"))
    os.fdopen(pipe_r).read(1)

    child = inject_syscall(pid, saved_regs, 57, 0, 0, 0, 0, 0, 0)  # fork(2)

    syscall.ptrace(syscall.PTRACE_SETREGS, pid, 0, saved_regs.byref())
    for tid in tids:
        syscall.ptrace(syscall.PTRACE_DETACH, tid, 0, 0)
    downtime = time.time() - start

    popen.kill()
    popen.wait()

    proc = subprocess.run(
        f"gdb --batch -ex gcore -p {child}".split(),
        capture_output=True,
    )
    os.kill(child, 9)
    print(proc.stdout.decode(), file=sys.stdout)
    print(proc.stderr.decode(), file=sys.stderr)

    downtime -= time.time()
    syscall.ptrace(syscall.PTRACE_ATTACH, pid, 0, 0)
    os.wait()
    saved_regs = syscall.UserRegsStruct()
    syscall.ptrace(syscall.PTRACE_GETREGS, pid, 0, saved_regs.byref())

    inject_syscall(pid, saved_regs, 61, child, 0, 0, 0, 0, 0)  # wait4(2)
    syscall.ptrace(syscall.PTRACE_SETREGS, pid, 0, saved_regs.byref())
    syscall.ptrace(syscall.PTRACE_DETACH, pid, 0, 0)
    downtime += time.time()
    print(f"the whole downtime is {downtime}")