kadence/desktop/kadence_desktop/blocking/processes.py

"""Application blocking by watching `/proc`.

No root needed: these are your own processes. The scan reads `/proc/*/comm` and
the cmdline, matches against the configured names, and either terminates the
process (SIGTERM, then SIGKILL if it ignores us) or just reports it, depending
on the setting.

Matching is deliberately forgiving — `steam` catches `steamwebhelper`, and
`telegram` catches `Telegram` — and an entry containing `*` or `?` is treated as
a glob (`steam*`, `*discord*`, `telegram-?`), matched against both the process
name and its command line. It never matches ourselves, PID 1, or anything
outside the current user's processes.

Every match carries its **start time** as well as its pid. A pid on its own is
not an identity: Linux reuses them, and the blocker remembers which processes it
has already told you about. Keyed by pid alone, a relaunch that happens to land
on a recycled number would inherit the old entry and be let through — which is
exactly the "it started the second time" this file is here to prevent.
"""

from __future__ import annotations

import logging
import os
import signal
from fnmatch import fnmatch
from dataclasses import dataclass
from pathlib import Path

log = logging.getLogger(__name__)

_SELF_MARKERS = ("kadence", "python3 -m kadence_desktop")


def matches(pattern: str, name: str, cmdline: str) -> bool:
    """Whether one block-list entry matches a process.

    A mask is anchored to the process *name* (so `steam*` does not match every
    command that merely mentions steam) but also tried against the whole command
    line, which is how `*discord*` catches an Electron app launched by path.
    """
    pattern = pattern.strip().lower()
    if not pattern:
        return False
    if any(char in pattern for char in "*?["):
        return fnmatch(name, pattern) or fnmatch(cmdline, pattern)
    return pattern in name or pattern in cmdline


@dataclass(slots=True)
class RunningApp:
    pid: int
    name: str
    cmdline: str
    matched: str
    #: Field 22 of `/proc/<pid>/stat` — clock ticks since boot. With the pid it
    #: makes an identity that survives pid reuse. 0 when it could not be read.
    start_time: int = 0

    @property
    def identity(self) -> tuple[int, int]:
        return (self.pid, self.start_time)


def _read(path: Path) -> str:
    try:
        return path.read_text("utf-8", errors="replace")
    except OSError:
        return ""


def start_time(pid: int) -> int:
    """Clock ticks since boot at which this pid started, 0 if unreadable.

    `comm` sits in `stat` wrapped in parentheses and may itself contain spaces
    and parentheses, so the fields are counted from the *last* `)` — the one
    place this file cannot naively split on whitespace.
    """
    raw = _read(Path(f"/proc/{pid}/stat"))
    cut = raw.rfind(")")
    if cut < 0:
        return 0
    fields = raw[cut + 2 :].split()
    # stat's field 3 is `state`, so field 22 is index 19 of what follows comm.
    if len(fields) < 20:
        return 0
    try:
        return int(fields[19])
    except ValueError:
        return 0


def alive(pid: int, started: int = 0) -> bool:
    """Whether this exact process is still running — not merely this pid."""
    if not Path(f"/proc/{pid}").exists():
        return False
    return started == 0 or start_time(pid) == started


def scan(patterns: list[str]) -> list[RunningApp]:
    """Every live process of this user matching one of `patterns`."""
    wanted = [pattern.strip().lower() for pattern in patterns if pattern.strip()]
    if not wanted:
        return []

    uid = os.getuid()
    self_pid = os.getpid()
    found: list[RunningApp] = []

    for entry in Path("/proc").iterdir():
        if not entry.name.isdigit():
            continue
        pid = int(entry.name)
        if pid in (1, self_pid):
            continue
        try:
            if entry.stat().st_uid != uid:
                continue
        except OSError:
            continue

        comm = _read(entry / "comm").strip()
        if not comm:
            continue
        cmdline = _read(entry / "cmdline").replace("\x00", " ").strip()
        name, command = comm.lower(), cmdline.lower()
        if any(marker in f"{name} {command}" for marker in _SELF_MARKERS):
            continue

        match = next((pattern for pattern in wanted if matches(pattern, name, command)), None)
        if match:
            found.append(
                RunningApp(
                    pid=pid,
                    name=comm,
                    cmdline=cmdline[:200],
                    matched=match,
                    start_time=start_time(pid),
                )
            )
    return found


def terminate(app: RunningApp, *, force: bool = False) -> bool:
    """Signal the process once. True when the signal landed (or it is gone).

    Deliberately **not** blocking. The previous version sent SIGTERM and then
    busy-waited two seconds for the process to die before escalating, on
    whichever thread happened to call it — which on the desktop is the one
    drawing the countdown. Since the blocker now rescans every few seconds
    anyway, the escalation is simply the *next* pass's job (see
    `BlockerManager._enforce_apps`), and nothing has to wait for anybody.

    A `PermissionError` on SIGTERM escalates immediately rather than next pass:
    a process that refused the polite signal on permission grounds is not about
    to quit on its own.
    """
    try:
        os.kill(app.pid, signal.SIGKILL if force else signal.SIGTERM)
    except ProcessLookupError:
        return True
    except PermissionError:
        if force:
            log.warning("not allowed to stop %s (pid %s)", app.name, app.pid)
            return False
        log.warning("not allowed to ask %s (pid %s) to quit", app.name, app.pid)
        try:
            os.kill(app.pid, signal.SIGKILL)
        except ProcessLookupError:
            return True
        except PermissionError:
            log.warning("not allowed to stop %s (pid %s)", app.name, app.pid)
            return False
    return True