mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-23 16:36:23 +00:00
00e9f6dd4d
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.
505 lines
21 KiB
Python
505 lines
21 KiB
Python
"""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 <bootstrap that runs hermes_cli.main>`; 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()
|