host.controller_web_editor.serve
HTTP server for the controller web editor.
stdlib http.server is sufficient: this is a localhost dev tool, not a
production service. Three concerns are wired together:
- Serve files out of
static/(the built SPA, or a placeholder page). - Expose a small JSON API under
/api/for config I/O. - Keep all filesystem access through
paths.safe_resolve.
Run with python -m host.controller_web_editor.
1"""HTTP server for the controller web editor. 2 3stdlib ``http.server`` is sufficient: this is a localhost dev tool, not a 4production service. Three concerns are wired together: 5 6* Serve files out of ``static/`` (the built SPA, or a placeholder page). 7* Expose a small JSON API under ``/api/`` for config I/O. 8* Keep all filesystem access through :mod:`paths.safe_resolve`. 9 10Run with ``python -m host.controller_web_editor``. 11""" 12 13from __future__ import annotations 14 15import argparse 16import json 17import logging 18from http import HTTPStatus 19from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer 20from pathlib import Path 21from urllib.parse import parse_qs, urlsplit 22 23from .build import maybe_build_spa 24from .paths import PathNotAllowedError, list_yaml_configs, safe_resolve 25from .routes.config import load_as_json, save_from_json 26from .routes.export import ExportRequestError, export_from_yaml 27from .routes.hitboxes import ( 28 LayoutNotAllowedError, 29 list_layouts, 30 load_layout, 31 save_layout, 32) 33from .routes.prefs import load_prefs, save_prefs 34 35 36log = logging.getLogger("controller_web_editor") 37 38STATIC_DIR = Path(__file__).resolve().parent / "static" 39 40# Shown in the browser when ``static/index.html`` is missing -- e.g. a 41# bare checkout without Node and without the committed bundle. Plain 42# string so we don't need a template file on disk. 43_MISSING_BUILD_HTML = """<!doctype html> 44<html><head><meta charset="utf-8"><title>Controller Editor: build missing</title> 45<style> 46body { font-family: system-ui, sans-serif; max-width: 40rem; margin: 4rem auto; 47 padding: 0 1rem; color: #222; } 48h1 { font-size: 1.4rem; } 49code, pre { background: #f3f3f3; padding: 0.1rem 0.4rem; border-radius: 4px; 50 font-family: Consolas, "SF Mono", monospace; } 51pre { padding: 0.6rem 0.8rem; overflow-x: auto; } 52.tip { color: #555; font-size: 0.9em; margin-top: 1.5rem; } 53</style></head><body> 54<h1>The SPA isn't built yet</h1> 55<p>The server is running but <code>static/index.html</code> doesn't exist, 56so there's nothing to serve at <code>/</code>.</p> 57<p>Build the frontend with:</p> 58<pre>cd host/controller_web_editor/web 59npm install 60npm run build</pre> 61<p>Then reload this page. After that, the server will rebuild 62automatically whenever you edit files under <code>web/src/</code>.</p> 63<p class="tip">If you don't have Node installed, grab it from 64<a href="https://nodejs.org">nodejs.org</a> (LTS is fine). Or pull a 65fresh checkout -- the compiled bundle is checked in.</p> 66</body></html> 67""" 68 69# MIME types we explicitly recognise. Anything else falls back to octet-stream. 70_MIME = { 71 ".html": "text/html; charset=utf-8", 72 ".js": "application/javascript; charset=utf-8", 73 ".mjs": "application/javascript; charset=utf-8", 74 ".css": "text/css; charset=utf-8", 75 ".json": "application/json; charset=utf-8", 76 ".svg": "image/svg+xml", 77 ".png": "image/png", 78 ".ico": "image/x-icon", 79 ".map": "application/json; charset=utf-8", 80} 81 82 83class Handler(BaseHTTPRequestHandler): 84 # Quieter access log; we'll route through logging instead of stderr. 85 def log_message(self, fmt, *args): # noqa: N802 (stdlib name) 86 log.info("%s - %s", self.address_string(), fmt % args) 87 88 # --- response helpers --- 89 90 def _send_json(self, status: int, payload: object) -> None: 91 body = json.dumps(payload).encode("utf-8") 92 self.send_response(status) 93 self.send_header("Content-Type", "application/json; charset=utf-8") 94 self.send_header("Content-Length", str(len(body))) 95 self.end_headers() 96 self.wfile.write(body) 97 98 def _send_error_json(self, status: int, message: str) -> None: 99 self._send_json(status, {"error": message}) 100 101 def _send_file(self, path: Path) -> None: 102 try: 103 data = path.read_bytes() 104 except OSError as exc: 105 self._send_error_json(HTTPStatus.NOT_FOUND, str(exc)) 106 return 107 self.send_response(HTTPStatus.OK) 108 self.send_header( 109 "Content-Type", _MIME.get(path.suffix.lower(), "application/octet-stream")) 110 self.send_header("Content-Length", str(len(data))) 111 self.send_header("Cache-Control", "no-store") 112 self.end_headers() 113 self.wfile.write(data) 114 115 # --- request entry points --- 116 117 def do_GET(self): # noqa: N802 (stdlib name) 118 parts = urlsplit(self.path) 119 query = parse_qs(parts.query) 120 121 if parts.path == "/api/configs": 122 self._send_json(HTTPStatus.OK, {"configs": list_yaml_configs()}) 123 return 124 125 if parts.path == "/api/config": 126 path_param = (query.get("path") or [""])[0] 127 try: 128 resolved = safe_resolve(path_param) 129 except PathNotAllowedError as exc: 130 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 131 return 132 if not resolved.is_file(): 133 self._send_error_json( 134 HTTPStatus.NOT_FOUND, f"no such file: {path_param}") 135 return 136 try: 137 self._send_json(HTTPStatus.OK, load_as_json(resolved)) 138 except Exception as exc: 139 log.exception("load failed for %s", resolved) 140 self._send_error_json( 141 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 142 return 143 144 if parts.path == "/api/hitboxes": 145 layout = (query.get("layout") or ["xbox"])[0] 146 try: 147 self._send_json(HTTPStatus.OK, load_layout(layout)) 148 except LayoutNotAllowedError as exc: 149 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 150 except FileNotFoundError as exc: 151 self._send_error_json(HTTPStatus.NOT_FOUND, str(exc)) 152 except Exception as exc: 153 log.exception("hitboxes load failed for %s", layout) 154 self._send_error_json( 155 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 156 return 157 158 if parts.path == "/api/hitboxes/layouts": 159 self._send_json(HTTPStatus.OK, {"layouts": list_layouts()}) 160 return 161 162 if parts.path == "/api/prefs": 163 self._send_json(HTTPStatus.OK, load_prefs()) 164 return 165 166 if parts.path == "/api/export": 167 path_param = (query.get("path") or [""])[0] 168 orientation = (query.get("orientation") or ["landscape"])[0] 169 fmt = (query.get("format") or ["pdf"])[0] 170 hide_raw = (query.get("hide_unassigned") or ["0"])[0].lower() 171 hide_unassigned = hide_raw in ("1", "true", "yes", "on") 172 try: 173 resolved = safe_resolve(path_param) 174 except PathNotAllowedError as exc: 175 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 176 return 177 if not resolved.is_file(): 178 self._send_error_json( 179 HTTPStatus.NOT_FOUND, f"no such file: {path_param}") 180 return 181 try: 182 data, content_type, filename = export_from_yaml( 183 resolved, orientation, fmt, hide_unassigned) 184 except ExportRequestError as exc: 185 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 186 return 187 except ValueError as exc: 188 # No controllers, multi-page PNG, etc. 189 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 190 return 191 except Exception as exc: 192 log.exception("export failed for %s", resolved) 193 self._send_error_json( 194 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 195 return 196 self.send_response(HTTPStatus.OK) 197 self.send_header("Content-Type", content_type) 198 self.send_header("Content-Length", str(len(data))) 199 self.send_header( 200 "Content-Disposition", f'attachment; filename="{filename}"') 201 self.send_header("Cache-Control", "no-store") 202 self.end_headers() 203 self.wfile.write(data) 204 return 205 206 # Static files. Default route serves index.html. 207 rel = parts.path.lstrip("/") or "index.html" 208 target = (STATIC_DIR / rel).resolve() 209 try: 210 target.relative_to(STATIC_DIR.resolve()) 211 except ValueError: 212 self._send_error_json(HTTPStatus.FORBIDDEN, "outside static root") 213 return 214 if not target.is_file(): 215 # SPA fallback — anything unknown returns index.html so the 216 # frontend router can decide. 217 target = STATIC_DIR / "index.html" 218 if not target.is_file(): 219 # Friendly explainer when nobody has built the bundle. 220 body = _MISSING_BUILD_HTML.encode("utf-8") 221 self.send_response(HTTPStatus.OK) 222 self.send_header("Content-Type", "text/html; charset=utf-8") 223 self.send_header("Content-Length", str(len(body))) 224 self.send_header("Cache-Control", "no-store") 225 self.end_headers() 226 self.wfile.write(body) 227 return 228 self._send_file(target) 229 230 def do_POST(self): # noqa: N802 (stdlib name) 231 parts = urlsplit(self.path) 232 query = parse_qs(parts.query) 233 234 if parts.path == "/api/config": 235 path_param = (query.get("path") or [""])[0] 236 try: 237 resolved = safe_resolve(path_param) 238 except PathNotAllowedError as exc: 239 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 240 return 241 length = int(self.headers.get("Content-Length", "0")) 242 try: 243 payload = json.loads(self.rfile.read(length).decode("utf-8")) 244 except json.JSONDecodeError as exc: 245 self._send_error_json( 246 HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}") 247 return 248 try: 249 save_from_json(resolved, payload) 250 except Exception as exc: 251 log.exception("save failed for %s", resolved) 252 self._send_error_json( 253 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 254 return 255 self._send_json(HTTPStatus.OK, {"saved": path_param}) 256 return 257 258 if parts.path == "/api/hitboxes": 259 layout = (query.get("layout") or ["xbox"])[0] 260 length = int(self.headers.get("Content-Length", "0")) 261 try: 262 payload = json.loads(self.rfile.read(length).decode("utf-8")) 263 except json.JSONDecodeError as exc: 264 self._send_error_json( 265 HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}") 266 return 267 try: 268 save_layout(layout, payload) 269 except LayoutNotAllowedError as exc: 270 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 271 return 272 except ValueError as exc: 273 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 274 return 275 except Exception as exc: 276 log.exception("hitboxes save failed for %s", layout) 277 self._send_error_json( 278 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 279 return 280 self._send_json(HTTPStatus.OK, {"saved": layout}) 281 return 282 283 if parts.path == "/api/prefs": 284 length = int(self.headers.get("Content-Length", "0")) 285 try: 286 payload = json.loads(self.rfile.read(length).decode("utf-8")) 287 except json.JSONDecodeError as exc: 288 self._send_error_json( 289 HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}") 290 return 291 if not isinstance(payload, dict): 292 self._send_error_json( 293 HTTPStatus.BAD_REQUEST, "payload must be a JSON object") 294 return 295 try: 296 updated = save_prefs(payload) 297 except ValueError as exc: 298 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 299 return 300 except Exception as exc: 301 log.exception("prefs save failed") 302 self._send_error_json( 303 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 304 return 305 self._send_json(HTTPStatus.OK, updated) 306 return 307 308 self._send_error_json(HTTPStatus.NOT_FOUND, parts.path) 309 310 311def run(host: str = "127.0.0.1", port: int = 8071) -> None: 312 logging.basicConfig( 313 level=logging.INFO, 314 format="%(asctime)s %(levelname)s %(name)s: %(message)s", 315 ) 316 # Auto-build the SPA if web/src/ is newer than static/. Best-effort; 317 # skipped silently when Node isn't installed or 318 # CONTROLLER_WEB_EDITOR_SKIP_BUILD is set. 319 status = maybe_build_spa() 320 log.info("spa build: %s", status) 321 server = ThreadingHTTPServer((host, port), Handler) 322 log.info("serving on http://%s:%d (static=%s)", host, port, STATIC_DIR) 323 try: 324 server.serve_forever() 325 except KeyboardInterrupt: 326 log.info("shutting down") 327 finally: 328 server.server_close() 329 330 331def main(argv: list[str] | None = None) -> int: 332 parser = argparse.ArgumentParser(description=__doc__) 333 parser.add_argument("--host", default="127.0.0.1") 334 parser.add_argument("--port", type=int, default=8071) 335 args = parser.parse_args(argv) 336 run(args.host, args.port) 337 return 0
84class Handler(BaseHTTPRequestHandler): 85 # Quieter access log; we'll route through logging instead of stderr. 86 def log_message(self, fmt, *args): # noqa: N802 (stdlib name) 87 log.info("%s - %s", self.address_string(), fmt % args) 88 89 # --- response helpers --- 90 91 def _send_json(self, status: int, payload: object) -> None: 92 body = json.dumps(payload).encode("utf-8") 93 self.send_response(status) 94 self.send_header("Content-Type", "application/json; charset=utf-8") 95 self.send_header("Content-Length", str(len(body))) 96 self.end_headers() 97 self.wfile.write(body) 98 99 def _send_error_json(self, status: int, message: str) -> None: 100 self._send_json(status, {"error": message}) 101 102 def _send_file(self, path: Path) -> None: 103 try: 104 data = path.read_bytes() 105 except OSError as exc: 106 self._send_error_json(HTTPStatus.NOT_FOUND, str(exc)) 107 return 108 self.send_response(HTTPStatus.OK) 109 self.send_header( 110 "Content-Type", _MIME.get(path.suffix.lower(), "application/octet-stream")) 111 self.send_header("Content-Length", str(len(data))) 112 self.send_header("Cache-Control", "no-store") 113 self.end_headers() 114 self.wfile.write(data) 115 116 # --- request entry points --- 117 118 def do_GET(self): # noqa: N802 (stdlib name) 119 parts = urlsplit(self.path) 120 query = parse_qs(parts.query) 121 122 if parts.path == "/api/configs": 123 self._send_json(HTTPStatus.OK, {"configs": list_yaml_configs()}) 124 return 125 126 if parts.path == "/api/config": 127 path_param = (query.get("path") or [""])[0] 128 try: 129 resolved = safe_resolve(path_param) 130 except PathNotAllowedError as exc: 131 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 132 return 133 if not resolved.is_file(): 134 self._send_error_json( 135 HTTPStatus.NOT_FOUND, f"no such file: {path_param}") 136 return 137 try: 138 self._send_json(HTTPStatus.OK, load_as_json(resolved)) 139 except Exception as exc: 140 log.exception("load failed for %s", resolved) 141 self._send_error_json( 142 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 143 return 144 145 if parts.path == "/api/hitboxes": 146 layout = (query.get("layout") or ["xbox"])[0] 147 try: 148 self._send_json(HTTPStatus.OK, load_layout(layout)) 149 except LayoutNotAllowedError as exc: 150 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 151 except FileNotFoundError as exc: 152 self._send_error_json(HTTPStatus.NOT_FOUND, str(exc)) 153 except Exception as exc: 154 log.exception("hitboxes load failed for %s", layout) 155 self._send_error_json( 156 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 157 return 158 159 if parts.path == "/api/hitboxes/layouts": 160 self._send_json(HTTPStatus.OK, {"layouts": list_layouts()}) 161 return 162 163 if parts.path == "/api/prefs": 164 self._send_json(HTTPStatus.OK, load_prefs()) 165 return 166 167 if parts.path == "/api/export": 168 path_param = (query.get("path") or [""])[0] 169 orientation = (query.get("orientation") or ["landscape"])[0] 170 fmt = (query.get("format") or ["pdf"])[0] 171 hide_raw = (query.get("hide_unassigned") or ["0"])[0].lower() 172 hide_unassigned = hide_raw in ("1", "true", "yes", "on") 173 try: 174 resolved = safe_resolve(path_param) 175 except PathNotAllowedError as exc: 176 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 177 return 178 if not resolved.is_file(): 179 self._send_error_json( 180 HTTPStatus.NOT_FOUND, f"no such file: {path_param}") 181 return 182 try: 183 data, content_type, filename = export_from_yaml( 184 resolved, orientation, fmt, hide_unassigned) 185 except ExportRequestError as exc: 186 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 187 return 188 except ValueError as exc: 189 # No controllers, multi-page PNG, etc. 190 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 191 return 192 except Exception as exc: 193 log.exception("export failed for %s", resolved) 194 self._send_error_json( 195 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 196 return 197 self.send_response(HTTPStatus.OK) 198 self.send_header("Content-Type", content_type) 199 self.send_header("Content-Length", str(len(data))) 200 self.send_header( 201 "Content-Disposition", f'attachment; filename="{filename}"') 202 self.send_header("Cache-Control", "no-store") 203 self.end_headers() 204 self.wfile.write(data) 205 return 206 207 # Static files. Default route serves index.html. 208 rel = parts.path.lstrip("/") or "index.html" 209 target = (STATIC_DIR / rel).resolve() 210 try: 211 target.relative_to(STATIC_DIR.resolve()) 212 except ValueError: 213 self._send_error_json(HTTPStatus.FORBIDDEN, "outside static root") 214 return 215 if not target.is_file(): 216 # SPA fallback — anything unknown returns index.html so the 217 # frontend router can decide. 218 target = STATIC_DIR / "index.html" 219 if not target.is_file(): 220 # Friendly explainer when nobody has built the bundle. 221 body = _MISSING_BUILD_HTML.encode("utf-8") 222 self.send_response(HTTPStatus.OK) 223 self.send_header("Content-Type", "text/html; charset=utf-8") 224 self.send_header("Content-Length", str(len(body))) 225 self.send_header("Cache-Control", "no-store") 226 self.end_headers() 227 self.wfile.write(body) 228 return 229 self._send_file(target) 230 231 def do_POST(self): # noqa: N802 (stdlib name) 232 parts = urlsplit(self.path) 233 query = parse_qs(parts.query) 234 235 if parts.path == "/api/config": 236 path_param = (query.get("path") or [""])[0] 237 try: 238 resolved = safe_resolve(path_param) 239 except PathNotAllowedError as exc: 240 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 241 return 242 length = int(self.headers.get("Content-Length", "0")) 243 try: 244 payload = json.loads(self.rfile.read(length).decode("utf-8")) 245 except json.JSONDecodeError as exc: 246 self._send_error_json( 247 HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}") 248 return 249 try: 250 save_from_json(resolved, payload) 251 except Exception as exc: 252 log.exception("save failed for %s", resolved) 253 self._send_error_json( 254 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 255 return 256 self._send_json(HTTPStatus.OK, {"saved": path_param}) 257 return 258 259 if parts.path == "/api/hitboxes": 260 layout = (query.get("layout") or ["xbox"])[0] 261 length = int(self.headers.get("Content-Length", "0")) 262 try: 263 payload = json.loads(self.rfile.read(length).decode("utf-8")) 264 except json.JSONDecodeError as exc: 265 self._send_error_json( 266 HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}") 267 return 268 try: 269 save_layout(layout, payload) 270 except LayoutNotAllowedError as exc: 271 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 272 return 273 except ValueError as exc: 274 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 275 return 276 except Exception as exc: 277 log.exception("hitboxes save failed for %s", layout) 278 self._send_error_json( 279 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 280 return 281 self._send_json(HTTPStatus.OK, {"saved": layout}) 282 return 283 284 if parts.path == "/api/prefs": 285 length = int(self.headers.get("Content-Length", "0")) 286 try: 287 payload = json.loads(self.rfile.read(length).decode("utf-8")) 288 except json.JSONDecodeError as exc: 289 self._send_error_json( 290 HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}") 291 return 292 if not isinstance(payload, dict): 293 self._send_error_json( 294 HTTPStatus.BAD_REQUEST, "payload must be a JSON object") 295 return 296 try: 297 updated = save_prefs(payload) 298 except ValueError as exc: 299 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 300 return 301 except Exception as exc: 302 log.exception("prefs save failed") 303 self._send_error_json( 304 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 305 return 306 self._send_json(HTTPStatus.OK, updated) 307 return 308 309 self._send_error_json(HTTPStatus.NOT_FOUND, parts.path)
HTTP request handler base class.
The following explanation of HTTP serves to guide you through the code as well as to expose any misunderstandings I may have about HTTP (so you don't need to read the code to figure out I'm wrong :-).
HTTP (HyperText Transfer Protocol) is an extensible protocol on top of a reliable stream transport (e.g. TCP/IP). The protocol recognizes three parts to a request:
- One line identifying the request type and path
- An optional set of RFC-822-style headers
- An optional data part
The headers and data are separated by a blank line.
The first line of the request has the form
where
The specification specifies that lines are separated by CRLF but for compatibility with the widest range of clients recommends servers also handle LF. Similarly, whitespace in the request line is treated sensibly (allowing multiple spaces between components and allowing trailing whitespace).
Similarly, for output, lines ought to be separated by CRLF pairs but most clients grok LF characters just fine.
If the first line of the request has the form
(i.e.
The reply form of the HTTP 1.x protocol again has three parts:
- One line giving the response code
- An optional set of RFC-822-style headers
- The data
Again, the headers and data are separated by a blank line.
The response code line has the form
where
This server parses the request and the headers, and then calls a
function specific to the request type (
do_SPAM()
Note that the request name is case sensitive (i.e. SPAM and spam are different requests).
The various request details are stored in instance variables:
client_address is the client IP address in the form (host, port);
command, path and version are the broken-down request line;
headers is an instance of email.message.Message (or a derived class) containing the header information;
rfile is a file object open for reading positioned at the start of the optional input data part;
wfile is a file object open for writing.
IT IS IMPORTANT TO ADHERE TO THE PROTOCOL FOR WRITING!
The first thing to be written must be the response line. Then follow 0 or more header lines, then a blank line, and then the actual data (if any). The meaning of the header lines depends on the command executed by the server; in most cases, when data is returned, there should be at least one header line of the form
Content-type:
where
86 def log_message(self, fmt, *args): # noqa: N802 (stdlib name) 87 log.info("%s - %s", self.address_string(), fmt % args)
Log an arbitrary message.
This is used by all other logging functions. Override it if you have specific logging wishes.
The first argument, FORMAT, is a format string for the message to be logged. If the format string contains any % escapes requiring parameters, they should be specified as subsequent arguments (it's just like printf!).
The client ip and current date/time are prefixed to every message.
Unicode control characters are replaced with escaped hex before writing the output to stderr.
118 def do_GET(self): # noqa: N802 (stdlib name) 119 parts = urlsplit(self.path) 120 query = parse_qs(parts.query) 121 122 if parts.path == "/api/configs": 123 self._send_json(HTTPStatus.OK, {"configs": list_yaml_configs()}) 124 return 125 126 if parts.path == "/api/config": 127 path_param = (query.get("path") or [""])[0] 128 try: 129 resolved = safe_resolve(path_param) 130 except PathNotAllowedError as exc: 131 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 132 return 133 if not resolved.is_file(): 134 self._send_error_json( 135 HTTPStatus.NOT_FOUND, f"no such file: {path_param}") 136 return 137 try: 138 self._send_json(HTTPStatus.OK, load_as_json(resolved)) 139 except Exception as exc: 140 log.exception("load failed for %s", resolved) 141 self._send_error_json( 142 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 143 return 144 145 if parts.path == "/api/hitboxes": 146 layout = (query.get("layout") or ["xbox"])[0] 147 try: 148 self._send_json(HTTPStatus.OK, load_layout(layout)) 149 except LayoutNotAllowedError as exc: 150 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 151 except FileNotFoundError as exc: 152 self._send_error_json(HTTPStatus.NOT_FOUND, str(exc)) 153 except Exception as exc: 154 log.exception("hitboxes load failed for %s", layout) 155 self._send_error_json( 156 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 157 return 158 159 if parts.path == "/api/hitboxes/layouts": 160 self._send_json(HTTPStatus.OK, {"layouts": list_layouts()}) 161 return 162 163 if parts.path == "/api/prefs": 164 self._send_json(HTTPStatus.OK, load_prefs()) 165 return 166 167 if parts.path == "/api/export": 168 path_param = (query.get("path") or [""])[0] 169 orientation = (query.get("orientation") or ["landscape"])[0] 170 fmt = (query.get("format") or ["pdf"])[0] 171 hide_raw = (query.get("hide_unassigned") or ["0"])[0].lower() 172 hide_unassigned = hide_raw in ("1", "true", "yes", "on") 173 try: 174 resolved = safe_resolve(path_param) 175 except PathNotAllowedError as exc: 176 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 177 return 178 if not resolved.is_file(): 179 self._send_error_json( 180 HTTPStatus.NOT_FOUND, f"no such file: {path_param}") 181 return 182 try: 183 data, content_type, filename = export_from_yaml( 184 resolved, orientation, fmt, hide_unassigned) 185 except ExportRequestError as exc: 186 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 187 return 188 except ValueError as exc: 189 # No controllers, multi-page PNG, etc. 190 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 191 return 192 except Exception as exc: 193 log.exception("export failed for %s", resolved) 194 self._send_error_json( 195 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 196 return 197 self.send_response(HTTPStatus.OK) 198 self.send_header("Content-Type", content_type) 199 self.send_header("Content-Length", str(len(data))) 200 self.send_header( 201 "Content-Disposition", f'attachment; filename="{filename}"') 202 self.send_header("Cache-Control", "no-store") 203 self.end_headers() 204 self.wfile.write(data) 205 return 206 207 # Static files. Default route serves index.html. 208 rel = parts.path.lstrip("/") or "index.html" 209 target = (STATIC_DIR / rel).resolve() 210 try: 211 target.relative_to(STATIC_DIR.resolve()) 212 except ValueError: 213 self._send_error_json(HTTPStatus.FORBIDDEN, "outside static root") 214 return 215 if not target.is_file(): 216 # SPA fallback — anything unknown returns index.html so the 217 # frontend router can decide. 218 target = STATIC_DIR / "index.html" 219 if not target.is_file(): 220 # Friendly explainer when nobody has built the bundle. 221 body = _MISSING_BUILD_HTML.encode("utf-8") 222 self.send_response(HTTPStatus.OK) 223 self.send_header("Content-Type", "text/html; charset=utf-8") 224 self.send_header("Content-Length", str(len(body))) 225 self.send_header("Cache-Control", "no-store") 226 self.end_headers() 227 self.wfile.write(body) 228 return 229 self._send_file(target)
231 def do_POST(self): # noqa: N802 (stdlib name) 232 parts = urlsplit(self.path) 233 query = parse_qs(parts.query) 234 235 if parts.path == "/api/config": 236 path_param = (query.get("path") or [""])[0] 237 try: 238 resolved = safe_resolve(path_param) 239 except PathNotAllowedError as exc: 240 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 241 return 242 length = int(self.headers.get("Content-Length", "0")) 243 try: 244 payload = json.loads(self.rfile.read(length).decode("utf-8")) 245 except json.JSONDecodeError as exc: 246 self._send_error_json( 247 HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}") 248 return 249 try: 250 save_from_json(resolved, payload) 251 except Exception as exc: 252 log.exception("save failed for %s", resolved) 253 self._send_error_json( 254 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 255 return 256 self._send_json(HTTPStatus.OK, {"saved": path_param}) 257 return 258 259 if parts.path == "/api/hitboxes": 260 layout = (query.get("layout") or ["xbox"])[0] 261 length = int(self.headers.get("Content-Length", "0")) 262 try: 263 payload = json.loads(self.rfile.read(length).decode("utf-8")) 264 except json.JSONDecodeError as exc: 265 self._send_error_json( 266 HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}") 267 return 268 try: 269 save_layout(layout, payload) 270 except LayoutNotAllowedError as exc: 271 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 272 return 273 except ValueError as exc: 274 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 275 return 276 except Exception as exc: 277 log.exception("hitboxes save failed for %s", layout) 278 self._send_error_json( 279 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 280 return 281 self._send_json(HTTPStatus.OK, {"saved": layout}) 282 return 283 284 if parts.path == "/api/prefs": 285 length = int(self.headers.get("Content-Length", "0")) 286 try: 287 payload = json.loads(self.rfile.read(length).decode("utf-8")) 288 except json.JSONDecodeError as exc: 289 self._send_error_json( 290 HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}") 291 return 292 if not isinstance(payload, dict): 293 self._send_error_json( 294 HTTPStatus.BAD_REQUEST, "payload must be a JSON object") 295 return 296 try: 297 updated = save_prefs(payload) 298 except ValueError as exc: 299 self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) 300 return 301 except Exception as exc: 302 log.exception("prefs save failed") 303 self._send_error_json( 304 HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) 305 return 306 self._send_json(HTTPStatus.OK, updated) 307 return 308 309 self._send_error_json(HTTPStatus.NOT_FOUND, parts.path)
312def run(host: str = "127.0.0.1", port: int = 8071) -> None: 313 logging.basicConfig( 314 level=logging.INFO, 315 format="%(asctime)s %(levelname)s %(name)s: %(message)s", 316 ) 317 # Auto-build the SPA if web/src/ is newer than static/. Best-effort; 318 # skipped silently when Node isn't installed or 319 # CONTROLLER_WEB_EDITOR_SKIP_BUILD is set. 320 status = maybe_build_spa() 321 log.info("spa build: %s", status) 322 server = ThreadingHTTPServer((host, port), Handler) 323 log.info("serving on http://%s:%d (static=%s)", host, port, STATIC_DIR) 324 try: 325 server.serve_forever() 326 except KeyboardInterrupt: 327 log.info("shutting down") 328 finally: 329 server.server_close()