#!/usr/bin/env python3
"""Quit / Force Quit's signal on Linux (src/platform/linux.js), sent only to the process the host listed.

    dokki-signal <pid> TERM|KILL <start> [<exe>]

<start> is field 22 of /proc/<pid>/stat (the start time in clock ticks) as the running list
read it, or "-" when it could not; <exe> the /proc/<pid>/exe link it read (optional).

Race-free where the kernel allows (Linux 5.3+, Python 3.9+): the process is pinned with
pidfd_open first, then /proc/<pid> is checked, then the signal goes through
pidfd_send_signal. A pid reused at any point is refused: either the pidfd already names
the old (ended) process, or the start time read after pinning is not the listed one.
Without pidfds: the same checks, then kill(2) right after (the old check-then-kill).

Exit status: 0 sent, 3 the process is gone, 9 the pid is now another process (or another
executable), 2 invalid arguments, 1 anything else (EPERM...). python3 stdlib only; unit
tests in native/linux/tests/test_dokki_signal.py.
"""

import errno
import os
import signal
import sys

EXIT_INVALID, EXIT_GONE, EXIT_IDENTITY = 2, 3, 9
SIGNALS = {"TERM": signal.SIGTERM, "KILL": signal.SIGKILL}


class Gone(Exception):
    pass


class Changed(Exception):
    pass


def parse_start(stat):
    """Field 22 of /proc/<pid>/stat (after the parenthesised comm, which may hold spaces)."""
    text = stat.decode("utf-8", "replace") if isinstance(stat, bytes) else str(stat)
    end = text.rfind(")")
    if end < 0:
        return ""
    fields = text[end + 2:].split()
    return fields[19] if len(fields) > 19 else ""


def parse_args(argv):
    if len(argv) not in (4, 5):
        raise ValueError("usage: dokki-signal <pid> TERM|KILL <start>|- [<exe>]")
    pid_text, name, start = argv[1], argv[2], argv[3]
    if not pid_text.isdigit() or len(pid_text) > 10 or int(pid_text) <= 1:
        raise ValueError("pid")
    if name not in SIGNALS:
        raise ValueError("signal")
    if start != "-" and (not start.isdigit() or len(start) > 20):
        raise ValueError("start")
    exe = argv[4] if len(argv) == 5 else ""
    if exe and (not exe.startswith("/") or "\0" in exe or len(exe) > 4096):
        raise ValueError("exe")
    return int(pid_text), SIGNALS[name], ("" if start == "-" else start), exe


class Proc:
    """The system calls, apart so tests can fake them."""

    def __init__(self, root="/proc"):
        self.root = root

    def pidfd_open(self, pid):
        opener = getattr(os, "pidfd_open", None)
        if opener is None:
            raise NotImplementedError
        try:
            return opener(pid)
        except OSError as error:
            if error.errno in (errno.ENOSYS, errno.EINVAL):
                raise NotImplementedError from error
            raise

    def stat(self, pid):
        with open("%s/%d/stat" % (self.root, pid), "rb") as f:
            return f.read()

    def exe(self, pid):
        return os.readlink("%s/%d/exe" % (self.root, pid))

    def pidfd_send_signal(self, fd, sig):
        sender = getattr(signal, "pidfd_send_signal", None)
        if sender is None:
            raise NotImplementedError
        sender(fd, sig)

    def kill(self, pid, sig):
        os.kill(pid, sig)

    def close(self, fd):
        os.close(fd)


def check(proc, pid, start, exe):
    try:
        got = parse_start(proc.stat(pid))
    except FileNotFoundError as error:
        raise Gone() from error
    except ProcessLookupError as error:
        raise Gone() from error
    if start and got != start:
        raise Changed()
    if exe:
        try:
            if proc.exe(pid) != exe:
                raise Changed()
        except FileNotFoundError as error:
            raise Gone() from error


def send(proc, pid, sig, start, exe):
    """-> "pidfd" or "kill" (how it went); raises Gone / Changed / OSError."""
    try:
        fd = proc.pidfd_open(pid)
    except NotImplementedError:
        fd = None
    except ProcessLookupError as error:
        raise Gone() from error
    if fd is None:
        check(proc, pid, start, exe)
        try:
            proc.kill(pid, sig)
        except ProcessLookupError as error:
            raise Gone() from error
        return "kill"
    try:
        # Pinned: from here on the pidfd names one process. Checked after pinning, so a reuse in
        # between shows as another start time (or as an ended process the signal cannot reach).
        check(proc, pid, start, exe)
        try:
            proc.pidfd_send_signal(fd, sig)
        except NotImplementedError:
            proc.kill(pid, sig)  # an old Python on a new kernel: still checked just before
            return "kill"
        except ProcessLookupError as error:
            raise Gone() from error
        return "pidfd"
    finally:
        proc.close(fd)


def main(argv, proc=None):
    try:
        pid, sig, start, exe = parse_args(argv)
    except ValueError as error:
        sys.stderr.write("invalid: %s\n" % error)
        return EXIT_INVALID
    try:
        how = send(proc or Proc(), pid, sig, start, exe)
    except Gone:
        sys.stderr.write("gone\n")
        return EXIT_GONE
    except Changed:
        sys.stderr.write("identity\n")
        return EXIT_IDENTITY
    except OSError as error:
        sys.stderr.write("failed: %s\n" % error.strerror)
        return 1
    sys.stdout.write("%s\n" % how)
    return 0


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