#!/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))