From 5192e0408863247617bc6936a49c88c6e8574b5d Mon Sep 17 00:00:00 2001 From: "jj.wang" Date: Thu, 22 Oct 2026 02:59:37 +0000 Subject: [PATCH] PAD-12: per-terminal transport (direct Tailscale SSH or Warpgate, pre-session fallback, pinned host keys, agent key from Vault), one-call win10 launch and one-call collect, remote call timings, agentd ping --- README.md | 2 +- agentd/cli.py | 40 ++ agentd/config.py | 110 +++- agentd/core.py | 6 +- agentd/executors/base.py | 20 +- agentd/executors/claude_code.py | 11 +- agentd/executors/codex.py | 12 +- agentd/executors/shell.py | 13 +- agentd/transports/__init__.py | 151 +++++- agentd/transports/direct_ssh.py | 313 +++++++++++ agentd/transports/local.py | 2 +- agentd/transports/warpgate_ssh.py | 268 ++++++++-- config/agentd.example.yaml | 7 +- config/agentd.yaml | 24 +- config/schema.json | 18 +- cron/agentd-tick-lxc.sh | 40 +- docs/deploy-lxc.md | 61 ++- docs/design.md | 18 +- docs/executor-contract.md | 6 +- docs/runbook.md | 16 +- docs/usage.md | 9 +- tests/test_direct_ssh.py | 827 ++++++++++++++++++++++++++++++ tests/test_posix_remote.py | 8 +- 23 files changed, 1850 insertions(+), 132 deletions(-) create mode 100644 agentd/transports/direct_ssh.py create mode 100644 tests/test_direct_ssh.py diff --git a/README.md b/README.md index 4c88555..517a369 100644 --- a/README.md +++ b/README.md @@ -9,5 +9,5 @@ Plane 看板驱动的通用 agent 调度服务：在 Plane 建工作项并指派 ## 使用 - 契约与扩展：`docs/executor-contract.md`；安装、cron、日志、故障：`docs/runbook.md`；LXC 部署（调度器 2026-10-21 起跑在 `agentd-01`）：`docs/deploy-lxc.md` -- 入口：`cli/agentd tick|new|show|comment|state|assign|smoke|check-config`（在 `agent-access.py exec --secret-env …` 下运行） +- 入口：`cli/agentd tick|new|show|comment|state|assign|smoke|ping|check-config`（在 `agent-access.py exec --secret-env …` 下运行；`ping` 验证到终端的 direct-ssh / warpgate-ssh 路径） - 测试：`python3 -m unittest`（离线，不访问 Plane/win10） diff --git a/agentd/cli.py b/agentd/cli.py index 5356c8c..c264741 100644 --- a/agentd/cli.py +++ b/agentd/cli.py @@ -9,6 +9,7 @@ from . import config as config_mod from .core import Agentd from .plane import Plane, brief_to_html, html_to_text, text_to_html from .store import Busy +from .transports import CallLog, TransportError, direct_ssh, make_transport def identity(cfg, wanted=None): @@ -159,6 +160,31 @@ def cmd_smoke(cfg, a): sys.exit(0 if ok else 1) +def cmd_ping(cfg, a): + """Trivial command on a terminal, timed per call (verifies direct-ssh / warpgate-ssh after a deployment). The + direct-ssh key comes from the environment like the Plane tokens (agent-access.py exec --secret-env).""" + if a.terminal not in cfg.terminals: + sys.exit("unknown terminal %s" % a.terminal) + t = cfg.terminals[a.terminal] + if (a.transport or t["transport"]) == "local": + sys.exit("%s is a local terminal: nothing to ping" % a.terminal) + if a.transport == "direct-ssh" and "ssh" not in t: + sys.exit("terminals.%s has no ssh section" % a.terminal) + log = CallLog(event=lambda m: print("event: " + m, flush=True), echo=True) + tr = make_transport(cfg, a.terminal, log=log, force=a.transport) + ok = True + for i in range(a.count): + t0 = time.monotonic() + try: + pong = tr.ping() + except TransportError as exc: + ok = False + print("ping %d: FAILED after %d ms: %s" % (i + 1, (time.monotonic() - t0) * 1000, exc), flush=True) + continue + print("ping %d: %d ms %s" % (i + 1, (time.monotonic() - t0) * 1000, pong), flush=True) + sys.exit(0 if ok else 1) + + def cmd_check_config(cfg, a): print("config %s is valid" % cfg.path) print("projects: %s" % ", ".join(cfg.projects)) @@ -166,6 +192,13 @@ def cmd_check_config(cfg, a): env = cfg.token_env(name) print("terminal %-10s %-13s %-8s executors=%s token %s: %s" % ( name, t["transport"], t["os"], ",".join(t["executors"]), env, "set" if os.environ.get(env) else "MISSING")) + if t["transport"] == "direct-ssh": + s = t["ssh"] + print(" direct-ssh %s@%s:%s, fallback %s" % (s["user"], s["host"], s.get("port", 22), t.get("fallback", "none"))) + if cfg.uses_direct_ssh(): + print("direct-ssh key %s: %s; ssh_dir %s" % ( + cfg.ssh_key_env, "set" if direct_ssh.key_present(cfg.ssh_key_env) else "MISSING (direct terminals use their fallback)", + cfg.ssh_dir)) print("cron/exec secret args: %s" % " ".join(cfg.secret_env_args())) if cfg.webhook: k = cfg.webhook["secret_key"] @@ -221,6 +254,12 @@ def parser(): s.add_argument("--port", type=int, help="port (default: webhook.port)") s.set_defaults(fn=cmd_webhook_serve) sub.add_parser("check-config", help="validate the config and list token env vars").set_defaults(fn=cmd_check_config) + s = sub.add_parser("ping", help="time a trivial command on a terminal (direct-ssh needs the key env var, see check-config)") + s.add_argument("terminal") + s.add_argument("--transport", choices=["direct-ssh", "warpgate-ssh"], + help="force one channel, no fallback (default: the terminal's transport with its fallback)") + s.add_argument("-n", "--count", type=int, default=3, help="calls (default 3: the 2nd+ reuse the SSH connection)") + s.set_defaults(fn=cmd_ping) return ap @@ -230,4 +269,5 @@ def main(argv=None): cfg = config_mod.load(a.config) except config_mod.ConfigError as exc: sys.exit(str(exc)) + direct_ssh.take_key(cfg.ssh_key_env) # out of os.environ before any child process starts a.fn(cfg, a) diff --git a/agentd/config.py b/agentd/config.py index 73ab95d..8e8b55f 100644 --- a/agentd/config.py +++ b/agentd/config.py @@ -2,8 +2,10 @@ The config only ever holds Vault *key names* (`token_key`). The token itself is read from the environment variable of the same name, which `agent-access.py exec --secret-env KEY=KEY` injects into the process.""" +import base64 import json import os +import struct from pathlib import Path import yaml @@ -11,6 +13,7 @@ import yaml from . import ROOT SCHEMA_PATH = ROOT / "config" / "schema.json" +DEFAULT_SSH_KEY = "AGENTD_SSH_KEY" # env var (== Vault key name) carrying the direct-ssh private key DEFAULT_PATHS = (ROOT / "config" / "agentd.local.yaml", ROOT / "config" / "agentd.yaml") @@ -83,6 +86,48 @@ def schema_errors(data, schema): for e in validator.iter_errors(data)] +def host_key_error(value): + """None if `value` is a usable pinned host key ('ssh-ed25519 ' whose blob names the same type).""" + parts = str(value).split() + if len(parts) < 2 or parts[0] != "ssh-ed25519": + return "must be 'ssh-ed25519 ' (the terminal's /etc/ssh/ssh_host_ed25519_key.pub, verified out of band)" + try: + blob = base64.b64decode(parts[1], validate=True) + n = struct.unpack(">I", blob[:4])[0] + ok = blob[4:4 + n] == b"ssh-ed25519" and len(blob) == 4 + n + 4 + 32 + except (ValueError, struct.error): + ok = False + return None if ok else "the base64 part is not an ssh-ed25519 public key (copy/paste damage?)" + + +def ssh_dir_of(data): + d = (data.get("agentd") or {}).get("ssh_dir") + if d: + p = Path(os.path.expanduser(d)) + return p if p.is_absolute() else ROOT / p + s = Path(os.path.expanduser((data.get("agentd") or {}).get("state_dir", "state"))) + return (s if s.is_absolute() else ROOT / s) / "ssh" + + +def remote_errors(data, tname, t, kind): + """Rules every remote transport (warpgate-ssh, direct-ssh, and the fallback channel) shares.""" + errs = [] + if t.get("os") == "windows": + if t.get("launch") != "schtasks": + errs.append("terminals.%s: %s on windows needs launch=schtasks" % (tname, kind)) + if kind == "warpgate-ssh" and t.get("target", tname) != "win10": + errs.append("terminals.%s: the Windows helper scripts only reach target 'win10' for now" % tname) + else: + if t.get("launch") != "nohup": + errs.append("terminals.%s: %s on %s needs launch=nohup" % (tname, kind, t.get("os"))) + for key in ("repo", "run_root"): + if key in t and not str(t[key]).startswith("/"): + errs.append("terminals.%s.%s: must be an absolute path on the remote (no ~)" % (tname, key)) + if kind == "warpgate-ssh" and not data.get("vault", {}).get("config"): + errs.append("terminals.%s: remote POSIX terminals need vault.config (relay identity)" % tname) + return errs + + def semantic_errors(data): """Cross-references the schema cannot express.""" errs = [] @@ -91,21 +136,36 @@ def semantic_errors(data): for ex in t.get("executors", []): if ex not in execs: errs.append("terminals.%s: unknown executor '%s'" % (tname, ex)) - if t.get("transport") == "warpgate-ssh" and t.get("os") == "windows": - if t.get("launch") != "schtasks": - errs.append("terminals.%s: warpgate-ssh on windows needs launch=schtasks" % tname) - if t.get("target", tname) != "win10": - errs.append("terminals.%s: the Windows helper scripts only reach target 'win10' for now" % tname) - elif t.get("transport") == "warpgate-ssh": - if t.get("launch") != "nohup": - errs.append("terminals.%s: warpgate-ssh on %s needs launch=nohup" % (tname, t.get("os"))) - for key in ("repo", "run_root"): - if key in t and not str(t[key]).startswith("/"): - errs.append("terminals.%s.%s: must be an absolute path on the remote (no ~)" % (tname, key)) - if not data.get("vault", {}).get("config"): - errs.append("terminals.%s: remote POSIX terminals need vault.config (relay identity)" % tname) - if t.get("transport") == "local" and t.get("os") != "linux": + kind = t.get("transport") + if kind == "warpgate-ssh": + errs += remote_errors(data, tname, t, kind) + elif kind == "direct-ssh": + errs += remote_errors(data, tname, t, kind) + s = t.get("ssh") + if not s: + errs.append("terminals.%s: direct-ssh needs ssh: {host, user, port, host_key}" % tname) + else: + if not s.get("host") or str(s["host"]).startswith("-") or not s.get("user"): + errs.append("terminals.%s.ssh: host and user must be non-empty (host not starting with '-')" % tname) + if s.get("port", 22) > 65535: + errs.append("terminals.%s.ssh.port: %r > 65535" % (tname, s["port"])) + why = host_key_error(s.get("host_key", "")) + if why: + errs.append("terminals.%s.ssh.host_key: %s" % (tname, why)) + if t.get("fallback") == "warpgate-ssh": + if "target" not in t: + errs.append("terminals.%s: fallback warpgate-ssh needs the Warpgate 'target'" % tname) + elif t.get("os") == "windows" and t["target"] != "win10": + errs.append("terminals.%s: the Windows helper scripts (fallback) only reach target 'win10' for now" % tname) + if t.get("os") != "windows" and not data.get("vault", {}).get("config"): + errs.append("terminals.%s: fallback warpgate-ssh needs vault.config (relay identity)" % tname) + elif t.get("fallback"): + errs.append("terminals.%s: 'fallback' only applies to transport direct-ssh" % tname) + if kind == "local" and t.get("os") != "linux": errs.append("terminals.%s: local transport supports only os=linux" % tname) + ctl = ssh_dir_of(data) # "/-<40-char %C>", plus ".<16 chars>" while ssh sets the socket up + if any(t.get("transport") == "direct-ssh" for t in terms.values()) and len(str(ctl)) + 1 + 9 + 40 + 17 > 104: + errs.append("agentd.ssh_dir: %s is too long for the ControlMaster socket path (UNIX limit 104)" % ctl) for ename, e in execs.items(): adapter = e.get("adapter", ename) if adapter not in ("codex", "claude-code", "shell"): @@ -156,6 +216,19 @@ class Config: p = Path(os.path.expanduser(self.agentd.get("state_dir", "state"))) return p if p.is_absolute() else ROOT / p + @property + def ssh_dir(self): + """direct-ssh ControlMaster sockets + generated known_hosts (default /ssh).""" + return ssh_dir_of(self.data) + + @property + def ssh_key_env(self): + """Env var (== Vault key name) holding the direct-ssh private key.""" + return self.agentd.get("ssh_key") or DEFAULT_SSH_KEY + + def uses_direct_ssh(self): + return any(t["transport"] == "direct-ssh" for t in self.terminals.values()) + def helpers_dir(self, terminal): d = self.terminals[terminal].get("helpers_dir") or self.agentd.get("helpers_dir") or "tools/win10" p = Path(os.path.expanduser(d)) @@ -185,13 +258,16 @@ class Config: return keys def scrubbed_env(self): - """os.environ without the secrets above, for every child process agentd starts.""" - drop = self.secret_keys() + """os.environ without the secrets above, for every child process agentd starts. The direct-ssh key variable + is always dropped, even with no direct-ssh terminal (a rollback may leave the cron script injecting it).""" + drop = self.secret_keys() | {self.ssh_key_env} return {k: v for k, v in os.environ.items() if k not in drop} def secret_env_args(self): - """`--secret-env` arguments for agent-access.py exec, one per distinct token key.""" + """`--secret-env` arguments for agent-access.py exec: one per distinct token key, plus the direct-ssh key.""" keys = sorted({t["token_key"] for t in self.terminals.values()}) + if self.uses_direct_ssh(): + keys.append(self.ssh_key_env) return ["--secret-env %s=%s" % (k, k) for k in keys] def project(self, name): diff --git a/agentd/core.py b/agentd/core.py index f028141..989d406 100644 --- a/agentd/core.py +++ b/agentd/core.py @@ -21,7 +21,7 @@ from .executors import ExecResult, Run, make_executor from .executors.base import fill from .plane import Plane, html_to_text, text_to_html from .store import Store -from .transports import TransportError, make_transport +from .transports import CallLog, TransportError, make_transport TAGS = ("review", "question", "blocked", "progress") WORKER = "[worker]" @@ -152,9 +152,11 @@ class Agentd: self.store = store or Store(cfg.state_dir, log_dir=cfg.log_dir) self.now = now self.log = self.store.log + # one line per remote call (terminal, channel, operation, ms, rc) -> remote.log; fallback/host-key -> agentd.log + self.calllog = CallLog(self.store.log_dir / "remote.log", self.log) self.plane_factory = plane_factory or (lambda term: Plane( cfg.plane["base_url"], cfg.plane["workspace"], cfg.token(term), cfg.plane.get("timeout", 60))) - self.transport_factory = transport_factory or (lambda term: make_transport(cfg, term)) + self.transport_factory = transport_factory or (lambda term: make_transport(cfg, term, log=self.calllog)) self.executor_factory = executor_factory or (lambda term, name: make_executor(cfg, term, name, self.transport(term))) self._reset_caches() diff --git a/agentd/executors/base.py b/agentd/executors/base.py index 181e87d..46c213a 100644 --- a/agentd/executors/base.py +++ b/agentd/executors/base.py @@ -6,6 +6,8 @@ import re from dataclasses import dataclass, field from typing import List, Optional +from ..transports import fetch_each + @dataclass class Run: @@ -55,6 +57,11 @@ class Executor: raise NotImplementedError +def parse_exit(text): + m = re.search(r"exit=(-?\d+)", text or "") + return int(m.group(1)) if m else None + + def fill(template, params): """Replace {name} placeholders that exist in params; leave anything else untouched.""" return re.sub(r"\{(\w+)\}", lambda m: str(params[m.group(1)]) if m.group(1) in params else m.group(0), template) @@ -101,7 +108,7 @@ class RunnerExecutor(Executor): run_dir, task = h["run_dir"], h["task"] script_name, script = t.runner(run_dir, run.repo, self.command(run, run_dir), self.stdout_name) files = {"prompt.md": run.prompt, script_name: script} - if hasattr(t, "start_run"): # remote POSIX: stage + launch in one relay round trip + if hasattr(t, "start_run"): # remote: stage + launch in one call (direct-ssh; POSIX also on the relay) t.start_run(task, run_dir, files, script_name) else: t.mkdir(run_dir, clean=True) @@ -121,9 +128,14 @@ class RunnerExecutor(Executor): return "finished" if text and "exit=" in text else "running" def exit_code(self, handle): - text = self.transport.read_text(self.transport.join(handle["run_dir"], "exit.txt")) or "" - m = re.search(r"exit=(-?\d+)", text) - return int(m.group(1)) if m else None + return parse_exit(self.transport.read_text(self.transport.join(handle["run_dir"], "exit.txt"))) + + def fetch(self, handle, ops): + """Reads of the run directory for `collect`: one call on remote transports (see transports.fetch).""" + t = self.transport + if hasattr(t, "fetch"): + return t.fetch(handle["run_dir"], ops) + return fetch_each(t, handle["run_dir"], ops) def kill(self, handle): self.transport.kill(handle["task"], handle["run_dir"]) diff --git a/agentd/executors/claude_code.py b/agentd/executors/claude_code.py index b4ec19c..f52243d 100644 --- a/agentd/executors/claude_code.py +++ b/agentd/executors/claude_code.py @@ -2,7 +2,7 @@ resume with `--resume `. The JSON result carries `result` (last message) and `session_id`.""" import json -from .base import ExecResult, RunnerExecutor +from .base import ExecResult, RunnerExecutor, parse_exit def parse_output(text): @@ -26,13 +26,14 @@ def parse_output(text): class ClaudeCodeExecutor(RunnerExecutor): stdout_name = "out.json" + COLLECT = [("read", "exit.txt"), ("read", "out.json"), ("tail", "stderr.txt", 5)] + def collect(self, handle): - t, d = self.transport, handle["run_dir"] - obj = parse_output(t.read_text(t.join(d, "out.json"))) + exit_text, out, tail = self.fetch(handle, self.COLLECT) # one remote call + obj = parse_output(out) err = bool(obj.get("is_error")) or obj.get("subtype", "success") != "success" last = "" if err else (obj.get("result") or "") - tail = self.stderr_tail(handle) if err: tail = ("%s %s" % (obj.get("subtype", ""), str(obj.get("result", ""))[:400]) + "\n" + tail).strip() - return ExecResult(last.strip(), self.exit_code(handle), obj.get("session_id") or handle.get("thread_id"), + return ExecResult(last.strip(), parse_exit(exit_text), obj.get("session_id") or handle.get("thread_id"), self.logs(handle, "out.json", "stderr.txt"), tail) diff --git a/agentd/executors/codex.py b/agentd/executors/codex.py index 2a43962..b88c1bb 100644 --- a/agentd/executors/codex.py +++ b/agentd/executors/codex.py @@ -2,22 +2,22 @@ Thread id comes from the `thread.started` event in events.jsonl; the last message from last.md.""" import json -from .base import ExecResult, RunnerExecutor +from .base import ExecResult, RunnerExecutor, parse_exit class CodexExecutor(RunnerExecutor): stdout_name = "events.jsonl" + COLLECT = [("read", "exit.txt"), ("read", "last.md"), ("grep", "events.jsonl", "thread.started"), + ("tail", "stderr.txt", 5), ("tail", "events.jsonl", 1)] def collect(self, handle): - t, d = self.transport, handle["run_dir"] - last = t.read_text(t.join(d, "last.md")) or "" + exit_text, last, line, err_tail, last_event = self.fetch(handle, self.COLLECT) # one remote call thread = handle.get("thread_id") - line = t.grep_first(t.join(d, "events.jsonl"), "thread.started") if line: try: thread = json.loads(line).get("thread_id") or thread except ValueError: pass - tail = (self.stderr_tail(handle) + "\n" + t.tail(t.join(d, "events.jsonl"), 1)[-600:]).strip() - return ExecResult(last.strip(), self.exit_code(handle), thread, + tail = (err_tail + "\n" + last_event[-600:]).strip() + return ExecResult((last or "").strip(), parse_exit(exit_text), thread, self.logs(handle, "events.jsonl", "stderr.txt", "last.md"), tail) diff --git a/agentd/executors/shell.py b/agentd/executors/shell.py index d8eca52..60e55ed 100644 --- a/agentd/executors/shell.py +++ b/agentd/executors/shell.py @@ -1,6 +1,6 @@ """Shell/script executor: runs `{script}` with the prompt on stdin. The last message is `last.md` in the run directory if the script wrote it (its path is passed as {out}), else the tail of stdout. No threads.""" -from .base import ExecResult, RunnerExecutor +from .base import ExecResult, RunnerExecutor, parse_exit class ShellExecutor(RunnerExecutor): @@ -10,10 +10,11 @@ class ShellExecutor(RunnerExecutor): def resume(self): return False + COLLECT = [("read", "exit.txt"), ("read", "last.md"), ("read", "stdout.txt"), ("tail", "stderr.txt", 5)] + def collect(self, handle): - t, d = self.transport, handle["run_dir"] - last = t.read_text(t.join(d, "last.md")) + exit_text, last, stdout, tail = self.fetch(handle, self.COLLECT) # one remote call if last is None: - last = (t.read_text(t.join(d, "stdout.txt")) or "")[-6000:] - return ExecResult(last.strip(), self.exit_code(handle), None, - self.logs(handle, "stdout.txt", "stderr.txt"), self.stderr_tail(handle)) + last = (stdout or "")[-6000:] + return ExecResult(last.strip(), parse_exit(exit_text), None, + self.logs(handle, "stdout.txt", "stderr.txt"), tail) diff --git a/agentd/transports/__init__.py b/agentd/transports/__init__.py index 908257a..90a0eb8 100644 --- a/agentd/transports/__init__.py +++ b/agentd/transports/__init__.py @@ -4,21 +4,162 @@ Interface (duck-typed, see docs/executor-contract.md): os, join(*parts), quote(s), default_run_root() mkdir(path, clean=False), write_text(path, text), read_text(path) -> str|None, grep_first(path, needle) -> str|None, tail(path, n) -> str + fetch(run_dir, ops) -> [result per op] (ops: ("read", name) | ("grep", name, needle) | ("tail", name, n)) runner(run_dir, cwd, command, stdout_name) -> (script_name, script_text) - launch(task, script_path, run_dir), kill(task, run_dir), forget(task) - process_count(name) -> int|None (None = unreachable)""" + start_run(task, run_dir, files, script_name) (remote only), launch(task, script_path, run_dir), kill(task, run_dir), + forget(task), process_count(name) -> int|None (None = unreachable) + +Remote terminals are reached through a channel (design §9): `warpgate-ssh` (agent-access.py run through the +Warpgate gateway) or `direct-ssh` (OpenSSH over Tailscale with a dedicated key, optional warpgate fallback).""" +import base64 +import functools +import re +import time class TransportError(Exception): pass -def make_transport(cfg, terminal): - kind = cfg.terminals[terminal]["transport"] +class Unreached(TransportError): + """The remote command certainly did not run (connect / auth / agent failure, direct path unavailable): the + only kind of failure a direct-ssh terminal may repeat on its fallback channel.""" + + +class HostKeyError(TransportError): + """The terminal presented a host key other than the pinned one: a security error, never retried elsewhere.""" + + +class CallLog: + """One line per remote call -> `path` (remote.log); notable events (fallback, host key) also -> `event`.""" + + def __init__(self, path=None, event=None, echo=False): + self.path, self.event_fn, self.echo = path, event, echo + self._once = set() + + def call(self, terminal, via, op, ms, rc, note=""): + line = "term=%s via=%s op=%s ms=%d rc=%s%s" % (terminal, via, op or "-", ms, rc, " " + note if note else "") + if self.echo: + print(line, flush=True) + if self.path: + try: + self.path.parent.mkdir(parents=True, exist_ok=True) + with open(self.path, "a", encoding="utf-8") as fh: + fh.write(time.strftime("%Y-%m-%dT%H:%M:%S") + " " + line + "\n") + except OSError: + pass + + def event(self, msg, once=None): + """`once` = a key: the message is logged only the first time per process.""" + if once is not None: + if once in self._once: + return + self._once.add(once) + if self.event_fn: + self.event_fn(msg) + + +NULL_LOG = CallLog() + + +def operation(name): + """Method decorator: names the remote operation for the call log (innermost wins, restored on return).""" + def deco(fn): + @functools.wraps(fn) + def wrapper(self, *args, **kwargs): + prev, self._op = getattr(self, "_op", None), name + try: + return fn(self, *args, **kwargs) + finally: + self._op = prev + return wrapper + return deco + + +def fetch_each(t, run_dir, ops): + """`fetch` built from the single-file primitives (one call per op): local transport and fallbacks.""" + out = [] + for o in ops: + path = t.join(run_dir, o[1]) + if o[0] == "read": + out.append(t.read_text(path)) + elif o[0] == "grep": + out.append(t.grep_first(path, o[2])) + elif o[0] == "tail": + out.append(t.tail(path, o[2])) + else: + raise ValueError("unknown fetch op %r" % (o,)) + return out + + +def parse_fetch(out, ops, what=""): + """Decode the `F=` / `M` (missing) lines of a remote fetch script into results per op: + read -> text|None, grep -> line|None, tail -> text ("" when missing).""" + res = [] + for i, o in enumerate(ops): + m = re.search(r"^F%d=([A-Za-z0-9+/=]*)$" % i, out, re.M) + if m: + text = base64.b64decode(m.group(1)).decode("utf-8", "replace") if m.group(1) else "" + elif re.search(r"^M%d$" % i, out, re.M): + text = None + else: + raise TransportError("fetch %s: unexpected output: %s" % (what, out.strip()[-200:])) + if o[0] == "grep": + text = (text or "").rstrip("\n") or None + elif o[0] == "tail": + text = (text or "").rstrip("\n") + res.append(text) + return res + + +class RoutedTransport: + """direct-ssh terminal: every operation runs on the primary (direct) transport; when it fails before the remote + command ran (`Unreached`) the operation is repeated on the fallback (warpgate-ssh) transport, which then serves + the rest of this transport's life (one tick). Each direct operation is a single remote call, so a repeat never + doubles a side effect. Host-key failures and failures after the command may have run propagate unchanged.""" + PURE = ("join", "quote", "default_run_root", "runner") + + def __init__(self, terminal, primary, fallback=None, log=None): + self.terminal, self.primary, self.fallback = terminal, primary, fallback + self.log = log or NULL_LOG + self.on_fallback = False + + def __getattr__(self, name): + attr = getattr(self.primary, name) + if name in self.PURE or not callable(attr): + return attr + + def routed(*args, **kwargs): + if self.on_fallback: + return getattr(self.fallback, name)(*args, **kwargs) + try: + return attr(*args, **kwargs) + except Unreached as exc: + if self.fallback is None: + raise + self.on_fallback = True + self.log.event("%s: direct-ssh failed before the command ran (%s); using the warpgate-ssh fallback " + "for the rest of this tick" % (self.terminal, exc)) + return getattr(self.fallback, name)(*args, **kwargs) + return routed + + +def make_transport(cfg, terminal, log=None, force=None): + """`force` = direct-ssh | warpgate-ssh: that channel only, no fallback (agentd ping).""" + t = cfg.terminals[terminal] + kind = force or t["transport"] if kind == "local": from .local import LocalTransport return LocalTransport(cfg, terminal) if kind == "warpgate-ssh": from .warpgate_ssh import WarpgateSshTransport - return WarpgateSshTransport(cfg, terminal) + return WarpgateSshTransport(cfg, terminal, log=log) + if kind == "direct-ssh": + from .direct_ssh import DirectChannel + from .warpgate_ssh import WarpgateSshTransport + primary = WarpgateSshTransport(cfg, terminal, channel=DirectChannel(cfg, terminal, log)) + fallback = None + if not force and t.get("fallback") == "warpgate-ssh": + fallback = WarpgateSshTransport(cfg, terminal, log=log, label="fallback") + return RoutedTransport(terminal, primary, fallback, log) raise TransportError("unknown transport %s" % kind) diff --git a/agentd/transports/direct_ssh.py b/agentd/transports/direct_ssh.py new file mode 100644 index 0000000..3c8865e --- /dev/null +++ b/agentd/transports/direct_ssh.py @@ -0,0 +1,313 @@ +"""direct-ssh channel (PAD-12): OpenSSH from the dispatcher straight to the terminal over Tailscale, no Warpgate. + +- Auth: one dedicated agentd key. Its private text arrives in the env var named by `agentd.ssh_key` (default + AGENTD_SSH_KEY, injected by `agent-access.py exec --secret-env`), is popped from os.environ (children never see + it) and loaded with `ssh-add -` into a per-process ssh-agent on a socket in a private 0700 temp dir. The key is + never written to disk, argv, logs or exceptions. No key / agent failure -> the channel is unavailable + (`Unreached`), so a terminal with `fallback: warpgate-ssh` uses Warpgate instead. +- Host key: pinned per terminal (`ssh.host_key`); a known_hosts file is generated from it (public data) and + checked strictly. A mismatch raises HostKeyError: never a fallback. +- Reuse: ControlMaster=auto / ControlPersist=60 under `agentd.ssh_dir` (root-only 0700), so the second call of a + tick (and the next tick within a minute) skips the handshake. The socket name carries a fingerprint of the pinned + host key: a changed pin never rides an open master that was verified against the old one. +- Payload on stdin: Windows runs a fixed PowerShell bootstrap (-EncodedCommand) that reads the base64 of the real + script from stdin; POSIX runs a fixed bash loader that reads the script from stdin. One call per operation, no + argv size limit.""" +import atexit +import base64 +import hashlib +import os +import re +import shutil +import signal +import subprocess +import tempfile +import time + +from . import NULL_LOG, HostKeyError, TransportError, Unreached +from .local import sh_quote + +CONNECT_TIMEOUT = 8 +CONTROL_PERSIST = 60 +SOCKET_MAX = 104 # sun_path; ssh adds ".<16 random chars>" while it sets up the master socket +CTL_NAME = 1 + 9 + 40 + 17 # "/-<%C: 40 hex>" + that suffix + +# Fixed PowerShell bootstrap: the operation's script arrives on stdin as base64 of its UTF-8 text (ASCII on the +# wire, so the console input code page does not matter). +BOOTSTRAP = ("$ProgressPreference='SilentlyContinue'\n" + "$ErrorActionPreference='Continue'\n" + "[Console]::OutputEncoding=[Text.Encoding]::UTF8\n" + "$s=[Text.Encoding]::UTF8.GetString([Convert]::FromBase64String([Console]::In.ReadToEnd().Trim()))\n" + "& ([ScriptBlock]::Create($s))\n") + + +def encoded_command(script): + """PowerShell -EncodedCommand value: base64 of the UTF-16LE text.""" + return base64.b64encode(script.encode("utf-16-le")).decode("ascii") + + +PS_COMMAND = "powershell.exe -NoProfile -NonInteractive -EncodedCommand " + encoded_command(BOOTSTRAP) +# POSIX loader: the script is read from stdin into a variable (no MAX_ARG_STRLEN), then runs in a subshell with +# stdin = /dev/null. Its exit 255 is reported as 254: 255 stays ssh's own "no session" code. (The PowerShell +# bootstrap exits 0/1 by $?; none of our PowerShell scripts calls `exit`.) +SH_COMMAND = "bash -c " + sh_quote('exec 3<&0 'ok' (the remote command ran, whatever its exit code), 'hostkey', 'unreached' (certainly not run) or + 'unknown' (ssh exit 255 without a recognisable pre-session error: the command may have run). Any stdout proves + that a session ran the command, so it is never 'unreached'.""" + if rc != 255: + return "ok" + if (out or "").strip(): + return "unknown" + if HOSTKEY.search(err or ""): + return "hostkey" + if UNREACHED.search(err or ""): + return "unreached" + return "unknown" + + +# ---------------------------------------------------------------- per-process key agent +def _die_with_parent(): + """preexec: the agent gets SIGTERM when agentd dies, even by SIGKILL (Linux PR_SET_PDEATHSIG).""" + try: + import ctypes + ctypes.CDLL("libc.so.6", use_errno=True).prctl(1, signal.SIGTERM) + except Exception: + pass + + +def normalise_key(text): + """Vault may hand the key with literal `\\n` escapes or without the final newline.""" + if "\n" not in text.strip() and "\\n" in text: + text = text.replace("\\n", "\n") + return text.strip() + "\n" + + +_KEYS = {} # env var name -> key text, taken out of os.environ at CLI start until the agent loads it + + +def take_key(env_name): + """Move the key out of os.environ right away (called by the CLI before anything else runs), so no child -- not + even one started before the first direct call -- can inherit it.""" + if env_name not in _KEYS: + _KEYS[env_name] = os.environ.pop(env_name, None) + else: + os.environ.pop(env_name, None) + + +def key_present(env_name): + return bool((_KEYS.get(env_name) or os.environ.get(env_name) or "").strip()) + + +class KeyAgent: + """ssh-agent of this process holding the agentd key. `error` set = direct-ssh unavailable here.""" + + def __init__(self, env_name): + self.env_name = env_name + self.dir = self.sock = self.pub = self.proc = None + self.error = None + + def start(self): + key = os.environ.pop(self.env_name, None) # popped even if loading fails: children never inherit it + key = _KEYS.pop(self.env_name, None) or key + if not key or not key.strip(): + self.error = "%s is not set (agent-access.py exec --secret-env %s=)" % (self.env_name, self.env_name) + return self + agent_bin, add_bin = shutil.which("ssh-agent"), shutil.which("ssh-add") + if not (agent_bin and add_bin and shutil.which("ssh")): + self.error = "ssh / ssh-agent / ssh-add not installed" + return self + env = {"PATH": os.environ.get("PATH", "/usr/bin:/bin"), "LANG": "C"} + try: + self.dir = tempfile.mkdtemp(prefix="agentd-agent-") # 0700 + self.sock = os.path.join(self.dir, "agent.sock") + atexit.register(self.close) + self.proc = subprocess.Popen([agent_bin, "-D", "-a", self.sock], stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env, + close_fds=True, preexec_fn=_die_with_parent) + end = time.monotonic() + 5 + while not os.path.exists(self.sock): + if self.proc.poll() is not None or time.monotonic() > end: + self.error = "ssh-agent did not start" + return self + time.sleep(0.02) + env["SSH_AUTH_SOCK"] = self.sock + add = subprocess.run([add_bin, "-q", "-"], input=normalise_key(key).encode("utf-8"), env=env, + stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=15) + key = None + if add.returncode != 0: + why = add.stderr.decode("utf-8", "replace").strip().splitlines()[:1] + why = why[0][:120] if why and "-----" not in why[0] else "" + self.error = "ssh-add failed (rc %d) %s" % (add.returncode, why) + return self + listed = subprocess.run([add_bin, "-L"], env=env, capture_output=True, timeout=15) + pub = next((l for l in listed.stdout.decode("ascii", "replace").splitlines() if l.startswith("ssh-")), None) + if not pub: + self.error = "ssh-agent holds no key after ssh-add" + return self + self.pub = os.path.join(self.dir, "agentd.pub") # public half: selects the agent key (IdentitiesOnly) + with open(self.pub, "w", encoding="ascii") as fh: + fh.write(pub + "\n") + except (OSError, subprocess.SubprocessError) as exc: + self.error = "ssh-agent setup failed: %s" % type(exc).__name__ + finally: + key = None + if self.error: + self.close() + return self + + def close(self): + if self.proc and self.proc.poll() is None: + self.proc.terminate() + try: + self.proc.wait(3) + except subprocess.TimeoutExpired: + self.proc.kill() + if self.dir: + shutil.rmtree(self.dir, ignore_errors=True) + + +_AGENTS = {} + + +def key_agent(cfg, log=NULL_LOG): + """The process-wide agent for cfg's key env var, started on first use; raises Unreached when unavailable.""" + name = cfg.ssh_key_env + agent = _AGENTS.get(name) + if agent is None: + agent = _AGENTS[name] = KeyAgent(name).start() + if agent.error: + log.event("direct-ssh unavailable in this process: %s" % agent.error, once="agent:" + name) + raise Unreached("direct-ssh unavailable: %s" % agent.error) + return agent + + +def control_dir(cfg): + """agentd.ssh_dir: ControlMaster sockets + generated known_hosts; created 0700, must be ours.""" + d = cfg.ssh_dir + try: + d.mkdir(parents=True, exist_ok=True) + os.chmod(d, 0o700) + if os.stat(d).st_uid != os.geteuid(): + raise OSError("not owned by this user") + except OSError as exc: + raise Unreached("ssh_dir %s unusable: %s" % (d, exc)) from None + return d + + +# ---------------------------------------------------------------- channel +class DirectChannel: + """One terminal's direct OpenSSH path. ps() / sh() mirror the warpgate channel; payloads go on stdin.""" + stdin = True + + def __init__(self, cfg, terminal, log=None, label="direct"): + self.cfg, self.terminal, self.log, self.via = cfg, terminal, log or NULL_LOG, label + s = cfg.terminals[terminal]["ssh"] + self.host, self.user, self.port = s["host"], s["user"], int(s.get("port", 22)) + self.host_key = " ".join(s["host_key"].split()[:2]) + self.pin = hashlib.sha256(self.host_key.encode("ascii")).hexdigest()[:8] + self.alias = "agentd-" + terminal + self.hostkey_failed = None + self.name = "ssh %s@%s" % (self.user, self.host) + + def known_hosts(self, d): + path = d / ("known_hosts." + self.terminal) + line = "%s %s\n" % (self.alias, self.host_key) + try: + if not path.exists() or path.read_text(encoding="ascii") != line: + tmp = path.with_name("%s.%d.tmp" % (path.name, os.getpid())) # per terminal and process + tmp.write_text(line, encoding="ascii") + os.replace(tmp, path) + except OSError as exc: + raise Unreached("known_hosts %s: %s" % (path, exc)) from None + return path + + def argv(self, agent, remote_command): + d = control_dir(self.cfg) + ctl = str(d / (self.pin + "-%C")) # pin fingerprint: a new pin never reuses a master verified on the old one + if len(str(d)) + CTL_NAME > SOCKET_MAX: + self.log.event("agentd.ssh_dir %s is too long for a control socket; connection reuse off" % d, once="ctl") + mux = ["-o", "ControlMaster=no", "-o", "ControlPath=none"] + else: + mux = ["-o", "ControlMaster=auto", "-o", "ControlPersist=%d" % CONTROL_PERSIST, "-o", "ControlPath=" + ctl] + return (["ssh", "-F", "/dev/null", "-T", "-p", str(self.port), "-l", self.user, + "-o", "BatchMode=yes", "-o", "IdentitiesOnly=yes", + "-o", "IdentityAgent=" + agent.sock, "-o", "IdentityFile=" + agent.pub, + "-o", "PreferredAuthentications=publickey", "-o", "PasswordAuthentication=no", + "-o", "KbdInteractiveAuthentication=no", + "-o", "StrictHostKeyChecking=yes", "-o", "UserKnownHostsFile=" + str(self.known_hosts(d)), + "-o", "GlobalKnownHostsFile=/dev/null", "-o", "HostKeyAlias=" + self.alias, + "-o", "HostKeyAlgorithms=" + self.host_key.split()[0], "-o", "UpdateHostKeys=no", + "-o", "ConnectTimeout=%d" % CONNECT_TIMEOUT, "-o", "ServerAliveInterval=15", + "-o", "ServerAliveCountMax=3", "-o", "ForwardAgent=no", "-o", "ForwardX11=no", + "-o", "ClearAllForwardings=yes", "-o", "LogLevel=ERROR", "-o", "EscapeChar=none"] + + mux + ["--", self.host, remote_command]) + + def run(self, remote_command, payload, timeout, op): + """-> (rc, stdout, stderr) of a call whose command ran; raises HostKeyError / Unreached / TransportError.""" + if self.hostkey_failed: + raise HostKeyError(self.hostkey_failed) + agent = key_agent(self.cfg, self.log) + argv = self.argv(agent, remote_command) + env = self.cfg.scrubbed_env() + env.pop("SSH_AUTH_SOCK", None) + t0 = time.monotonic() + ms = lambda: (time.monotonic() - t0) * 1000 # noqa: E731 + # output goes to files, not pipes: a ControlPersist master forked by this call must never hold our pipes open + with tempfile.TemporaryFile() as fo, tempfile.TemporaryFile() as fe: + try: + p = subprocess.run(argv, input=payload, stdout=fo, stderr=fe, env=env, timeout=timeout) + except subprocess.TimeoutExpired: + self.log.call(self.terminal, self.via, op, ms(), "timeout") + raise TransportError("%s timeout after %ss" % (self.name, timeout)) from None # may have run + except OSError as exc: + self.log.call(self.terminal, self.via, op, ms(), "launch-failed") + raise Unreached("%s: ssh launch failed: %s" % (self.name, type(exc).__name__)) from None + fo.seek(0) + fe.seek(0) + dec = lambda b: b.decode("utf-8", "replace").replace("\r", "") # noqa: E731 + out, err = dec(fo.read()), dec(fe.read()) + kind = classify(p.returncode, err, out) + self.log.call(self.terminal, self.via, op, ms(), p.returncode, "" if kind == "ok" else kind) + if kind == "hostkey": + self.hostkey_failed = ("%s:%d host key verification FAILED (pinned %s...; other key or no such key type " + "offered): not falling back; check terminals.%s.ssh.host_key out of band" + % (self.host, self.port, self.host_key[:40], self.terminal)) + self.log.event("SECURITY %s: %s" % (self.terminal, self.hostkey_failed), once="hostkey:" + self.terminal) + raise HostKeyError(self.hostkey_failed) + if kind == "unreached": + raise Unreached("%s exit 255: %s" % (self.name, err.strip()[-300:])) + if kind == "unknown": + raise TransportError("%s exit 255: %s" % (self.name, err.strip()[-300:] or "(no message)")) + return p.returncode, out, err + + def ps(self, script, timeout, op=None): + rc, out, err = self.run(PS_COMMAND, ps_payload(script), timeout, op) + if rc != 0 and not out.strip(): + raise TransportError("%s exit %s: %s" % (self.name, rc, err.strip()[-300:])) + return out + + def sh(self, script, timeout, op=None): + return self.run(SH_COMMAND, script.encode("utf-8"), timeout, op) diff --git a/agentd/transports/local.py b/agentd/transports/local.py index 3a328c5..761d9ea 100644 --- a/agentd/transports/local.py +++ b/agentd/transports/local.py @@ -91,5 +91,5 @@ class LocalTransport: pass def process_count(self, name): - out = subprocess.run(["pgrep", "-c", "-x", name], capture_output=True, text=True) + out = subprocess.run(["pgrep", "-c", "-x", name], capture_output=True, text=True, env=self.cfg.scrubbed_env()) return int(out.stdout.strip() or 0) diff --git a/agentd/transports/warpgate_ssh.py b/agentd/transports/warpgate_ssh.py index 2114536..1de0ffc 100644 --- a/agentd/transports/warpgate_ssh.py +++ b/agentd/transports/warpgate_ssh.py @@ -1,17 +1,25 @@ -"""Terminals behind Warpgate, reached through the server-access relay client: +"""Remote terminals (Windows and Linux/macOS) and the Warpgate channel. + +The transport classes build the scripts; a *channel* carries them (design §9): +- WarpgateChannel (`transport: warpgate-ssh`, also the direct-ssh fallback): the server-access relay client agent-access.py --config run --target --timeout N -- -The client shlex-joins argv and the target account's login shell runs it (cmd.exe on Windows); stdin is -/dev/null, so everything travels in argv. `relay()` is the single place that builds that call. + The client shlex-joins argv and the target account's login shell runs it (cmd.exe on Windows); stdin is + /dev/null (agent-access.py run_remote uses stdin=DEVNULL), so everything travels in argv. `relay()` is the single + place that builds that call. +- DirectChannel (`transport: direct-ssh`, agentd/transports/direct_ssh.py): OpenSSH over Tailscale; the script goes + on stdin, so every operation (including a launch with its prompt) is one call. -- os linux / darwin -> PosixWarpgateTransport: every operation is one `bash -c '