EtherGold is serving :31234, but local DF HTTP "
f"({_df_local_base()}) did not respond.
Restart DyzineFrame from the HDMI Supervisor, then retry " "/ohd/setup or " "/ohd/modes.
" "EtherGold Status still works.
" "" ) return html.encode("utf-8"), "text/html; charset=utf-8", 502 def proxy_ohd_to_dyzineframe( *, method: str, path_with_query: str, body: Optional[bytes] = None, content_type: Optional[str] = None, ) -> tuple[int, Dict[str, str], bytes]: """Forward one OHD request to DF localhost HTTP. Raises OSError/URLError on failure.""" parsed = urlparse(path_with_query if path_with_query.startswith("/") else f"/{path_with_query}") path = parsed.path or "/ohd" query = f"?{parsed.query}" if parsed.query else "" url = f"{_df_local_base()}{path}{query}" data = body if method.upper() in ("POST", "PUT", "PATCH") else None headers: Dict[str, str] = {"Accept": "*/*"} if data is not None and content_type: headers["Content-Type"] = content_type req = urllib.request.Request(url, data=data, headers=headers, method=method.upper()) with urllib.request.urlopen(req, timeout=DF_OHD_PROXY_TIMEOUT_S) as resp: out_headers = { "Content-Type": resp.headers.get("Content-Type") or "application/octet-stream", } loc = resp.headers.get("Location") if loc: out_headers["Location"] = loc return int(resp.status), out_headers, resp.read() def _purge_ui_sessions(now: float | None = None) -> None: stamp = time.time() if now is None else now expired = [token for token, exp in _UI_SESSIONS.items() if exp < stamp] for token in expired: _UI_SESSIONS.pop(token, None) def mint_ui_session() -> str: """Opaque UI session; lives only in this HTTP process (dies on reboot/restart).""" token = secrets.token_urlsafe(24) with _UI_SESSION_LOCK: _purge_ui_sessions() _UI_SESSIONS[token] = time.time() + _UI_SESSION_TTL_S return token def ui_session_ok(token: str | None) -> bool: if not token: return False with _UI_SESSION_LOCK: _purge_ui_sessions() exp = _UI_SESSIONS.get(token) return exp is not None and exp >= time.time() def reset_ui_sessions_for_tests() -> None: with _UI_SESSION_LOCK: _UI_SESSIONS.clear() _LISTEN_HEALTH_CACHE: tuple[float, Dict[str, Any]] | None = None _LISTEN_HEALTH_TTL_S = 2.0 def _etc_dir() -> Path: raw = os.environ.get("ETHERGOLD_ETC", "").strip() return Path(raw) if raw else Path("/etc/ethergold") def _env_file_get(key: str) -> str: etc = _etc_dir() candidates = [ etc / "ethergold.env", etc / "role.env", etc / "edge.env", Path(os.environ.get("ETHERGOLD_ROOT", "/home/topbox/EtherGold")) / "fixture-bridge.env", Path.home() / "EtherGold" / "fixture-bridge.env", ] for path in candidates: if not path.is_file(): continue try: for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.strip() or line.lstrip().startswith("#") or "=" not in line: continue k, _, v = line.partition("=") if k.strip() == key: return v.strip().strip('"').strip("'") except OSError: continue return "" def resolve_health_mode() -> Dict[str, Any]: """Map field config / role.env to Health card mode: listen | bridge.""" profile = ( os.environ.get("FIELD_PROFILE") or _env_file_get("FIELD_PROFILE") or "" ).strip() role = ( os.environ.get("ETHERGOLD_ROLE") or _env_file_get("ETHERGOLD_ROLE") or "" ).strip().lower() edge_mode = ( os.environ.get("EDGE_MODE") or _env_file_get("EDGE_MODE") or "0" ).strip() node_role = ( os.environ.get("EDGE_ROLE") or os.environ.get("NODE_ROLE") or _env_file_get("EDGE_ROLE") or _env_file_get("NODE_ROLE") or "" ).strip() wants_df = False cfg = None if load_field_config is not None: try: cfg = load_field_config(_etc_dir()) profile = (cfg.profile or profile or "listen_ohd").strip() if not node_role: node_role = (cfg.node_role or "").strip() if is_listen_profile is not None and is_listen_profile(cfg): role = "listen" elif not role: role = "edge" if profile_wants_dyzineframe is not None: wants_df = bool(profile_wants_dyzineframe(cfg)) except Exception: cfg = None if not profile: profile = "listen_ohd" if role == "listen" else "atlas_wifi_star" listen_profiles = ("listen_ohd", "listen_ohd_server", "listen_ohd_mini") if profile in listen_profiles or role == "listen": mode = "listen" else: mode = "bridge" if edge_mode == "1" and mode != "listen": mode = "bridge" if profile == "listen_ohd_server": wants_df = False elif profile in ("listen_ohd", "listen_ohd_mini", "atlas_wifi_star") and cfg is None: wants_df = True return { "mode": mode, "profile": profile, "role": role or ("listen" if mode == "listen" else "edge"), "edge_mode": edge_mode, "node_role": node_role, "wants_dyzineframe": wants_df, "wants_collector": mode == "listen", "wants_listen": mode == "listen", "expect_inject_0": mode == "listen", } def _uartd_socks(pos: int) -> list[str]: named = [ f"/tmp/pog422_uart_{pos}.sock", f"/tmp/pog422_uart_{pos}_machine.sock", f"/tmp/pog422_uart_{pos}_ois.sock", os.environ.get("POG422_UARTD_SOCK", "").strip(), ] found = [p for p in named if p and os.path.exists(p)] if found: return found return sorted(glob.glob("/tmp/pog422_uart*.sock")) def _live_uartd_inject_flag() -> str: """Detect inject mode from running bridge/listen, else env. This is the **software** flag (POG422_UARTD_INJECT / --uartd-inject): whether uartd is allowed to TX/inject on copper — not a live packet counter. """ # Prefer live cmdline (Start bridge may enable inject without rewriting env). for pat in ("pog422_bridge.py", "pog422_listen.py", "pog422_uartd"): try: out = subprocess.run( ["pgrep", "-af", pat], capture_output=True, text=True, check=False, timeout=2, ) except (OSError, subprocess.TimeoutExpired): continue for line in (out.stdout or "").splitlines(): low = line.lower() if "--uartd-inject" in low or "pog422_uartd_inject=1" in low: return "1" if "pog422_uartd_inject=0" in low: return "0" # Bridge with --uartd but without --uartd-inject is still often inject=1 # via env; do not infer 0 from cmdline alone. env = ( os.environ.get("POG422_UARTD_INJECT") or _env_file_get("POG422_UARTD_INJECT") or "" ).strip().lower() return env def _normalize_star_role(node_role: str, profile: str) -> str: """Return hub | spoke | side_a | side_b | '' for Health cards.""" r = (node_role or "").strip().lower() if r in ("master_hub", "hub"): return "hub" if r in ("slave_spoke", "spoke"): return "spoke" if r in ("side_a", "side_b"): return r if profile == "atlas_wifi_star": return "" # unknown — don't guess hub vs spoke return r def _pgrep_running(pattern: str) -> list[int]: try: out = subprocess.run( ["pgrep", "-f", pattern], capture_output=True, text=True, check=False, timeout=2, ) except (OSError, subprocess.TimeoutExpired): return [] return [int(line.strip()) for line in out.stdout.splitlines() if line.strip().isdigit()] def _systemctl_show(unit: str, prop: str) -> str: try: out = subprocess.run( ["systemctl", "show", unit, f"-p{prop}", "--value"], capture_output=True, text=True, check=False, timeout=3, ) except (OSError, subprocess.TimeoutExpired): return "" return (out.stdout or "").strip() def _systemctl_enabled(unit: str) -> str: """`systemctl is-enabled` — enabled / disabled / not-found / masked / "".""" try: out = subprocess.run( ["systemctl", "is-enabled", unit], capture_output=True, text=True, check=False, timeout=3, ) except (OSError, subprocess.TimeoutExpired): return "" return (out.stdout or "").strip() def _file_age_s(path: str) -> float | None: try: return max(0.0, time.time() - os.path.getmtime(path)) except OSError: return None def _csv_path() -> str: return os.environ.get( "ETHERGOLD_PROGRESSIVES_CSV", os.path.join(os.environ.get("ETHERGOLD_ROOT", "/home/topbox/EtherGold"), "progressives.csv"), ) def collect_listen_health() -> Dict[str, Any]: """Runtime health for GET /health — listen or bridge by field config/role.""" global _LISTEN_HEALTH_CACHE now = time.time() if _LISTEN_HEALTH_CACHE is not None: cached_at, cached = _LISTEN_HEALTH_CACHE if now - cached_at < _LISTEN_HEALTH_TTL_S: return cached mode_info = resolve_health_mode() mode = mode_info["mode"] pos_raw = ( os.environ.get("FIXTURE_POS") or _env_file_get("FIXTURE_POS") or "1" ) try: pos = int(pos_raw) except ValueError: pos = 1 inject_raw = _live_uartd_inject_flag() inject_env = ( os.environ.get("POG422_UARTD_INJECT") or _env_file_get("POG422_UARTD_INJECT") or "" ).strip().lower() star_role = _normalize_star_role( str(mode_info.get("node_role") or ""), str(mode_info.get("profile") or ""), ) tty = sorted(glob.glob("/dev/ttyXRUSB*")) socks = _uartd_socks(pos) collector_active = _systemctl_show("ethergold-collector.service", "ActiveState") == "active" collector_user = _systemctl_show("ethergold-collector.service", "User") collector_enabled = _systemctl_enabled("ethergold-collector.service") == "enabled" supervisor_pids = _pgrep_running("ethergold_listen_supervisor.py") listen_pids = _pgrep_running("pog422_listen.py") bridge_pids = _pgrep_running("pog422_bridge.py") run_eg_pids = _pgrep_running("run_ethergold.sh") df_pids = _pgrep_running("DyzineFrame.py") + _pgrep_running("dyzineframe_supervisor.py") csv_age = _file_age_s(_csv_path()) cg_age = _file_age_s(resolve_current_game_path()) reasons: list[str] = [] inject_ok = False # Who is expected to own listen. A bench seat runs it by hand (Run EG / # run_ethergold.sh) with the collector unit left disabled — that is a valid # configuration, not a fault, so long as pog422_listen.py is actually up. # A provisioned field seat has the unit enabled; there, inactive is a fault. manual_listen = ( mode == "listen" and not collector_active and not collector_enabled and bool(listen_pids) ) worker_owner = "collector" if collector_active else ("manual" if manual_listen else "none") if mode == "listen": inject_ok = inject_raw in ("0", "false", "no", "off") if mode_info.get("wants_collector") and not collector_active and not manual_listen: reasons.append("collector_inactive") # The listen supervisor belongs to the systemd-managed seat only. if not supervisor_pids and not manual_listen: reasons.append("supervisor_down") if not listen_pids: reasons.append("listen_down") if not inject_ok: reasons.append("inject_must_be_0") # Local DF expected on listen_ohd / mini (not listen_ohd_server). if mode_info.get("wants_dyzineframe") and not df_pids: reasons.append("dyzineframe_down") else: # Edge / wifi star / wifi bank bridge: software inject path must be on. # Prefer live --uartd-inject; env=0 alone is not a fault if live is 1. inject_ok = inject_raw in ("1", "true", "yes", "on") if not inject_ok and inject_raw == "" and bridge_pids: # Bridge running without an explicit flag — treat as OK if cmdline # had --uartd-inject (already folded into inject_raw) else warn. inject_ok = False reasons.append("inject_expected_1") elif not inject_ok: reasons.append("inject_expected_1") if not bridge_pids and not run_eg_pids: reasons.append("bridge_down") # Bridge health does not require local DF (OHD / no-monitor soak OK). if not tty: reasons.append("no_ttyXRUSB") if not socks: reasons.append("no_uartd_sock") if csv_age is None: reasons.append("progressives_csv_missing") else: try: if os.path.getsize(_csv_path()) <= 0: reasons.append("progressives_csv_missing") except OSError: reasons.append("progressives_csv_missing") payload: Dict[str, Any] = { "ok": not reasons, "reasons": reasons, "mode": mode, "profile": mode_info.get("profile"), "role": mode_info.get("role"), "edge_mode": mode_info.get("edge_mode"), "node_role": mode_info.get("node_role"), "star_role": star_role, "wants_dyzineframe": bool(mode_info.get("wants_dyzineframe")), "collector_active": collector_active, "collector_enabled": collector_enabled, "collector_user": collector_user or None, "manual_listen": manual_listen, "worker_owner": worker_owner, "supervisor_pids": supervisor_pids, "listen_pids": listen_pids, "bridge_pids": bridge_pids, "run_ethergold_pids": run_eg_pids, "dyzineframe_pids": df_pids, "tty_xrusb": tty, "uartd_socks": socks, # Software TX path flag (not "packets on the wire right now"). "inject": inject_raw, "inject_env": inject_env, "inject_ok": inject_ok, "inject_meaning": ( "uartd_software_inject_path — 1 allows eth→UART TX on copper; " "0 is listen-only (must not TX on daisy)" ), "progressives_csv_age_s": csv_age, "current_game_age_s": cg_age, "fixture_pos": pos, } _LISTEN_HEALTH_CACHE = (now, payload) return payload def _pogcc_com_owner() -> str: path = os.path.expanduser("~/DyzineFrame/config.ini") try: text = Path(path).read_text(encoding="utf-8", errors="replace") except OSError: return "" for line in text.splitlines(): stripped = line.strip() if stripped.lower().startswith("pogcc_com_owner"): _, _, val = stripped.partition("=") return val.strip().lower() return "" def _read_json_obj(path: str) -> Dict[str, Any]: try: with open(path, encoding="utf-8") as handle: payload = json.load(handle) return payload if isinstance(payload, dict) else {} except (OSError, ValueError, json.JSONDecodeError): return {} def collect_services_health(listen: Dict[str, Any] | None = None) -> Dict[str, Any]: """Per-service breakdown for GET /health. Product stamp stays EG_VERSION.""" listen = listen if listen is not None else collect_listen_health() owner = _pogcc_com_owner() pogcc_active = _systemctl_show("ethergold-pogcc-com.service", "ActiveState") == "active" status = _read_json_obj("/run/ethergold/pogcc-com.status") if not status: status = _read_json_obj(os.path.expanduser("~/EtherGold/status/pogcc-com.status")) pogcc_reasons: list[str] = [] if owner == "ethergold": pogcc_ok = pogcc_active if not pogcc_active: pogcc_reasons.append("inactive") if pogcc_active and status.get("usb_up") is False: pogcc_ok = False pogcc_reasons.append("heltec_cdc_down") else: pogcc_ok = not pogcc_active if pogcc_active: pogcc_ok = False pogcc_reasons.append("two_cdc_owners") mode = (listen.get("mode") or "listen").strip() listen_ok = bool(listen.get("ok")) if mode == "listen" else True bridge_active = bool(listen.get("bridge_pids") or listen.get("run_ethergold_pids")) bridge_ok = True if mode == "bridge": bridge_ok = bridge_active and "bridge_down" not in (listen.get("reasons") or []) listen_ok = True # not applicable collector_applicable = mode == "listen" manual_listen = bool(listen.get("manual_listen")) # Manual bench listen: the unit is deliberately not the owner, so it is not # a fault. Field seats (unit enabled) still fail when it is inactive. collector_ok = ( bool(listen.get("collector_active")) or manual_listen if collector_applicable else True ) star_role = (listen.get("star_role") or "").strip().lower() profile = (listen.get("profile") or "").strip() hub_applicable = profile == "atlas_wifi_star" and star_role == "hub" spoke_applicable = profile == "atlas_wifi_star" and star_role == "spoke" star_generic = profile == "atlas_wifi_star" and star_role not in ("hub", "spoke") bank_a = profile == "wifi_bank_bridge" and star_role == "side_a" bank_b = profile == "wifi_bank_bridge" and star_role == "side_b" star_ok = bridge_ok if mode == "bridge" else True return { "mode": mode, "profile": listen.get("profile"), "star_role": star_role, "http": { "ok": True, "active": True, "unit": "ethergold-http.service", }, "listen": { "ok": listen_ok if mode == "listen" else True, "active": bool(listen.get("listen_pids")), "applicable": mode == "listen", "reasons": list(listen.get("reasons") or []) if mode == "listen" else [], }, "bridge": { "ok": bridge_ok if mode == "bridge" else True, "active": bridge_active, "applicable": mode == "bridge" and profile not in ("atlas_wifi_star", "wifi_bank_bridge"), "pids": list(listen.get("bridge_pids") or []), "reasons": [r for r in (listen.get("reasons") or []) if r.startswith("bridge") or r == "inject_expected_1"] if mode == "bridge" else [], }, "wifi_hub": { "ok": star_ok if hub_applicable else True, "active": bridge_active if hub_applicable else False, "applicable": hub_applicable, "label": "WiFi Hub", "node_role": listen.get("node_role"), }, "wifi_spoke": { "ok": star_ok if spoke_applicable else True, "active": bridge_active if spoke_applicable else False, "applicable": spoke_applicable, "label": "WiFi Spoke", "node_role": listen.get("node_role"), }, "wifi_star": { "ok": star_ok if star_generic else True, "active": bridge_active if star_generic else False, "applicable": star_generic, "label": "WiFi Star", "node_role": listen.get("node_role"), }, "wifi_side_a": { "ok": star_ok if bank_a else True, "active": bridge_active if bank_a else False, "applicable": bank_a, "label": "WiFi Side A", }, "wifi_side_b": { "ok": star_ok if bank_b else True, "active": bridge_active if bank_b else False, "applicable": bank_b, "label": "WiFi Side B", }, "collector": { "ok": collector_ok, "active": bool(listen.get("collector_active")), "applicable": collector_applicable, "user": listen.get("collector_user"), "unit": "ethergold-collector.service", "enabled": bool(listen.get("collector_enabled")), "worker_owner": listen.get("worker_owner") or "none", }, "pogcc_com": { "ok": pogcc_ok, "active": pogcc_active, "owner": owner or "dyzineframe", "unit": "ethergold-pogcc-com.service", "usb_up": status.get("usb_up") if status else None, "build": status.get("build") if status else None, "reasons": pogcc_reasons, }, } def _fireworks_path() -> str: return os.environ.get( "ETHERGOLD_FIREWORKS_REQUEST_PATH", os.path.join( os.environ.get("ETHERGOLD_ROOT", "/home/topbox/EtherGold"), "fireworks-request.json", ), ) def _ui_html_path() -> Path: return Path(__file__).resolve().parent / "field_config_ui.html" def _status_ui_html_path() -> Path: return Path(__file__).resolve().parent / "field_status_ui.html" def _home_ui_html_path() -> Path: return Path(__file__).resolve().parent / "field_home_ui.html" def _health_ui_html_path() -> Path: return Path(__file__).resolve().parent / "field_health_ui.html" def _p3_capture_ui_html_path() -> Path: return Path(__file__).resolve().parent / "field_p3_capture_ui.html" def _resolve_ota_channel() -> str: try: from eg_deploy_control import resolve_ota_channel return resolve_ota_channel() except Exception: return (os.environ.get("OTA_CHANNEL") or "field").strip().lower() or "field" def build_http_health_payload() -> Dict[str, Any]: """Public JSON for GET /health. HTML twin is GET /EtherGold/health.""" listen = collect_listen_health() return { "ok": True, "product": "ethergold", "version": HTTP_BUILD, "service": "ethergold-http", "field_config": FieldConfig is not None, "boot_id": _HTTP_BOOT_ID, "ui": "/EtherGold/health", "ota_channel": _resolve_ota_channel(), "mode": listen.get("mode") or "listen", "profile": listen.get("profile"), "role": listen.get("role"), "listen": listen, "runtime": listen, "services": collect_services_health(listen), } def _etc_override() -> Path | None: raw = os.environ.get("ETHERGOLD_ETC", "").strip() return Path(raw) if raw else None def _read_progressives_sync_state(csv_path: str) -> Dict[str, Any]: state_path = os.path.join(os.path.dirname(csv_path) or ".", PROGRESSIVE_SYNC_STATE_NAME) if not os.path.isfile(state_path): return {} try: with open(state_path, encoding="utf-8") as handle: payload = json.load(handle) if isinstance(payload, dict): return payload except (OSError, ValueError, json.JSONDecodeError): pass return {} def build_progressives_sync_payload( *, csv_path: str, broadcast_interval_s: float = REMOTE_POLL_INTERVAL_S, version: str = "", ) -> Dict[str, float | str]: if os.path.isfile(csv_path): last_epoch = os.path.getmtime(csv_path) else: last_epoch = 0.0 interval = max(0.5, float(broadcast_interval_s)) poll_after = (last_epoch + interval) if last_epoch > 0 else time.time() state = _read_progressives_sync_state(csv_path) sync_epoch = float(state.get("sync_epoch", 0.0) or 0.0) payload: Dict[str, float | str] = { "last_update_epoch": last_epoch, "broadcast_interval_s": interval, "poll_after_epoch": poll_after, "server_now_epoch": time.time(), "version": version or HTTP_BUILD, } if sync_epoch > 0: payload["sync_epoch"] = sync_epoch return payload def _issue_csrf() -> str: token = secrets.token_urlsafe(24) now = time.time() expired = [k for k, exp in _CSRF_TOKENS.items() if exp < now] for k in expired: _CSRF_TOKENS.pop(k, None) _CSRF_TOKENS[token] = now + _CSRF_TTL_S return token def _check_csrf(token: str | None) -> bool: if not token: return False exp = _CSRF_TOKENS.get(token) if exp is None or exp < time.time(): _CSRF_TOKENS.pop(token or "", None) return False return True class EtherGoldHTTPHandler(BaseHTTPRequestHandler): server_version = HTTP_BUILD def log_message(self, format: str, *args) -> None: try: request_line = args[0] if args else "" status = str(args[1]) if len(args) > 1 else "" path = request_line.split(" ", 2)[1] if request_line else "" path = path.split("?", 1)[0] if path in _QUIET_PATHS and status in ("200", "304"): return except Exception: pass super().log_message(format, *args) def _send_bytes( self, status: int, body: bytes, *, content_type: str, cache_control: str | None = None, last_modified: float | None = None, extra_headers: dict[str, str] | None = None, ) -> None: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) if cache_control: self.send_header("Cache-Control", cache_control) if last_modified is not None and last_modified > 0: self.send_header("Last-Modified", formatdate(last_modified, usegmt=True)) if extra_headers: for key, val in extra_headers.items(): self.send_header(key, val) self.end_headers() self.wfile.write(body) def _send_json(self, payload: Dict[str, Any], status: int = 200) -> None: if status == 200 and isinstance(payload, dict): session = getattr(self, "_ui_session", "") if session: payload = dict(payload) payload["http_session"] = session payload["boot_id"] = _HTTP_BOOT_ID body = json.dumps(payload).encode("utf-8") self._send_bytes(status, body, content_type="application/json", cache_control="no-cache") def _send_file(self, path: str, mimetype: str) -> None: if not os.path.isfile(path): self.send_error(404, f"missing {path}") return with open(path, "rb") as handle: body = handle.read() self._send_bytes( 200, body, content_type=mimetype, last_modified=os.path.getmtime(path), ) def _read_json_body(self) -> dict[str, Any]: length = int(self.headers.get("Content-Length") or "0") raw = self.rfile.read(length) if length > 0 else b"{}" try: data = json.loads(raw.decode("utf-8") or "{}") except (UnicodeDecodeError, json.JSONDecodeError, ValueError): return {} return data if isinstance(data, dict) else {} def _authorized(self) -> bool: if FieldConfig is None: return False header = self.headers.get("Authorization", "") if not header.startswith("Basic "): return False try: decoded = base64.b64decode(header[6:].strip()).decode("utf-8") user, _, password = decoded.partition(":") except Exception: return False expect_user, expect_pass = load_auth_credentials(_etc_override()) return secrets.compare_digest(user, expect_user) and secrets.compare_digest( password, expect_pass ) def _require_auth(self) -> bool: self._ui_session = "" token = (self.headers.get(_SESSION_HEADER) or "").strip() if ui_session_ok(token): self._ui_session = token return True login_intent = (self.headers.get(_LOGIN_HEADER) or "").strip().lower() in ( "1", "true", "yes", ) if login_intent and self._authorized(): self._ui_session = mint_ui_session() return True extra_headers: dict[str, str] = {} error = "session_stale" if login_intent: error = "unauthorized" extra_headers["WWW-Authenticate"] = 'Basic realm="EtherGold Field Config"' body = json.dumps({"error": error}).encode("utf-8") self._send_bytes( 401, body, content_type="application/json", cache_control="no-cache", extra_headers=extra_headers or None, ) return False def _proxy_ohd_request(self) -> bool: """If path is /ohd/*, proxy to DF localhost. Returns True when handled.""" if not is_ohd_proxy_path(urlparse(self.path).path): return False method = self.command.upper() body: Optional[bytes] = None content_type: Optional[str] = None if method == "POST": try: length = int(self.headers.get("Content-Length") or "0") except ValueError: length = 0 body = self.rfile.read(length) if length > 0 else b"" content_type = self.headers.get("Content-Type") or "application/x-www-form-urlencoded" want_json = "/api/" in (urlparse(self.path).path or "") try: status, headers, payload = proxy_ohd_to_dyzineframe( method=method, path_with_query=self.path, body=body, content_type=content_type, ) extra = {k: v for k, v in headers.items() if k.lower() != "content-type"} self._send_bytes( status, payload, content_type=headers.get("Content-Type") or "application/octet-stream", cache_control="no-store", extra_headers=extra or None, ) except urllib.error.HTTPError as exc: # DF answered with an HTTP error — forward body/status, not "DF down". try: payload = exc.read() or b"" except OSError: payload = b"" ctype = exc.headers.get("Content-Type") if exc.headers else None self._send_bytes( int(exc.code), payload, content_type=ctype or "text/plain; charset=utf-8", cache_control="no-store", ) except (urllib.error.URLError, TimeoutError, OSError, ConnectionError) as exc: down, ctype, code = _ohd_df_down_body(want_json=want_json) note = f"{type(exc).__name__}: {exc}" if want_json: try: data = json.loads(down.decode("utf-8")) data["detail"] = note down = json.dumps(data).encode("utf-8") except (ValueError, UnicodeDecodeError): pass self._send_bytes(code, down, content_type=ctype, cache_control="no-store") return True def do_GET(self) -> None: if self._proxy_ohd_request(): return parsed = urlparse(self.path) route = parsed.path.rstrip("/") if parsed.path != "/" else parsed.path if route == "/EtherGold/progressives.csv": self._send_file(_csv_path(), "text/csv; charset=utf-8") return if route == "/EtherGold/progressives-sync.json": payload = build_progressives_sync_payload(csv_path=_csv_path()) self._send_json(payload) return if route == "/EtherGold/fireworks-request.json": path = _fireworks_path() payload: Dict[str, Any] = {} if os.path.isfile(path): try: with open(path, encoding="utf-8") as handle: loaded = json.load(handle) if isinstance(loaded, dict): payload = loaded except (OSError, ValueError, json.JSONDecodeError): pass # Empty JSON is the normal idle state; avoid a 404 access-log line # every 500 ms on every OHD sign. self._send_json(payload) return if route == "/EtherGold/floor-games.json": floor_path = default_floor_games_path() cached = resolve_serve_payload(floor_path) self._send_json(cached) return if route in ("/health", "/health/"): self._send_json(build_http_health_payload()) return if route in ("/", "/home", "/home/", "/EtherGold", "/EtherGold/", "/EtherGold/home"): path = _home_ui_html_path() if not path.is_file(): self.send_error(404, "field_home_ui.html missing") return body = path.read_text(encoding="utf-8").encode("utf-8") self._send_bytes( 200, body, content_type="text/html; charset=utf-8", cache_control="no-cache", ) return if route in ("/EtherGold/config", "/EtherGold/config/", "/config", "/config/"): # HTML is unauthenticated. No session → /?next=/EtherGold/config. # APIs require a process-scoped UI session (Sign in), not leftover Basic. path = _ui_html_path() if not path.is_file(): self.send_error(404, "field_config_ui.html missing") return body = path.read_text(encoding="utf-8").encode("utf-8") self._send_bytes( 200, body, content_type="text/html; charset=utf-8", cache_control="no-store", ) return if route in ("/EtherGold/status", "/EtherGold/status/"): # HTML is unauthenticated. No session → /?next=/EtherGold/status. # /api/runtime still requires Basic. path = _status_ui_html_path() if not path.is_file(): self.send_error(404, "field_status_ui.html missing") return body = path.read_text(encoding="utf-8").encode("utf-8") self._send_bytes( 200, body, content_type="text/html; charset=utf-8", cache_control="no-store", ) return if route in ("/EtherGold/health", "/EtherGold/health/"): # Public HTML (same payload as /health JSON). Do not content-negotiate # /health — field UI boot_id watches fetch("/health") as JSON. path = _health_ui_html_path() if not path.is_file(): self.send_error(404, "field_health_ui.html missing") return body = path.read_text(encoding="utf-8").encode("utf-8") self._send_bytes( 200, body, content_type="text/html; charset=utf-8", cache_control="no-cache", ) return if route in ("/EtherGold/p3-capture", "/EtherGold/p3-capture/"): path = _p3_capture_ui_html_path() if not path.is_file(): self.send_error(404, "field_p3_capture_ui.html missing") return body = path.read_text(encoding="utf-8").encode("utf-8") self._send_bytes( 200, body, content_type="text/html; charset=utf-8", cache_control="no-cache", ) return if route == "/EtherGold/api/status": if not self._require_auth(): return etc = _etc_override() cfg = load_field_config(etc) self._send_json(status_snapshot(cfg, etc)) return if route == "/EtherGold/api/runtime": if not self._require_auth(): return try: from runtime_status import build_runtime_payload except ImportError: self._send_json({"error": "runtime_status module missing"}, status=500) return self._send_json(build_runtime_payload(etc=_etc_override())) return if route == "/EtherGold/api/config": if not self._require_auth(): return etc = _etc_override() ensure_default_auth_file(etc) cfg = load_field_config(etc) self._send_json( { "ok": True, "http_build": HTTP_BUILD, "config": mask_secrets(cfg), "status": status_snapshot(cfg, etc), "csrf": _issue_csrf(), } ) return if route == "/EtherGold/api/dyzineframe/deploy": if not self._require_auth(): return if df_deploy_status is None: self._send_json({"error": "df_deploy_control module missing"}, status=500) return self._send_json(df_deploy_status()) return if route == "/EtherGold/api/ota/status": if not self._require_auth(): return payload: Dict[str, Any] = {"ok": True} payload["dyzineframe"] = df_deploy_status() if df_deploy_status else {"ok": False, "error": "missing"} payload["ethergold"] = eg_deploy_status() if eg_deploy_status else {"ok": False, "error": "missing"} self._send_json(payload) return if route == "/EtherGold/api/enrollment/locations": if not self._require_auth(): return if enrollment_locations_payload is None: self._send_json({"error": "enrollment locations unavailable"}, status=500) return self._send_json(enrollment_locations_payload()) return if route == "/EtherGold/api/bank/roster": if not self._require_auth(): return try: from bank_roster import load_roster except ImportError: self._send_json({"error": "bank_roster module missing"}, status=500) return self._send_json({"ok": True, "roster": load_roster()}) return if route == "/EtherGold/api/p3-capture": if not self._require_auth(): return try: from p3_screen_capture import public_status except ImportError: self._send_json({"error": "p3_screen_capture module missing"}, status=500) return payload = public_status() payload["csrf"] = _issue_csrf() self._send_json(payload) return if route == "/EtherGold/api/p3-capture/image": if not self._require_auth(): return try: from p3_screen_capture import image_bytes except ImportError: self._send_json({"error": "p3_screen_capture module missing"}, status=500) return body, filename, err = image_bytes() if body is None: self._send_json({"error": err or "no PNG yet"}, status=404) return safe_name = (filename or "p3-capture.png").replace('"', "") self._send_bytes( 200, body, content_type="image/png", cache_control="no-store", extra_headers={ "Content-Disposition": f'attachment; filename="{safe_name}"', }, ) return self.send_error(404, f"not found: {route}") def do_POST(self) -> None: if self._proxy_ohd_request(): return parsed = urlparse(self.path) route = parsed.path.rstrip("/") if parsed.path != "/" else parsed.path if FieldConfig is None: self._send_json({"error": "field_config module missing"}, status=500) return if not self._require_auth(): return csrf = self.headers.get("X-CSRF-Token") or self.headers.get("X-Csrf-Token") if not _check_csrf(csrf): self._send_json({"error": "invalid or missing CSRF token"}, status=403) return body = self._read_json_body() etc = _etc_override() if route == "/EtherGold/api/config/validate": raw = body.get("config") if "config" in body else body cfg = FieldConfig.from_dict(raw) role_hint = None if isinstance(raw, dict): role_hint = raw.get("machine_role") if role_hint is None: role_hint = body.get("machine_role") self._send_json(preview_apply(cfg, machine_role=role_hint)) return if route == "/EtherGold/api/config": action = (body.get("action") or "save").strip().lower() raw = body.get("config") or body cfg = FieldConfig.from_dict(raw) role_hint = None if isinstance(raw, dict): role_hint = raw.get("machine_role") if role_hint is None: role_hint = body.get("machine_role") errors = validate_field_config(cfg, machine_role=role_hint) if errors: self._send_json({"ok": False, "errors": errors}, status=400) return if action == "save": path = save_field_config(cfg, etc) self._send_json({"ok": True, "saved": str(path), "config": mask_secrets(cfg)}) return self._send_json({"error": f"unknown action {action}"}, status=400) return if route == "/EtherGold/api/config/apply": raw = body.get("config") or body cfg = FieldConfig.from_dict(raw) role_hint = None if isinstance(raw, dict): role_hint = raw.get("machine_role") if role_hint is None: role_hint = body.get("machine_role") dry = bool(body.get("dry_run", False)) # Write env files only. Never reboot. Never restart listen/collector. result = apply_field_config( cfg, etc=etc, invoke_role_script=False, dry_run=dry, machine_role=role_hint, ) status = 200 if result.get("ok") else 400 self._send_json(result, status=status) return if route == "/EtherGold/api/dyzineframe/check-updates": if df_check_for_updates is None: self._send_json({"error": "df_deploy_control module missing"}, status=500) return result = df_check_for_updates() self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/dyzineframe/update-now": if df_update_now is None: self._send_json({"error": "df_deploy_control module missing"}, status=500) return result = df_update_now() self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/dyzineframe/rollback": if df_rollback_now is None: self._send_json({"error": "df_deploy_control module missing"}, status=500) return result = df_rollback_now() self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/dyzineframe/promote": if df_promote_now is None: self._send_json({"error": "df_deploy_control module missing"}, status=500) return confirm = bool(body.get("confirm_hold")) result = df_promote_now(confirm_hold=confirm) self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/dyzineframe/restart": if df_restart_now is None: self._send_json({"error": "df_deploy_control module missing"}, status=500) return result = df_restart_now() self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/field/recover": if df_recover_stack_now is None: self._send_json({"error": "df_deploy_control module missing"}, status=500) return result = df_recover_stack_now() self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/ethergold/check-updates": if eg_check_for_updates is None: self._send_json({"error": "eg_deploy_control module missing"}, status=500) return result = eg_check_for_updates() self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/ethergold/update-now": if eg_update_now is None: self._send_json({"error": "eg_deploy_control module missing"}, status=500) return result = eg_update_now() status = 200 if result.get("ok") or result.get("pending") else 400 self._send_json(result, status=status) return if route == "/EtherGold/api/ethergold/rollback": if eg_rollback_now is None: self._send_json({"error": "eg_deploy_control module missing"}, status=500) return result = eg_rollback_now() self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/ethergold/promote": if eg_promote_now is None: self._send_json({"error": "eg_deploy_control module missing"}, status=500) return confirm = bool(body.get("confirm_hold")) result = eg_promote_now(confirm_hold=confirm) # pending USB handshake is not a hard failure status = 200 if result.get("ok") or result.get("pending") else 400 self._send_json(result, status=status) return if route == "/EtherGold/api/ethergold/apply/usb-unplugged": if eg_apply_usb_unplugged is None: self._send_json({"error": "eg_deploy_control module missing"}, status=500) return result = eg_apply_usb_unplugged() status = 200 if result.get("ok") or result.get("pending") else 400 self._send_json(result, status=status) return if route == "/EtherGold/api/ethergold/apply/usb-replugged": if eg_apply_usb_replugged is None: self._send_json({"error": "eg_deploy_control module missing"}, status=500) return result = eg_apply_usb_replugged() status = 200 if result.get("ok") or result.get("pending") else 400 self._send_json(result, status=status) return if route == "/EtherGold/api/ethergold/apply/usb-cancel": if eg_apply_usb_cancel is None: self._send_json({"error": "eg_deploy_control module missing"}, status=500) return result = eg_apply_usb_cancel() self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/ethergold/bridge/start": if eg_bridge_start is None: self._send_json({"error": "eg_deploy_control module missing"}, status=500) return result = eg_bridge_start() self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/ethergold/bridge/stop": if eg_bridge_stop is None: self._send_json({"error": "eg_deploy_control module missing"}, status=500) return result = eg_bridge_stop() self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/ethergold/sync_capture": if eg_sync_capture is None: self._send_json({"error": "eg_deploy_control module missing"}, status=500) return result = eg_sync_capture() self._send_json(result, status=200 if result.get("ok") else 400) return if route == "/EtherGold/api/ethergold/job": if eg_job_status is None: self._send_json({"error": "eg_deploy_control module missing"}, status=500) return name = str(body.get("name") or "bridge") result = eg_job_status(name) self._send_json(result, status=200) return if route == "/EtherGold/api/bank/register": try: from bank_roster import load_roster, save_roster, upsert_member except ImportError: self._send_json({"error": "bank_roster module missing"}, status=500) return member = body.get("member") if isinstance(body.get("member"), dict) else body roster = upsert_member(load_roster(), member) save_roster(roster) self._send_json({"ok": True, "roster": roster}) return if route == "/EtherGold/api/identity/p3-grab": try: from p3_identity_grab import grab_p3_identity except ImportError: self._send_json({"error": "p3_identity_grab module missing"}, status=500) return result = grab_p3_identity( force=bool(body.get("force") or body.get("force")) ) cfg = load_field_config(etc) result["status"] = status_snapshot(cfg, etc) self._send_json(result) return if route == "/EtherGold/api/p3-capture": try: from p3_screen_capture import request_capture except ImportError: self._send_json({"error": "p3_screen_capture module missing"}, status=500) return result = request_capture() result["csrf"] = _issue_csrf() status = 200 if result.get("ok") else 400 if result.get("error") == "A capture is already in progress": status = 409 self._send_json(result, status=status) return self.send_error(404, f"not found: {route}") def serve_forever(host: str, port: int) -> None: csv_path = _csv_path() floor_path = default_floor_games_path() try: ensure_default_auth_file(_etc_override()) except OSError as exc: print(f"WARN: could not write field-ui.auth: {exc}", flush=True) cleared = clear_preboot_progressive_runtime() if cleared: print( f" cleared pre-boot progressive runtime: {', '.join(cleared)}", flush=True, ) print(f"ethergold {HTTP_BUILD} listening on {host}:{port}", flush=True) try: from cabinet_identity import load_cabinet_identity from bank_roster import advertise_payload, start_master_advertiser ident = load_cabinet_identity() if ident.get("machine_role") == "M": hn = ident.get("topbox_host") or socket.gethostname() start_master_advertiser( lambda: advertise_payload( bank=int(ident.get("bank_number") or 0), hostname=hn, http_port=port, asset_id=str(ident.get("asset_id") or ""), term_number=int(ident.get("term_number") or 0), ) ) print(" bank master advertiser on UDP 41236", flush=True) except Exception as exc: print(f"WARN: bank advertiser not started: {exc}", flush=True) print(f" progressives.csv → {csv_path}", flush=True) print(f" floor-games.json → {floor_path}", flush=True) print(f" field UI login → http://{host}:{port}/", flush=True) print(f" field config UI → http://{host}:{port}/EtherGold/config", flush=True) print(f" health JSON → http://{host}:{port}/health", flush=True) print(f" health HTML → http://{host}:{port}/EtherGold/health", flush=True) print(f" P3 capture UI → http://{host}:{port}/EtherGold/p3-capture", flush=True) server = ThreadingHTTPServer((host, port), EtherGoldHTTPHandler) try: server.serve_forever() except KeyboardInterrupt: print("ethergold-http shutdown", flush=True) finally: server.server_close() def main() -> int: parser = argparse.ArgumentParser(description="EtherGold HTTP for OHD progressives + floor games + field config") parser.add_argument("--host", default=os.environ.get("ETHERGOLD_HTTP_HOST", "0.0.0.0")) parser.add_argument( "--port", type=int, default=int(os.environ.get("ETHERGOLD_HTTP_PORT", str(DEFAULT_PORT))), ) args = parser.parse_args() serve_forever(args.host, args.port) return 0 if __name__ == "__main__": raise SystemExit(main())