flatpak/scripts/flatpak-coredumpctl

#!/usr/bin/env python3
import sys

if sys.version_info < (3, 10):
    print(
        f'A minimum Python version of at least 3.10 is required, '
        f'{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro} was found',
        file=sys.stderr
    )
    sys.exit(1)

import argparse
import json
import os
import re
import shutil
import subprocess
import shlex
import tempfile
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from signal import Signals
from typing import IO

### 'debug' subcommand ###

class DebugDumpArgs:
    @dataclass
    class AppId:
        app_id: str

    @dataclass
    class BuildDir:
        directory: str

    target: AppId | BuildDir
    coredumpctl_matches: str
    extra_flatpak_args: str
    gdb_arguments: str

    def __init__(self, ns: argparse.Namespace) -> None:
        if ns.app is None and ns.build_directory is None:
            raise ValueError("Either `--build-directory` or `APP` must be specified.")
        if ns.app is not None and ns.build_directory is not None:
            raise ValueError("`--build-directory` and `APP` are mutually exclusive.")
        if ns.app is not None:
            self.target = self.AppId(ns.app)
        else:
            self.target = self.BuildDir(ns.build_directory)
        self.coredumpctl_matches= ns.coredumpctl_matches
        self.extra_flatpak_args = ns.extra_flatpak_args
        self.gdb_arguments = ns.gdb_arguments

def debug_dump(args: DebugDumpArgs) -> int:
    match args.target:
        case DebugDumpArgs.AppId(app_id):
            flatpak_subcommand = "run"
            target_args = ["--command=gdb", "--devel", app_id]
        case DebugDumpArgs.BuildDir(build_dir):
            flatpak_subcommand = "build"
            target_args = [build_dir, "gdb"]

    # We need access to the host from the sandbox to run.
    flatpak_command = [
        "flatpak",
        flatpak_subcommand,
        "--filesystem=home",
        f"--filesystem={tempfile.gettempdir()}",
        *shlex.split(args.extra_flatpak_args),
        *target_args
    ]

    with tempfile.NamedTemporaryFile() as coredump:
        dumpres = subprocess.run(
            ["coredumpctl", "dump"] + shlex.split(args.coredumpctl_matches),
            stdout=coredump, stderr=subprocess.PIPE, text=True
        )
        if dumpres.returncode != 0:
            print("Failed to retrieve coredump via coredumpctl.", file=sys.stderr)
            if dumpres.stderr:
                print(f"Reason:\n{dumpres.stderr}", file=sys.stderr)

            return dumpres.returncode

        matches = re.findall(r".*Executable: (.*)", dumpres.stderr)
        if len(matches) != 1:
            print(f"Could not determine executable from coredumpctl output "
                  f"(found {len(matches)} 'Executable:' line(s)).",
                  file=sys.stderr)
            return 1
        executable = matches[0]
        if not executable.startswith(("/newroot/", "/app/")):
            print(f"Executable {executable} doesn't seem to be a flatpaked application.",
                file=sys.stderr)
        executable = executable.replace("/newroot", "", 1)
        flatpak_command.extend([executable, coredump.name])
        flatpak_command.extend(shlex.split(args.gdb_arguments))

        print(f"Running: `{shlex.join(flatpak_command)}`")
        return subprocess.run(flatpak_command).returncode

### 'list' subcommand ###

@contextmanager
def pager_stream(use_pager: bool) -> Iterator[IO[str]]:
    proc = None
    out: IO[str] = sys.stdout

    if use_pager and sys.stdout.isatty():
        pager = shlex.split(os.getenv("PAGER", "less"))
        if pager and shutil.which(pager[0]):
            env = os.environ.copy()
            # Use env instead of cli flags because other pagers
            # sometime also read the LESS variable
            env["LESS"] = "FRSX"
            proc = subprocess.Popen(
                pager,
                stdin=subprocess.PIPE,
                text=True,
                env=env
            )
            # subprocess.PIPE guarantees stdin is not None
            assert proc.stdin is not None
            out = proc.stdin

    try:
        yield out
    finally:
        if proc is not None and proc.stdin is not None:
            proc.stdin.close()
            proc.wait()

class TextStyle(Enum):
    RESET = "0"
    UNDERLINE = "4"
    GRAY = "90"

class TextStyler:
    def __init__(self) -> None:
        term = os.getenv("TERM")
        self.__use_color = (
            sys.stdout.isatty()
            and not os.getenv("NO_COLOR")
            and term not in (None, "dumb")
        )
        if os.getenv("FORCE_COLOR"):
            self.__use_color = True

    def style_to_ansi(self, style: TextStyle) -> str:
        if not self.__use_color:
            return ""
        return f"\033[{style.value}m"

    def style(self, style: TextStyle, string: str) -> str:
        return f"{self.style_to_ansi(style)}{string}{self.style_to_ansi(TextStyle.RESET)}"

def microseconds_to_datetime(us: int) -> str:
    return (
        datetime
            .fromtimestamp(us / 1_000_000)
            .astimezone()
            .strftime("%a %Y-%m-%d %H:%M:%S %Z")
    )

def format_bytes(num: float | None) -> str:
    if num is None:
        return "-"

    for unit in ['', 'K', 'M', 'G', 'T', 'P', 'E', 'Z']:
        if abs(num) < 1024.0:
            return f"{num:3.1f}{unit}"
        num /= 1024.0

    return f"{num:.1f}Yi"

def list_dumps(show_inaccessible: bool, use_pager: bool, reverse: bool, matches: list[str]) -> int:
    coredumpctl_command = ["coredumpctl", "list", "--json=short"]
    if reverse:
        coredumpctl_command.append("--reverse")
    coredumpctl_command.extend(matches)

    result = subprocess.run(
        coredumpctl_command,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )

    if result.returncode != 0:
        print("Failed to retrieve coredump list from coredumpctl.", file=sys.stderr)
        if result.stderr:
            print(f"Reason:\n{result.stderr}", file=sys.stderr)
        return result.returncode

    entries = json.loads(result.stdout)
    rows: list[dict[str, str]] = []
    widths: dict[str, int] = {
        "time": len("TIME"),
        "pid": len("PID"),
        "uid": len("UID"),
        "gid": len("GID"),
        "sig": len("SIG"),
        "corefile": len("COREFILE"),
        "exe": len("EXE"),
        "size": len("SIZE"),
    }
    for entry in entries:
        corefile = entry["corefile"]
        exe = entry["exe"]
        if not exe.startswith(("/newroot/", "/app/")):
            continue
        if not show_inaccessible and corefile != "present":
            continue
        row: dict[str, str] = {
            "time": microseconds_to_datetime(entry["time"]),
            "pid": str(entry["pid"]),
            "uid": str(entry["uid"]),
            "gid": str(entry["gid"]),
            "sig": Signals(entry["sig"]).name,
            "corefile": corefile,
            "exe": exe,
            "size": format_bytes(entry["size"])
        }
        rows.append(row)
        for key, value in row.items():
            widths[key] = max(widths[key], len(value))

    if len(rows) == 0:
        if show_inaccessible:
            print("No coredumps were found. Maybe try again with '--show-inaccessible'")
        else:
            print("No coredumps were found.")
        return 0

    with pager_stream(use_pager) as output_file:
        styler = TextStyler()

        print(
            styler.style(TextStyle.UNDERLINE,
                f"{'TIME':<{widths['time']}} "
                f"{'PID':>{widths['pid']}} "
                f"{'UID':>{widths['uid']}} "
                f"{'GID':>{widths['gid']}} "
                f"{'SIG':<{widths['sig']}} "
                f"{'COREFILE':<{widths['corefile']}} "
                f"{'EXE':<{widths['exe']}} "
                f"{'SIZE':<{widths['size']}}"
            ),
            file=output_file
        )

        for r in rows:
            corefile = r['corefile']
            if corefile != "present":
                corefile = styler.style(TextStyle.GRAY, corefile)
            print(
                f"{r['time']:<{widths['time']}} "
                f"{r['pid']:>{widths['pid']}} "
                f"{r['uid']:>{widths['uid']}} "
                f"{r['gid']:>{widths['gid']}} "
                f"{r['sig']:>{widths['sig']}} "
                f"{corefile:<{widths['corefile']}} "
                f"{r['exe']:<{widths['exe']}} "
                f"{r['size']:>{widths['size']}}",
                file=output_file
            )

    return 0

def main() -> int:
    if not shutil.which("coredumpctl"):
        print("'coredumpctl' is required but could not be found in PATH",
              file=sys.stderr)
        return 1

    parser = argparse.ArgumentParser(description=
        "Inspect and debug application coredumps from Flatpaks. "
        "Requires coredumpctl to retrieve crash dumps.")
    subparsers = parser.add_subparsers(dest="command")

    debug_parser = subparsers.add_parser("debug", help="Debug coredump inside a flatpak",
                                         description="Debug in gdb an application that crashed inside a flatpak")
    debug_parser.add_argument("-b", "--build-directory", default=None,
                              help="The build directory to use. It allows you to retrieve a coredump"
                              " for applications being built")
    debug_parser.add_argument("--extra-flatpak-args", default="", metavar="ARGS",
                              help="Extra argument to pass to flatpak")
    debug_parser.add_argument("app", nargs="?",
                              help="The flatpak application to use. eg. `org.gnome.Epiphany//3.28`")
    debug_parser.add_argument("-m", "--coredumpctl-matches", default="", metavar="MATCHES",
                              help="Coredumpctl matches, see `man coredumpctl` for more information")
    debug_parser.add_argument("--gdb-arguments", default="", metavar="ARGS",
                              help="Arguments to pass to gdb")

    list_parser = subparsers.add_parser("list", help="List available coredumps",
                                        description="List all coredumps from crashes that occurred inside flatpaks")
    list_parser.add_argument("--show-inaccessible", action="store_true",
                             help="List coredumps that are missing or inaccessible")
    list_parser.add_argument("-r", "--reverse", action="store_true",
                             help="Show the newest entries first")
    list_parser.add_argument("--no-pager", action="store_false", dest="use_pager",
                             help="Don't pipe output to a pager")
    list_parser.add_argument("matches", nargs="*",
                             help="Coredumpctl matches, see `man coredumpctl` for more information")

    args = parser.parse_args()
    match args.command:
        case "debug":
            try:
                validated_args = DebugDumpArgs(args)
            except ValueError as e:
                print(e, file=sys.stderr)
                debug_parser.print_help(sys.stderr)
                return 1
            exit_code = debug_dump(validated_args)
        case "list":
            exit_code = list_dumps(
                args.show_inaccessible,
                args.use_pager,
                args.reverse,
                args.matches
            )
        case _:
            parser.print_help(sys.stderr)
            return 1

    return exit_code

if __name__ == "__main__":
    sys.exit(main())