"""Serve the PRD Plugin control surface locally (REQ-141). A snapshot cannot toggle anything: a `file://` page has no way to write `.prd_plugin/config.json`. So the UI that actually controls the plugin needs a process that owns the write path, and this is the smallest one that does — stdlib only, one page, two endpoints. Security shape, because this rewrites configuration: - **Loopback only, and it refuses to be told otherwise.** The public-service rule (REQ-122) is that a *declared* service must bind its manifest's `bind_host` and never silently fall back to loopback. This is the mirror case: a local control surface is not a declared service, and loopback is precisely what keeps a config-mutating endpoint off the network. Asking it to bind `0.0.0.0` is refused rather than honoured. - **No static file handler.** Unknown paths 404. A server sitting in a repo must not become a file browser for it. - **Validation is not reimplemented.** Every write goes through `prd_config.set_toggle`, the same validated authority the CLI uses, so the HTTP surface cannot become a way around a rule the catalog enforces. """ import argparse import json import sys from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path SCRIPTS = Path(__file__).resolve().parent if str(SCRIPTS) not in sys.path: sys.path.insert(0, str(SCRIPTS)) DEFAULT_HOST = "127.0.0.1" DEFAULT_PORT = 7717 LOOPBACK = {"127.0.0.1", "localhost", "::1"} MAX_BODY = 64 * 1024 # Changes that cannot take effect in the session that is already running. # Grounded, not guessed: the hook dispatcher re-reads configuration on every # event, so hook behaviour changes at the next event. What it cannot undo is # guidance already injected into the running session's context, or the # Substrate handshake performed at session start. That needs a NEW SESSION — # which is not the same as restarting the app. SESSION_SCOPED_PREFIXES = ("hooks.nudge", "integrations.substrate") class BindRefused(RuntimeError): """Refused to expose a configuration-mutating server beyond loopback.""" def apply_scope(key): """`session` if the change only lands in a new session, else `immediate`.""" return "session" if str(key).startswith(SESSION_SCOPED_PREFIXES) else "immediate" def _notice(key, scope): if scope != "session": return "" if str(key).startswith("integrations.substrate"): return ("Applies in a new session — the Substrate handshake happens at " "session start.") return ("Applies in a new session — guidance already injected into the " "running session stays in its context.") def apply_toggle(root, key, value): """Write one setting through the validated authority, and say when it lands.""" import prd_config try: prd_config.set_toggle(str(root), key, value) except Exception as exc: # surfaced to the caller as 400, never a traceback return {"ok": False, "key": key, "error": str(exc)[:400]} scope = apply_scope(key) return {"ok": True, "key": key, "value": prd_config.get(str(root), key), "apply_scope": scope, "notice": _notice(key, scope)} def _state(root): import prd_ui_export snapshot = prd_ui_export.build_snapshot(root) for toggle in snapshot.get("toggles", []): toggle["apply_scope"] = apply_scope(toggle["key"]) toggle["notice"] = _notice(toggle["key"], toggle["apply_scope"]) return snapshot def _handler(root): class Handler(BaseHTTPRequestHandler): server_version = "PRDPluginUI" def log_message(self, *args): # keep the console for the operator pass def _send(self, status, body, content_type): payload = body.encode("utf-8") if isinstance(body, str) else body self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(payload))) # This page is meant to be framed by a host shell on the same # machine, so framing is allowed - but only from a local origin. self.send_header("Content-Security-Policy", "frame-ancestors 'self' http://localhost:* http://127.0.0.1:*") self.send_header("X-Content-Type-Options", "nosniff") self.end_headers() self.wfile.write(payload) def _json(self, status, obj): self._send(status, json.dumps(obj, ensure_ascii=False), "application/json") def do_GET(self): path = self.path.split("?", 1)[0].rstrip("/") or "/" if path == "/": import prd_ui_export self._send(200, prd_ui_export.build_html(root, live=True), "text/html; charset=utf-8") elif path == "/api/state": self._json(200, _state(root)) else: # No file serving, deliberately. self._json(404, {"ok": False, "error": "not found"}) def do_POST(self): if self.path.split("?", 1)[0].rstrip("/") != "/api/toggle": self._json(404, {"ok": False, "error": "not found"}) return try: length = int(self.headers.get("Content-Length") or 0) except ValueError: length = 0 if length <= 0 or length > MAX_BODY: self._json(400, {"ok": False, "error": "missing or oversized body"}) return try: payload = json.loads(self.rfile.read(length).decode("utf-8")) except (ValueError, UnicodeDecodeError): self._json(400, {"ok": False, "error": "body must be JSON"}) return if not isinstance(payload, dict) or "key" not in payload: self._json(400, {"ok": False, "error": "expected {key, value}"}) return result = apply_toggle(root, payload.get("key"), payload.get("value")) self._json(200 if result.get("ok") else 400, result) return Handler def build_server(root, host=DEFAULT_HOST, port=DEFAULT_PORT): if host not in LOOPBACK: raise BindRefused( f"refusing to bind {host!r}: this server rewrites PRD Plugin configuration and " f"is loopback-only by design. Bind 127.0.0.1, or put a reviewed proxy in front " f"of it if a host app on another machine genuinely needs it.") return ThreadingHTTPServer((host, port), _handler(Path(root))) def main(argv=None): parser = argparse.ArgumentParser( description="Serve the PRD Plugin control surface on loopback.") parser.add_argument("--repo-root", default=".") parser.add_argument("--host", default=DEFAULT_HOST) parser.add_argument("--port", type=int, default=DEFAULT_PORT) args = parser.parse_args(argv) try: httpd = build_server(args.repo_root, host=args.host, port=args.port) except BindRefused as exc: print(f"[PRD Plugin] {exc}") return 2 host, port = httpd.server_address[0], httpd.server_address[1] print(f"[PRD Plugin] control surface on http://{host}:{port} (Ctrl+C to stop)") print(f"[PRD Plugin] embed it with