felhom-agent/configs/felhom-crash-guard

#!/usr/bin/python3
# felhom-crash-guard — a crashed host restarts by itself, but not forever (`09` decision 88, R-851, `11` §5.9).
#
# Install as /usr/local/sbin/felhom-crash-guard (0755 root:root), with felhom-crash-guard.service (boot / clean-stop)
# and felhom-crash-guard-check.timer (hourly re-arm check). Python 3, standard library only.
# Tests: configs/test_felhom_crash_guard.py (temp dirs; nothing real is touched).
#
# WHAT IT DOES
#   boot        early at every boot. Was the previous boot ended CLEANLY? (the clean-stop marker exists). If not, this
#               boot follows an UNCLEAN stop — a kernel crash, a power cut or a hard reset (they cannot be told apart
#               on these boxes: measured 2026-10-04 on demo-hp, efi_pstore is on yet saved NOTHING for a real panic;
#               the journal and `last` show only "no shutdown"). It records the unclean boot, counts those in the last
#               WINDOW_MINUTES, and sets kernel.panic:
#                 - fewer than LIMIT-1 recent unclean boots → kernel.panic = PANIC_SECONDS (a crash restarts the box);
#                 - LIMIT-1 or more → the guard TRIPS: kernel.panic = 0, so the LIMIT-th crash within the window
#                   leaves the box OFF (operator's own words: "if it crashes 3 times within one hour, it stays off").
#               A tripped guard stays tripped across further boots until it re-arms.
#   clean-stop  ExecStop of the service: writes the clean-stop marker during an orderly shutdown or reboot.
#   check       hourly: a tripped guard re-arms after REARM_HOURS of normal running (since the trip AND since boot).
#   rearm       the operator re-arms by hand (`felhom-crash-guard rearm`).
#   status      prints the state.
# The state is /var/lib/felhom-crash-guard/state.json (0644: the non-root agent reads it into its host report).
# Before the service runs (very early boot) the kernel default kernel.panic = 0 applies, so a crash THAT early leaves
# the box off — the safe side: a box that cannot reach userspace must not loop.
import json
import os
import sys
import time

CONF = "/etc/felhom/crash-guard.conf"
STATE_DIR = "/var/lib/felhom-crash-guard"
DEFAULTS = {"LIMIT": 3, "WINDOW_MINUTES": 60, "PANIC_SECONDS": 10, "REARM_HOURS": 24}


class Env:
    """Paths and clock; tests replace them."""

    def __init__(self, conf=CONF, state_dir=STATE_DIR, panic_path="/proc/sys/kernel/panic",
                 uptime_path="/proc/uptime", boot_id_path="/proc/sys/kernel/random/boot_id"):
        self.conf, self.state_dir = conf, state_dir
        self.panic_path, self.uptime_path, self.boot_id_path = panic_path, uptime_path, boot_id_path

    def now(self):
        return time.time()

    def log(self, line):
        print(line, file=sys.stderr, flush=True)
        try:
            import subprocess
            subprocess.run(["logger", "-t", "felhom-crash-guard", line], timeout=10)
        except Exception:
            pass


def iso(t):
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t))


def parse_iso(s):
    import calendar
    return calendar.timegm(time.strptime(s, "%Y-%m-%dT%H:%M:%SZ"))


def load_conf(env):
    c = dict(DEFAULTS)
    try:
        for line in open(env.conf):
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, v = (x.strip() for x in line.split("=", 1))
            if k in c and v.isdigit() and int(v) >= (1 if k != "PANIC_SECONDS" else 1):
                c[k] = int(v)
    except OSError:
        pass
    return c


def state_path(env):
    return os.path.join(env.state_dir, "state.json")


def marker_path(env):
    return os.path.join(env.state_dir, "clean-stop")


def load_state(env):
    try:
        with open(state_path(env)) as f:
            s = json.load(f)
        return s if isinstance(s, dict) else None
    except (OSError, ValueError):
        return None


def save_state(env, s):
    os.makedirs(env.state_dir, mode=0o755, exist_ok=True)
    tmp = state_path(env) + ".tmp"
    with open(tmp, "w") as f:
        json.dump(s, f, indent=2, sort_keys=True)
        f.write("\n")
    os.chmod(tmp, 0o644)
    os.replace(tmp, state_path(env))


def set_panic(env, seconds):
    with open(env.panic_path, "w") as f:
        f.write(f"{seconds}\n")


def read(path, default=""):
    try:
        with open(path) as f:
            return f.read().strip()
    except OSError:
        return default


def summarize(s, c, now):
    window = c["WINDOW_MINUTES"] * 60
    times = [parse_iso(t) for t in s.get("unclean_boots", [])]
    after = parse_iso(s["rearmed_at"]) if s.get("rearmed_at") else 0
    # a re-arm starts a fresh window (or the next unclean boot would trip again at once); the history stays
    s["unclean_boots_in_window"] = sum(1 for t in times if now - t <= window and t > after)
    s["unclean_boots_24h"] = sum(1 for t in times if now - t <= 86400)
    s["config"] = c
    s["updated_at"] = iso(now)


def boot(env):
    c = load_conf(env)
    now = env.now()
    try:
        up = float(read(env.uptime_path, "0").split()[0])
    except (ValueError, IndexError):
        up = 0.0
    boot_at = now - up
    prev = load_state(env)
    first = prev is None
    s = prev or {"version": 1, "unclean_boots": [], "tripped": False}
    clean = os.path.exists(marker_path(env))
    unclean = (not first) and (not clean)
    try:
        os.remove(marker_path(env))
    except OSError:
        pass
    # keep 7 days of history (the 24 h figure and the operator's view), drop older
    s["unclean_boots"] = [t for t in s.get("unclean_boots", []) if now - parse_iso(t) <= 7 * 86400]
    if unclean:
        s["unclean_boots"].append(iso(boot_at))
    s["last_boot_at"] = iso(boot_at)
    s["last_boot_unclean"] = unclean
    s["boot_id"] = read(env.boot_id_path, "unknown")
    summarize(s, c, now)
    if not s.get("tripped") and s["unclean_boots_in_window"] >= c["LIMIT"] - 1:
        s["tripped"], s["tripped_at"] = True, iso(now)
        s["tripped_reason"] = (f"{s['unclean_boots_in_window']} unclean boots within {c['WINDOW_MINUTES']} minutes — "
                               f"the next crash leaves the box off (limit {c['LIMIT']})")
        env.log(f"crash-guard: TRIPPED: {s['tripped_reason']}")
    panic = 0 if s.get("tripped") else c["PANIC_SECONDS"]
    set_panic(env, panic)
    s["kernel_panic"] = panic
    s["armed"] = not s.get("tripped")
    save_state(env, s)
    env.log(f"crash-guard: boot first={first} unclean={unclean} in-window={s['unclean_boots_in_window']} "
            f"tripped={s.get('tripped')} kernel.panic={panic}")
    return 0


def clean_stop(env):
    os.makedirs(env.state_dir, mode=0o755, exist_ok=True)
    with open(marker_path(env), "w") as f:
        f.write(iso(env.now()) + "\n")
    env.log("crash-guard: clean stop recorded")
    return 0


def rearm(env, by):
    c = load_conf(env)
    now = env.now()
    s = load_state(env) or {"version": 1, "unclean_boots": []}
    was = bool(s.get("tripped"))
    s["tripped"] = False
    s["armed"] = True
    s["rearmed_at"], s["rearmed_by"] = iso(now), by
    if was:
        s["last_trip"] = {"at": s.get("tripped_at"), "reason": s.get("tripped_reason")}
    s.pop("tripped_at", None)
    s.pop("tripped_reason", None)
    summarize(s, c, now)
    set_panic(env, c["PANIC_SECONDS"])
    s["kernel_panic"] = c["PANIC_SECONDS"]
    save_state(env, s)
    env.log(f"crash-guard: RE-ARMED by {by} (was tripped: {was}); kernel.panic={c['PANIC_SECONDS']}")
    return 0


def check(env):
    c = load_conf(env)
    now = env.now()
    s = load_state(env)
    if not s:
        return 0
    if s.get("tripped"):
        since = max(parse_iso(s["tripped_at"]), parse_iso(s.get("last_boot_at", s["tripped_at"])))
        if now - since >= c["REARM_HOURS"] * 3600:
            return rearm(env, f"timer ({c['REARM_HOURS']} h of normal running)")
    summarize(s, c, now)
    save_state(env, s)
    return 0


def main(argv, env=None):
    env = env or Env()
    cmd = argv[1] if len(argv) == 2 else ""
    if cmd == "boot":
        return boot(env)
    if cmd == "clean-stop":
        return clean_stop(env)
    if cmd == "check":
        return check(env)
    if cmd == "rearm":
        return rearm(env, "operator")
    if cmd == "status":
        print(json.dumps(load_state(env), indent=2, sort_keys=True))
        return 0
    print("usage: felhom-crash-guard boot|clean-stop|check|rearm|status", file=sys.stderr)
    return 2


if __name__ == "__main__":
    if os.geteuid() != 0:
        print("felhom-crash-guard: must run as root", file=sys.stderr)
        sys.exit(2)
    sys.exit(main(sys.argv))