diff --git a/skillopt_sleep/__main__.py b/skillopt_sleep/__main__.py index d6fac84f..49625d0f 100644 --- a/skillopt_sleep/__main__.py +++ b/skillopt_sleep/__main__.py @@ -5,6 +5,7 @@ python -m skillopt_sleep status # show state + latest staged proposal python -m skillopt_sleep adopt # apply the latest staged proposal (with backup) python -m skillopt_sleep harvest # just print what would be mined (debug) + python -m skillopt_sleep dashboard # local control panel on 127.0.0.1 Common flags: --project PATH project to evolve (default: cwd) @@ -531,6 +532,12 @@ def main(argv=None) -> int: p_unsched = sub.add_parser("unschedule", help="remove the nightly cron entry") _add_common(p_unsched) p_unsched.add_argument("--all", action="store_true", help="remove all managed entries") + p_dash = sub.add_parser( + "dashboard", help="serve the local control-panel dashboard (127.0.0.1)") + _add_common(p_dash) + p_dash.add_argument("--port", type=int, default=8321) + p_dash.add_argument("--no-browser", action="store_true", + help="don't auto-open the browser") args = parser.parse_args(argv) if args.cmd == "run": @@ -543,6 +550,10 @@ def main(argv=None) -> int: return cmd_adopt(args) if args.cmd == "harvest": return cmd_harvest(args) + if args.cmd == "dashboard": + from skillopt_sleep.dashboard import serve + return serve(project=args.project or os.getcwd(), port=args.port, + open_browser=not args.no_browser) if args.cmd == "schedule": return cmd_schedule(args) if args.cmd == "unschedule": diff --git a/skillopt_sleep/dashboard.html b/skillopt_sleep/dashboard.html new file mode 100644 index 00000000..99062d0f --- /dev/null +++ b/skillopt_sleep/dashboard.html @@ -0,0 +1,607 @@ + + + + + +SkillOpt-Sleep — Control Panel + + + +
+ +
+
idle
+ + +
+ +
+ + +
+ +

Pipeline — one sleep night, left to right

+
+ + +
+
⚙Models & Knobs + + + saved ✓
+
+
+
TARGET the model whose skill is being evolved — runs every attempt (replay rollout)
+
OPTIMIZER the model that mines tasks, judges rubrics and writes edits (reflect)
+
Leave the optimizer/target overrides empty to have the base backend play all roles. Changes apply from the next run. Prompt edits (below, per stage) apply to the very next model call.
+
+
+
+
+ +
+ + +
+
▤Run log
+ 
(no run this session)
+
+
+
+ + + + diff --git a/skillopt_sleep/dashboard.py b/skillopt_sleep/dashboard.py new file mode 100644 index 00000000..6c2be3ef --- /dev/null +++ b/skillopt_sleep/dashboard.py @@ -0,0 +1,545 @@ +"""SkillOpt-Sleep — local control-panel dashboard. + +A zero-dependency (stdlib ``http.server``) web UI over one project's sleep +pipeline. It is arranged to mirror the actual data flow — + + transcripts -> harvest -> mine -> split -> replay -> reflect -> gate + -> stage -> adopt + +— and for every stage shows: which agent role runs it (target / optimizer / +pure code), which model that role resolves to, the exact prompt template it +receives (editable, live), and the selected night's evidence events for that +stage (from ``evidence.jsonl``). Config changes are written to the user +config file and apply from the next run; prompt overrides apply to the very +next model call (the registry re-reads its override file on mtime change). + +Serves on 127.0.0.1 only. + + python -m skillopt_sleep dashboard [--project DIR] [--port N] + +Binding to loopback is not by itself an authorization boundary: any page in +the user's browser can send cross-origin requests to 127.0.0.1, and a hostile +name that resolves to loopback (DNS rebinding) can make them look same-origin. +Since these endpoints start real runs, adopt artifacts, and rewrite config and +prompts, every request is checked before it reaches a handler — see +``_authorize``. No CORS headers are ever emitted, so a foreign origin cannot +read responses either. +""" +from __future__ import annotations + +import hmac +import json +import os +import secrets +import subprocess +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Dict, List, Optional, Tuple +from urllib.parse import unquote + +from skillopt_sleep import prompts as prompt_registry +from skillopt_sleep.config import DEFAULTS, HOME_STATE_DIR, load_config +from skillopt_sleep.evidence import read_events +from skillopt_sleep.staging import adopt as adopt_staging +from skillopt_sleep.staging import staging_root + +_HTML_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dashboard.html") + +# The served HTML carries the capability token in place of this placeholder, +# so the token never travels in a URL (referrers, shell history, server logs). +_TOKEN_PLACEHOLDER = "__SKILLOPT_DASHBOARD_TOKEN__" +# A custom header cannot be set by a cross-origin HTML form, so requiring it +# forces a preflight that this server deliberately does not answer. +_TOKEN_HEADER = "X-SkillOpt-Dashboard-Token" +_MAX_BODY_BYTES = 1 << 20 # 1 MiB: these payloads are prompts and config, not uploads + +# Hostnames that really mean "this machine". A rebound name like evil.com +# resolving to 127.0.0.1 arrives with its own Host and is rejected. +_LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1", "[::1]"}) + +# Config keys the dashboard may write (safety allowlist: everything else in +# the user config file is preserved untouched). +_EDITABLE_KEYS = { + "backend", "model", + "optimizer_backend", "optimizer_model", "target_backend", "target_model", + "azure_endpoint", + "gate_mode", "gate_metric", "gate_mixed_weight", + "edit_budget", "holdout_fraction", "lookback_hours", + "max_tasks_per_night", "max_sessions_per_night", "max_tokens_per_night", + "dream_rollouts", "dream_factor", "recall_k", + "evolve_skill", "evolve_memory", "llm_mine", "target_skill_path", + "preferences", "evidence_log", "evidence_max_chars", "auto_adopt", + "transcript_source", +} + + +def _read_json(path: str) -> Optional[Any]: + try: + with open(path, encoding="utf-8") as f: + return json.load(f) + except Exception: + return None + + +def _read_text(path: str, limit: int = 200_000) -> str: + try: + with open(path, encoding="utf-8") as f: + return f.read(limit) + except Exception: + return "" + + +def _user_config_file() -> str: + return os.path.join(HOME_STATE_DIR, "config.json") + + +def _coerce_config_value(key: str, value: Any) -> Any: + """Coerce a submitted value to the type of the built-in default. + + ``load_config`` does not type-coerce, so a string where a number belongs + reaches the arithmetic downstream (``edit_budget: "4"``). The default's + type is the schema — this is the only place that knows it, and it is + enforced server-side because the client is not the only possible caller. + """ + default = DEFAULTS.get(key) + if isinstance(default, bool): # before int: bool is a subclass of int + if isinstance(value, bool): + return value + text = str(value).strip().lower() + if text in {"true", "1", "yes", "on"}: + return True + if text in {"false", "0", "no", "off"}: + return False + raise ValueError(f"{key} must be a boolean") + if isinstance(default, int): + try: + return int(str(value).strip()) + except (TypeError, ValueError): + raise ValueError(f"{key} must be an integer") from None + if isinstance(default, float): + try: + return float(str(value).strip()) + except (TypeError, ValueError): + raise ValueError(f"{key} must be a number") from None + if isinstance(default, str): + if isinstance(value, (dict, list)): + raise ValueError(f"{key} must be a string") + return str(value) + return value + + +def _write_config(updates: Dict[str, Any]) -> Dict[str, Any]: + """Merge validated updates into the user config file. + + Raises ValueError if any submitted value cannot be coerced; nothing is + written in that case, so a bad field cannot half-apply a form. + """ + path = _user_config_file() + current = _read_json(path) or {} + accepted: Dict[str, Any] = {} + removed = [] + for k, v in updates.items(): + if k not in _EDITABLE_KEYS: + continue + if v is None or v == "": + removed.append(k) # empty resets the key to the built-in default + else: + accepted[k] = _coerce_config_value(k, v) + for k in removed: + current.pop(k, None) + current.update(accepted) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + json.dump(current, f, ensure_ascii=False, indent=2) + return current + + +def _list_nights(project: str) -> List[Dict[str, Any]]: + root = staging_root(project) + out: List[Dict[str, Any]] = [] + if not os.path.isdir(root): + return out + for name in sorted(os.listdir(root), reverse=True): + d = os.path.join(root, name) + if not os.path.isdir(d): + continue + report = _read_json(os.path.join(d, "report.json")) or {} + entry = { + "ts": name, + "night": report.get("night"), + "accepted": report.get("accepted"), + "gate_action": report.get("gate_action", ""), + "baseline": report.get("baseline_score"), + "candidate": report.get("candidate_score"), + "n_tasks": report.get("n_tasks"), + "n_sessions": report.get("n_sessions"), + "tokens_used": report.get("tokens_used"), + "has_report": bool(report), + "has_evidence": os.path.exists(os.path.join(d, "evidence.jsonl")), + "has_manifest": os.path.exists(os.path.join(d, "manifest.json")), + "adopted": os.path.isdir(os.path.join(d, "backup")), + } + out.append(entry) + return out + + +class _RunState: + """At most one pipeline subprocess at a time, log tailed to a file.""" + + def __init__(self) -> None: + self.proc: Optional[subprocess.Popen] = None + self.log_path = "" + self.mode = "" + self.lock = threading.Lock() + + def running(self) -> bool: + return self.proc is not None and self.proc.poll() is None + + def start(self, project: str, dry_run: bool) -> Dict[str, Any]: + with self.lock: + if self.running(): + return {"ok": False, "error": "a run is already in progress"} + cfg = load_config(invoked_project=project) + self.log_path = os.path.join(cfg.state_dir, "dashboard-run.log") + os.makedirs(os.path.dirname(self.log_path), exist_ok=True) + mode = "dry-run" if dry_run else "run" + cmd = [sys.executable, "-m", "skillopt_sleep", mode, + "--project", project, "--progress"] + no_window = getattr(subprocess, "CREATE_NO_WINDOW", 0) if os.name == "nt" else 0 + try: + # The child inherits a dup of this descriptor, so the parent + # closes its own copy immediately rather than leaving the log + # held open (and, on Windows, locked) for the server's life. + with open(self.log_path, "w", encoding="utf-8") as log: + self.proc = subprocess.Popen( + cmd, stdout=log, stderr=subprocess.STDOUT, + creationflags=no_window, cwd=project or None, + ) + except OSError as exc: + # A failed spawn must not raise out of the request thread — + # that would drop the connection and leave the UI hanging. + self.proc = None + return {"ok": False, "error": f"could not start {mode}: {exc}"} + self.mode = mode + return {"ok": True, "mode": self.mode} + + def status(self) -> Dict[str, Any]: + tail = "" + if self.log_path: + text = _read_text(self.log_path) + tail = text[-6000:] + rc = None + if self.proc is not None: + rc = self.proc.poll() + return {"running": self.running(), "returncode": rc, + "mode": self.mode, "tail": tail} + + +def _night_dir(project: str, ts: str) -> Optional[str]: + """Resolve a night id to its staging directory, or None if it is not one. + + ``os.path.basename`` alone is not containment: it leaves ``".."`` intact, + so ``/api/night/..`` resolved to the staging parent and ``/api/adopt`` + would have copied whatever it found there over the live SKILL.md and + CLAUDE.md. Reject anything that is not a plain single path component, then + confirm the *resolved* path is still under the staging root so a symlink + planted in staging cannot redirect the read either. + """ + name = str(ts or "") + if not name or name in {".", ".."}: + return None + if name != os.path.basename(name): # separators, drive letters, absolutes + return None + if os.sep in name or (os.altsep and os.altsep in name): + return None + + root = staging_root(project) + candidate = os.path.join(root, name) + try: + real_root = os.path.realpath(root) + real_candidate = os.path.realpath(candidate) + if os.path.commonpath([real_root, real_candidate]) != real_root: + return None + if real_candidate == real_root: + return None + except (OSError, ValueError): # unrelated roots / different drives + return None + if not os.path.isdir(real_candidate): + return None + return candidate + + +def _split_host(value: str) -> Tuple[str, str]: + """Split a Host header into (hostname, port), tolerating IPv6 brackets.""" + host = (value or "").strip() + if host.startswith("["): # [::1]:8321 + close = host.find("]") + if close < 0: + return "", "" + return host[:close + 1].lower(), host[close + 2:] if host[close + 1:close + 2] == ":" else "" + if host.count(":") > 1: # bare IPv6 with no brackets + return host.lower(), "" + name, _, port = host.partition(":") + return name.lower(), port + + +def _is_loopback_host(value: str, port: int) -> bool: + name, host_port = _split_host(value) + if name not in _LOOPBACK_HOSTS: + return False + # A Host with no port means the default port for the scheme, which is + # never the port this server was given. + return host_port == str(port) + + +def _allowed_origins(port: int) -> frozenset: + return frozenset( + f"http://{host}:{port}" + for host in ("127.0.0.1", "localhost", "[::1]") + ) + + +class DashboardHandler(BaseHTTPRequestHandler): + project: str = "" + run_state: _RunState + token: str = "" + port: int = 0 + + # ── plumbing ────────────────────────────────────────────────────────── + def log_message(self, fmt: str, *args: Any) -> None: # quiet server + pass + + # ── request authorization ───────────────────────────────────────────── + def _authorize(self, *, mutating: bool) -> bool: + """Gate one request. Returns False after writing the error response. + + Loopback ``Host`` is required on every request, which is what defeats + DNS rebinding: the browser sends the attacker's name, not ours. State + changing requests additionally need a trusted ``Origin``, a JSON + content type, and the per-process capability token — each of which a + cross-origin page is independently unable to produce. + """ + if not _is_loopback_host(self.headers.get("Host", ""), self.port): + self._json({"error": "forbidden: bad host"}, 403) + return False + if not mutating: + return True + + origin = self.headers.get("Origin") + if not origin or origin not in _allowed_origins(self.port): + # Missing Origin is rejected too: it is not evidence of a + # same-origin caller, and the real page always sends one. + self._json({"error": "forbidden: bad origin"}, 403) + return False + + ctype = (self.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower() + if ctype != "application/json": + # Form and text/plain bodies are exactly the shapes a cross-origin + #
can send without a preflight. + self._json({"error": "unsupported media type: expected application/json"}, 415) + return False + + presented = self.headers.get(_TOKEN_HEADER) or "" + if not self.token or not hmac.compare_digest(presented, self.token): + self._json({"error": "forbidden: bad token"}, 403) + return False + return True + + def _send(self, code: int, body: bytes, ctype: str) -> None: + self.send_response(code) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def _json(self, obj: Any, code: int = 200) -> None: + self._send(code, json.dumps(obj, ensure_ascii=False, default=str).encode("utf-8"), + "application/json; charset=utf-8") + + def _drain(self, limit: int) -> None: + """Discard up to ``limit`` bytes of an unwanted request body.""" + remaining = limit + while remaining > 0: + chunk = self.rfile.read(min(65536, remaining)) + if not chunk: + return + remaining -= len(chunk) + + def _body(self) -> Optional[Dict[str, Any]]: + """Parse a required JSON object body, or write a 4xx and return None. + + An absent, empty, oversized or non-object body is an error rather than + an empty dict: treating ``{}`` as "no arguments supplied" is what let a + contentless POST to /api/run start a real run. + """ + try: + declared = int(self.headers.get("Content-Length", "") or "") + except ValueError: + self._json({"error": "invalid Content-Length"}, 400) + return None + if declared <= 0: + self._json({"error": "request body required"}, 400) + return None + if declared > _MAX_BODY_BYTES: + # Drain a bounded amount first so a well-behaved client can read + # the 413 rather than seeing a connection reset mid-upload, then + # hang up instead of consuming an unbounded stream. + self._drain(min(declared, _MAX_BODY_BYTES * 2)) + self.close_connection = True + self._json({"error": "request body too large"}, 413) + return None + + raw = self.rfile.read(declared) + if len(raw) != declared: + self._json({"error": "truncated request body"}, 400) + return None + try: + obj = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, ValueError): + self._json({"error": "malformed JSON body"}, 400) + return None + if not isinstance(obj, dict): + self._json({"error": "request body must be a JSON object"}, 400) + return None + return obj + + # ── GET ─────────────────────────────────────────────────────────────── + def do_GET(self) -> None: # noqa: N802 (http.server API) + if not self._authorize(mutating=False): + return + path = self.path.split("?", 1)[0] + if path in {"/", "/index.html"}: + html = _read_text(_HTML_PATH, limit=5_000_000) + # Hand the page its capability token. Only a same-origin document + # can read this body, so only it can make mutating calls. + html = html.replace(_TOKEN_PLACEHOLDER, self.token) + self._send(200, html.encode("utf-8"), "text/html; charset=utf-8") + return + if path == "/api/overview": + cfg = load_config(invoked_project=self.project) + effective = {k: cfg.get(k) for k in sorted(_EDITABLE_KEYS)} + self._json({ + "project": self.project, + "config": effective, + "defaults": {k: DEFAULTS.get(k) for k in sorted(_EDITABLE_KEYS)}, + "config_path": _user_config_file(), + "prompts": prompt_registry.describe(), + "prompts_path": prompt_registry.overrides_path(), + "nights": _list_nights(self.project), + }) + return + if path.startswith("/api/night/"): + ts = unquote(path[len("/api/night/"):]) + d = _night_dir(self.project, ts) + if d is None: + self._json({"error": "unknown night"}, 404) + return + self._json({ + "ts": ts, + "dir": d, + "report": _read_json(os.path.join(d, "report.json")), + "manifest": _read_json(os.path.join(d, "manifest.json")), + "diagnostics": _read_json(os.path.join(d, "diagnostics.json")), + "report_md": _read_text(os.path.join(d, "report.md")), + "proposed_skill": _read_text(os.path.join(d, "proposed_SKILL.md")), + "proposed_memory": _read_text(os.path.join(d, "proposed_CLAUDE.md")), + "evidence": read_events(os.path.join(d, "evidence.jsonl")), + "adopted": os.path.isdir(os.path.join(d, "backup")), + }) + return + if path == "/api/run/status": + self._json(self.run_state.status()) + return + self._json({"error": "not found"}, 404) + + # ── POST ────────────────────────────────────────────────────────────── + def do_POST(self) -> None: # noqa: N802 + if not self._authorize(mutating=True): + return + path = self.path.split("?", 1)[0] + body = self._body() + if body is None: # _body already answered with the specific 4xx + return + if path == "/api/config": + updates = body.get("updates") or {} + if not isinstance(updates, dict): + self._json({"error": "updates must be an object"}, 400) + return + try: + saved = _write_config(updates) + except ValueError as exc: + self._json({"error": str(exc)}, 400) + return + cfg = load_config(invoked_project=self.project) + self._json({"ok": True, "saved": saved, + "config": {k: cfg.get(k) for k in sorted(_EDITABLE_KEYS)}}) + return + if path == "/api/prompts": + updates = body.get("updates") or {} + if not isinstance(updates, dict): + self._json({"error": "updates must be an object"}, 400) + return + prompt_registry.save_overrides(updates) + self._json({"ok": True, "prompts": prompt_registry.describe()}) + return + if path == "/api/run": + self._json(self.run_state.start(self.project, bool(body.get("dry_run")))) + return + if path == "/api/adopt": + d = _night_dir(self.project, body.get("ts", "")) + if d is None: + self._json({"error": "unknown night"}, 404) + return + try: + updated = adopt_staging(d) + except Exception as exc: # surface, don't crash the server + self._json({"ok": False, "error": str(exc)}, 500) + return + self._json({"ok": True, "updated": updated}) + return + self._json({"error": "not found"}, 404) + + +def make_server(project: str = "", port: int = 8321) -> ThreadingHTTPServer: + """Bind a dashboard server on loopback with a fresh capability token. + + The handler is a per-server subclass so its token and port cannot leak + between servers (the tests run several), and so the token is scoped to one + process lifetime: restarting the dashboard invalidates every old page. + """ + project = os.path.abspath(project or os.getcwd()) + token = secrets.token_urlsafe(32) + + class _Handler(DashboardHandler): + pass + + _Handler.project = project + _Handler.run_state = _RunState() + _Handler.token = token + httpd = ThreadingHTTPServer(("127.0.0.1", port), _Handler) + # Bound port, which may be ephemeral when port=0 was requested. Host and + # Origin are validated against this exact value. + _Handler.port = httpd.server_address[1] + return httpd + + +def serve(project: str = "", port: int = 8321, open_browser: bool = True) -> int: + project = os.path.abspath(project or os.getcwd()) + httpd = make_server(project, port) + url = f"http://127.0.0.1:{httpd.server_address[1]}/" + print(f"[sleep] dashboard for {project}\n[sleep] serving {url} (Ctrl+C to stop)") + if open_browser: + try: + import webbrowser + threading.Timer(0.4, webbrowser.open, args=(url,)).start() + except Exception: + pass + try: + httpd.serve_forever() + except KeyboardInterrupt: + pass + finally: + httpd.server_close() + return 0 diff --git a/tests/test_dashboard_security.py b/tests/test_dashboard_security.py new file mode 100644 index 00000000..bdef9689 --- /dev/null +++ b/tests/test_dashboard_security.py @@ -0,0 +1,554 @@ +"""Tests for the local dashboard's request-authorization boundary. + +The dashboard's POST endpoints start real runs, adopt staged artifacts, and +rewrite config and prompts. Binding to loopback does not protect them: any +page in the user's browser can POST cross-origin to 127.0.0.1, and a hostile +name resolving to loopback (DNS rebinding) makes those requests look local. + +Every negative case here asserts twice: that the request was refused, *and* +that the side effect did not happen — a 403 that still started a run would +pass a status-code-only test. + +Pure stdlib, deterministic, no network beyond 127.0.0.1, no third-party deps. + +Run: python -m unittest tests.test_dashboard_security +""" +from __future__ import annotations + +import http.client +import json +import os +import tempfile +import threading +import unittest +from unittest import mock + +from skillopt_sleep import dashboard +from skillopt_sleep import prompts as prompt_registry + +_TOKEN_HEADER = "X-SkillOpt-Dashboard-Token" + + +class _FakeRunState: + """Records launches instead of spawning a real pipeline subprocess.""" + + def __init__(self) -> None: + self.starts = [] + + def running(self) -> bool: + return False + + def start(self, project, dry_run): + self.starts.append((project, bool(dry_run))) + return {"ok": True, "mode": "dry-run" if dry_run else "run"} + + def status(self): + return {"running": False, "returncode": None, "mode": "", "tail": ""} + + +class DashboardSecurityTestCase(unittest.TestCase): + """A live loopback dashboard with every side effect redirected to tmp.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + root = self.tmp.name + + self.config_path = os.path.join(root, "config.json") + self.prompts_path = os.path.join(root, "prompts.json") + + patches = [ + mock.patch.dict(os.environ, + {"SKILLOPT_SLEEP_PROMPTS_PATH": self.prompts_path}), + mock.patch.object(dashboard, "_user_config_file", + return_value=self.config_path), + ] + for patch in patches: + patch.start() + self.addCleanup(patch.stop) + + self.adopted = [] + adopt_patch = mock.patch.object( + dashboard, "adopt_staging", + side_effect=lambda d: self.adopted.append(d) or {"skill": "ok"}) + adopt_patch.start() + self.addCleanup(adopt_patch.stop) + + # A staged night that /api/adopt would accept if it got through. + self.night = "20260801-000000" + staging = os.path.join(root, ".skillopt-sleep", "staging", self.night) + os.makedirs(staging) + with open(os.path.join(staging, "manifest.json"), "w", encoding="utf-8") as f: + json.dump({"files": []}, f) + + self.httpd = dashboard.make_server(root, 0) + self.handler = self.httpd.RequestHandlerClass + self.run_state = _FakeRunState() + self.handler.run_state = self.run_state + self.port = self.httpd.server_address[1] + self.token = self.handler.token + thread = threading.Thread(target=self.httpd.serve_forever, daemon=True) + thread.start() + + def _shutdown(): + self.httpd.shutdown() + self.httpd.server_close() + self.addCleanup(_shutdown) + + # ── request helpers ─────────────────────────────────────────────────── + def request(self, method, path, *, body=None, headers=None, + host=None, origin="", ctype="application/json", token=None): + """Issue one request with full control over the security-relevant bits.""" + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5) + sent = {"Host": host if host is not None else f"127.0.0.1:{self.port}"} + if ctype: + sent["Content-Type"] = ctype + if origin: + sent["Origin"] = origin + resolved = self.token if token is None else token + if resolved: + sent[_TOKEN_HEADER] = resolved + sent.update(headers or {}) + payload = body.encode("utf-8") if isinstance(body, str) else body + conn.request(method, path, body=payload, headers=sent) + response = conn.getresponse() + raw = response.read() + conn.close() + try: + parsed = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, ValueError): + parsed = raw + return response.status, parsed, response + + def post(self, path, obj=None, **kw): + body = kw.pop("body", None) + if body is None: + body = json.dumps({} if obj is None else obj) + return self.request("POST", path, body=body, + origin=kw.pop("origin", self.origin()), **kw) + + def origin(self): + return f"http://127.0.0.1:{self.port}" + + # ── side-effect probes ──────────────────────────────────────────────── + def saved_config(self): + try: + with open(self.config_path, encoding="utf-8") as f: + return json.load(f) + except (OSError, ValueError): + return {} + + def miner_override(self): + for entry in prompt_registry.describe(): + if entry["name"] == "miner": + return entry.get("override") + return None + + def assertNothingChanged(self, msg=""): + self.assertEqual(self.saved_config(), {}, f"config was written {msg}") + self.assertIn(self.miner_override(), (None, ""), f"prompt was written {msg}") + self.assertEqual(self.run_state.starts, [], f"a run was started {msg}") + self.assertEqual(self.adopted, [], f"a night was adopted {msg}") + + +class TestHappyPath(DashboardSecurityTestCase): + """Same-origin JSON with the token still does everything it should.""" + + def test_html_carries_a_substituted_token(self): + status, _body, response = self.request("GET", "/", ctype="") + html = _body if isinstance(_body, bytes) else json.dumps(_body).encode() + self.assertEqual(status, 200) + self.assertIn(b"Control Panel", html) + self.assertIn(self.token.encode(), html) + self.assertNotIn(b"__SKILLOPT_DASHBOARD_TOKEN__", html) + self.assertNotIn("Access-Control-Allow-Origin", dict(response.getheaders())) + + def test_overview_and_night_reads(self): + status, body, _r = self.request("GET", "/api/overview", ctype="") + self.assertEqual(status, 200) + self.assertEqual({p["name"] for p in body["prompts"]}, + {"miner", "attempt", "judge", "reflect"}) + status, _body, _r = self.request("GET", "/api/night/nope", ctype="") + self.assertEqual(status, 404) + + def test_config_write(self): + status, body, _r = self.post("/api/config", {"updates": {"edit_budget": 7}}) + self.assertEqual(status, 200) + self.assertTrue(body["ok"]) + self.assertEqual(self.saved_config(), {"edit_budget": 7}) + + def test_config_ignores_keys_outside_the_allowlist(self): + status, _body, _r = self.post( + "/api/config", {"updates": {"edit_budget": 3, "claude_home": "/etc"}}) + self.assertEqual(status, 200) + self.assertEqual(self.saved_config(), {"edit_budget": 3}) + + def test_prompt_roundtrip(self): + status, body, _r = self.post("/api/prompts", + {"updates": {"miner": "X __PROMPTS__"}}) + self.assertEqual(status, 200) + mined = [p for p in body["prompts"] if p["name"] == "miner"][0] + self.assertEqual(mined["override"], "X __PROMPTS__") + self.post("/api/prompts", {"updates": {"miner": None}}) + self.assertIn(self.miner_override(), (None, "")) + + def test_run_and_adopt(self): + status, body, _r = self.post("/api/run", {"dry_run": True}) + self.assertEqual(status, 200) + self.assertTrue(body["ok"]) + self.assertEqual(self.run_state.starts, [(self.tmp.name, True)]) + + status, body, _r = self.post("/api/adopt", {"ts": self.night}) + self.assertEqual(status, 200) + self.assertTrue(body["ok"]) + self.assertEqual(len(self.adopted), 1) + + def test_localhost_origin_and_host_are_accepted(self): + status, _body, _r = self.request( + "POST", "/api/config", body=json.dumps({"updates": {"edit_budget": 5}}), + host=f"localhost:{self.port}", origin=f"http://localhost:{self.port}") + self.assertEqual(status, 200) + self.assertEqual(self.saved_config(), {"edit_budget": 5}) + + +class TestHostValidation(DashboardSecurityTestCase): + """DNS rebinding: the browser sends the attacker's name, not ours.""" + + def test_foreign_host_is_rejected_on_every_endpoint(self): + for method, path in [ + ("GET", "/"), ("GET", "/api/overview"), ("GET", "/api/run/status"), + ("POST", "/api/config"), ("POST", "/api/prompts"), + ("POST", "/api/run"), ("POST", "/api/adopt"), + ]: + with self.subTest(method=method, path=path): + status, _body, _r = self.request( + method, path, + body=json.dumps({"updates": {"edit_budget": 9}, "ts": self.night, + "dry_run": True}) if method == "POST" else None, + host=f"attacker.example.com:{self.port}", origin=self.origin()) + self.assertEqual(status, 403) + self.assertNothingChanged("after foreign Host") + + def test_host_with_wrong_port_is_rejected(self): + status, _body, _r = self.post("/api/config", {"updates": {"edit_budget": 9}}, + host=f"127.0.0.1:{self.port + 1}") + self.assertEqual(status, 403) + self.assertNothingChanged("after wrong-port Host") + + def test_host_without_port_is_rejected(self): + status, _body, _r = self.post("/api/config", {"updates": {"edit_budget": 9}}, + host="127.0.0.1") + self.assertEqual(status, 403) + self.assertNothingChanged("after portless Host") + + def test_rebinding_name_resolving_to_loopback_is_rejected(self): + # The attacker's DNS points at 127.0.0.1, so the connection succeeds; + # only the Host header distinguishes it from the real dashboard. + status, _body, _r = self.post("/api/run", {"dry_run": False}, + host=f"rebind.attacker.test:{self.port}", + origin=f"http://rebind.attacker.test:{self.port}") + self.assertEqual(status, 403) + self.assertNothingChanged("after DNS-rebinding request") + + +class TestOriginValidation(DashboardSecurityTestCase): + MUTATING = [ + ("/api/config", {"updates": {"edit_budget": 9}}), + ("/api/prompts", {"updates": {"miner": "pwned"}}), + ("/api/run", {"dry_run": False}), + ("/api/adopt", {"ts": "20260801-000000"}), + ] + + def test_foreign_origin_is_rejected(self): + for path, payload in self.MUTATING: + with self.subTest(path=path): + status, _body, _r = self.post(path, payload, + origin="http://evil.example.com") + self.assertEqual(status, 403) + self.assertNothingChanged("after foreign Origin") + + def test_missing_origin_is_rejected(self): + for path, payload in self.MUTATING: + with self.subTest(path=path): + status, _body, _r = self.post(path, payload, origin="") + self.assertEqual(status, 403) + self.assertNothingChanged("after missing Origin") + + def test_null_and_lookalike_origins_are_rejected(self): + for origin in ("null", + f"https://127.0.0.1:{self.port}", + f"http://127.0.0.1.evil.com:{self.port}", + f"http://127.0.0.1:{self.port}.evil.com", + f"http://127.0.0.1:{self.port + 1}"): + with self.subTest(origin=origin): + status, _body, _r = self.post( + "/api/config", {"updates": {"edit_budget": 9}}, origin=origin) + self.assertEqual(status, 403) + self.assertNothingChanged("after lookalike Origin") + + +class TestContentTypeValidation(DashboardSecurityTestCase): + """The body shapes a cross-origin can send without a preflight.""" + + def test_form_and_text_bodies_are_rejected(self): + for ctype, body in [ + ("application/x-www-form-urlencoded", "dry_run=1"), + ("text/plain", json.dumps({"dry_run": True})), + ("text/plain;charset=UTF-8", json.dumps({"dry_run": True})), + ("multipart/form-data; boundary=x", "--x--"), + ("", json.dumps({"dry_run": True})), + ]: + with self.subTest(ctype=ctype): + status, _body, _r = self.post("/api/run", body=body, ctype=ctype) + self.assertEqual(status, 415) + self.assertNothingChanged("after non-JSON content type") + + def test_the_documented_csrf_vector_cannot_start_a_run(self): + """The reported case: an empty cross-origin form POST to /api/run.""" + status, _body, _r = self.request( + "POST", "/api/run", body=b"", + origin="http://evil.example.com", + ctype="application/x-www-form-urlencoded", token="") + self.assertEqual(status, 403) + self.assertEqual(self.run_state.starts, []) + + def test_json_content_type_with_parameters_is_accepted(self): + status, _body, _r = self.post("/api/config", {"updates": {"edit_budget": 4}}, + ctype="application/json; charset=utf-8") + self.assertEqual(status, 200) + self.assertEqual(self.saved_config(), {"edit_budget": 4}) + + +class TestTokenValidation(DashboardSecurityTestCase): + def test_missing_and_wrong_tokens_are_rejected(self): + for token in ("", "not-the-token", "x" * len(self.token) if self.token else "x"): + with self.subTest(token=token[:12]): + status, _body, _r = self.post( + "/api/config", {"updates": {"edit_budget": 9}}, token=token) + self.assertEqual(status, 403) + self.assertNothingChanged("after bad token") + + def test_token_is_unguessable_and_per_process(self): + other = dashboard.make_server(self.tmp.name, 0) + try: + self.assertNotEqual(other.RequestHandlerClass.token, self.token) + self.assertGreaterEqual(len(self.token), 32) + # A token from another dashboard process must not work here. + status, _body, _r = self.post("/api/config", {"updates": {"edit_budget": 9}}, + token=other.RequestHandlerClass.token) + self.assertEqual(status, 403) + finally: + other.server_close() + self.assertNothingChanged("after cross-process token") + + +class TestBodyValidation(DashboardSecurityTestCase): + def test_empty_body_is_not_coerced_to_an_empty_object(self): + for path in ("/api/run", "/api/config", "/api/prompts", "/api/adopt"): + with self.subTest(path=path): + status, _body, _r = self.post(path, body=b"") + self.assertEqual(status, 400) + self.assertNothingChanged("after empty body") + + def test_malformed_and_non_object_bodies_are_rejected(self): + for body in ("{not json", "[]", '"a string"', "null", "42", ""): + with self.subTest(body=body): + status, _body, _r = self.post("/api/config", body=body) + self.assertEqual(status, 400) + self.assertNothingChanged("after malformed body") + + def test_oversized_body_is_rejected(self): + payload = json.dumps({"updates": {"preferences": "A" * (1 << 21)}}) + status, _body, _r = self.post("/api/config", body=payload) + self.assertEqual(status, 413) + self.assertNothingChanged("after oversized body") + + def test_non_object_updates_are_rejected(self): + for path in ("/api/config", "/api/prompts"): + with self.subTest(path=path): + status, _body, _r = self.post(path, {"updates": "everything"}) + self.assertEqual(status, 400) + self.assertNothingChanged("after non-object updates") + + +class TestStagingPathContainment(DashboardSecurityTestCase): + """`os.path.basename` is not containment: basename("..") == "..". + + Before this was fixed, /api/night/.. read the staging parent and + /api/adopt would copy whatever it found there over the live SKILL.md. + """ + + ESCAPES = ["..", "../..", "../" * 4, "/etc", "\\Windows", + ".", "", "C:\\Windows", "%2e%2e", "..%2f.."] + + def test_night_read_cannot_escape_the_staging_root(self): + for ts in self.ESCAPES: + with self.subTest(ts=ts): + status, body, _r = self.request( + "GET", "/api/night/" + ts, ctype="") + self.assertEqual(status, 404) + self.assertNotIn("report", body if isinstance(body, dict) else {}) + + def test_adopt_cannot_escape_the_staging_root(self): + for ts in self.ESCAPES: + with self.subTest(ts=ts): + status, _body, _r = self.post("/api/adopt", {"ts": ts}) + self.assertEqual(status, 404) + self.assertEqual(self.adopted, [], "adopt escaped the staging root") + + def test_adopt_rejects_non_string_ts(self): + for ts in [None, 42, ["a"], {"a": 1}, True]: + with self.subTest(ts=ts): + status, _body, _r = self.post("/api/adopt", {"ts": ts}) + self.assertEqual(status, 404) + self.assertEqual(self.adopted, []) + + def test_linked_night_pointing_outside_is_rejected(self): + """A link planted inside staging must not redirect the read out of it. + + The name is a plain component, so only resolving it catches this. + Unprivileged Windows cannot create symlinks; a directory junction + exercises the same containment check there. + """ + outside = os.path.join(self.tmp.name, "outside") + os.makedirs(outside, exist_ok=True) + with open(os.path.join(outside, "report.json"), "w", encoding="utf-8") as f: + f.write("{}") + link = os.path.join(self.tmp.name, ".skillopt-sleep", "staging", "sneaky") + try: + os.symlink(outside, link, target_is_directory=True) + except (OSError, NotImplementedError, AttributeError): + if os.name != "nt": + self.skipTest("symlink creation not permitted on this platform") + import subprocess + created = subprocess.run(["cmd", "/c", "mklink", "/J", link, outside], + capture_output=True, text=True) + if not os.path.isdir(link): + self.skipTest(f"cannot create a link here: {created.stderr.strip()}") + + status, _body, _r = self.request("GET", "/api/night/sneaky", ctype="") + self.assertEqual(status, 404) + status, _body, _r = self.post("/api/adopt", {"ts": "sneaky"}) + self.assertEqual(status, 404) + self.assertEqual(self.adopted, []) + + def test_the_real_night_is_still_reachable(self): + status, body, _r = self.request("GET", "/api/night/" + self.night, ctype="") + self.assertEqual(status, 200) + self.assertEqual(body["ts"], self.night) + status, body, _r = self.post("/api/adopt", {"ts": self.night}) + self.assertEqual(status, 200) + self.assertEqual(len(self.adopted), 1) + + +class TestConfigValueTyping(DashboardSecurityTestCase): + """`load_config` does not type-coerce, so the API must.""" + + def test_numeric_strings_are_stored_as_numbers(self): + status, _body, _r = self.post("/api/config", {"updates": { + "edit_budget": "7", "gate_mixed_weight": "0.25", + "max_tasks_per_night": "12"}}) + self.assertEqual(status, 200) + saved = self.saved_config() + self.assertEqual(saved["edit_budget"], 7) + self.assertIsInstance(saved["edit_budget"], int) + self.assertEqual(saved["gate_mixed_weight"], 0.25) + self.assertIsInstance(saved["gate_mixed_weight"], float) + self.assertIsInstance(saved["max_tasks_per_night"], int) + + def test_boolean_strings_are_stored_as_booleans(self): + status, _body, _r = self.post("/api/config", {"updates": { + "llm_mine": "false", "evolve_skill": "true", "auto_adopt": False}}) + self.assertEqual(status, 200) + saved = self.saved_config() + self.assertIs(saved["llm_mine"], False) + self.assertIs(saved["evolve_skill"], True) + self.assertIs(saved["auto_adopt"], False) + + def test_free_text_is_not_coerced_to_a_boolean(self): + """The old blanket 'true' -> True turned house rules into a bool.""" + status, _body, _r = self.post( + "/api/config", {"updates": {"preferences": "true"}}) + self.assertEqual(status, 200) + self.assertEqual(self.saved_config()["preferences"], "true") + + def test_unparseable_numbers_are_rejected_and_nothing_is_written(self): + for key, value in [("edit_budget", "not-a-number"), + ("gate_mixed_weight", "high"), + ("max_tasks_per_night", "12 tasks"), + ("llm_mine", "maybe")]: + with self.subTest(key=key): + status, body, _r = self.post( + "/api/config", {"updates": {key: value}}) + self.assertEqual(status, 400) + self.assertIn(key, body["error"]) + self.assertEqual(self.saved_config(), {}) + + def test_a_bad_field_does_not_half_apply_the_form(self): + status, _body, _r = self.post("/api/config", {"updates": { + "edit_budget": "5", "gate_mixed_weight": "nonsense"}}) + self.assertEqual(status, 400) + self.assertEqual(self.saved_config(), {}, + "a rejected form must not persist its valid fields") + + +class TestRunLauncherResilience(unittest.TestCase): + """A failed spawn must not raise out of the request-handler thread.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + stub = mock.Mock() + stub.state_dir = self.tmp.name + patch = mock.patch.object(dashboard, "load_config", return_value=stub) + patch.start() + self.addCleanup(patch.stop) + + def test_spawn_failure_returns_a_structured_error(self): + state = dashboard._RunState() + with mock.patch.object(dashboard.subprocess, "Popen", + side_effect=OSError("no interpreter")): + result = state.start(self.tmp.name, dry_run=True) + self.assertFalse(result["ok"]) + self.assertIn("no interpreter", result["error"]) + self.assertFalse(state.running()) + # The launcher stays usable after a failure. + self.assertEqual(state.status()["running"], False) + + def test_log_handle_is_released_to_the_child(self): + state = dashboard._RunState() + with mock.patch.object(dashboard.subprocess, "Popen") as popen: + popen.return_value.poll.return_value = 0 + result = state.start(self.tmp.name, dry_run=True) + self.assertTrue(result["ok"]) + # Windows refuses to remove a file the parent still holds open, so a + # successful unlink proves the parent closed its copy. + os.unlink(state.log_path) + + +class TestNoPermissiveCors(DashboardSecurityTestCase): + def test_no_cors_headers_are_ever_emitted(self): + probes = [ + ("GET", "/", None), ("GET", "/api/overview", None), + ("POST", "/api/config", json.dumps({"updates": {"edit_budget": 2}})), + ] + for method, path, body in probes: + with self.subTest(path=path): + _status, _body, response = self.request( + method, path, body=body, origin=self.origin()) + headers = {k.lower() for k in dict(response.getheaders())} + self.assertNotIn("access-control-allow-origin", headers) + self.assertNotIn("access-control-allow-credentials", headers) + self.assertNotIn("access-control-allow-headers", headers) + + def test_preflight_is_not_answered_permissively(self): + status, _body, response = self.request( + "OPTIONS", "/api/run", origin="http://evil.example.com", ctype="") + headers = {k.lower() for k in dict(response.getheaders())} + self.assertNotIn("access-control-allow-origin", headers) + self.assertNotEqual(status, 204) + self.assertGreaterEqual(status, 400) + + +if __name__ == "__main__": + unittest.main()