mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
unsupportedpastels
98527bf2c5
connectWindowsRemote ignored devServer: the spawn payload never asked windows_ssh_runtime to set HERMES_DESKTOP_DEV_SERVER, the ownership record did not name a renderer, and reuse did not compare one. A dev build's renderer sockets through the SSH forward were then refused unless the Windows host had been configured by hand. Send devServer in the spawn payload and have the runtime helper set (or clear) HERMES_DESKTOP_DEV_SERVER from it, record it in both ownership writes, and decide reuse with the POSIX lockServesConnect predicate. Records without the field count as no dev server, so packaged builds reuse as before.
509 lines
22 KiB
Python
509 lines
22 KiB
Python
"""Native Windows trust boundary for Desktop SSH backend lifecycle."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import importlib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
from hermes_constants import get_default_hermes_root
|
|
|
|
_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() -> Any:
|
|
"""Namespace of the pywin32 modules (ntsecuritycon, pywintypes, win32api, win32con, win32file,
|
|
win32security); import is deferred so the module imports on non-Windows hosts."""
|
|
if sys.platform != "win32":
|
|
raise RuntimeError("Windows SSH runtime is only available on Windows")
|
|
names = ("ntsecuritycon", "pywintypes", "win32api", "win32con", "win32file", "win32security")
|
|
return SimpleNamespace(**{name: importlib.import_module(name) for name in names})
|
|
|
|
|
|
def _check(pattern: re.Pattern, value: str, message: str) -> str:
|
|
if not pattern.fullmatch(value or ""):
|
|
raise ValueError(message)
|
|
return value
|
|
|
|
|
|
def _ownership(value: str) -> str:
|
|
return _check(_HEX32, value, "invalid ownership ID")
|
|
|
|
|
|
def _nonce(value: str) -> str:
|
|
return _check(_HEX16, value, "invalid spawn nonce")
|
|
|
|
|
|
def _root() -> Path:
|
|
# The helper uploads the token before the child applies `--profile`; read_token() runs after
|
|
# profile activation. Anchor both to the machine root so a named profile (or custom
|
|
# HERMES_HOME) cannot move the reader away from the helper's token.
|
|
return get_default_hermes_root() / "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():
|
|
w = _win32()
|
|
token = w.win32security.OpenProcessToken(w.win32api.GetCurrentProcess(), w.win32con.TOKEN_QUERY)
|
|
return w.win32security.GetTokenInformation(token, w.win32security.TokenUser)[0]
|
|
|
|
|
|
def _system_sid():
|
|
return _win32().win32security.ConvertStringSidToSid("S-1-5-18")
|
|
|
|
|
|
def _sid_str(sid) -> str:
|
|
return _win32().win32security.ConvertSidToStringSid(sid)
|
|
|
|
|
|
def _security_attributes():
|
|
win32security = _win32().win32security
|
|
owner = _sid_str(_current_sid())
|
|
# D:P blocks inherited grants at creation. SDDL also avoids pywin32 311's
|
|
# SetSecurityDescriptorControl argument-width bug on ARM64.
|
|
descriptor = win32security.ConvertStringSecurityDescriptorToSecurityDescriptor(
|
|
f"O:{owner}D:P(A;;FA;;;{owner})(A;;FA;;;SY)", win32security.SDDL_REVISION_1)
|
|
attributes = win32security.SECURITY_ATTRIBUTES()
|
|
attributes.SECURITY_DESCRIPTOR = descriptor
|
|
return attributes
|
|
|
|
|
|
def _allowed_sids():
|
|
return {_sid_str(_current_sid()), _sid_str(_system_sid())}
|
|
|
|
|
|
def _verify_security(handle) -> None:
|
|
win32security = _win32().win32security
|
|
info = win32security.OWNER_SECURITY_INFORMATION | win32security.DACL_SECURITY_INFORMATION
|
|
descriptor = win32security.GetSecurityInfo(handle, win32security.SE_FILE_OBJECT, info)
|
|
allowed = _allowed_sids()
|
|
if _sid_str(descriptor.GetSecurityDescriptorOwner()) not in allowed:
|
|
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")
|
|
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)
|
|
if ace[0][0] in allow_types and ace[1] and _sid_str(ace[-1]) 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().win32file
|
|
handle = win32file.CreateFile(str(path), access, share, _security_attributes(), creation, flags, None)
|
|
try:
|
|
actual = win32file.GetFinalPathNameByHandle(handle, 0).removeprefix("\\\\?\\")
|
|
if os.path.normcase(actual) != os.path.normcase(os.path.abspath(str(path))):
|
|
raise OSError("Windows SSH runtime handle escaped its expected path")
|
|
if win32file.GetFileInformationByHandle(handle)[0] & 0x400: # FILE_ATTRIBUTE_REPARSE_POINT
|
|
raise OSError("Windows SSH runtime path contains a reparse point")
|
|
_verify_security(handle)
|
|
return handle
|
|
except BaseException:
|
|
win32file.CloseHandle(handle)
|
|
raise
|
|
|
|
|
|
def _open_existing(path: Path, access: int, extra_flags: int = 0, share: int = 0):
|
|
"""``_open`` an existing file (FILE_ATTRIBUTE_NORMAL | reparse guard); None when it is missing."""
|
|
w = _win32()
|
|
try:
|
|
return _open(path, access, w.win32con.OPEN_EXISTING,
|
|
w.win32con.FILE_ATTRIBUTE_NORMAL | _OPEN_REPARSE_POINT | extra_flags, share)
|
|
except w.pywintypes.error as exc:
|
|
if exc.winerror in (2, 3):
|
|
return None
|
|
raise
|
|
|
|
|
|
def _read_shared(path: Path, limit: int, share: int) -> bytes | None:
|
|
"""Read up to ``limit`` bytes of ``path`` opened read-only with ``share``; None when missing."""
|
|
w = _win32()
|
|
handle = _open_existing(path, w.win32con.GENERIC_READ | w.win32con.READ_CONTROL, share=share)
|
|
if handle is None:
|
|
return None
|
|
try:
|
|
return w.win32file.ReadFile(handle, limit)[1]
|
|
finally:
|
|
w.win32file.CloseHandle(handle)
|
|
|
|
|
|
def _write_new(path: Path, data: bytes, share: int = 0) -> None:
|
|
"""Create ``path`` (CREATE_NEW), write ``data`` and flush."""
|
|
w = _win32()
|
|
win32con, win32file = w.win32con, w.win32file
|
|
handle = _open(path, win32con.GENERIC_WRITE | win32con.READ_CONTROL, win32con.CREATE_NEW,
|
|
win32con.FILE_ATTRIBUTE_NORMAL | _OPEN_REPARSE_POINT, share)
|
|
try:
|
|
win32file.WriteFile(handle, data)
|
|
win32file.FlushFileBuffers(handle)
|
|
finally:
|
|
win32file.CloseHandle(handle)
|
|
|
|
|
|
def write_private_file(path: Path, data: bytes) -> None:
|
|
"""Create/replace ``path`` with an owner+SYSTEM-only protected DACL (Windows only).
|
|
|
|
The repo's writer for any credential landing on a Windows disk: ``os.open(..., 0o600)`` sets
|
|
no ACLs there at all, so a token written that way inherits whatever the parent directory
|
|
grants. CreateFile ignores the security descriptor when the file already exists, so an
|
|
existing file is removed first rather than re-opened — which also avoids ``os.replace``
|
|
failing against a reader that still holds the old file open.
|
|
"""
|
|
with contextlib.suppress(FileNotFoundError):
|
|
os.unlink(str(path))
|
|
_write_new(path, data)
|
|
|
|
|
|
def _ensure_directory(path: Path) -> None:
|
|
w = _win32()
|
|
pywintypes, win32con, win32file = w.pywintypes, w.win32con, w.win32file
|
|
if path.parent not in (Path(path.anchor), path) and not path.parent.exists():
|
|
_ensure_directory(path.parent)
|
|
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:
|
|
_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]:
|
|
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:
|
|
_write_new(path, token)
|
|
except BaseException:
|
|
path.unlink(missing_ok=True)
|
|
raise
|
|
return {"path": str(path)}
|
|
|
|
|
|
def read_token(path_value: str) -> str:
|
|
w = _win32()
|
|
win32con, win32file = w.win32con, w.win32file
|
|
path = Path(path_value)
|
|
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:
|
|
win32con = _win32().win32con
|
|
_ensure_scope(ownership_id)
|
|
data = _read_shared(_lock_path(ownership_id), _MAX_JSON + 1, win32con.FILE_SHARE_READ)
|
|
if data is None or 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 _lock_record(ownership_id: str, payload: dict[str, Any]) -> bytes:
|
|
# The backend's owner watchdog only trusts a lock whose logPath ends in /<ownershipId>/<nonce>.log
|
|
# (dashboard_procs._valid_lockfile_payload); without it a superseded Windows backend never retires.
|
|
record = {**payload, "logPath": _log_path(ownership_id, str(payload.get("spawnNonce"))).as_posix()}
|
|
data = json.dumps(record, separators=(",", ":")).encode()
|
|
if len(data) > _MAX_JSON:
|
|
raise ValueError("lock payload is too large")
|
|
return data
|
|
|
|
|
|
def write_lock(ownership_id: str, payload: dict[str, Any]) -> None:
|
|
win32file = _win32().win32file
|
|
directory = _ensure_scope(ownership_id)
|
|
data = _lock_record(ownership_id, payload)
|
|
temporary = directory / f".{os.urandom(8).hex()}.lock.tmp"
|
|
_write_new(temporary, data)
|
|
win32file.MoveFileEx(str(temporary), str(_lock_path(ownership_id)),
|
|
_MOVE_REPLACE_EXISTING | _MOVE_WRITE_THROUGH)
|
|
|
|
|
|
def remove_artifact(path: Path) -> bool:
|
|
w = _win32()
|
|
handle = _open_existing(path, w.win32con.DELETE | w.win32con.READ_CONTROL, _DELETE_ON_CLOSE)
|
|
if handle is None:
|
|
return False
|
|
w.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)
|
|
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 the hermes exe or (normally) the base Python, whose path varies by venv/uv
|
|
# layout — so match "a python running our module" (`-c` bootstrap or plain `-m`). Identity
|
|
# is anchored by the unforgeable creation-time + secret owner-nonce below.
|
|
is_python = os.path.basename(arg0).startswith("python")
|
|
from gateway.status_inline_source import inline_bootstrap_argv
|
|
logical_argv = inline_bootstrap_argv(argv) or argv
|
|
launches_module = logical_argv[1:3] == ["-m", "hermes_cli.main"]
|
|
executable_match = arg0 == expected or (is_python and launches_module)
|
|
try:
|
|
serve = logical_argv.index("serve")
|
|
owner = logical_argv.index("--ssh-owner-nonce", serve + 1)
|
|
owned = executable_match and "--isolated" in logical_argv[serve + 1:] and logical_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_command(hermes_path: str) -> list[str]:
|
|
"""Ask the configured installation for its direct, boot-selecting command.
|
|
|
|
The port owner must be the process we lock, not a console-launcher child.
|
|
No assumption about python.exe beside an external bin launcher is valid.
|
|
"""
|
|
out = subprocess.run([hermes_path, "--print-runtime-command"], capture_output=True,
|
|
text=True, encoding="utf-8", errors="replace", timeout=30, check=False)
|
|
if out.returncode != 0:
|
|
raise ValueError("could not resolve Hermes runtime; refresh this installation's launcher")
|
|
try:
|
|
command = json.loads(out.stdout)
|
|
except ValueError as exc:
|
|
raise ValueError("Hermes launcher did not report a runtime command") from exc
|
|
if (not isinstance(command, list) or not command
|
|
or not all(isinstance(part, str) and "\x00" not in part for part in command)
|
|
or not os.path.isabs(command[0]) or not os.path.isfile(command[0])):
|
|
raise ValueError("Hermes launcher reported an invalid runtime command")
|
|
return command
|
|
|
|
|
|
def _spawn_env(payload: dict[str, Any]) -> dict[str, str]:
|
|
"""The serve's environment. A Desktop dev build's renderer is an http origin the backend trusts
|
|
only from the HERMES_DESKTOP_DEV_SERVER it inherits, so the payload's ``devServer`` is handed
|
|
down exactly as the POSIX spawn command does; without one the variable is not set at all."""
|
|
dev_server = str(payload.get("devServer") or "")
|
|
if len(dev_server) > 2048 or any(ch in dev_server for ch in "\x00\r\n"):
|
|
raise ValueError("invalid devServer")
|
|
env = dict(os.environ)
|
|
env.pop("VIRTUAL_ENV", None)
|
|
env.pop("PYTHONPATH", None)
|
|
env.pop("HERMES_DESKTOP_DEV_SERVER", None)
|
|
if dev_server:
|
|
env["HERMES_DESKTOP_DEV_SERVER"] = dev_server
|
|
return env
|
|
|
|
|
|
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))
|
|
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")
|
|
args = _resolve_direct_command(hermes_path)
|
|
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])
|
|
env = _spawn_env(payload)
|
|
_ensure_scope(ownership_id)
|
|
log_path = _log_path(ownership_id, spawn_nonce)
|
|
win32con = _win32().win32con
|
|
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:
|
|
# DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP | CREATE_BREAKAWAY_FROM_JOB
|
|
process = subprocess.Popen(args, stdin=subprocess.DEVNULL, stdout=log_stream, stderr=log_stream,
|
|
close_fds=True, creationflags=0x00000008 | 0x00000200 | 0x01000000, 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, encoding="utf-8", errors="replace", timeout=20, check=False)
|
|
help_result = subprocess.run([path, "serve", "--help"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=20, check=False)
|
|
help_text = help_result.stdout + help_result.stderr
|
|
return {
|
|
"path": path,
|
|
"version": (version.stdout + version.stderr).splitlines()[0] if version.returncode == 0 else "",
|
|
"runtimeBootstrapOwnership": 1,
|
|
"supported": "--ssh-session-token-file" in help_text and "--ssh-owner-nonce" in help_text}
|
|
|
|
|
|
def _probe(*_: str) -> dict[str, Any]:
|
|
import platform
|
|
return {"os": "Windows", "arch": platform.machine(), "hermesHome": str(get_default_hermes_root()), "python": sys.executable}
|
|
|
|
|
|
def _inventory(*_: str) -> dict[str, Any]:
|
|
"""Desktop's read-only roster: the canonical root and profile predicate; never mints an identity."""
|
|
from hermes_cli.install_identity import _INSTALL_ID_FILENAME, _read_existing
|
|
from hermes_cli.profiles import _iter_named_profile_dirs
|
|
install_id, _ = _read_existing(get_default_hermes_root() / _INSTALL_ID_FILENAME)
|
|
return {"profiles": ["default", *[p.name for p in _iter_named_profile_dirs()]], "installId": install_id}
|
|
|
|
|
|
def _read_log(ownership_id: str, spawn_nonce: str) -> dict[str, Any]:
|
|
win32con = _win32().win32con
|
|
data = _read_shared(_log_path(ownership_id, spawn_nonce), _MAX_LOG,
|
|
win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE)
|
|
return {"content": "" if data is None else data.decode(errors="replace")}
|
|
|
|
|
|
def _write_lock_op(ownership_id: str) -> dict[str, Any]:
|
|
write_lock(ownership_id, _read_json_stdin())
|
|
return {"ok": True}
|
|
|
|
|
|
# operation -> (argument count or None for "any", handler(*args)).
|
|
_OPERATIONS: dict[str, tuple[int | None, Any]] = {
|
|
"probe": (None, _probe),
|
|
"inventory": (None, _inventory),
|
|
"upload-token": (2, lambda o, n: upload_token(o, n, sys.stdin.buffer.read(65))),
|
|
"read-lock": (1, read_lock),
|
|
"write-lock": (1, _write_lock_op),
|
|
"remove-lock": (1, lambda o: {"removed": remove_artifact(_lock_path(o))}),
|
|
"remove-token": (2, lambda o, n: {"removed": remove_artifact(_token_path(o, n))}),
|
|
"read-log": (2, _read_log),
|
|
"remove-log": (2, lambda o, n: {"removed": remove_artifact(_log_path(o, n))}),
|
|
"spawn": (None, lambda *_: spawn_backend(_read_json_stdin())),
|
|
"inspect": (1, inspect_hermes),
|
|
"process-state": (4, lambda p, c, h, n: process_state(int(p), int(c), h, n)),
|
|
"terminate": (4, lambda p, c, h, n: {"terminated": terminate_owned(int(p), int(c), h, n)})}
|
|
|
|
|
|
def dispatch(argv: list[str]) -> Any:
|
|
if not argv:
|
|
raise ValueError("missing operation")
|
|
operation, args = argv[0], argv[1:]
|
|
entry = _OPERATIONS.get(operation)
|
|
if entry is None or (entry[0] is not None and len(args) != entry[0]):
|
|
raise ValueError("invalid operation")
|
|
return entry[1](*args)
|
|
|
|
|
|
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()
|