From 30fbc1fb805ca35ae2f5156bb3a768f5ab60862b Mon Sep 17 00:00:00 2001 From: Zijian Zhang <35801754+futrime@users.noreply.github.com> Date: Mon, 12 Oct 2026 23:05:56 +0800 Subject: [PATCH] feat(web): reach daemons on other machines over ssh, installing hmz there Each daemon under `daemons` in settings.yaml is reached by running `hmz web --attached` over ssh, with a loopback port forwarded to it, and proxied at /d//api/... (SSE and uploads streamed) with its own key; /d/local/api/... is this machine. A missing or other hmz there is installed at this version with uv tool install, never while its daemon may hold running epics. Adds /api/daemons/{name}/state and /connect. Co-Authored-By: Claude Opus 5.5 --- src/hmz/web/__init__.py | 4 + src/hmz/web/proxy.py | 133 ++++++ src/hmz/web/remote.py | 677 +++++++++++++++++++++++++++ src/hmz/web/site.py | 39 +- tests/integration/doubles_ssh.py | 216 +++++++++ tests/integration/test_web_remote.py | 270 +++++++++++ tests/unit/web/test_remote.py | 27 ++ 7 files changed, 1365 insertions(+), 1 deletion(-) create mode 100644 src/hmz/web/proxy.py create mode 100644 src/hmz/web/remote.py create mode 100644 tests/integration/doubles_ssh.py create mode 100644 tests/integration/test_web_remote.py create mode 100644 tests/unit/web/test_remote.py diff --git a/src/hmz/web/__init__.py b/src/hmz/web/__init__.py index 5c6d7197..efdcb16e 100644 --- a/src/hmz/web/__init__.py +++ b/src/hmz/web/__init__.py @@ -60,6 +60,7 @@ def listening(*, port: int = 0, apart: bool = True) -> Generator[Site]: from hmz.coganchor import releases from .places import Places + from .remote import Remotes from .runs import Runs from .site import Site from .verses import fetches @@ -73,6 +74,9 @@ def listening(*, port: int = 0, apart: bool = True) -> Generator[Site]: # And which release of humanize is the newest, where what is kept has got old. releases.refresh() held_open.callback(site.server_close) + # Every daemon on another machine let go of as the block ends, and what reached it. + site.remotes = Remotes() + held_open.callback(site.remotes.close) # Set first as the block ends, which is what ends every stream being sent. held_open.callback(site.going.set) yield site diff --git a/src/hmz/web/proxy.py b/src/hmz/web/proxy.py new file mode 100644 index 00000000..1bb712f6 --- /dev/null +++ b/src/hmz/web/proxy.py @@ -0,0 +1,133 @@ +"""A request of this interface's page, asked of another machine's interface instead. + +What a page asks of a daemon on another machine, under `/d//api/...`, is asked of that +machine's own interface at `/api/...`, through the port forwarded to it, and its answer handed +back as it comes -- a stream of events as each event comes, a file sent or downloaded a piece +at a time. The request is let in there as that interface's own page's: with its key, and saying +it comes from its own origin. Nothing of this interface's own key or cookies goes with it, and +nothing of what that one sets comes back. +""" + +from __future__ import annotations + +import contextlib +import http.client +from typing import TYPE_CHECKING + +from .routing import Refusal + +if TYPE_CHECKING: + from http.server import BaseHTTPRequestHandler + +__all__ = ["forwards"] + +#: What a request carries on: what it sends and what it asks for, never who it is. +_ASKED = ( + "Accept", + "Content-Type", + "Content-Length", + "If-None-Match", + "Last-Event-ID", + "X-Hmz-Filename", +) + +#: What an answer does not carry back: what is the connection's own, and the other's cookies. +_KEPT = frozenset( + { + "connection", + "keep-alive", + "transfer-encoding", + "set-cookie", + "server", + "date", + } +) + +#: How much is carried at a time. +_CHUNK = 1 << 16 + +#: How long, in seconds, the other side may say nothing before it is let go. A stream of +#: events says something more often than this. +_PATIENCE = 300.0 + + +def forwards( + handler: BaseHTTPRequestHandler, + port: int, + said: dict[str, str], + method: str, + path: str, +) -> None: + """Asks the interface on a port what a request asked, and answers the request with it. + + Args: + handler: The request, not yet answered. + port: Where the other interface is reached, on this machine's loopback. + said: What a request to it carries to be let in, by header. + method: The request's method. + path: What to ask for there: its path and query. + + Raises: + Refusal: If it could not be asked, before anything of an answer was sent. + """ + connection = http.client.HTTPConnection("127.0.0.1", port, timeout=_PATIENCE) + try: + try: + answer = _asks(handler, connection, said, method, path) + except (OSError, http.client.HTTPException) as why: + raise Refusal("The daemon did not answer.", 502) from why + handler.send_response(answer.status, answer.reason) + for name, value in answer.getheaders(): + if name.lower() not in _KEPT: + handler.send_header(name, value) + if answer.getheader("Content-Length") is None: + # Without a length, the answer ends where the connection does. + handler.close_connection = True + handler.end_headers() + while chunk := answer.read1(_CHUNK): + handler.wfile.write(chunk) + handler.wfile.flush() + except (OSError, http.client.HTTPException): + # Cut off once answering had begun: nothing is left to say but to stop. + handler.close_connection = True + finally: + connection.close() + + +def _asks( + handler: BaseHTTPRequestHandler, + connection: http.client.HTTPConnection, + said: dict[str, str], + method: str, + path: str, +) -> http.client.HTTPResponse: + """Sends the request on, what it sends a piece at a time, and reads how it is answered.""" + if handler.headers.get("Transfer-Encoding") is not None: + raise Refusal("Say how long what is sent is.", 411) + try: + length = int(handler.headers.get("Content-Length") or 0) + except ValueError as why: + raise Refusal("Say how long what is sent is.", 411) from why + if length < 0: + raise Refusal("Say how long what is sent is.", 411) + connection.putrequest(method, path, skip_host=True, skip_accept_encoding=True) + for name in _ASKED: + value = handler.headers.get(name) + if value is not None: + connection.putheader(name, value) + for name, value in said.items(): + connection.putheader(name, value) + connection.endheaders() + left = length + # An answer given before everything was sent -- a refusal -- is still the answer. + with contextlib.suppress(BrokenPipeError, ConnectionResetError): + while left: + chunk = handler.rfile.read(min(left, _CHUNK)) + if not chunk: + raise Refusal("What was sent was cut short.", 400) + connection.send(chunk) + left -= len(chunk) + if left: + # What was not read of what was sent is no next request. + handler.close_connection = True + return connection.getresponse() diff --git a/src/hmz/web/remote.py b/src/hmz/web/remote.py new file mode 100644 index 00000000..b179d76f --- /dev/null +++ b/src/hmz/web/remote.py @@ -0,0 +1,677 @@ +"""Daemons on other machines, reached over ssh and answered through this interface. + +Each daemon written down under `daemons` in humanize's settings is a host that `ssh` reaches. +What is reached there is that machine's own web interface: `hmz web --attached`, started over +ssh and stopped by the ssh going, with a port of this machine's loopback forwarded on to it. A +page asks this interface, which asks that one with its key: the remote interface keeps its own +key, its own runs and its own refusals, and nothing of it is served to anything but this page. + +An `hmz` missing there, or of another version than this one, is installed at this version with +`uv tool install`, uv itself fetched first where the host has none -- but never in place of one +the host's daemon may be running epics under, which keep running the release they started +with. A development build, `0.0.0`, installs nothing: it uses whatever `hmz` is there. + +A daemon is reached when a page first asks for it, or is told to connect, and reached again, +waiting longer each time, whenever it is lost. +""" + +from __future__ import annotations + +import contextlib +import dataclasses +import queue +import re +import socket +import subprocess +import threading +import time +from typing import TYPE_CHECKING, Any, cast + +from .routing import Refusal, routes + +if TYPE_CHECKING: + from collections.abc import Iterator, Sequence + from typing import IO + + from hmz.coganchor.transport import Road + + from .routing import Asked + +__all__ = ["ROUTES", "Remotes", "forwarding", "headers", "road", "said"] + +#: The release of uv fetched onto a host that has none, through astral's installer for it: +#: the one the desktop app pins. +UV = "0.12.24" + +#: What a daemon's state may be. +IDLE, CONNECTING, INSTALLING, UP, FAILED = ( + "idle", + "connecting", + "installing", + "up", + "failed", +) + +#: What every command there runs with: where `uv tool` and astral's installer put what they +#: install, which a shell that is not a login one may not have on its `PATH`. +_PATH = 'export PATH="$HOME/.local/bin:$HOME/.cargo/bin:$PATH"; ' + +#: What is asked of a host before anything is started there: the version of its `hmz`, and +#: whether its daemon is running -- by the process `daemon.json` names in `humanize-` of +#: any temporary directory one may have been started with, as the desktop app tells -- +#: `unknown` where that cannot be read. +_PROBE = ( + _PATH + + r""" +echo "hmz: $(hmz --version 2>/dev/null | sed -n 's/^hmz //p' | tail -n 1)" +daemon=none +for under in "$TMPDIR" "$TEMP" "$TMP" "$(getconf DARWIN_USER_TEMP_DIR 2>/dev/null)" \ + /tmp /var/tmp; do + [ -n "$under" ] || continue + record="${under%/}/humanize-$(id -u)/daemon.json" + [ -e "$record" ] || continue + if [ ! -r "$record" ]; then daemon=unknown; continue; fi + pid=$(sed -n 's/.*"pid": *\([0-9][0-9]*\).*/\1/p' "$record" | head -n 1) + if [ -z "$pid" ]; then daemon=unknown + elif kill -0 "$pid" 2>/dev/null || ps -p "$pid" >/dev/null 2>&1; then + daemon=running; break + fi +done +echo "daemon: $daemon" +""" +) + +#: Installs `hmz` at a version, `{version}`, fetching uv first where there is none. +_INSTALL = _PATH + ( + "export UV_NO_MODIFY_PATH=1; " + "if ! command -v uv >/dev/null 2>&1; then " + f"echo 'Getting uv {UV}'; " + "if command -v curl >/dev/null 2>&1; " + f"then curl -LsSf https://astral.sh/uv/{UV}/install.sh | sh; " + f"else wget -qO- https://astral.sh/uv/{UV}/install.sh | sh; fi; " + "command -v uv >/dev/null 2>&1 || exit 1; fi; " + "exec uv tool install --force --python '>=3.12' 'hmz=={version}'" +) + +#: Starts the interface there, for as long as the ssh holding its standard input lives. +_SERVE = _PATH + "exec hmz web --no-open --attached --port 0" + +#: The line it says where it listens with. +_ADDRESS = re.compile(r"hmz web: http://127\.0\.0\.1:(\d+)/\?key=(\S+)") + +#: What `ssh` is told for every command here before what the host's own settings and humanize's +#: say beside them: nothing is asked of anybody, since nobody is at a terminal to answer. +_UNASKED = ("-o", "BatchMode=yes", "-o", "ConnectTimeout=20") + +#: How long, in seconds, asking a host what it has may take. +_PROBING = 60.0 + +#: How long installing `hmz` there may take. +_INSTALLING = 900.0 + +#: How long the interface there may take to say where it listens. +_STARTING = 60.0 + +#: How long the forwarded port may take to be listened on. +_FORWARDING = 20.0 + +#: How long a page asking for a daemon not yet reached waits for it. +_WAITING = 15.0 + +#: The wait before reaching a daemon again, doubled each time it fails, up to the last. +_BACKOFF = (1.0, 60.0) + +#: How long a process is given to stop before it is killed. +_STOPPING = 3.0 + + +class _Failed(Exception): # noqa: N818 -- the sentence a daemon's state says + """Reaching a daemon went wrong, as its state says it.""" + + +def _here_version() -> str: + """This humanize's version, as PyPI has it.""" + from importlib.metadata import version + + return version("hmz") + + +def _entries() -> list[dict[str, Any]]: + """Every daemon written down, as humanize's settings hold them.""" + from hmz.coganchor import settings + + said = settings.read().get("daemons") + if not isinstance(said, list): + return [] + return [ + cast("dict[str, Any]", one) + for one in cast("list[object]", said) + if isinstance(one, dict) + ] + + +def _entry(name: str) -> dict[str, Any] | None: + """The daemon written down by that name, or None.""" + if name == "local": # This machine, whatever is written down. + return None + return next((one for one in _entries() if one.get("name") == name), None) + + +def road(said: dict[str, Any]) -> Road: + """How a command is run on a daemon's host, as its entry says to reach it. + + Raises: + ValueError: For an entry that says no host, or says one wrongly. + """ + from hmz.coganchor.machines.store import SSHRuntime + from hmz.coganchor.transport import Road, Target, ssh_flags + + options: Any = said.get("options") or [] + if isinstance(options, dict): + options = cast("dict[Any, Any]", options).items() + try: + pairs = cast("list[tuple[Any, Any]]", list(options)) + runtime = SSHRuntime( + name="daemon", + host=str(said.get("host") or ""), + user=str(said.get("user") or ""), + port=int(said.get("port") or 0), + identity_file=str(said.get("identity_file") or ""), + proxy_jump=str(said.get("proxy_jump") or ""), + options={str(key): str(value) for key, value in pairs}, + alias=str(said.get("alias") or ""), + config=str(said.get("config") or ""), + ) + except (TypeError, ValueError) as why: + raise ValueError(str(why).removeprefix("daemon: ")) from why + target = Target( + "ssh", host=runtime.login(), port=runtime.port, options=runtime.settings() + ) + road = Road.to(target) + # After what the host's own settings say, which win, and before what humanize assumes. + own = 1 + len(ssh_flags(target.options)) + prefix = (*road.prefix[:own], *_UNASKED, *road.prefix[own:]) + return dataclasses.replace(road, prefix=prefix) + + +def forwarding(on: Road, local: int, remote: int) -> list[str]: + """The `ssh` that forwards a port of this machine's loopback to one of the host's. + + A connection of its own rather than one riding a shared master, whose forwards outlive + the client that asked for them; and one that lasts as long as its standard input, which + whatever started it holds, so that it goes with it however that goes. + """ + ssh, *options, host = on.prefix + alone = ( + ssh, + "-o", + "ControlMaster=no", + "-o", + "ControlPath=none", + "-o", + "ExitOnForwardFailure=yes", + *options, + "-L", + f"127.0.0.1:{local}:127.0.0.1:{remote}", + host, + ) + return dataclasses.replace(on, prefix=alone).line(["cat"]) + + +def said(lines: Sequence[str]) -> dict[str, str]: + """What a host answered the probe with, by what each line says it is.""" + found: dict[str, str] = {} + for line in lines: + key, colon, value = line.partition(":") + if colon and key in ("hmz", "daemon"): + found[key] = value.strip() + return found + + +def headers(port: int, key: str) -> dict[str, str]: + """What a request to an interface listening on `port` carries to be let in, as its page's.""" + return { + "Host": f"127.0.0.1:{port}", + "Origin": f"http://127.0.0.1:{port}", + "Sec-Fetch-Site": "same-origin", + "Cookie": f"hmz-{port}={key}", + } + + +def _free(wanted: int) -> int: + """A port free on this machine's loopback: `wanted` where it is, else any. + + The one the other interface listens on where it can be had, so that a cookie for one is + named as a cookie for the other. + """ + for port in (wanted, 0): + with socket.socket() as trying: + try: + trying.bind(("127.0.0.1", port)) + except OSError: + continue + return int(trying.getsockname()[1]) + raise _Failed("No port is free here to forward.") + + +def _last(lines: Sequence[str], otherwise: str) -> str: + """The last line said, or what is said otherwise.""" + return next((line for line in reversed(lines) if line.strip()), otherwise).strip() + + +class _Remote: + """One daemon, reached and reached again on a thread of its own until it is closed.""" + + def __init__(self, name: str, version: str, changed: threading.Condition) -> None: + self.name = name + self._version = version + self._changed = changed + self.state = IDLE + self.detail = "" + self.there = "" + self.port = 0 + self.remote = 0 + self.key = "" + self.entry: dict[str, Any] | None = None + #: What is said of it while it is up: why its hmz is not this one's, where it is not. + self._note = "" + self._children: set[subprocess.Popen[str]] = set() + self._closed = threading.Event() + self._again = threading.Event() + self._thread: threading.Thread | None = None + + def seen(self) -> dict[str, str]: + return {"state": self.state, "detail": self.detail, "version": self.there} + + def starts(self, said: dict[str, Any], *, now: bool) -> None: + """Reaches it, as `said` is its entry. + + Again at once where that has changed, or where it failed and is wanted `now` rather + than after its wait. + """ + if self._thread is None or not self._thread.is_alive(): + self._closed.clear() + self._thread = threading.Thread( + target=self._keeps, daemon=True, name=f"hmz-web-remote-{self.name}" + ) + self._thread.start() + elif (now and self.state == FAILED) or said != self.entry: + self._again.set() + + def close(self) -> None: + self._closed.set() + self._again.set() + with self._changed: + children = list(self._children) + for child in children: + _stops(child) + if self._thread is not None: + self._thread.join(timeout=_STOPPING * 3) + + def _says(self, state: str, detail: str = "") -> None: + with self._changed: + self.state, self.detail = state, detail + self._changed.notify_all() + + def _keeps(self) -> None: + """Reaches it, holds it while it lasts, and reaches it again after a wait.""" + wait = _BACKOFF[0] + while not self._closed.is_set(): + self._again.clear() + self._says(CONNECTING) + try: + web, forward = self._reaches() + except Exception as why: # noqa: BLE001 -- whatever it was is what it says + self._ends() + if self._closed.is_set(): + break + self._says(FAILED, str(why)) + else: + self._says(UP, self._note) + wait = _BACKOFF[0] + while not self._again.wait(0.25): + if web.poll() is not None or forward.poll() is not None: + break + self._ends() + if self._closed.is_set(): + break + if self._again.is_set(): + continue + self._says(CONNECTING, "Lost") + self._again.wait(wait) + wait = min(wait * 2, _BACKOFF[1]) + self._says(IDLE) + + def _reaches(self) -> tuple[subprocess.Popen[str], subprocess.Popen[str]]: + """Reaches it. + + Probes the host, installs `hmz` there where it must, starts the interface there and + forwards a port to it. + + Returns: + The interface there, and the forward. + + Raises: + _Failed: Saying what went wrong. + """ + self.entry = _entry(self.name) + if self.entry is None: + raise _Failed(f"No daemon is called {self.name}.") + try: + on = road(self.entry) + except ValueError as why: + raise _Failed(str(why)) from why + version = self._installed(on) + self.there = version + web = self._started(on.line(["sh", "-c", _SERVE]), apart=True) + address = self._listens(web) + self.remote, self.key = int(address[1]), address[2] + self.port = _free(self.remote) + forward = self._started(forwarding(on, self.port, self.remote)) + told = _drained(forward) + ending = time.monotonic() + _FORWARDING + while not _listened(self.port): + if forward.poll() is not None: + time.sleep(0.1) # For what it said last to be read. + raise _Failed(_last(told, "The port could not be forwarded.")) + if self._closed.is_set() or time.monotonic() > ending: + raise _Failed("The port was not forwarded in time.") + time.sleep(0.05) + return web, forward + + def _installed(self, on: Road) -> str: + """The version of `hmz` there, once it is this one where it may be made so.""" + self._note = "" + found = self._probed(on) + there = found.get("hmz", "") + wanted = self._version + if wanted in ("0.0.0", there): + if not there: + raise _Failed(f"hmz is not installed on {self.name}.") + return there + if there and found.get("daemon") != "none": + if found.get("daemon") == "running": + why = f"hmz {there} there holds running epics; hmz {wanted} waits" + else: + why = f"Whether hmz {there} there runs epics is unknown; not replaced" + self._note = why + return there + self._says(INSTALLING, f"hmz {wanted}") + lines: list[str] = [] + installing = self._started( + on.line(["sh", "-c", _INSTALL.format(version=wanted)]) + ) + for line in _lines(installing, _INSTALLING): + lines.append(line) + if line.strip(): + self._says(INSTALLING, line.strip()) + if installing.wait() != 0: + raise _Failed(_last(lines, f"hmz {wanted} could not be installed.")) + there = self._probed(on).get("hmz", "") + if there != wanted: + raise _Failed(f"hmz {wanted} was installed, and hmz {there} answers.") + return there + + def _probed(self, on: Road) -> dict[str, str]: + probing = self._started(on.line(["sh", "-c", _PROBE])) + lines = list(_lines(probing, _PROBING)) + if probing.wait() != 0 or "daemon" not in said(lines): + raise _Failed(_last(lines, f"{self.name} did not answer.")) + return said(lines) + + def _listens(self, web: subprocess.Popen[str]) -> re.Match[str]: + """Where the interface there listens, once it says.""" + stdout = web.stdout + assert stdout is not None # noqa: S101 -- it was started with one + assert web.stderr is not None # noqa: S101 + heard: queue.Queue[str | None] = queue.Queue() + errors = _drained(web, web.stderr) + + def reads() -> None: + # Read to its end, which it says nothing more before. + with contextlib.suppress(OSError, ValueError), stdout: + for line in stdout: + heard.put(line) + heard.put(None) + + threading.Thread(target=reads, daemon=True, name="hmz-web-remote").start() + ending = time.monotonic() + _STARTING + while True: + try: + line = heard.get(timeout=max(0.0, ending - time.monotonic())) + except queue.Empty: + raise _Failed("hmz web there did not say where it listens.") from None + if line is None: + web.wait() + raise _Failed(_last(errors, "hmz web there stopped.")) + found = _ADDRESS.search(line) + if found: + return found + + def _started( + self, argv: list[str], *, apart: bool = False + ) -> subprocess.Popen[str]: + """A process, which is stopped as this daemon is closed or lost. + + Its standard input is held open until then, so that it goes with this one however + that goes; what it says on stderr is read `apart` from its stdout, or with it. + """ + with self._changed: + if self._closed.is_set(): + raise _Failed("Closed.") + try: + child = subprocess.Popen( + argv, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE if apart else subprocess.STDOUT, + text=True, + errors="replace", + start_new_session=True, + ) + except OSError as why: + raise _Failed(f"ssh could not be run: {why}") from why + self._children.add(child) + return child + + def _ends(self) -> None: + """Stops and reaps every process started for it.""" + with self._changed: + children = list(self._children) + self._children.clear() + for child in children: + _stops(child) + + +def _lines(child: subprocess.Popen[str], patience: float) -> Iterator[str]: + """Each line a process says, until it ends or has taken longer than `patience`.""" + assert child.stdout is not None # noqa: S101 -- it was started with one + timer = threading.Timer(patience, child.kill) + timer.start() + try: + with child.stdout: + yield from (line.rstrip("\n") for line in child.stdout) + finally: + timer.cancel() + + +def _drained(child: subprocess.Popen[str], said: IO[str] | None = None) -> list[str]: + """The last lines a process says, read as it says them so that it is never held up. + + Args: + child: The process. + said: What of it is read: its stdout, by default. + """ + lines: list[str] = [] + held = said or child.stdout + assert held is not None # noqa: S101 -- it was started with one + + def drains() -> None: + with held, contextlib.suppress(OSError, ValueError): + for line in held: + lines.append(line.rstrip("\n")) + del lines[:-20] + + threading.Thread(target=drains, daemon=True, name="hmz-web-remote").start() + return lines + + +def _listened(port: int) -> bool: + """Whether something listens on a port of this machine's loopback.""" + with socket.socket() as trying: + trying.settimeout(1.0) + return trying.connect_ex(("127.0.0.1", port)) == 0 + + +def _stops(child: subprocess.Popen[str]) -> None: + """Stops a process and reaps it: its standard input closed, then terminated, then killed.""" + if child.stdin is not None: + with contextlib.suppress(OSError): + child.stdin.close() + with contextlib.suppress(subprocess.TimeoutExpired): + child.wait(timeout=1.0) + if child.poll() is None: + with contextlib.suppress(OSError): + child.terminate() + try: + child.wait(timeout=_STOPPING) + except subprocess.TimeoutExpired: + child.kill() + child.wait() + + +class Remotes: + """Every daemon on another machine, as each is reached. + + Args: + version: The version of `hmz` installed where another is found; this one's by default. + """ + + def __init__(self, version: str | None = None) -> None: + self._version = version or _here_version() + self._changed = threading.Condition() + self._held: dict[str, _Remote] = {} + self._closed = False + + def state(self, name: str) -> dict[str, str]: + """How a daemon stands: `{state, detail, version}`. + + Its state is `idle` where it has not been asked for, then `connecting`, `installing`, + `up` or `failed`. + """ + with self._changed: + held = self._held.get(name) + return held.seen() if held else {"state": IDLE, "detail": "", "version": ""} + + def connect(self, name: str) -> dict[str, str]: + """Starts reaching a daemon, or reaches it again at once where it failed. + + Returns: + How it stands. + + Raises: + Refusal: If no daemon is written down by that name. + """ + self._starts(name, now=True) + return self.state(name) + + def reached(self, name: str) -> tuple[int, str]: + """Where a daemon is reached here, and the key it lets in. + + Reached first where it was not, waiting a while for it. + + Returns: + The port of this machine's loopback forwarded to it, and its key. + + Raises: + Refusal: If it is not up, with its state; or if none is called that. + """ + held = self._up(name) + return held.port, held.key + + def asks(self, name: str) -> tuple[int, dict[str, str]]: + """Where a daemon is reached here, and what a request to it carries to be let in. + + Raises: + Refusal: As `reached` does. + """ + held = self._up(name) + return held.port, headers(held.remote, held.key) + + def close(self) -> None: + """Stops reaching every daemon, every process started for one stopped and reaped.""" + with self._changed: + self._closed = True + held = list(self._held.values()) + stopping = [threading.Thread(target=one.close) for one in held] + for one in stopping: + one.start() + for one in stopping: + one.join() + + def _starts(self, name: str, *, now: bool) -> _Remote: + """A daemon, being reached: again where what reaches it has changed. + + Raises: + Refusal: If none is written down by that name -- one that was is let go of. + """ + said = _entry(name) + with self._changed: + if self._closed: + raise Refusal("This interface is stopping.", 503) + if said is None: + gone = self._held.pop(name, None) + if gone is not None: + threading.Thread(target=gone.close, daemon=True).start() + raise Refusal("There is no daemon by that name.", 404) + held = self._held.setdefault( + name, _Remote(name, self._version, self._changed) + ) + held.starts(said, now=now) + return held + + def _up(self, name: str) -> _Remote: + """A daemon once it is up, waiting a while for one being reached. + + Raises: + Refusal: If it is not up, with its state; or if none is called that. + """ + held = self._starts(name, now=False) + with self._changed: + # Not past a failure, which is reached again after its wait rather than at once. + self._changed.wait_for( + lambda: held.state in (UP, FAILED, INSTALLING), timeout=_WAITING + ) + if held.state == UP: + return held + raise Refusal( + held.detail or f"{name} is not reached yet.", 503, **held.seen() + ) + + +def _local() -> dict[str, str]: + return {"state": UP, "detail": "", "version": _here_version()} + + +def _remotes(asked: Asked) -> Remotes: + """The daemons on other machines, where the interface reaches any by the name asked.""" + if asked.site.remotes is None or _entry(asked.matched["name"]) is None: + raise Refusal("There is no daemon by that name.", 404) + return asked.site.remotes + + +def _state(asked: Asked) -> dict[str, str]: + name = asked.matched["name"] + return _local() if name == "local" else _remotes(asked).state(name) + + +def _connect(asked: Asked) -> dict[str, str]: + name = asked.matched["name"] + return _local() if name == "local" else _remotes(asked).connect(name) + + +#: How a page reads how a daemon stands, and has it reached. +ROUTES = routes( + ("GET", "/api/daemons/{name}/state", _state), + ("POST", "/api/daemons/{name}/connect", _connect), +) diff --git a/src/hmz/web/site.py b/src/hmz/web/site.py index e6204457..40ea50d6 100644 --- a/src/hmz/web/site.py +++ b/src/hmz/web/site.py @@ -44,6 +44,8 @@ held, paths, places, + proxy, + remote, runs, runtimes, settings, @@ -55,6 +57,7 @@ from collections.abc import Generator from .places import Places + from .remote import Remotes from .runs import Runs __all__ = ["Site"] @@ -71,8 +74,12 @@ *paths.ROUTES, *places.ROUTES, *verses.ROUTES, + *remote.ROUTES, ) +#: What a daemon is called in the paths a page asks another machine's daemon by. +_DAEMON = frozenset(string.ascii_letters + string.digits + "._-") + #: How many ports any would do for are tried before one free on both loopbacks is given up on. _PORTS = 8 @@ -152,6 +159,8 @@ def __init__(self, port: int, *, places: Places, runs: Runs) -> None: self.places = places self.hmz: Any = places.hmz self.runs = runs + #: The daemons on other machines, as each is reached, once `listening` has them. + self.remotes: Remotes | None = None #: Set as the server stops, which is what ends every stream it is sending. self.going = threading.Event() self._key = secrets.token_urlsafe(32) @@ -284,6 +293,8 @@ def _routes(self, method: str, path: str, query: dict[str, str]) -> None: ) if path.startswith("/api/"): self._api(method, path, query) + elif path.startswith("/d/"): + self._daemon(method, path, query) elif method != "GET": raise Refusal("Only the page's own files are here.", 405) elif path == "/" and "key" in query: @@ -352,13 +363,39 @@ def _signs_in(self, key: str) -> None: ) self._ends(0) - def _api(self, method: str, path: str, query: dict[str, str]) -> None: + def _admits(self) -> None: + """Refuses a request to the API from a browser not let in, or from another page.""" if not self._signed_in(): raise Refusal( "Open the address hmz web printed to let this browser in.", 401 ) if not self._asked_here(): raise Refusal("Use this server's own page to make this request.", 403) + + def _daemon(self, method: str, path: str, query: dict[str, str]) -> None: + """The API of a daemon by name, `/d//api/...`. + + This machine's own for `local`, and otherwise asked of the interface on the machine + it names. + """ + name, _, rest = path.removeprefix("/d/").partition("/") + rest = f"/{rest}" + if not name or not set(name) <= _DAEMON or not rest.startswith("/api/"): + raise Refusal("There is nothing here by that name.", 404) + if name == "local": + self._api(method, rest, query) + return + self._admits() + if method == "POST" and not self._from_its_page(): + raise Refusal("Use this server's own page to make this request.", 403) + if self.site.remotes is None: + raise Refusal("There is no daemon by that name.", 404) + port, said = self.site.remotes.asks(name) + asked = urllib.parse.urlsplit(self.path).query + proxy.forwards(self, port, said, method, rest + (f"?{asked}" if asked else "")) + + def _api(self, method: str, path: str, query: dict[str, str]) -> None: + self._admits() for route in ROUTES: matched = route.match(method, path) if matched is None: diff --git a/tests/integration/doubles_ssh.py b/tests/integration/doubles_ssh.py new file mode 100644 index 00000000..6f37500d --- /dev/null +++ b/tests/integration/doubles_ssh.py @@ -0,0 +1,216 @@ +"""A stand-in for `ssh`, and the machine it reaches: this one, under a home of its own. + +The `ssh` put first on `PATH` takes what the real one is told -- its options, the destination, +the command -- drops the options and the destination, and runs the command here, with `sh`, in +the other machine's home: its own `HOME`, `HUMANIZE_HOME` and `TMPDIR`. A port it is told to +forward with `-L` it forwards over TCP, for as long as it runs. A destination with `unreachable` +in it is one it cannot reach, as `ssh` says so. Every time it is run it writes its pid and what +it was told to `ssh.log` beside that home. + +The machine has an `hmz` in `~/.local/bin` which is this one, saying whichever version it is +said to be installed at, or none, and a `uv` whose `tool install` installs the version asked. +""" + +from __future__ import annotations + +import os +import json +import sys +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING + +import yaml + +if TYPE_CHECKING: + import pytest + +_SSH = """#!{python} +import json, os, socket, subprocess, sys, threading + +HOME, HUMANIZE, TMP, LOG = {home!r}, {humanize!r}, {tmp!r}, {log!r} +TAKES = set("BbcDEeFIiJLlmOopQRSWw") + +args, forwards, at = sys.argv[1:], [], 0 +while at < len(args) and args[at].startswith("-"): + flag = args[at][1:2] + if flag in TAKES: + value = args[at][2:] or args[at + 1] + at += 1 if args[at][2:] else 2 + if flag == "L": + forwards.append(value) + else: + at += 1 +host, command = args[at], " ".join(args[at + 1:]) +with open(LOG, "a") as log: + log.write(json.dumps([os.getpid(), *sys.argv[1:]]) + "\\n") +if "unreachable" in host: + print(f"ssh: Could not resolve hostname {{host}}: Name or service not known", + file=sys.stderr) + sys.exit(255) + + +def pumps(source, sink): + try: + while chunk := source.recv(65536): + sink.sendall(chunk) + except OSError: + pass + for one in (source, sink): + try: + one.shutdown(socket.SHUT_RDWR) + except OSError: + pass + + +def serves(listening, to): + while True: + near, _ = listening.accept() + try: + far = socket.create_connection(to) + except OSError: + near.close() + continue + for there in ((near, far), (far, near)): + threading.Thread(target=pumps, args=there, daemon=True).start() + + +for spec in forwards: + *bind, port, to, to_port = spec.split(":") + listening = socket.socket() + listening.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + listening.bind((bind[0] if bind else "127.0.0.1", int(port))) + except OSError as why: + print(f"bind: {{why}}", file=sys.stderr) + sys.exit(255) + listening.listen() + threading.Thread( + target=serves, args=(listening, (to, int(to_port))), daemon=True + ).start() + +env = dict(os.environ, HOME=HOME, HUMANIZE_HOME=HUMANIZE, TMPDIR=TMP) +os.chdir(HOME) +if not command: + threading.Event().wait() +if not forwards: + os.execvpe("sh", ["sh", "-c", command], env) +sys.exit(subprocess.run(["sh", "-c", command], env=env).returncode) +""" + +_HMZ = """#!/bin/sh +if [ ! -f "$HOME/.hmz-version" ]; then echo "sh: hmz: not found" >&2; exit 127; fi +if [ "$1" = "--version" ]; then echo "hmz $(cat "$HOME/.hmz-version")"; exit 0; fi +exec {python} -m hmz "$@" +""" + +_UV = """#!/bin/sh +echo "$@" >> "$HOME/uv.log" +for said; do case "$said" in hmz==*) version="${said#hmz==}";; esac; done +echo "Resolved 1 package in 3ms" +printf '%s' "$version" > "$HOME/.hmz-version" +echo "Installed 1 executable: hmz" +""" + + +@dataclass(frozen=True) +class Host: + """The machine the stand-in reaches. + + Attributes: + home: Its `HOME`. + humanize: Its `HUMANIZE_HOME`. + tmp: Its `TMPDIR`. + log: Where each `ssh` run writes its pid and what it was told. + """ + + home: Path + humanize: Path + tmp: Path + log: Path + + def installs(self, version: str | None) -> None: + """Has its `hmz` say this version, or be missing.""" + held = self.home / ".hmz-version" + if version is None: + held.unlink(missing_ok=True) + else: + held.write_text(version) + + def runs_epics(self) -> None: + """Has its daemon running, which holds its epics: this process, as one alive.""" + mine = self.tmp / f"humanize-{os.getuid()}" + mine.mkdir(mode=0o700, exist_ok=True) + (mine / "daemon.json").write_text(f'{{\n "pid": {os.getpid()}\n}}\n') + + def installed(self) -> str: + """What `uv` was asked to install, one line a time.""" + held = self.home / "uv.log" + return held.read_text() if held.exists() else "" + + def pids(self) -> list[int]: + """Every `ssh` run so far, by pid.""" + if not self.log.exists(): + return [] + return [json.loads(line)[0] for line in self.log.read_text().splitlines()] + + def told(self) -> list[list[str]]: + """What every `ssh` run so far was told.""" + if not self.log.exists(): + return [] + return [json.loads(line)[1:] for line in self.log.read_text().splitlines()] + + +def faked( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, version: str | None = "0.0.0" +) -> Host: + """Puts the stand-in `ssh` first on `PATH`, reaching a machine of its own. + + Args: + tmp_path: The test's own directory. + monkeypatch: What puts `PATH` back. + version: What its `hmz` says it is, or None for none at all. + + Returns: + The machine. + """ + there = tmp_path / "there" + host = Host( + there / "home", + there / "humanize", + # Short, as a socket's path under it must be. + Path(tempfile.mkdtemp(prefix="r", dir=os.environ["TMPDIR"])), + there / "ssh.log", + ) + (host.home / ".local" / "bin").mkdir(parents=True) + host.humanize.mkdir() + made = tmp_path / "bin" + made.mkdir() + for at, script in ( + ( + made / "ssh", + _SSH.format( + python=sys.executable, + home=str(host.home), + humanize=str(host.humanize), + tmp=str(host.tmp), + log=str(host.log), + ), + ), + (host.home / ".local" / "bin" / "hmz", _HMZ.format(python=sys.executable)), + (host.home / ".local" / "bin" / "uv", _UV), + ): + at.write_text(script) + at.chmod(0o755) + monkeypatch.setenv("PATH", f"{made}{os.pathsep}{os.environ['PATH']}") + host.installs(version) + return host + + +def written(*daemons: dict[str, object]) -> None: + """Writes these daemons down in this machine's settings, as `daemons`.""" + from hmz.coganchor import settings + + settings.changes(lambda held: held.__setitem__("daemons", list(daemons))) + assert yaml.safe_load(settings.where().read_text())["daemons"] diff --git a/tests/integration/test_web_remote.py b/tests/integration/test_web_remote.py new file mode 100644 index 00000000..0968d84a --- /dev/null +++ b/tests/integration/test_web_remote.py @@ -0,0 +1,270 @@ +"""A daemon on another machine, reached over ssh and answered through this interface. + +The other machine is this one under a home of its own, reached by the stand-in `ssh` of +`doubles_ssh`, running this `hmz web` there as the real one would run the one installed. +""" + +from __future__ import annotations + +import http.client +import os +import socket +import time +from typing import TYPE_CHECKING, Any + +import pytest + +from hmz.web.remote import Remotes + +from .doubles_ssh import Host, faked, written +from .doubles_web import PATIENCE, Answer, Browser, served + +if TYPE_CHECKING: + from pathlib import Path + + from hmz.web.site import Site + + +@pytest.fixture +def host(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Host: + """The other machine, written down as `box`, with this version of `hmz`.""" + found = faked(tmp_path, monkeypatch) + written( + { + "name": "box", + "host": "box.example", + "user": "me", + "options": [["Compression", "yes"]], + } + ) + return found + + +def _stands(remotes: Remotes, name: str, *states: str) -> dict[str, str]: + """How a daemon stands, once it stands as one of `states`.""" + ending = time.monotonic() + PATIENCE + while (said := remotes.state(name))["state"] not in states: + assert time.monotonic() < ending, f"{name} never came to {states}: {said}" + time.sleep(0.05) + return said + + +def _asked(browser: Browser, path: str) -> Answer: + """What the interface answers a read with, once the daemon is reached.""" + ending = time.monotonic() + PATIENCE + while (answer := browser.get(path)).status == 503: + assert time.monotonic() < ending, answer.body + time.sleep(0.1) + return answer + + +def _gone(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return True + return False + + +def test_a_page_reads_and_writes_a_daemon_on_another_machine(host: Host) -> None: + with served() as site: + browser = Browser(site) + browser.signs_in() + listed = _asked(browser, "/d/box/api/workspaces") + assert listed.status == 200, listed.body + assert str(host.home) in str(listed.json()) + state = browser.get("/api/daemons/box/state").json() + assert state == {"state": "up", "detail": "", "version": "0.0.0"} + # Written there, and not here. + kept = host.home / "project" + kept.mkdir() + wrote = browser.post("/d/box/api/workspaces", {"workspace": str(kept)}) + assert wrote.status == 200, wrote.body + assert wrote.json()["kept"] == str(kept) + assert str(kept) in (host.humanize / "settings.yaml").read_text() + # A file sent there a piece at a time, refused there for what it is. + sent = browser.sends( + "/d/box/api/runs/import", b"not an epic" * 1000, "x.hmzepic" + ) + assert sent.status not in (502, 503), sent.body + assert "error" in sent.json() + # Its runs, as a page reads them. + assert _asked(browser, "/d/box/api/runs").status == 200 + # And what it was told: nothing asked of anybody, the host's own options first. + told = host.told()[0] + assert told.index("Compression=yes") < told.index("BatchMode=yes") + assert told.index("BatchMode=yes") < told.index("BatchMode=no") + assert "me@box.example" in told + + +def test_local_is_this_machine_by_either_path(host: Host) -> None: + with served() as site: + browser = Browser(site) + browser.signs_in() + assert ( + browser.get("/d/local/api/workspaces").json() + == browser.get("/api/workspaces").json() + ) + assert browser.get("/api/daemons/local/state").json()["state"] == "up" + assert browser.get("/d/nobody/api/runs").status == 404 + assert browser.get("/api/daemons/nobody/state").status == 404 + assert not host.pids() + + +def test_only_the_page_let_in_reaches_another_machine(host: Host) -> None: + with served() as site: + browser = Browser(site) + assert browser.get("/d/box/api/runs").status == 401 + browser.signs_in() + assert ( + browser.get("/d/box/api/runs", **{"Sec-Fetch-Site": "cross-site"}).status + == 403 + ) + assert ( + browser.post( + "/d/box/api/workspaces", {}, Origin="http://elsewhere.example" + ).status + == 403 + ) + assert not host.pids() + + +def test_a_stream_of_events_comes_through_as_it_comes(host: Host) -> None: + with served() as site: + browser = Browser(site) + browser.signs_in() + _asked(browser, "/d/box/api/runs") + connection = http.client.HTTPConnection( + "127.0.0.1", site.port, timeout=PATIENCE + ) + try: + connection.request( + "GET", + "/d/box/api/held/stream", + headers={"Host": f"127.0.0.1:{site.port}", "Cookie": browser.cookie}, + ) + stream = connection.getresponse() + assert stream.status == 200 + assert stream.getheader("Content-Type", "").startswith("text/event-stream") + assert stream.getheader("Set-Cookie") is None + events: list[str] = [] + while not events: + line = stream.readline().decode() + assert line, "the stream ended before any event" + if line.startswith("event: "): + events.append(line.removeprefix("event: ").strip()) + finally: + connection.close() + + +def test_a_daemon_that_cannot_be_reached_says_why( + host: Host, monkeypatch: pytest.MonkeyPatch +) -> None: + del host + written({"name": "gone", "host": "unreachable.example"}) + with served() as site: + browser = Browser(site) + browser.signs_in() + connected = browser.post("/api/daemons/gone/connect") + assert connected.status == 200 + remotes = _remotes(site) + said = _stands(remotes, "gone", "failed") + assert "Could not resolve hostname" in said["detail"] + refused = browser.get("/d/gone/api/runs") + assert refused.status == 503 + assert refused.json()["state"] == "failed" + + +def test_closing_stops_and_reaps_every_process_started(host: Host) -> None: + remotes = Remotes() + remotes.connect("box") + _stands(remotes, "box", "up") + port, key = remotes.reached("box") + assert key + pids = host.pids() + # The probe, the interface there, the forward. + assert len(pids) == 3 + forward = host.told()[-1] + assert "ControlPath=none" in forward + assert any(word.startswith(f"127.0.0.1:{port}:127.0.0.1:") for word in forward) + remotes.close() + assert all(_gone(pid) for pid in pids) + assert remotes.state("box")["state"] == "idle" + with socket.socket() as again: + assert again.connect_ex(("127.0.0.1", port)) != 0 + + +def test_another_version_there_is_replaced_with_this_one(host: Host) -> None: + host.installs("1.0.0") + remotes = Remotes(version="1.2.3") + try: + remotes.connect("box") + said = _stands(remotes, "box", "up", "failed") + assert said == {"state": "up", "detail": "", "version": "1.2.3"} + assert host.installed().split() == [ + "tool", + "install", + "--force", + "--python", + ">=3.12", + "hmz==1.2.3", + ] + finally: + remotes.close() + + +def test_one_missing_there_is_installed(host: Host) -> None: + host.installs(None) + remotes = Remotes(version="1.2.3") + try: + remotes.connect("box") + assert _stands(remotes, "box", "up", "failed")["version"] == "1.2.3" + finally: + remotes.close() + + +def test_one_running_epics_there_is_left_alone(host: Host) -> None: + host.installs("1.0.0") + host.runs_epics() + remotes = Remotes(version="1.2.3") + try: + remotes.connect("box") + said = _stands(remotes, "box", "up", "failed") + assert said["state"] == "up" + assert said["version"] == "1.0.0" + assert "running epics" in said["detail"] + assert not host.installed() + finally: + remotes.close() + + +def test_a_development_build_installs_nothing(host: Host) -> None: + host.installs(None) + remotes = Remotes(version="0.0.0") + try: + remotes.connect("box") + said = _stands(remotes, "box", "failed") + assert said["detail"] == "hmz is not installed on box." + assert not host.installed() + finally: + remotes.close() + + +def test_a_daemon_no_longer_written_down_is_let_go(host: Host) -> None: + with served() as site: + browser = Browser(site) + browser.signs_in() + assert _asked(browser, "/d/box/api/runs").status == 200 + pids = host.pids() + written({"name": "other", "host": "other.example"}) + assert browser.get("/d/box/api/runs").status == 404 + assert browser.get("/api/daemons/box/state").status == 404 + ending = time.monotonic() + PATIENCE + while not all(_gone(pid) for pid in pids): + assert time.monotonic() < ending, "what reached it was never stopped" + time.sleep(0.05) + + +def _remotes(site: Site) -> Any: + assert site.remotes is not None + return site.remotes diff --git a/tests/unit/web/test_remote.py b/tests/unit/web/test_remote.py new file mode 100644 index 00000000..2ad215c7 --- /dev/null +++ b/tests/unit/web/test_remote.py @@ -0,0 +1,27 @@ +"""What a host says it has, and what a request to its interface carries to be let in.""" + +from __future__ import annotations + +from hmz.web.remote import headers, said + + +def test_a_probe_is_read_by_what_each_line_says() -> None: + assert said( + [ + "Warning: Permanently added 'box' to the list of known hosts.", + "hmz: 0.5.0rc1", + "daemon: running", + "something: else", + ] + ) == {"hmz": "0.5.0rc1", "daemon": "running"} + assert said(["hmz: ", "daemon: none"]) == {"hmz": "", "daemon": "none"} + assert said([]) == {} + + +def test_a_request_there_is_its_own_page_s() -> None: + assert headers(8123, "k3y") == { + "Host": "127.0.0.1:8123", + "Origin": "http://127.0.0.1:8123", + "Sec-Fetch-Site": "same-origin", + "Cookie": "hmz-8123=k3y", + }