Fenris/scripts/fenris

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


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 print_status
    print_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()