From 00e9f6dd4d5760c2a0cce4222639513de75e6d50 Mon Sep 17 00:00:00 2001 From: yoniebans Date: Fri, 17 Jul 2026 11:37:11 +0200 Subject: [PATCH] feat(serve): add Windows remote backend runtime for Desktop SSH MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add hermes_cli/windows_ssh_runtime.py: a native Windows trust boundary invoked over SSH to manage the Desktop SSH backend lifecycle — ACL-locked one-shot token file (owner+SYSTEM only, SE_DACL_PROTECTED, read-once via DELETE_ON_CLOSE), ownership lock, and detached serve spawn with CREATE_BREAKAWAY_FROM_JOB so the backend survives the SSH session's Job Object teardown. Process ownership is proven by exact pid + creation-time + argv + owner-nonce; state is tri-state (owned / stale / indeterminate) so a live-but-unreadable process is never mistaken for stale. Route _read_ssh_session_token_file to the Windows reader on win32, and move the POSIX token root from a hardcoded ~/.hermes to get_hermes_home() so both platforms honor the active profile. --- hermes_cli/main.py | 9 +- hermes_cli/windows_ssh_runtime.py | 505 ++++++++++++++++++ .../test_ssh_session_token_parser.py | 40 +- 3 files changed, 536 insertions(+), 18 deletions(-) create mode 100644 hermes_cli/windows_ssh_runtime.py diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 21b9ebf74cb0..5b54d8ee7635 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -12073,18 +12073,23 @@ def _maybe_setup_dashboard_auth_interactively(args) -> None: def _read_ssh_session_token_file(path: str) -> str: """Read and unlink a Desktop SSH token from its private runtime directory.""" + if sys.platform == "win32": + from hermes_cli.windows_ssh_runtime import read_token + return read_token(path) + import stat as _stat from pathlib import Path as _Path + from hermes_constants import get_hermes_home as _get_hermes_home if not os.path.isabs(path): raise SystemExit("--ssh-session-token-file must be absolute") token_path = _Path(path) - token_root = _Path.home() / ".hermes" / "desktop-ssh" + token_root = _get_hermes_home() / "desktop-ssh" try: relative = token_path.relative_to(token_root) except ValueError as exc: - raise SystemExit("--ssh-session-token-file must be under ~/.hermes/desktop-ssh") from exc + raise SystemExit("--ssh-session-token-file must be under the desktop-ssh directory") from exc if len(relative.parts) != 2 or not re.fullmatch(r"[0-9a-f]{32}", relative.parts[0]): raise SystemExit("--ssh-session-token-file has an invalid runtime path") if not re.fullmatch(r"[0-9a-f]{16}\.token", relative.parts[1]): diff --git a/hermes_cli/windows_ssh_runtime.py b/hermes_cli/windows_ssh_runtime.py new file mode 100644 index 000000000000..b80b81e85ba4 --- /dev/null +++ b/hermes_cli/windows_ssh_runtime.py @@ -0,0 +1,505 @@ +"""Native Windows trust boundary for Desktop SSH backend lifecycle.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import subprocess +import sys +from pathlib import Path +from typing import Any + +from hermes_constants import get_hermes_home + +_HEX32 = re.compile(r"[0-9a-f]{32}\Z") +_HEX16 = re.compile(r"[0-9a-f]{16}\Z") +_HEX64 = re.compile(r"[0-9a-f]{64}\Z") +_MAX_JSON = 1024 * 1024 +_MAX_LOG = 512 * 1024 +_OPEN_REPARSE_POINT = 0x00200000 +_DELETE_ON_CLOSE = 0x04000000 +_MOVE_REPLACE_EXISTING = 0x00000001 +_MOVE_WRITE_THROUGH = 0x00000008 + + +def _win32() -> tuple[Any, ...]: + if sys.platform != "win32": + raise RuntimeError("Windows SSH runtime is only available on Windows") + import ntsecuritycon + import pywintypes + import win32api + import win32con + import win32file + import win32process + import win32security + return ntsecuritycon, pywintypes, win32api, win32con, win32file, win32process, win32security + + +def _ownership(value: str) -> str: + if not _HEX32.fullmatch(value or ""): + raise ValueError("invalid ownership ID") + return value + + +def _nonce(value: str) -> str: + if not _HEX16.fullmatch(value or ""): + raise ValueError("invalid spawn nonce") + return value + + +def _root() -> Path: + return get_hermes_home() / "desktop-ssh" + + +def _directory(ownership_id: str) -> Path: + return _root() / _ownership(ownership_id) + + +def _token_path(ownership_id: str, spawn_nonce: str) -> Path: + return _directory(ownership_id) / f"{_nonce(spawn_nonce)}.token" + + +def _log_path(ownership_id: str, spawn_nonce: str) -> Path: + return _directory(ownership_id) / f"{_nonce(spawn_nonce)}.log" + + +def _lock_path(ownership_id: str) -> Path: + return _directory(ownership_id) / "backend.lock.json" + + +def _current_sid(): + _, _, win32api, win32con, _, _, win32security = _win32() + token = win32security.OpenProcessToken(win32api.GetCurrentProcess(), win32con.TOKEN_QUERY) + return win32security.GetTokenInformation(token, win32security.TokenUser)[0] + + +def _system_sid(): + return _win32()[6].ConvertStringSidToSid("S-1-5-18") + + +def _security_attributes(): + ntsecuritycon, _, _, _, _, _, win32security = _win32() + owner = _current_sid() + acl = win32security.ACL() + for sid in (owner, _system_sid()): + acl.AddAccessAllowedAceEx(win32security.ACL_REVISION, 0, ntsecuritycon.FILE_ALL_ACCESS, sid) + descriptor = win32security.SECURITY_DESCRIPTOR() + descriptor.SetSecurityDescriptorOwner(owner, False) + descriptor.SetSecurityDescriptorDacl(True, acl, False) + # Protect the DACL so inheritable parent ACEs (%LOCALAPPDATA% grants) are not merged in. + descriptor.SetSecurityDescriptorControl(win32security.SE_DACL_PROTECTED, win32security.SE_DACL_PROTECTED) + attributes = win32security.SECURITY_ATTRIBUTES() + attributes.SECURITY_DESCRIPTOR = descriptor + return attributes + + +def _allowed_sids(): + win32security = _win32()[6] + return {win32security.ConvertSidToStringSid(_current_sid()), + win32security.ConvertSidToStringSid(_system_sid())} + + +def _verify_security(handle) -> None: + _, _, _, _, _, _, win32security = _win32() + info = win32security.OWNER_SECURITY_INFORMATION | win32security.DACL_SECURITY_INFORMATION + descriptor = win32security.GetSecurityInfo(handle, win32security.SE_FILE_OBJECT, info) + owner = descriptor.GetSecurityDescriptorOwner() + owner_value = win32security.ConvertSidToStringSid(owner) + if owner_value not in _allowed_sids(): + raise OSError("Windows SSH runtime object has the wrong owner") + dacl = descriptor.GetSecurityDescriptorDacl() + if dacl is None: + raise OSError("Windows SSH runtime object has a null DACL") + allowed = _allowed_sids() + allow_types = { + win32security.ACCESS_ALLOWED_ACE_TYPE, + win32security.ACCESS_ALLOWED_OBJECT_ACE_TYPE, + getattr(win32security, "ACCESS_ALLOWED_CALLBACK_ACE_TYPE", 9), + getattr(win32security, "ACCESS_ALLOWED_CALLBACK_OBJECT_ACE_TYPE", 11), + } + for index in range(dacl.GetAceCount()): + ace = dacl.GetAce(index) + ace_type = ace[0][0] + mask = ace[1] + sid = ace[-1] + if ace_type in allow_types and mask: + if win32security.ConvertSidToStringSid(sid) not in allowed: + raise OSError("Windows SSH runtime object has a permissive DACL") + + +def _open(path: Path, access: int, creation: int, flags: int, share: int = 0): + _, _, _, _, win32file, _, _ = _win32() + handle = win32file.CreateFile(str(path), access, share, _security_attributes(), creation, flags, None) + try: + actual = win32file.GetFinalPathNameByHandle(handle, 0) + if actual.startswith("\\\\?\\"): + actual = actual[4:] + expected = os.path.abspath(str(path)) + if os.path.normcase(actual) != os.path.normcase(expected): + raise OSError("Windows SSH runtime handle escaped its expected path") + attributes = win32file.GetFileInformationByHandle(handle)[0] + if attributes & 0x400: + raise OSError("Windows SSH runtime path contains a reparse point") + _verify_security(handle) + return handle + except BaseException: + win32file.CloseHandle(handle) + raise + + +def _ensure_directory(path: Path) -> None: + _, pywintypes, _, win32con, win32file, _, _ = _win32() + if path.parent != path: + _ensure_directory(path.parent) if path.parent not in (Path(path.anchor), path) and not path.parent.exists() else None + if not path.exists(): + try: + win32file.CreateDirectory(str(path), _security_attributes()) + except pywintypes.error as exc: + if exc.winerror != 183: + raise + handle = _open(path, win32con.GENERIC_READ | win32con.READ_CONTROL, win32con.OPEN_EXISTING, + win32con.FILE_FLAG_BACKUP_SEMANTICS | _OPEN_REPARSE_POINT, + win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE | win32con.FILE_SHARE_DELETE) + win32file.CloseHandle(handle) + + +def _ensure_scope(ownership_id: str) -> Path: + root = _root() + _ensure_directory(root) + directory = _directory(ownership_id) + _ensure_directory(directory) + return directory + + +def upload_token(ownership_id: str, spawn_nonce: str, token: bytes) -> dict[str, Any]: + _, _, _, win32con, win32file, _, _ = _win32() + if len(token) != 64 or not _HEX64.fullmatch(token.decode("ascii", errors="ignore")): + raise ValueError("invalid session token") + _ensure_scope(ownership_id) + path = _token_path(ownership_id, spawn_nonce) + try: + handle = _open(path, win32con.GENERIC_WRITE | win32con.READ_CONTROL, win32con.CREATE_NEW, + win32con.FILE_ATTRIBUTE_NORMAL | _OPEN_REPARSE_POINT) + try: + win32file.WriteFile(handle, token) + win32file.FlushFileBuffers(handle) + finally: + win32file.CloseHandle(handle) + except BaseException: + path.unlink(missing_ok=True) + raise + return {"path": str(path)} + + +def read_token(path_value: str) -> str: + _, _, _, win32con, win32file, _, _ = _win32() + path = Path(path_value) + root = _root() + try: + relative = path.relative_to(root) + except ValueError as exc: + raise SystemExit("--ssh-session-token-file must be under the desktop-ssh directory") from exc + if len(relative.parts) != 2 or not _HEX32.fullmatch(relative.parts[0]) or not re.fullmatch(r"[0-9a-f]{16}\.token", relative.parts[1]): + raise SystemExit("--ssh-session-token-file has an invalid runtime path") + flags = win32con.FILE_ATTRIBUTE_NORMAL | _OPEN_REPARSE_POINT | _DELETE_ON_CLOSE + try: + handle = _open(path, win32con.GENERIC_READ | win32con.READ_CONTROL | win32con.DELETE, + win32con.OPEN_EXISTING, flags) + except Exception as exc: + raise SystemExit("--ssh-session-token-file is not accessible") from exc + try: + _, data = win32file.ReadFile(handle, 65) + finally: + win32file.CloseHandle(handle) + token = data.decode("ascii", errors="ignore") + if len(token) != 64 or not _HEX64.fullmatch(token): + raise SystemExit("--ssh-session-token-file contains an invalid token") + return token + + +def _read_json_stdin() -> dict[str, Any]: + raw = sys.stdin.buffer.read(_MAX_JSON + 1) + if len(raw) > _MAX_JSON: + raise ValueError("runtime payload is too large") + parsed = json.loads(raw) + if not isinstance(parsed, dict): + raise ValueError("runtime payload must be an object") + return parsed + + +def read_lock(ownership_id: str) -> dict[str, Any] | None: + _, pywintypes, _, win32con, win32file, _, _ = _win32() + _ensure_scope(ownership_id) + path = _lock_path(ownership_id) + try: + handle = _open(path, win32con.GENERIC_READ | win32con.READ_CONTROL, win32con.OPEN_EXISTING, + win32con.FILE_ATTRIBUTE_NORMAL | _OPEN_REPARSE_POINT, + win32con.FILE_SHARE_READ) + except pywintypes.error as exc: + if exc.winerror in (2, 3): + return None + raise + try: + _, data = win32file.ReadFile(handle, _MAX_JSON + 1) + finally: + win32file.CloseHandle(handle) + if len(data) > _MAX_JSON: + return None + try: + parsed = json.loads(data) + except (UnicodeDecodeError, json.JSONDecodeError): + return None + return parsed if isinstance(parsed, dict) else None + + +def write_lock(ownership_id: str, payload: dict[str, Any]) -> None: + _, _, _, win32con, win32file, _, _ = _win32() + directory = _ensure_scope(ownership_id) + data = json.dumps(payload, separators=(",", ":")).encode() + if len(data) > _MAX_JSON: + raise ValueError("lock payload is too large") + temporary = directory / f".{os.urandom(8).hex()}.lock.tmp" + handle = _open(temporary, win32con.GENERIC_WRITE | win32con.READ_CONTROL, win32con.CREATE_NEW, + win32con.FILE_ATTRIBUTE_NORMAL | _OPEN_REPARSE_POINT) + try: + win32file.WriteFile(handle, data) + win32file.FlushFileBuffers(handle) + finally: + win32file.CloseHandle(handle) + win32file.MoveFileEx(str(temporary), str(_lock_path(ownership_id)), + _MOVE_REPLACE_EXISTING | _MOVE_WRITE_THROUGH) + + +def remove_artifact(path: Path) -> bool: + _, pywintypes, _, win32con, win32file, _, _ = _win32() + try: + handle = _open(path, win32con.DELETE | win32con.READ_CONTROL, win32con.OPEN_EXISTING, + win32con.FILE_ATTRIBUTE_NORMAL | _OPEN_REPARSE_POINT | _DELETE_ON_CLOSE) + except pywintypes.error as exc: + if exc.winerror in (2, 3): + return False + raise + win32file.CloseHandle(handle) + return True + + +def process_state(pid: int, creation_time_ns: int, hermes_path: str, spawn_nonce: str) -> dict[str, Any]: + import psutil + _nonce(spawn_nonce) + try: + process = psutil.Process(pid) + except psutil.NoSuchProcess as exc: + return {"alive": False, "owned": False, "indeterminate": False, "reason": type(exc).__name__} + try: + actual_creation = int(process.create_time() * 1_000_000_000) + argv = process.cmdline() + except psutil.NoSuchProcess as exc: + return {"alive": False, "owned": False, "indeterminate": False, "reason": type(exc).__name__} + except psutil.AccessDenied as exc: + return {"alive": True, "owned": False, "indeterminate": True, "reason": type(exc).__name__} + if actual_creation != creation_time_ns: + return {"alive": False, "owned": False, "indeterminate": False, "reason": "creation-time", + "actualCreationTimeNs": str(actual_creation), "expectedCreationTimeNs": str(creation_time_ns)} + if not argv: + return {"alive": True, "owned": False, "indeterminate": True, "reason": "argv-unavailable"} + expected = os.path.normcase(os.path.abspath(hermes_path)) + arg0 = os.path.normcase(os.path.abspath(argv[0])) + # argv[0] is either the hermes exe directly, or (normal case) the base Python + # interpreter -- its exact path varies by venv/uv layout, so match on "a python + # running our module". We launch via `-c` bootstrap, so it shows as + # `-c `; also accept a plain `-m` launch. + # Identity is anchored by the unforgeable creation-time + secret owner-nonce below. + is_python = os.path.basename(arg0).startswith("python") + launches_module = ( + argv[1:3] == ["-m", "hermes_cli.main"] + or (len(argv) > 2 and argv[1] == "-c" and "hermes_cli.main" in argv[2]) + ) + executable_match = arg0 == expected or (is_python and launches_module) + try: + serve = argv.index("serve") + owner = argv.index("--ssh-owner-nonce", serve + 1) + owned = executable_match and "--isolated" in argv[serve + 1:] and argv[owner + 1] == spawn_nonce + except (ValueError, IndexError): + owned = False + return {"alive": process.is_running(), "owned": owned, "indeterminate": False, + "creationTimeNs": str(actual_creation), "reason": "owned" if owned else "argv", + "argv": argv[:20], "expectedExecutable": expected} + + +def terminate_owned(pid: int, creation_time_ns: int, hermes_path: str, spawn_nonce: str) -> bool: + state = process_state(pid, creation_time_ns, hermes_path, spawn_nonce) + if not state["alive"] or not state["owned"]: + return False + import psutil + process = psutil.Process(pid) + if int(process.create_time() * 1_000_000_000) != creation_time_ns: + return False + process.terminate() + try: + process.wait(5) + except psutil.TimeoutExpired: + process.kill() + process.wait(5) + return True + + +def _resolve_direct_interpreter(python_entry: str) -> tuple[str, list[str]]: + """Resolve the venv launcher to the base interpreter it would exec, plus the + sys.path to reproduce. On Windows a venv Scripts\\python.exe is a launcher stub + that spawns the real interpreter as a CHILD (two PIDs); spawning the base + interpreter directly with the launcher's sys.path injected yields ONE process + that both owns the port and is the process we lock. + + Also resolves hermes_cli's own location and prepends its parent, because the + launcher finds the module via cwd / an editable-install path hook that a bare + PYTHONPATH does not reproduce.""" + query = ( + "import sys,json,os,importlib.util as u;" + "s=u.find_spec('hermes_cli');" + "root=os.path.dirname(os.path.dirname(s.origin)) if s and s.origin else '';" + "print(json.dumps({'base':getattr(sys,'_base_executable','') or sys.executable," + "'path':[p for p in sys.path if p],'root':root}))" + ) + out = subprocess.run([python_entry, "-c", query], capture_output=True, text=True, timeout=30) + if out.returncode != 0: + raise ValueError("could not resolve the base Python interpreter") + info = json.loads(out.stdout.strip().splitlines()[-1]) + base = info["base"] + if not base or not os.path.isfile(base): + raise ValueError("base Python interpreter was not found") + # drop editable-install finder markers ('__editable__.*.finder.__path_hook__') + # that PYTHONPATH cannot reproduce; keep only real filesystem entries. + py_path = [p for p in info.get("path", []) if os.path.exists(p)] + # the hermes_cli package parent must be explicit -- the launcher resolves it via + # cwd or an editable path hook, neither of which survives a bare PYTHONPATH spawn. + root = info.get("root") or "" + if root and os.path.isdir(root) and root not in py_path: + py_path.insert(0, root) + if not root or not os.path.isdir(root): + raise ValueError("could not locate the hermes_cli package") + return base, py_path + + +def spawn_backend(payload: dict[str, Any]) -> dict[str, Any]: + ownership_id = _ownership(str(payload["ownershipId"])) + spawn_nonce = _nonce(str(payload["spawnNonce"])) + configured_path = str(payload["hermesPath"]) + if not os.path.isabs(configured_path): + raise ValueError("Hermes path must be absolute") + hermes_path = os.path.abspath(configured_path) + token_path = str(_token_path(ownership_id, spawn_nonce)) + log_path = _log_path(ownership_id, spawn_nonce) + profile = str(payload.get("profile") or "") + if len(profile) > 256 or any(ch in profile for ch in "\x00\r\n"): + raise ValueError("invalid profile") + venv_dir = os.path.dirname(hermes_path) + python_entry = os.path.join(venv_dir, "python.exe") + if not os.path.isfile(python_entry): + raise ValueError("Hermes Python runtime was not found") + base_python, sys_path = _resolve_direct_interpreter(python_entry) + # Seed sys.path IN-PROCESS via a -c bootstrap rather than exporting PYTHONPATH: + # PYTHONPATH would be inherited by every subprocess the running backend spawns + # (terminal tool, user scripts), shadowing their imports. This keeps the path + # scoped to the backend process alone. + bootstrap = ( + "import sys,runpy;" + f"sys.path[:0]={sys_path!r};" + "runpy.run_module('hermes_cli.main',run_name='__main__',alter_sys=True)" + ) + args = [base_python, "-c", bootstrap] + if profile: + args.extend(["--profile", profile]) + args.extend(["serve", "--isolated", "--host", "127.0.0.1", "--port", "0", + "--ssh-session-token-file", token_path, "--ssh-owner-nonce", spawn_nonce]) + # VIRTUAL_ENV preserves venv identity; PYTHONPATH is deliberately NOT set (see above). + env = dict(os.environ) + env["VIRTUAL_ENV"] = os.path.dirname(venv_dir) + env.pop("PYTHONPATH", None) + _ensure_scope(ownership_id) + creationflags = 0x00000008 | 0x00000200 | 0x01000000 + _, _, _, win32con, win32file, _, _ = _win32() + log_handle = _open(log_path, win32con.GENERIC_WRITE | win32con.READ_CONTROL, + win32con.CREATE_NEW, win32con.FILE_ATTRIBUTE_NORMAL | _OPEN_REPARSE_POINT, + win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE) + import msvcrt + log_fd = msvcrt.open_osfhandle(int(log_handle), os.O_WRONLY) + with os.fdopen(log_fd, "wb", buffering=0) as log_stream: + process = subprocess.Popen(args, stdin=subprocess.DEVNULL, stdout=log_stream, stderr=log_stream, + close_fds=True, creationflags=creationflags, env=env) + creation_time_ns = int(__import__("psutil").Process(process.pid).create_time() * 1_000_000_000) + return {"pid": process.pid, "creationTimeNs": str(creation_time_ns), + "logPath": str(log_path), "tokenPath": token_path} + + +def inspect_hermes(hermes_path: str) -> dict[str, Any]: + path = os.path.abspath(hermes_path) + if not os.path.isabs(hermes_path) or not os.path.isfile(path): + raise ValueError("Hermes path is not an executable file") + version = subprocess.run([path, "--version"], capture_output=True, text=True, timeout=20) + help_result = subprocess.run([path, "serve", "--help"], capture_output=True, text=True, timeout=20) + help_text = help_result.stdout + help_result.stderr + return { + "path": path, + "version": (version.stdout + version.stderr).splitlines()[0] if version.returncode == 0 else "", + "supported": "--ssh-session-token-file" in help_text and "--ssh-owner-nonce" in help_text, + } + + +def dispatch(argv: list[str]) -> Any: + if not argv: + raise ValueError("missing operation") + operation = argv[0] + if operation == "probe": + import platform + return {"os": "Windows", "arch": platform.machine(), "hermesHome": str(get_hermes_home()), "python": sys.executable} + if operation == "upload-token" and len(argv) == 3: + return upload_token(argv[1], argv[2], sys.stdin.buffer.read(65)) + if operation == "read-lock" and len(argv) == 2: + return read_lock(argv[1]) + if operation == "write-lock" and len(argv) == 2: + write_lock(argv[1], _read_json_stdin()); return {"ok": True} + if operation == "remove-lock" and len(argv) == 2: + return {"removed": remove_artifact(_lock_path(argv[1]))} + if operation == "remove-token" and len(argv) == 3: + return {"removed": remove_artifact(_token_path(argv[1], argv[2]))} + if operation == "read-log" and len(argv) == 3: + path = _log_path(argv[1], argv[2]) + _, pywintypes, _, win32con, win32file, _, _ = _win32() + try: + handle = _open(path, win32con.GENERIC_READ | win32con.READ_CONTROL, + win32con.OPEN_EXISTING, win32con.FILE_ATTRIBUTE_NORMAL | _OPEN_REPARSE_POINT, + win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE) + except pywintypes.error as exc: + if exc.winerror in (2, 3): + return {"content": ""} + raise + try: + _, data = win32file.ReadFile(handle, _MAX_LOG) + finally: + win32file.CloseHandle(handle) + return {"content": data.decode(errors="replace")} + if operation == "remove-log" and len(argv) == 3: + return {"removed": remove_artifact(_log_path(argv[1], argv[2]))} + if operation == "spawn": + return spawn_backend(_read_json_stdin()) + if operation == "inspect" and len(argv) == 2: + return inspect_hermes(argv[1]) + if operation == "process-state" and len(argv) == 5: + return process_state(int(argv[1]), int(argv[2]), argv[3], argv[4]) + if operation == "terminate" and len(argv) == 5: + return {"terminated": terminate_owned(int(argv[1]), int(argv[2]), argv[3], argv[4])} + raise ValueError("invalid operation") + + +def main() -> None: + try: + print(json.dumps(dispatch(sys.argv[1:]), separators=(",", ":"))) + except Exception as exc: + print(json.dumps({"error": str(exc)}, separators=(",", ":")), file=sys.stderr) + raise SystemExit(1) + + +if __name__ == "__main__": + main() diff --git a/tests/hermes_cli/test_ssh_session_token_parser.py b/tests/hermes_cli/test_ssh_session_token_parser.py index 4ac02b38d1ea..197d8a713a66 100644 --- a/tests/hermes_cli/test_ssh_session_token_parser.py +++ b/tests/hermes_cli/test_ssh_session_token_parser.py @@ -2,6 +2,7 @@ import argparse import os import pytest +from hermes_constants import set_hermes_home_override, reset_hermes_home_override from hermes_cli.main import _read_ssh_session_token_file, cmd_dashboard from hermes_cli.subcommands.dashboard import build_dashboard_parser @@ -43,15 +44,18 @@ def test_one_shot_token_file_rejects_non_starting_operations(operation): def test_token_file_is_read_and_unlinked_through_private_directory(tmp_path, monkeypatch): home = tmp_path / "home" - token_dir = home / ".hermes" / "desktop-ssh" / ("a" * 32) + hermes_home = home / ".hermes" + token_dir = hermes_home / "desktop-ssh" / ("a" * 32) token_dir.mkdir(parents=True, mode=0o700) token_path = token_dir / "0123456789abcdef.token" token_path.write_text("b" * 64) token_path.chmod(0o600) - monkeypatch.setattr("pathlib.Path.home", lambda: home) - - assert _read_ssh_session_token_file(str(token_path)) == "b" * 64 - assert not token_path.exists() + override = set_hermes_home_override(hermes_home) + try: + assert _read_ssh_session_token_file(str(token_path)) == "b" * 64 + assert not token_path.exists() + finally: + reset_hermes_home_override(override) @pytest.mark.skipif(os.name == "nt", reason="POSIX symlink contract") @@ -64,12 +68,14 @@ def test_token_file_rejects_symlink(tmp_path, monkeypatch): target.chmod(0o600) token_path = token_dir / "0123456789abcdef.token" token_path.symlink_to(target) - monkeypatch.setattr("pathlib.Path.home", lambda: home) - - with pytest.raises(SystemExit, match="symlink|not accessible"): - _read_ssh_session_token_file(str(token_path)) - assert not token_path.exists() - assert target.read_text() == "b" * 64 + override = set_hermes_home_override(home / ".hermes") + try: + with pytest.raises(SystemExit, match="symlink|not accessible"): + _read_ssh_session_token_file(str(token_path)) + assert not token_path.exists() + assert target.read_text() == "b" * 64 + finally: + reset_hermes_home_override(override) def test_token_file_rejects_parent_escape(tmp_path, monkeypatch): @@ -79,8 +85,10 @@ def test_token_file_rejects_parent_escape(tmp_path, monkeypatch): escaped = token_root.parent / "0123456789abcdef.token" escaped.write_text("b" * 64) escaped.chmod(0o600) - monkeypatch.setattr("pathlib.Path.home", lambda: home) - - with pytest.raises(SystemExit, match="invalid runtime path"): - _read_ssh_session_token_file(str(token_root / ".." / escaped.name)) - assert escaped.exists() + override = set_hermes_home_override(home / ".hermes") + try: + with pytest.raises(SystemExit, match="invalid runtime path"): + _read_ssh_session_token_file(str(token_root / ".." / escaped.name)) + assert escaped.exists() + finally: + reset_hermes_home_override(override)