fix(ga_ultraplan): require Bearer token on /exec (closes #729)

assets/ga_ultraplan.py

@@ -1,7 +1,7 @@
 from contextlib import contextmanager, redirect_stdout, redirect_stderr
 from concurrent.futures import ThreadPoolExecutor
 from time import time, sleep
-import html, io, json, os, re, subprocess, sys, tempfile, threading, traceback, urllib.request, webbrowser
+import html, io, json, os, re, secrets, subprocess, sys, tempfile, threading, traceback, urllib.request, webbrowser
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 __all__ = ["plan", "phase", "parallel", "mapchain"]
@@ -12,6 +12,8 @@
 _TASK_SLUG = "task"; _FUNC_SEQ = 0; _PLANNED = False; _SESSION = None; _sessions = {}
 _RUN_DIR = os.path.abspath(os.environ.get("GA_ULTRAPLAN_RUNDIR", os.path.join(_ROOT, "temp", "ultraplan_default")))
 os.makedirs(_RUN_DIR, exist_ok=True)
+_TOKEN = os.environ.get("GA_ULTRAPLAN_TOKEN") or ""
+_TOKEN_FILE = os.path.join(_RUN_DIR, ".ultraplan_token")
 
 def _bind(rundir):
     global _SESSION, _RUN_DIR, _phases, _phase_stack, _tasks, _current, _events, _FUNC_SEQ, _TASK_SLUG
@@ -71,6 +73,12 @@ def do_GET(self):
     def do_POST(self):
         global _TASK_SLUG, _PLANNED
         if self.path != "/exec": self.send_response(404); self.end_headers(); return
+        if _TOKEN:
+            got = ""
+            for k, v in self.headers.items():
+                if k.lower() == "authorization": got = v; break
+            if not got.startswith("Bearer ") or not secrets.compare_digest(got[7:], _TOKEN):
+                self.send_response(401); self.end_headers(); return
         n = int(self.headers.get("Content-Length", "0")); req = json.loads(self.rfile.read(n).decode("utf-8"))
         out = io.StringIO(); err = io.StringIO(); rc = 0
         with _exec_lock, redirect_stdout(out), redirect_stderr(err):
@@ -91,8 +99,14 @@ def do_POST(self):
     def log_message(self, *a): pass
 
 def _serve_daemon():
-    global _srv
+    global _srv, _TOKEN
     sys.modules.setdefault("assets.ga_ultraplan", sys.modules[__name__])
+    if not _TOKEN:
+        _TOKEN = secrets.token_urlsafe(32)
+        os.makedirs(_RUN_DIR, exist_ok=True)
+        with open(_TOKEN_FILE, "w") as f: f.write(_TOKEN)
+        try: os.chmod(_TOKEN_FILE, 0o600)
+        except OSError: pass
     _srv = ThreadingHTTPServer(("127.0.0.1", _PORT), _H); _srv.timeout = 60; url = f"http://127.0.0.1:{_PORT}/"
     print(f"[ultraplan] {url}", flush=True)
     if os.environ.get("GA_ULTRAPLAN_BROWSER") != "0": webbrowser.open(url)
@@ -105,7 +119,13 @@ def _ping():
 def _show():
     if os.environ.get("GA_ULTRAPLAN_DAEMON") == "1" or os.environ.get("GA_ULTRAPLAN_HTML") == "0": return
     if not _ping():
-        subprocess.Popen([sys.executable, __file__, "--daemon"], cwd=_ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env={**os.environ, "GA_ULTRAPLAN_DAEMON":"1"})
+        tok = secrets.token_urlsafe(32) if not _TOKEN else _TOKEN
+        if not _TOKEN:
+            os.makedirs(_RUN_DIR, exist_ok=True)
+            with open(_TOKEN_FILE, "w") as f: f.write(tok)
+            try: os.chmod(_TOKEN_FILE, 0o600)
+            except OSError: pass
+        subprocess.Popen([sys.executable, __file__, "--daemon"], cwd=_ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env={**os.environ, "GA_ULTRAPLAN_DAEMON":"1", "GA_ULTRAPLAN_TOKEN": tok})
         for _ in range(20):
             if _ping(): break
             sleep(0.25)
@@ -117,7 +137,10 @@ def plan(rundir):
     if os.environ.get("GA_ULTRAPLAN_DAEMON") == "1": return
     _show(); path = os.path.abspath(sys.argv[0]); code = open(path, encoding="utf-8").read()
     data = json.dumps({"path": path, "cwd": os.getcwd(), "rundir": _RUN_DIR, "task": _task_slug(path), "code": code}).encode("utf-8")
-    r = urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{_PORT}/exec", data=data, headers={"Content-Type":"application/json"}), timeout=None)
+    tok = os.environ.get("GA_ULTRAPLAN_TOKEN") or (open(_TOKEN_FILE).read().strip() if os.path.exists(_TOKEN_FILE) else "")
+    headers = {"Content-Type": "application/json"}
+    if tok: headers["Authorization"] = "Bearer " + tok
+    r = urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{_PORT}/exec", data=data, headers=headers), timeout=None)
     resp = json.loads(r.read().decode("utf-8")); sys.stdout.write(resp.get("stdout", "")); sys.stderr.write(resp.get("stderr", "")); sys.exit(resp.get("returncode", 1))
 
 @contextmanager

test_issue_729_ultraplan_exec_auth.py

@@ -0,0 +1,105 @@
+"""
+Verification for fix/issue-729: require Authorization: Bearer <token> on /exec.
+
+Before fix: ga_ultraplan.py do_POST accepted any POST to /exec with no
+authentication — any local process that could reach 127.0.0.1:47831
+could execute arbitrary Python in the daemon process.
+
+After fix: a per-start token is minted in _serve_daemon (and propagated
+through _show when auto-spawning). do_POST now requires
+`Authorization: Bearer <token>` and uses secrets.compare_digest to
+verify it. Without the header (or with a wrong header) the daemon
+returns 401 and never enters the exec body.
+
+This test exercises the daemon end-to-end in two ways:
+  1. POST without Authorization header → 401 (the bug, now fixed).
+  2. POST with the correct Bearer token → 200 with our canary file
+     proving the request reached the exec sink.
+  3. POST with a wrong Bearer token → 401 (compare_digest rejects).
+"""
+import json, os, secrets, socket, subprocess, sys, tempfile, threading, time, urllib.request
+from pathlib import Path
+
+sys.path.insert(0, "/root/repos/GenericAgent")
+import importlib.util, sys
+spec = importlib.util.spec_from_file_location("ga_ultraplan", "/root/repos/GenericAgent/assets/ga_ultraplan.py")
+ga_ultraplan = importlib.util.module_from_spec(spec)
+sys.modules["ga_ultraplan"] = ga_ultraplan
+spec.loader.exec_module(ga_ultraplan)
+
+import socket
+# pick a free port
+with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as _s:
+    _s.bind(("127.0.0.1", 0))
+    PORT = _s.getsockname()[1]
+
+RUNDIR = tempfile.mkdtemp(prefix="ga-ultraplan-test-")
+os.makedirs(RUNDIR, exist_ok=True)
+
+# Configure the module for this test: distinct port + distinct run dir
+ga_ultraplan._PORT = PORT
+ga_ultraplan._RUN_DIR = RUNDIR
+ga_ultraplan._TOKEN_FILE = os.path.join(RUNDIR, ".ultraplan_token")
+ga_ultraplan._TOKEN = ""
+
+# Start the daemon in-process via _serve_daemon in a thread so the test
+# can drive it without launching a subprocess.
+daemon_thread = threading.Thread(target=ga_ultraplan._serve_daemon, daemon=True)
+daemon_thread.start()
+for _ in range(40):
+    try:
+        urllib.request.urlopen(f"http://127.0.0.1:{PORT}/", timeout=0.5).read(1)
+        break
+    except Exception:
+        time.sleep(0.1)
+
+token = open(ga_ultraplan._TOKEN_FILE).read().strip()
+assert len(token) > 20, f"token too short: {token!r}"
+
+
+def _post(headers):
+    nonce = "CANARY_" + secrets.token_hex(8)
+    marker = os.path.join(tempfile.gettempdir(), nonce)
+    code = f"from pathlib import Path; Path({marker!r}).write_text({nonce!r})\n"
+    body = json.dumps({"rundir": RUNDIR, "code": code, "path": "<test>"}).encode()
+    h = {"Content-Type": "application/json"}
+    h.update(headers)
+    req = urllib.request.Request(
+        f"http://127.0.0.1:{PORT}/exec",
+        data=body, headers=h, method="POST",
+    )
+    return urllib.request.urlopen(req), nonce, marker
+
+
+# Case 1: no Authorization header → must fail with 401.
+try:
+    resp, _, _ = _post({})
+    raise AssertionError(f"expected 401, got {resp.status}")
+except urllib.error.HTTPError as e:
+    assert e.code == 401, f"expected 401, got {e.code}"
+print("✓ no Authorization header → 401")
+
+
+# Case 2: wrong Bearer token → must fail with 401.
+try:
+    resp, _, _ = _post({"Authorization": "Bearer wrongtoken1234567"})
+    raise AssertionError(f"expected 401, got {resp.status}")
+except urllib.error.HTTPError as e:
+    assert e.code == 401, f"expected 401, got {e.code}"
+print("✓ wrong Bearer token → 401")
+
+
+# Case 3: correct Bearer token → 200, canary file written.
+resp, nonce, marker = _post({"Authorization": "Bearer " + token})
+assert resp.status == 200, f"expected 200, got {resp.status}"
+assert Path(marker).exists(), f"canary marker {marker} missing"
+assert Path(marker).read_text() == nonce, f"canary content mismatch"
+print("✓ correct Bearer token → 200, exec sink reached, canary written")
+
+
+# Cleanup: shutdown the server so the daemon thread can exit.
+try: ga_ultraplan._srv.shutdown()
+except Exception: pass
+time.sleep(0.2)
+print("ALL CHECKS PASSED")
+import os; os._exit(0)