From 00d42267e55e4147f56fe9e643ce9a62aafd3c52 Mon Sep 17 00:00:00 2001 From: "jj.wang" Date: Wed, 21 Oct 2026 07:10:41 +0000 Subject: [PATCH] Harden transports and state files - Executors and relay helpers get an environment without any configured Plane token / webhook secret (design 8: executors hold no Plane credentials). - Windows runner: Set-Location -LiteralPath, and a failed cd writes exit=97 instead of running the agent in the wrong directory. - schtasks launch: escape apostrophes and pass the script path as \"...\" so paths with spaces survive PowerShell 5.1 native quoting. - State and heartbeat files are written temp + fsync + rename. - Tests for the above plus codex run/resume command shapes and unique run dirs. --- agentd/config.py | 11 +++ agentd/store.py | 18 +++-- agentd/transports/local.py | 2 + agentd/transports/warpgate_ssh.py | 24 ++++-- tests/test_transports.py | 122 ++++++++++++++++++++++++++++++ 5 files changed, 163 insertions(+), 14 deletions(-) create mode 100644 tests/test_transports.py diff --git a/agentd/config.py b/agentd/config.py index 0ff4c59..652b638 100644 --- a/agentd/config.py +++ b/agentd/config.py @@ -157,6 +157,17 @@ class Config: def token(self, terminal): return os.environ.get(self.token_env(terminal)) + def secret_keys(self): + """Every env var name that carries a secret (Plane tokens, webhook secrets): never handed to executors.""" + keys = {t["token_key"] for t in self.terminals.values()} + keys |= {p["webhook_secret_key"] for p in self.projects.values() if p.get("webhook_secret_key")} + return keys + + def scrubbed_env(self): + """os.environ without the secrets above, for every child process agentd starts.""" + drop = self.secret_keys() + 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.""" keys = sorted({t["token_key"] for t in self.terminals.values()}) diff --git a/agentd/store.py b/agentd/store.py index 2772df8..2a0a565 100644 --- a/agentd/store.py +++ b/agentd/store.py @@ -31,11 +31,18 @@ class Store: data.setdefault("paused_groups", {}) return data - def save(self, data): + def _atomic_write(self, path, obj): + """Write temp + fsync + rename, so a reader (or a crash) never sees half a file.""" self.dir.mkdir(parents=True, exist_ok=True) - tmp = self.path.with_suffix(".tmp") - tmp.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8") - os.replace(tmp, self.path) + tmp = path.with_suffix(".tmp") + with open(tmp, "w", encoding="utf-8") as fh: + fh.write(json.dumps(obj, ensure_ascii=False, indent=1)) + fh.flush() + os.fsync(fh.fileno()) + os.replace(tmp, path) + + def save(self, data): + self._atomic_write(self.path, data) @contextlib.contextmanager def lock(self): @@ -48,9 +55,8 @@ class Store: yield def heartbeat(self, info): - self.dir.mkdir(parents=True, exist_ok=True) info = dict(info, ts=time.time(), at=time.strftime("%Y-%m-%dT%H:%M:%S")) - (self.dir / "heartbeat.json").write_text(json.dumps(info, ensure_ascii=False, indent=1), encoding="utf-8") + self._atomic_write(self.dir / "heartbeat.json", info) def log(self, msg): self.dir.mkdir(parents=True, exist_ok=True) diff --git a/agentd/transports/local.py b/agentd/transports/local.py index 9009abd..5d49830 100644 --- a/agentd/transports/local.py +++ b/agentd/transports/local.py @@ -59,7 +59,9 @@ class LocalTransport: def launch(self, task, script_path, run_dir): # `setsid -f` double-forks: the runner leads its own session (pgid == pid, see kill) and is reparented # to init, so a long-running `tick --loop` never accumulates zombies. + # The executor must not hold Plane credentials (design ยง8): drop every configured secret from its env. subprocess.run(["setsid", "-f", "bash", script_path], cwd=run_dir, check=True, close_fds=True, + env=self.cfg.scrubbed_env(), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def kill(self, task, run_dir): diff --git a/agentd/transports/warpgate_ssh.py b/agentd/transports/warpgate_ssh.py index e1fedcb..82ae0f3 100644 --- a/agentd/transports/warpgate_ssh.py +++ b/agentd/transports/warpgate_ssh.py @@ -15,7 +15,8 @@ from . import TransportError RUNNER = r"""$ErrorActionPreference = 'Continue' [Console]::OutputEncoding = [Text.Encoding]::UTF8 $OutputEncoding = [Text.Encoding]::UTF8 -Set-Location {cwd} +try {{ Set-Location -LiteralPath {cwd} -ErrorAction Stop }} catch {{ + Set-Content -Path '{d}\exit.txt' -Value ('exit=97 at ' + (Get-Date).ToString('s') + ' (cd failed)'); exit 97 }} Set-Content -Path '{d}\pid.txt' -Value $PID Set-Content -Path '{d}\started.txt' -Value (Get-Date).ToString('s') $prompt = Get-Content -Raw -Encoding UTF8 '{d}\prompt.md' @@ -53,7 +54,7 @@ class WarpgateSshTransport: # ---------------------------------------------------------------- relay primitives def ps(self, script, timeout=150): - env = dict(os.environ, WIN10_TIMEOUT=str(timeout)) + env = dict(self.cfg.scrubbed_env(), WIN10_TIMEOUT=str(timeout)) try: out = subprocess.run([self.ps_script, script], capture_output=True, text=True, timeout=timeout + 60, env=env) except subprocess.TimeoutExpired: @@ -64,7 +65,8 @@ class WarpgateSshTransport: def put(self, local, remote): sha = hashlib.sha256(open(local, "rb").read()).hexdigest() - out = subprocess.run([self.put_script, local, remote], capture_output=True, text=True, timeout=600) + out = subprocess.run([self.put_script, local, remote], capture_output=True, text=True, timeout=600, + env=self.cfg.scrubbed_env()) if ("remote sha256 " + sha) not in out.stdout: raise TransportError("upload of %s failed: %s" % (remote, (out.stdout + out.stderr).strip()[-300:])) @@ -113,12 +115,18 @@ class WarpgateSshTransport: def runner(self, run_dir, cwd, command, stdout_name): return "run.ps1", RUNNER.format(cwd=self.quote(cwd), d=run_dir.replace("'", "''"), command=command, stdout=stdout_name) + def launch_script(self, task, script_path): + """PowerShell that (re)creates and runs the one-shot task. Values sit in '...' literals (`'` doubled). The + script path is wrapped in \\" so that Windows PowerShell 5.1 (legacy native quoting) hands schtasks + /TR `... -File ""` intact even when the path contains spaces.""" + return ("schtasks /Delete /F /TN '%(t)s' 2>$null | Out-Null\n" + "$tr = 'powershell.exe -NoProfile -NonInteractive -ExecutionPolicy Bypass -WindowStyle Hidden -File \\\"%(s)s\\\"'\n" + "schtasks /Create /F /TN '%(t)s' /SC ONCE /ST 00:00 /TR $tr | Out-Null\n" + "schtasks /Run /TN '%(t)s' | Out-Null\n" + "'LAUNCHED=' + $LASTEXITCODE" % dict(t=task.replace("'", "''"), s=script_path.replace("'", "''"))) + def launch(self, task, script_path, run_dir): - out = self.ps("schtasks /Delete /F /TN '%(t)s' 2>$null | Out-Null\n" - "$tr = 'powershell.exe -NoProfile -NonInteractive -ExecutionPolicy Bypass -WindowStyle Hidden -File \"%(s)s\"'\n" - "schtasks /Create /F /TN '%(t)s' /SC ONCE /ST 00:00 /TR $tr | Out-Null\n" - "schtasks /Run /TN '%(t)s' | Out-Null\n" - "'LAUNCHED=' + $LASTEXITCODE" % dict(t=task, s=script_path)) + out = self.ps(self.launch_script(task, script_path)) if "LAUNCHED=0" not in out: raise TransportError("schtasks launch of %s failed: %s" % (task, out.strip()[-300:])) diff --git a/tests/test_transports.py b/tests/test_transports.py new file mode 100644 index 0000000..d119626 --- /dev/null +++ b/tests/test_transports.py @@ -0,0 +1,122 @@ +"""Transport hardening: secrets never reach executors, Windows script quoting, command shapes, atomic state files.""" +import json +import os +import tempfile +import time +import unittest +from pathlib import Path +from unittest import mock + +from agentd import ROOT, config +from agentd.config import Config +from agentd.executors import Run, make_executor +from agentd.store import Store +from agentd.transports.local import LocalTransport +from agentd.transports.warpgate_ssh import WarpgateSshTransport +from tests.fakes import config_data + + +class RecordingWin(WarpgateSshTransport): + """Windows transport with the relay replaced by an in-memory recorder.""" + def __init__(self, cfg, terminal): + super().__init__(cfg, terminal) + self.files, self.launched = {}, [] + + def mkdir(self, path, clean=False): + pass + + def write_text(self, path, text): + self.files[path] = text + + def launch(self, task, script_path, run_dir): + self.launched.append((task, script_path, run_dir)) + + +class LocalSecretsTest(unittest.TestCase): + def test_executor_env_has_no_plane_tokens(self): + with tempfile.TemporaryDirectory() as tmp: + data = config_data() + data["agentd"]["state_dir"] = str(Path(tmp) / "state") + data["terminals"]["oregon"]["repo"] = tmp + data["projects"]["GWENT"]["webhook_secret_key"] = "HOOK_GW" + cfg = Config(data) + self.assertEqual(cfg.secret_keys(), {"TOK_OR", "TOK_W10", "HOOK_GW"}) + t = LocalTransport(cfg, "oregon") + ex = make_executor(cfg, "oregon", "shell", t) + script = 'echo "or=${TOK_OR:-none} w10=${TOK_W10:-none} hook=${HOOK_GW:-none} keep=${KEEP_ME:-none}"' + with mock.patch.dict(os.environ, {"TOK_OR": "s1", "TOK_W10": "s2", "HOOK_GW": "s3", "KEEP_ME": "yes"}): + h = ex.start(Run("GW-9-1", tmp, "", script=script)) + end = time.time() + 10 + while ex.poll(h) != "finished" and time.time() < end: + time.sleep(0.05) + self.assertEqual(ex.collect(h).last_message, "or=none w10=none hook=none keep=yes") + + +class WindowsScriptTest(unittest.TestCase): + def setUp(self): + self.cfg = config.load(ROOT / "config" / "agentd.example.yaml") + self.t = RecordingWin(self.cfg, "win10") + self.ex = make_executor(self.cfg, "win10", "codex", self.t) + + def test_codex_run_and_resume_commands(self): + run_dir = self.t.join(self.ex.run_root(), "GW-5-1") + fresh = self.ex.command(Run("GW-5-1", self.t.repo, "p", "gpt-6-astra", "high"), run_dir) + self.assertTrue(fresh.startswith("& 'C:\\Users\\Administrator\\AppData\\Local\\Programs\\OpenAI\\Codex\\bin\\codex.exe' exec --json")) + self.assertIn("-C '%s' -s workspace-write" % self.t.repo, fresh) + self.assertIn("-o '%s\\last.md'" % run_dir, fresh) + resumed = self.ex.command(Run("GW-5-2", self.t.repo, "p", "gpt-6-astra", "high", thread_id="th-9"), run_dir) + self.assertIn(" exec resume th-9 --json -m gpt-6-astra", resumed) + self.assertNotIn(" -C ", resumed) # `codex exec resume` rejects -C/-s; the runner already cd's into the repo + self.assertNotIn(" -s ", resumed) + tiered = self.ex.command(Run("GW-5-3", self.t.repo, "p", "gpt-6.1-sol", "medium", tier="fast"), run_dir) + self.assertIn("-c 'service_tier=\"fast\"'", tiered) + + def test_run_dirs_are_unique_per_run(self): + a = self.ex.start(Run("GW-5-1", self.t.repo, "p", "m", "e")) + b = self.ex.start(Run("GW-5-2", self.t.repo, "p", "m", "e")) + self.assertNotEqual(a["run_dir"], b["run_dir"]) + self.assertEqual([x[0] for x in self.t.launched], ["agentd-GW-5-1", "agentd-GW-5-2"]) + self.assertTrue(a["run_dir"].endswith("\\runtime\\agentd\\GW-5-1")) + + def test_runner_quotes_paths_with_spaces_and_apostrophes(self): + _, script = self.t.runner(r"C:\a b\O'Neil\run", r"C:\my repo\O'Neil", "& 'C:\\x\\codex.exe' exec -", "events.jsonl") + self.assertIn(r"Set-Location -LiteralPath 'C:\my repo\O''Neil' -ErrorAction Stop", script) + self.assertIn(r"'C:\a b\O''Neil\run\prompt.md'", script) + self.assertIn(r"'C:\a b\O''Neil\run\exit.txt' -Value ('exit=97", script) # cd failure still ends the run + self.assertIn(r"Add-Content -Path 'C:\a b\O''Neil\run\events.jsonl'", script) + + def test_launch_script_quoting(self): + s = self.t.launch_script("agentd-GW-1-1", r"C:\My Docs\O'Neil\run.ps1") + self.assertIn(r"-File \"C:\My Docs\O''Neil\run.ps1\"'", s) + self.assertIn("/TN 'agentd-GW-1-1'", s) + + def test_relay_env_has_no_plane_tokens(self): + seen = {} + + def fake_run(args, **kw): + seen.update(kw["env"]) + return mock.Mock(returncode=0, stdout="X=ok", stderr="") + with mock.patch.dict(os.environ, {"PLANE_WINDOW10_API_TOKEN": "s"}), \ + mock.patch("agentd.transports.warpgate_ssh.subprocess.run", fake_run): + WarpgateSshTransport(self.cfg, "win10").ps("'X=ok'") + self.assertNotIn("PLANE_WINDOW10_API_TOKEN", seen) + self.assertIn("WIN10_TIMEOUT", seen) + + +class StoreTest(unittest.TestCase): + def test_save_is_atomic_and_heartbeat_is_complete_json(self): + with tempfile.TemporaryDirectory() as tmp: + st = Store(tmp, echo=False) + st.save({"projects": {"A": {"issues": {}}}}) + with mock.patch("agentd.store.os.replace", side_effect=OSError("disk full")): + with self.assertRaises(OSError): + st.save({"projects": {"B": {}}}) + self.assertIn("A", st.load()["projects"]) # old state survives a failed write + st.heartbeat({"errors": [], "inflight": []}) + hb = json.loads((Path(tmp) / "heartbeat.json").read_text()) + self.assertEqual(hb["errors"], []) + self.assertIn("ts", hb) + + +if __name__ == "__main__": + unittest.main()