#!/usr/bin/env python3 """fenris: unprivileged entry point for the Fenris TUI and CLI. With no arguments, opens the TUI. Subcommands route through fenris-monitor for privileged operations. Spec: §1.2, §8.4 """ import argparse import os import subprocess import sys from pathlib import Path def add_runtime_packages() -> None: """Make the package-owned, pure-Python dependencies importable. RPM and deb installations deliberately use the target system's Python. Their dependencies are vendored without a copied interpreter so a distro Python minor-version update cannot leave Fenris linked to a removed ABI. The legacy development install keeps its venv fallback. """ runtime_dir = Path("/opt/fenris") vendor_dir = runtime_dir / "vendor" if vendor_dir.is_dir(): sys.path.insert(0, str(vendor_dir)) return site_packages = next((runtime_dir / "lib").glob("python*/site-packages"), None) if site_packages: sys.path.insert(0, str(site_packages)) add_runtime_packages() def is_root() -> bool: """Check if running as root.""" return os.geteuid() == 0 def run_monitor(*args: str) -> None: """Run fenris-monitor with the given arguments. If not root, re-exec under pkexec. """ monitor_cmd = "/usr/libexec/fenris/fenris-monitor" if is_root(): result = subprocess.run([monitor_cmd] + list(args)) sys.exit(result.returncode) else: # Use pkexec to elevate pkexec = subprocess.run( ["which", "pkexec"], capture_output=True ) if pkexec.returncode != 0: print( "Error: No polkit agent available. " "Run as root: sudo fenris-monitor ...", file=sys.stderr, ) sys.exit(1) result = subprocess.run(["pkexec", monitor_cmd] + list(args)) sys.exit(result.returncode) def cmd_tui(args: argparse.Namespace) -> None: """Open the TUI.""" from fenris.tui import run_tui run_tui() def cmd_status(args: argparse.Namespace) -> None: """Show status.""" from fenris.status import render_status print(render_status()) def cmd_sample(args: argparse.Namespace) -> None: """Trigger on-demand collection.""" run_monitor("collect") def cmd_monitor_pause(args: argparse.Namespace) -> None: """Pause monitoring.""" # Pause asks confirmation (§7.4) if not args.yes: response = input("Pause monitoring? [y/N] ") if response.lower() not in ("y", "yes"): print("Aborted.") return run_monitor("disable", "--now") def cmd_monitor_resume(args: argparse.Namespace) -> None: """Resume monitoring.""" # Resume does not ask confirmation (§7.4) run_monitor("enable", "--now") def cmd_baseline_set(args: argparse.Namespace) -> None: """Set baseline.""" run_monitor("baseline", "set", args.baseline_json) def cmd_baseline_clear(args: argparse.Namespace) -> None: """Clear baseline.""" run_monitor("baseline", "clear") def cmd_import(args: argparse.Namespace) -> None: """Import legacy history.""" # This is a one-off migration, not a privileged operation print("Legacy import: use fenris-import directly") def cmd_migrate(args: argparse.Namespace) -> None: """Apply forward-only schema migrations (IN-5, IN-6). Called by 'sudo make upgrade'. Raises on newer-schema store. """ from fenris.store import migrate_to_latest from pathlib import Path store_path = Path("/var/lib/fenris/observations.db") if not store_path.exists(): print("No observation store found — nothing to migrate.") return steps = migrate_to_latest(store_path) if steps: print(f"Migration complete: {steps} step(s) applied.") else: print("Schema already current.") def main() -> None: parser = argparse.ArgumentParser( prog="fenris", description="Fenris NVMe endurance monitor", ) parser.add_argument( "--version", action="version", version="%(prog)s 0.3.0" ) subparsers = parser.add_subparsers(dest="command") # Default: TUI (no subcommand) subparsers.add_parser("tui", help="Open the TUI (default)") # Status subparsers.add_parser("status", help="Show status") # Sample (on-demand collection) subparsers.add_parser("sample", help="Trigger on-demand collection") # Monitor subcommand monitor_parser = subparsers.add_parser("monitor", help="Monitor control") monitor_sub = monitor_parser.add_subparsers(dest="monitor_action") # monitor pause pause_parser = monitor_sub.add_parser("pause", help="Pause monitoring") pause_parser.add_argument( "-y", "--yes", action="store_true", help="Skip confirmation" ) pause_parser.set_defaults(func=cmd_monitor_pause) # monitor resume resume_parser = monitor_sub.add_parser("resume", help="Resume monitoring") resume_parser.set_defaults(func=cmd_monitor_resume) # Baseline subcommand baseline_parser = subparsers.add_parser("baseline", help="Baseline operations") baseline_sub = baseline_parser.add_subparsers(dest="baseline_action") baseline_set = baseline_sub.add_parser("set", help="Set baseline") baseline_set.add_argument("baseline_json", help="Baseline JSON data") baseline_set.set_defaults(func=cmd_baseline_set) baseline_clear = baseline_sub.add_parser("clear", help="Clear baseline") baseline_clear.set_defaults(func=cmd_baseline_clear) # Import import_parser = subparsers.add_parser("import", help="Import legacy history") import_parser.add_argument("path", help="Path to history.jsonl") import_parser.set_defaults(func=cmd_import) # Migrate (IN-5, IN-6) — called by upgrade, not for human use migrate_parser = subparsers.add_parser("migrate", help=argparse.SUPPRESS) migrate_parser.set_defaults(func=cmd_migrate) # Rejected commands for cmd in ["start", "stop", "run"]: reject_parser = subparsers.add_parser(cmd, help=argparse.SUPPRESS) reject_parser.set_defaults(func=lambda a: print( f"'{cmd}' is not a valid command. " f"Use 'fenris monitor resume' instead.", file=sys.stderr, )) args = parser.parse_args() if args.command is None or args.command == "tui": cmd_tui(args) elif args.command == "status": cmd_status(args) elif args.command == "sample": cmd_sample(args) elif args.command == "monitor": if args.monitor_action is None: monitor_parser.error("a subcommand is required") args.func(args) elif args.command == "baseline": if args.baseline_action is None: baseline_parser.error("a subcommand is required") args.func(args) elif args.command == "import": cmd_import(args) elif args.command == "migrate": cmd_migrate(args) if __name__ == "__main__": main()