"""Terminal-attached invocation of Fenris's fixed privileged operations. Both human entry points use this module. Authentication has no frontend deadline; collection runtime is bounded by the native scheduler (ADR 0003). """ import os import shlex import subprocess MONITOR_HELPER = "/usr/libexec/fenris/fenris-monitor" class MonitorError(Exception): """An action failed, with a message and exit status for either renderer.""" def __init__(self, message: str, exit_code: int = 1): super().__init__(message) self.exit_code = exit_code def run_monitor(*args: str, helper_path: str = MONITOR_HELPER) -> None: """Run one helper operation, inheriting the terminal for authentication. Never invoke a shell, retry an action, or fall back to sudo automatically. The helper owns the operation allow-list and privileged state changes. """ helper_command = [helper_path, *args] needs_auth = os.geteuid() != 0 command = ["pkexec", *helper_command] if needs_auth else helper_command root_hint = ( " If authentication is unavailable, run in your terminal: " + shlex.join(["sudo", *helper_command]) ) if needs_auth else "" try: # The collector owns its 90-second runtime limit. A frontend timeout # would also count time spent authenticating or waiting for a run. result = subprocess.run(command) except FileNotFoundError as exc: raise MonitorError( "Command not found: %s.%s" % (exc.filename or command[0], root_hint), 127, ) from exc except OSError as exc: raise MonitorError("Cannot run monitoring action: %s.%s" % (exc, root_hint)) from exc except KeyboardInterrupt as exc: raise MonitorError( "Action interrupted. Check fenris status before retrying.", 130, ) from exc if result.returncode: exit_code = result.returncode if result.returncode > 0 else 128 - result.returncode raise MonitorError( "Action failed (exit %d). Check fenris status before retrying.%s" % (exit_code, root_hint), exit_code, )