Merge 3d8d93e3c9 into ae38fe70cd

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":

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)