diff --git a/strix/interface/main.py b/strix/interface/main.py index c9bd55961..8ac1c81aa 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -6,6 +6,8 @@ Strix Agent Interface import argparse import asyncio import contextlib +import os +import signal import sys from pathlib import Path @@ -418,7 +420,75 @@ def _bootstrap_scan(args: argparse.Namespace) -> None: telemetry_start(args) +def _install_cleanup_sigterm_handler() -> None: + """Install a cleanup-aware SIGTERM fallback before the scan engine starts. + + ``run_cli`` installs its own SIGINT/SIGTERM/SIGHUP handler once a report + state exists, but the interactive Go TUI installs none Python-side. Without + an early handler, a SIGTERM that arrives mid-setup or during an interactive + scan terminates the process by the default action, skipping report-state + cleanup. This fallback runs ``cleanup(status="interrupted")`` when a report + state exists, then exits; scan paths replace it with their own handler as + soon as it is safe to do so. + """ + + def _on_sigterm(_signum: int, _frame: object) -> None: + from strix.report.state import get_global_report_state + + state = get_global_report_state() + if state is not None: + state.cleanup(status="interrupted") + sys.exit(1) + + signal.signal(signal.SIGTERM, _on_sigterm) + + +def _enable_parent_death_signal() -> None: + """On Linux frozen binaries, ask the kernel to deliver SIGTERM when our parent dies. + + Strix ships as a PyInstaller onefile binary: the process users see is an + outer bootloader that spawns the real Python app as a child. When a caller + stops a scan by killing the direct child -- e.g. ``subprocess.run(..., timeout=)`` + sends SIGKILL, or a container/pipeline sends SIGKILL -- the bootloader dies + but the inner app cannot receive or forward that SIGKILL: it is reparented + to PID 1 and keeps scanning as an orphaned background process. + + PR_SET_PDEATHSIG makes the kernel deliver SIGTERM to this process the moment + the bootloader dies, so the app's SIGTERM handler (cleanup + exit) runs + instead of leaking an orphaned scan. Only meaningful when running as the + inner process of a PyInstaller onefile binary; pip/source installs keep the + user's shell as their parent and must not be killed when that shell exits. + No-op outside Linux or when not frozen. + """ + if sys.platform != "linux": + return + if not is_binary_install(): + return + try: + import ctypes # stdlib; late import keeps startup lean + + libc = ctypes.CDLL(None, use_errno=True) + pr_set_pdeathsig = 1 + parent_pid = os.getppid() + if libc.prctl(pr_set_pdeathsig, signal.SIGTERM) != 0: + logger.warning( + "prctl(PR_SET_PDEATHSIG, SIGTERM) failed: %s", + ctypes.get_errno(), + ) + return + # The bootloader can die between spawning this process and the prctl + # above; the process is then reparented before the kernel snapshots the + # parent, and PDEATHSIG would watch the wrong parent. Deliver SIGTERM + # ourselves when that happened so the scan still cleans up and exits. + if os.getppid() != parent_pid: + os.kill(os.getpid(), signal.SIGTERM) + except Exception: + logger.debug("PR_SET_PDEATHSIG unavailable", exc_info=True) + + def main() -> None: + _install_cleanup_sigterm_handler() + _enable_parent_death_signal() configure_dependency_logging() if sys.platform == "win32": diff --git a/tests/test_parent_death_signal.py b/tests/test_parent_death_signal.py new file mode 100644 index 000000000..634547242 --- /dev/null +++ b/tests/test_parent_death_signal.py @@ -0,0 +1,115 @@ +"""Tests for the Linux parent-death signal helper in interface/main.py.""" + +from __future__ import annotations + +import os +import signal +from unittest import mock + +from strix.interface.main import ( + _enable_parent_death_signal, + _install_cleanup_sigterm_handler, +) + + +def _ctypes_fixture() -> tuple[mock.Mock, mock.Mock, mock._patch]: + """Return (ctypes_mock, libc_mock, sysmods_patch) for intercepting ctypes. + + ``_enable_parent_death_signal`` binds ``ctypes`` as a function-local name via + ``import ctypes``, so patching the module attribute does not intercept it. + Replacing the module in ``sys.modules`` does. + """ + + ctypes_mock = mock.Mock() + libc_mock = ctypes_mock.CDLL.return_value + return ctypes_mock, libc_mock, mock.patch.dict("sys.modules", {"ctypes": ctypes_mock}) + + +@mock.patch("strix.interface.main.is_binary_install", return_value=False) +def test_noop_for_source_install(_mock_binary: mock.Mock) -> None: + ctypes_mock, _, sysmods = _ctypes_fixture() + with mock.patch("strix.interface.main.sys.platform", "linux"), sysmods: + _enable_parent_death_signal() + ctypes_mock.CDLL.assert_not_called() + + +@mock.patch("strix.interface.main.is_binary_install", return_value=True) +def test_noop_for_non_linux(_mock_binary: mock.Mock) -> None: + ctypes_mock, _, sysmods = _ctypes_fixture() + with mock.patch("strix.interface.main.sys.platform", "darwin"), sysmods: + _enable_parent_death_signal() + ctypes_mock.CDLL.assert_not_called() + + +@mock.patch("strix.interface.main.is_binary_install", return_value=True) +def test_registers_pdeathsig_when_frozen(_mock_binary: mock.Mock) -> None: + _, libc_mock, sysmods = _ctypes_fixture() + libc_mock.prctl.return_value = 0 + with ( + mock.patch("strix.interface.main.sys.platform", "linux"), + mock.patch("strix.interface.main.os.getppid", return_value=1234), + mock.patch("strix.interface.main.os.kill") as mock_kill, + sysmods, + ): + _enable_parent_death_signal() + libc_mock.prctl.assert_called_once_with(1, signal.SIGTERM) + mock_kill.assert_not_called() + + +@mock.patch("strix.interface.main.is_binary_install", return_value=True) +def test_self_terminates_when_parent_died_before_registration( + _mock_binary: mock.Mock, +) -> None: + _, libc_mock, sysmods = _ctypes_fixture() + libc_mock.prctl.return_value = 0 + with ( + mock.patch("strix.interface.main.sys.platform", "linux"), + mock.patch("strix.interface.main.os.getppid", side_effect=[1234, 1]), + mock.patch("strix.interface.main.os.kill") as mock_kill, + sysmods, + ): + _enable_parent_death_signal() + mock_kill.assert_called_once_with(os.getpid(), signal.SIGTERM) + + +@mock.patch("strix.interface.main.is_binary_install", return_value=True) +def test_warns_and_returns_when_prctl_fails(_mock_binary: mock.Mock) -> None: + _, libc_mock, sysmods = _ctypes_fixture() + libc_mock.prctl.return_value = -1 + with ( + mock.patch("strix.interface.main.sys.platform", "linux"), + mock.patch("strix.interface.main.os.getppid", return_value=1234), + mock.patch("strix.interface.main.os.kill") as mock_kill, + sysmods, + ): + _enable_parent_death_signal() + libc_mock.prctl.assert_called_once_with(1, signal.SIGTERM) + mock_kill.assert_not_called() + + +def test_cleanup_sigterm_handler_cleans_state_and_exits() -> None: + with mock.patch("strix.interface.main.signal.signal") as mock_signal: + _install_cleanup_sigterm_handler() + handler = mock_signal.call_args.args[1] + + state = mock.Mock() + with ( + mock.patch("strix.report.state.get_global_report_state", return_value=state), + mock.patch("sys.exit") as mock_exit, + ): + handler(signal.SIGTERM, None) + state.cleanup.assert_called_once_with(status="interrupted") + mock_exit.assert_called_once_with(1) + + +def test_cleanup_sigterm_handler_exits_without_state() -> None: + with mock.patch("strix.interface.main.signal.signal") as mock_signal: + _install_cleanup_sigterm_handler() + handler = mock_signal.call_args.args[1] + + with ( + mock.patch("strix.report.state.get_global_report_state", return_value=None), + mock.patch("sys.exit") as mock_exit, + ): + handler(signal.SIGTERM, None) + mock_exit.assert_called_once_with(1)