replacement for coredumpctl

coredumpselect.py

#!/usr/bin/env python3
"""
Allows the user to select one of the coredumps from the journal using fzf,

places the coredump at the given path (default /tmp/coredump),
links the related program binary to the given path (default /tmp/coredump-program),

and optionally attaches gdb.
"""
import argparse
import os
import shlex
import shutil
import signal
import subprocess
import tempfile

import systemd.journal


def get_journal_entries(match):
    """
    Returns all journal entries that match the given matcher.
    """
    reader = systemd.journal.Reader()
    reader.add_match(match)
    yield from reader


cli = argparse.ArgumentParser()
cli.add_argument("--gdb", action="store_true")
cli.add_argument("--coredump-filename", default="/tmp/coredump")
cli.add_argument("--coredump-program", default="/tmp/coredump-program")
cli.add_argument("--decompressor")
args = cli.parse_args()

# iterate over the journal to get a list of all choices
choices = {}
for entry in get_journal_entries("CODE_FUNC=submit_coredump"):
    try:
        coredump = entry["COREDUMP_FILENAME"]
        program = entry["COREDUMP_EXE"]
        timestamp = entry["COREDUMP_TIMESTAMP"]
    except KeyError:
        continue

    if not os.path.isfile(coredump):
        continue
    if not os.path.isfile(program):
        continue

    try:
        sig_name = signal.Signals(entry["COREDUMP_SIGNAL"]).name
    except:
        sig_name = "<unknown signal>"

    choice_string = f"{timestamp} {sig_name} {program}"

    choices[choice_string] = program, coredump

if not choices:
    print("no coredumps are available")
    raise SystemExit(1)

# allow the user to make a choice using fzf
with tempfile.NamedTemporaryFile("w") as tempfile:
    tempfile.write("\0".join(reversed(choices)))
    tempfile.flush()
    selector = subprocess.Popen(
        f"fzf --read0 --print0 < {shlex.quote(tempfile.name)}",
        shell=True,
        env=os.environ,
        stdout=subprocess.PIPE,
    )
    choice_string = selector.communicate()[0].decode().rstrip("\0")
chosen_program, chosen_coredump = choices[choice_string]

# print all metadata of the chosen coredump
reader = systemd.journal.Reader()
for entry in get_journal_entries(f"COREDUMP_FILENAME={chosen_coredump}"):
    oneline = []
    multiline = []
    for key, value in entry.items():
        if key.startswith("COREDUMP_"):
            key = key[9:]
            if isinstance(value, str) and "\n" in value:
                multiline.append((key, value))
            else:
                oneline.append((key, value))
    for key, value in multiline:
        print(f"\N{ESCAPE}[1m{key}\N{ESCAPE}[m\n{value}")
    for key, value in oneline:
        print(f"\N{ESCAPE}[1m{key}\N{ESCAPE}[m = {value}")
    print(f'\nJournal message:\n\n{entry.get("MESSAGE")}')
    break

# determine the decompression utility
if args.decompressor:
    decompressor = shlex.split(args.decompressor)
else:
    decompressor = {
        '.lz4': ['lz4', '-d'],
        '.zst': ['zstd', '-d'],
        '.xz': ['xz', '-d'],
        '.gz': ['gzip', '-d'],
    }.get(os.path.splitext(chosen_coredump)[1].lower())
    if decompressor is None:
        print(f'unknown extension {ext!r}')
        raise SystemExit(1)

if shutil.which(decompressor[0]) is None:
    print(f"decompressor {decompressor[0]!r} is not installed")
    raise SystemExit(1)

# unpack the compressed coredump to the chosen path
with open(chosen_coredump, "rb") as rfile, open(args.coredump_filename, "wb") as wfile:
    subprocess.check_call(decompressor, stdin=rfile, stdout=wfile)
print(f"coredump unpacked at: {shlex.quote(args.coredump_filename)}")

# link the coredump program binary to the chosen path
if os.path.isfile(args.coredump_program):
    os.unlink(args.coredump_program)
os.symlink(chosen_program, args.coredump_program)
print(f"executable linked at: {shlex.quote(args.coredump_program)}")

if args.gdb:
    # the user wants us to attach gdb to the coredump
    os.execvp("gdb", ["gdb", args.coredump_program, "-c", args.coredump_filename])