Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 29 additions & 24 deletions hermes_cli/gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -1141,6 +1141,27 @@ def is_windows() -> bool:
return sys.platform == 'win32'


def _windows_gateway_should_absorb_console_controls() -> bool:
"""Return True for detached Windows gateway runs that should ignore Ctrl+C.

Foreground ``hermes gateway run`` must remain interruptible from
PowerShell/CMD. Detached service-style launches opt in via
``HERMES_GATEWAY_DETACHED=1``; older wrappers without the env marker are
treated as detached when no interactive stdin is attached.
"""
if not is_windows():
return False

detached = os.getenv("HERMES_GATEWAY_DETACHED", "").strip().lower()
if detached in {"1", "true", "yes", "on"}:
return True

try:
return not bool(sys.stdin and sys.stdin.isatty())
except (ValueError, OSError):
return True


# =============================================================================
# Service Configuration
# =============================================================================
Expand Down Expand Up @@ -2978,34 +2999,17 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False):
_guard_official_docker_root_gateway()
sys.path.insert(0, str(PROJECT_ROOT))

# On Windows, when the gateway is launched as a detached background
# process (via ``hermes gateway install`` β†’ Scheduled Task / Startup
# folder / direct pythonw.exe spawn) there is no console attached. In
# that case Windows can still deliver CTRL_C_EVENT / CTRL_BREAK_EVENT
# to the process group under some circumstances (e.g. when *another*
# process in the same group sends one), which Python 3.11 translates
# into KeyboardInterrupt inside asyncio.run(). The outer handler below
# catches that and exits cleanly β€” silently killing the gateway. On
# detached boots we must absorb those spurious signals so the gateway
# stays alive; real user Ctrl+C still comes through prompt_toolkit /
# the asyncio signal handler when running in a real console.
#
# IMPORTANT lesson (May 2026): we originally gated this on "stdin is
# NOT a TTY" assuming only detached pythonw runs would be vulnerable.
# Wrong. When the user runs `hermes gateway start` from a PowerShell
# console, the gateway inherits that console and stdin IS a TTY β€”
# but it's STILL vulnerable to CTRL_C_EVENT broadcast by any sibling
# `hermes` invocation (like `hermes gateway status` 30 seconds later)
# because Windows routes console events to all processes sharing the
# console. Every hermes CLI process after that sibling fires is a
# potential drive-by killer. So on Windows, for `gateway run`
# specifically (never interactive by design), always install the
# SIGINT absorber regardless of TTY state.
# Detached Windows gateway runs must ignore console-control broadcasts
# from sibling CLI processes, but foreground `hermes gateway run` still
# needs to obey the banner's "Press Ctrl+C to stop" contract.
# Service-style launchers set HERMES_GATEWAY_DETACHED=1; older wrappers
# without the marker are handled by the non-TTY fallback.
try:
_stdin_is_tty = bool(sys.stdin and sys.stdin.isatty())
except (ValueError, OSError):
_stdin_is_tty = False
if is_windows():
_absorb_windows_console_controls = _windows_gateway_should_absorb_console_controls()
if _absorb_windows_console_controls:
try:
signal.signal(signal.SIGINT, signal.SIG_IGN)
if hasattr(signal, "SIGBREAK"):
Expand Down Expand Up @@ -3103,6 +3107,7 @@ def _exit_diag(tag: str, **extra: object) -> None:
replace=replace,
argv=sys.argv,
stdin_is_tty=_stdin_is_tty,
absorb_windows_console_controls=_absorb_windows_console_controls,
)

def _atexit_hook() -> None:
Expand Down
2 changes: 2 additions & 0 deletions hermes_cli/gateway_windows.py
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,7 @@ def _build_gateway_cmd_script(
lines.append(f"cd /d {_quote_cmd_script_arg(working_dir)}")
lines.append(f'set "HERMES_HOME={hermes_home}"')
lines.append('set "PYTHONIOENCODING=utf-8"')
lines.append('set "HERMES_GATEWAY_DETACHED=1"')
# VIRTUAL_ENV lets the gateway's own python detection find the venv
# if someone imports hermes_constants-based logic during startup.
venv_dir = str(Path(python_path).resolve().parent.parent)
Expand Down Expand Up @@ -371,6 +372,7 @@ def _build_gateway_argv() -> tuple[list[str], str, dict[str, str]]:
env_overlay = {
"HERMES_HOME": hermes_home,
"PYTHONIOENCODING": "utf-8",
"HERMES_GATEWAY_DETACHED": "1",
"VIRTUAL_ENV": str(Path(python_exe).resolve().parent.parent),
}
return argv, working_dir, env_overlay
Expand Down
60 changes: 60 additions & 0 deletions tests/hermes_cli/test_gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,66 @@ def fake_start_gateway(*, replace, verbosity):
assert calls == [(True, 2)]


def test_run_gateway_windows_foreground_keeps_ctrl_c_enabled(monkeypatch):
calls = []

def fake_start_gateway(*, replace, verbosity):
calls.append((replace, verbosity))
return object()

class _TTY:
def isatty(self):
return True

signal_calls = []

def fake_signal(sig, handler):
signal_calls.append((sig, handler))

_install_fake_gateway_run(monkeypatch, fake_start_gateway)
monkeypatch.setattr(gateway, "is_windows", lambda: True)
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
monkeypatch.setattr(gateway.sys, "stdin", _TTY())
monkeypatch.delenv("HERMES_GATEWAY_DETACHED", raising=False)
monkeypatch.setattr(gateway.signal, "signal", fake_signal)
monkeypatch.setattr(gateway.asyncio, "run", lambda coro: True)

gateway.run_gateway()

assert calls == [(False, 0)]
assert (gateway.signal.SIGINT, gateway.signal.SIG_IGN) not in signal_calls


def test_run_gateway_windows_detached_absorbs_console_controls(monkeypatch):
calls = []

def fake_start_gateway(*, replace, verbosity):
calls.append((replace, verbosity))
return object()

class _TTY:
def isatty(self):
return True

signal_calls = []

def fake_signal(sig, handler):
signal_calls.append((sig, handler))

_install_fake_gateway_run(monkeypatch, fake_start_gateway)
monkeypatch.setattr(gateway, "is_windows", lambda: True)
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
monkeypatch.setattr(gateway.sys, "stdin", _TTY())
monkeypatch.setenv("HERMES_GATEWAY_DETACHED", "1")
monkeypatch.setattr(gateway.signal, "signal", fake_signal)
monkeypatch.setattr(gateway.asyncio, "run", lambda coro: True)

gateway.run_gateway()

assert calls == [(False, 0)]
assert (gateway.signal.SIGINT, gateway.signal.SIG_IGN) in signal_calls


class TestSystemdLingerStatus:
def test_reports_enabled(self, monkeypatch):
monkeypatch.setattr(gateway, "is_linux", lambda: True)
Expand Down
Loading