From 3f23736c30c8741c915c6c41af6018f429a776c6 Mon Sep 17 00:00:00 2001 From: Alfred Kolakkal Date: Sat, 1 Aug 2026 11:59:40 -0600 Subject: [PATCH 1/2] feat(sleep): local control-panel dashboard for the sleep pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A zero-dependency (stdlib http.server) web UI over one project's sleep pipeline, arranged to mirror the data flow: per-stage role/model/prompt display with live prompt editing, a per-night evidentiary chain browser over evidence.jsonl, the gate arithmetic spelled out, a config editor, a run/dry-run launcher, and explicit adopt (nothing auto-adopts from the UI). python -m skillopt_sleep dashboard [--project DIR] [--port N] Addresses the review on #152: * Rebased onto current main; the evidence-chain commit landed as #151 and is dropped here. This commit is only the dashboard and its tests. * Request authorization. Binding to 127.0.0.1 is not an authorization boundary — any page in the user's browser can POST cross-origin to loopback, and a hostile name resolving to loopback (DNS rebinding) makes those requests look local. Every request now passes through _authorize: - loopback Host is required on every request, GET included, with the port matching the bound port. This is what defeats rebinding: the browser sends the attacker's name, not ours. - state-changing POSTs additionally require a trusted Origin (missing and foreign are both rejected), an application/json content type, and an unguessable per-process capability token in a custom header. Each is independently sufficient to stop a cross-origin page: a form cannot set a custom header or a JSON content type without a preflight, which this server deliberately does not answer. No CORS headers are emitted anywhere, so foreign origins cannot read responses either. The token is minted per server process and substituted into the served HTML, so it never travels in a URL and restarting the dashboard invalidates every previously served page. * Request bodies are validated, not coerced. An absent, empty, malformed, non-object or oversized body is now a specific 4xx instead of silently becoming {}. That coercion is what let a contentless cross-origin POST to /api/run start a real run. Oversized bodies are drained within a bound before the 413 so well-behaved clients read the status rather than a connection reset. tests/test_dashboard_security.py adds 25 tests. Every negative case asserts twice — that the request was refused, and that the side effect did not happen, since a 403 that still started a run would pass a status-only test. Covers foreign Host (all seven endpoints), wrong/absent port, rebinding names, foreign/missing/null/lookalike Origins, form, text/plain, multipart and absent content types, missing/wrong/ cross-process tokens, empty, malformed, non-object and oversized bodies, absence of CORS headers, unanswered preflight, and the reported vector verbatim (empty cross-origin form POST to /api/run). Same-origin JSON happy paths for config, prompts, run and adopt are retained. --- skillopt_sleep/__main__.py | 11 + skillopt_sleep/dashboard.html | 587 +++++++++++++++++++++++++++++++ skillopt_sleep/dashboard.py | 453 ++++++++++++++++++++++++ tests/test_dashboard_security.py | 399 +++++++++++++++++++++ 4 files changed, 1450 insertions(+) create mode 100644 skillopt_sleep/dashboard.html create mode 100644 skillopt_sleep/dashboard.py create mode 100644 tests/test_dashboard_security.py 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..231d8f0d --- /dev/null +++ b/skillopt_sleep/dashboard.html @@ -0,0 +1,587 @@ + + + + + +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..411f4d03 --- /dev/null +++ b/skillopt_sleep/dashboard.py @@ -0,0 +1,453 @@ +"""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 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 _write_config(updates: Dict[str, Any]) -> Dict[str, Any]: + path = _user_config_file() + current = _read_json(path) or {} + for k, v in updates.items(): + if k not in _EDITABLE_KEYS: + continue + if v is None or v == "": + # empty resets the key to the built-in default + current.pop(k, None) + else: + current[k] = v + 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) + self.mode = "dry-run" if dry_run else "run" + cmd = [sys.executable, "-m", "skillopt_sleep", self.mode, + "--project", project, "--progress"] + log = open(self.log_path, "w", encoding="utf-8") + no_window = getattr(subprocess, "CREATE_NO_WINDOW", 0) if os.name == "nt" else 0 + self.proc = subprocess.Popen( + cmd, stdout=log, stderr=subprocess.STDOUT, + creationflags=no_window, cwd=project or None, + ) + 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 _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 = os.path.basename(path[len("/api/night/"):]) + d = os.path.join(staging_root(self.project), ts) + if not os.path.isdir(d): + 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 + saved = _write_config(updates) + 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": + ts = os.path.basename(str(body.get("ts", ""))) + d = os.path.join(staging_root(self.project), ts) + if not os.path.isdir(d): + 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..b3ece60c --- /dev/null +++ b/tests/test_dashboard_security.py @@ -0,0 +1,399 @@ +"""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 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() From b3db5ef892fba3eed28da3893bc81a0144e69399 Mon Sep 17 00:00:00 2001 From: Alfred Kolakkal Date: Sat, 1 Aug 2026 12:26:35 -0600 Subject: [PATCH 2/2] fix(sleep): contain staging paths, harden run launcher, type config writes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses three review findings on the dashboard. * Staging path containment. `os.path.basename` is not containment: `basename("..")` is `".."`, so `/api/night/..` resolved to the staging parent and `/api/adopt` would have handed that directory to adopt_staging, copying whatever it found over the live SKILL.md and CLAUDE.md. Both endpoints now share `_night_dir`, which percent-decodes first, rejects anything that is not a plain single path component, and then confirms the *resolved* path is still under the staging root — so a symlink or junction planted inside staging cannot redirect the read either. Non-string `ts` values are rejected rather than stringified. * Run launcher resilience. A failing `subprocess.Popen` raised straight out of the request-handler thread, dropping the connection and leaving the UI waiting; it now returns {ok: false, error} and the launcher stays usable. The parent also closes its own copy of the log descriptor after spawning instead of relying on refcounting to release it, which on Windows kept the file locked. * Config value typing. The editor declared fields as "number" but rendered every one as `type="text"` (a no-op ternary) and decided what to coerce by matching a regex against the *label text* — so renaming a label would silently start writing strings into config.json, and `load_config` does not type-coerce. Fields now declare a kind (text/select/bool/int/float), rendering and parsing both key off that kind, and the server validates independently in `_coerce_config_value` against the type of each built-in default, since the client is not the only possible caller. A value that cannot be coerced is a 400 and nothing is written, so a bad field cannot half-apply a form. Free text is no longer coerced: "true" in the house rules stays the string "true". Adds 12 tests: traversal and link-escape attempts on both endpoints (asserting nothing was adopted, with a junction fallback so the link case runs on unprivileged Windows too), spawn-failure handling and log-handle release, and config typing including the partial-write and free-text cases. --- skillopt_sleep/dashboard.html | 62 ++++++++----- skillopt_sleep/dashboard.py | 126 +++++++++++++++++++++---- tests/test_dashboard_security.py | 155 +++++++++++++++++++++++++++++++ 3 files changed, 305 insertions(+), 38 deletions(-) diff --git a/skillopt_sleep/dashboard.html b/skillopt_sleep/dashboard.html index 231d8f0d..99062d0f 100644 --- a/skillopt_sleep/dashboard.html +++ b/skillopt_sleep/dashboard.html @@ -517,47 +517,67 @@

Evidence — night ${esc(SEL||"—")}${esc(label)}"; + const choices = kind==="bool" ? (opts || ["true","false"]) : opts; + if(kind==="select" || kind==="bool"){ + d.innerHTML = `"; + } else if(kind==="int" || kind==="float"){ + d.innerHTML = ``; } else { - d.innerHTML = ``; + d.innerHTML = ``; } g.appendChild(d); } } +function parseCfgValue(kind, raw){ + const v = String(raw ?? ""); + if(v === "") return ""; // empty resets to the default + if(kind === "bool") return v === "true"; + if(kind === "int" || kind === "float"){ + const n = Number(v); + if(!Number.isFinite(n)) return null; // signals a validation failure + return kind === "int" ? Math.trunc(n) : n; + } + return v; // text/select stay strings +} async function saveConfig(){ const updates = {}; + const bad = []; document.querySelectorAll("#cfgGrid [data-k]").forEach(el=>{ - let v = el.value; - if(v==="true") v = true; else if(v==="false") v = false; - else if(v!=="" && !isNaN(Number(v)) && el.closest(".cfgitem").querySelector("label").textContent.match(/weight|budget|fraction|hours|tasks|rollouts|factor|K$|chars/i)) v = Number(v); - updates[el.dataset.k] = v; + const parsed = parseCfgValue(el.dataset.kind, el.value); + if(parsed === null){ bad.push(el.dataset.k); return; } + updates[el.dataset.k] = parsed; }); + if(bad.length){ alert("Not a valid number: " + bad.join(", ")); return; } const r = await post("/api/config", {updates}); if(r.ok){ OV.config = r.config; const m = $("#cfgMsg"); m.classList.add("show"); setTimeout(()=>m.classList.remove("show"), 1600); renderConfig(); renderMap(); renderStages(); } + else if(r.error){ alert("Config not saved: " + r.error); } } /* ── run control ────────────────────────────────────────────────────── */ diff --git a/skillopt_sleep/dashboard.py b/skillopt_sleep/dashboard.py index 411f4d03..6c2be3ef 100644 --- a/skillopt_sleep/dashboard.py +++ b/skillopt_sleep/dashboard.py @@ -36,6 +36,7 @@ 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 @@ -93,17 +94,61 @@ 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 == "": - # empty resets the key to the built-in default - current.pop(k, None) + removed.append(k) # empty resets the key to the built-in default else: - current[k] = v + 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) @@ -158,15 +203,25 @@ def start(self, project: str, dry_run: bool) -> Dict[str, Any]: 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) - self.mode = "dry-run" if dry_run else "run" - cmd = [sys.executable, "-m", "skillopt_sleep", self.mode, + mode = "dry-run" if dry_run else "run" + cmd = [sys.executable, "-m", "skillopt_sleep", mode, "--project", project, "--progress"] - log = open(self.log_path, "w", encoding="utf-8") no_window = getattr(subprocess, "CREATE_NO_WINDOW", 0) if os.name == "nt" else 0 - self.proc = subprocess.Popen( - cmd, stdout=log, stderr=subprocess.STDOUT, - creationflags=no_window, cwd=project or None, - ) + 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]: @@ -181,6 +236,40 @@ def status(self) -> Dict[str, Any]: "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() @@ -342,9 +431,9 @@ def do_GET(self) -> None: # noqa: N802 (http.server API) }) return if path.startswith("/api/night/"): - ts = os.path.basename(path[len("/api/night/"):]) - d = os.path.join(staging_root(self.project), ts) - if not os.path.isdir(d): + 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({ @@ -378,7 +467,11 @@ def do_POST(self) -> None: # noqa: N802 if not isinstance(updates, dict): self._json({"error": "updates must be an object"}, 400) return - saved = _write_config(updates) + 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)}}) @@ -395,9 +488,8 @@ def do_POST(self) -> None: # noqa: N802 self._json(self.run_state.start(self.project, bool(body.get("dry_run")))) return if path == "/api/adopt": - ts = os.path.basename(str(body.get("ts", ""))) - d = os.path.join(staging_root(self.project), ts) - if not os.path.isdir(d): + d = _night_dir(self.project, body.get("ts", "")) + if d is None: self._json({"error": "unknown night"}, 404) return try: diff --git a/tests/test_dashboard_security.py b/tests/test_dashboard_security.py index b3ece60c..bdef9689 100644 --- a/tests/test_dashboard_security.py +++ b/tests/test_dashboard_security.py @@ -371,6 +371,161 @@ def test_non_object_updates_are_rejected(self): 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 = [