Spaces:
Sleeping
Sleeping
| import json | |
| import logging | |
| import os | |
| import time | |
| from http.server import BaseHTTPRequestHandler, HTTPServer, ThreadingHTTPServer | |
| from typing import Any, ClassVar | |
| from bot.channel_control import is_dashboard_admin_authorized, parse_terminal_command | |
| from bot.dashboard_server_routes import ( | |
| CHANNEL_CONTROL_IRC_ONLY_ACTIONS, | |
| build_observability_payload, | |
| handle_get, | |
| handle_post, | |
| handle_put, | |
| ) | |
| from bot.persistence_layer import persistence | |
| from bot.post_stream_report import build_post_stream_report | |
| from bot.runtime_config import ( | |
| BYTE_DASHBOARD_ADMIN_TOKEN, | |
| DASHBOARD_DIR, | |
| TWITCH_CHAT_MODE, | |
| irc_channel_control, | |
| ) | |
| class HealthHandler(BaseHTTPRequestHandler): | |
| MAX_CONTROL_BODY_BYTES = 4096 | |
| CHANNEL_CONTROL_IRC_ONLY_ACTIONS = CHANNEL_CONTROL_IRC_ONLY_ACTIONS | |
| _rate_limit_state: ClassVar[dict[str, list[float]]] = {} | |
| _rate_limit_max_requests = 100 | |
| _rate_limit_window_seconds = 60 | |
| _logger = logging.getLogger("byte.dashboard.server") | |
| def _check_rate_limit(self) -> bool: | |
| ip = self.client_address[0] | |
| now = time.time() | |
| # Cleanup old entries periodically (every 100 requests roughly) | |
| if len(self._rate_limit_state) > 1000: | |
| self._rate_limit_state.clear() | |
| history = self._rate_limit_state.get(ip, []) | |
| history = [ts for ts in history if now - ts < self._rate_limit_window_seconds] | |
| if len(history) >= self._rate_limit_max_requests: | |
| return False | |
| history.append(now) | |
| self._rate_limit_state[ip] = history | |
| return True | |
| def _send_cors_headers(self) -> None: | |
| self.send_header("Access-Control-Allow-Origin", "*") | |
| self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, OPTIONS") | |
| self.send_header( | |
| "Access-Control-Allow-Headers", "Content-Type, X-Byte-Admin-Token, Authorization" | |
| ) | |
| def _send_bytes(self, payload: bytes, content_type: str, status_code: int = 200) -> None: | |
| self.send_response(status_code) | |
| self._send_cors_headers() | |
| self.send_header("Content-Type", content_type) | |
| self.send_header("Cache-Control", "no-store") | |
| self.send_header("Content-Length", str(len(payload))) | |
| self.end_headers() | |
| self.wfile.write(payload) | |
| def _send_json(self, payload: dict[str, Any], status_code: int = 200) -> None: | |
| serialized = json.dumps(payload, ensure_ascii=False).encode("utf-8") | |
| self._send_bytes(serialized, "application/json; charset=utf-8", status_code=status_code) | |
| def _send_text(self, text: str, status_code: int = 200) -> None: | |
| self._send_bytes(text.encode("utf-8"), "text/plain; charset=utf-8", status_code=status_code) | |
| def _dashboard_authorized(self) -> bool: | |
| if not BYTE_DASHBOARD_ADMIN_TOKEN or len(BYTE_DASHBOARD_ADMIN_TOKEN.strip()) < 8: | |
| self._logger.error( | |
| "DASHBOARD_AUTH_CRITICAL: BYTE_DASHBOARD_ADMIN_TOKEN is missing or too short. Denying all access for security." | |
| ) | |
| return False | |
| authorized = is_dashboard_admin_authorized(self.headers, BYTE_DASHBOARD_ADMIN_TOKEN) | |
| if not authorized: | |
| import hmac | |
| import urllib.parse | |
| parsed_path = urllib.parse.urlparse(self.path) | |
| query_params = urllib.parse.parse_qs(parsed_path.query) | |
| if "auth" in query_params: | |
| provided_token = query_params["auth"][0].strip() | |
| authorized = hmac.compare_digest(provided_token, BYTE_DASHBOARD_ADMIN_TOKEN.strip()) | |
| if not authorized: | |
| from bot.runtime_config import logger | |
| logger.warning("Auth rejection for route %s from %s", self.path, self.address_string()) | |
| return authorized | |
| def _send_dashboard_auth_challenge(self) -> None: | |
| payload = b"Unauthorized" | |
| self.send_response(401) | |
| self.send_header("Content-Type", "text/plain; charset=utf-8") | |
| self.send_header("Cache-Control", "no-store") | |
| self.send_header("WWW-Authenticate", 'Basic realm="Byte Dashboard"') | |
| self.send_header("Content-Length", str(len(payload))) | |
| self.end_headers() | |
| self.wfile.write(payload) | |
| def _send_forbidden(self) -> None: | |
| self._send_json( | |
| {"ok": False, "error": "forbidden", "message": "Forbidden"}, | |
| status_code=403, | |
| ) | |
| def _read_json_payload(self, *, allow_empty: bool = False) -> dict[str, Any]: | |
| raw_length = str(self.headers.get("Content-Length", "0") or "0") | |
| try: | |
| content_length = int(raw_length) | |
| except ValueError as error: | |
| raise ValueError("Invalid Content-Length header.") from error | |
| if content_length <= 0: | |
| if allow_empty: | |
| return {} | |
| raise ValueError("Request body is required.") | |
| if content_length > self.MAX_CONTROL_BODY_BYTES: | |
| raise ValueError("Request body is too large.") | |
| payload_bytes = self.rfile.read(content_length) | |
| if not payload_bytes: | |
| if allow_empty: | |
| return {} | |
| raise ValueError("Request body is empty.") | |
| try: | |
| payload = json.loads(payload_bytes.decode("utf-8")) | |
| except (UnicodeDecodeError, json.JSONDecodeError) as error: | |
| raise ValueError("Invalid JSON payload.") from error | |
| if not isinstance(payload, dict): | |
| raise ValueError("JSON payload must be an object.") | |
| return payload | |
| def _send_dashboard_asset(self, relative_path: str, content_type: str) -> bool: | |
| if ".." in relative_path or relative_path.startswith("/"): | |
| self._send_text("Invalid path", status_code=400) | |
| return True | |
| target_path = (DASHBOARD_DIR / relative_path).resolve() | |
| if DASHBOARD_DIR not in target_path.parents: | |
| self._send_text("Not Found", status_code=404) | |
| return True | |
| if not target_path.is_file(): | |
| self._send_text("Not Found", status_code=404) | |
| return True | |
| self._send_bytes(target_path.read_bytes(), content_type=content_type, status_code=200) | |
| return True | |
| def _build_observability_payload(self, channel_id: str | None = None) -> dict[str, Any]: | |
| return build_observability_payload(channel_id) | |
| def _handle_channel_control(self, payload: dict[str, Any]) -> tuple[dict[str, Any], int]: | |
| action = str(payload.get("action", "") or "").strip().lower() | |
| channel_login = str(payload.get("channel", "") or "").strip() | |
| command_text = str(payload.get("command", "") or "").strip() | |
| if command_text: | |
| action, channel_login = parse_terminal_command(command_text) | |
| if TWITCH_CHAT_MODE != "irc": | |
| if action in self.CHANNEL_CONTROL_IRC_ONLY_ACTIONS: | |
| return ( | |
| { | |
| "ok": False, | |
| "error": "unsupported_mode", | |
| "message": "Channel control de runtime so funciona em TWITCH_CHAT_MODE=irc.", | |
| "mode": TWITCH_CHAT_MODE, | |
| "action": action, | |
| }, | |
| 409, | |
| ) | |
| if action == "list": | |
| return ( | |
| { | |
| "ok": True, | |
| "action": "list", | |
| "channels": [], | |
| "mode": TWITCH_CHAT_MODE, | |
| "message": "Sem runtime IRC ativo neste modo. join/part ficam bloqueados em eventsub.", | |
| }, | |
| 200, | |
| ) | |
| result = irc_channel_control.execute(action=action, channel_login=channel_login) | |
| result["mode"] = TWITCH_CHAT_MODE | |
| if result.get("ok") and action == "part": | |
| report_channel_id = ( | |
| str(channel_login or result.get("channel") or "default").strip().lower() | |
| ) | |
| report_channel_id = report_channel_id or "default" | |
| try: | |
| snapshot = build_observability_payload(report_channel_id) | |
| history_points = persistence.load_observability_channel_history_sync( | |
| report_channel_id, | |
| limit=120, | |
| ) | |
| generated_report = build_post_stream_report( | |
| channel_id=report_channel_id, | |
| history_points=list(history_points or []), | |
| observability_snapshot=snapshot, | |
| trigger="auto_part_success", | |
| ) | |
| persistence.save_post_stream_report_sync( | |
| report_channel_id, | |
| generated_report, | |
| trigger="auto_part_success", | |
| ) | |
| result["post_stream_report"] = { | |
| "generated": True, | |
| "channel_id": report_channel_id, | |
| "trigger": "auto_part_success", | |
| } | |
| except Exception as error: | |
| self._logger.error( | |
| "Falha ao gerar post_stream_report automatico para %s: %s", | |
| report_channel_id, | |
| error, | |
| ) | |
| if result.get("ok"): | |
| return result, 200 | |
| error_code = str(result.get("error", "") or "") | |
| if error_code in {"runtime_unavailable", "timeout"}: | |
| status_code = 503 | |
| elif error_code in {"runtime_error"}: | |
| status_code = 500 | |
| else: | |
| status_code = 400 | |
| return result, status_code | |
| def do_OPTIONS(self) -> None: | |
| self.send_response(204) | |
| self._send_cors_headers() | |
| self.end_headers() | |
| def do_GET(self) -> None: | |
| if not self._check_rate_limit(): | |
| self._send_text("Too Many Requests", status_code=429) | |
| return | |
| _run_route_safely(self, handle_get) | |
| def do_PUT(self) -> None: | |
| if not self._check_rate_limit(): | |
| self._send_text("Too Many Requests", status_code=429) | |
| return | |
| _run_route_safely(self, handle_put) | |
| def do_POST(self) -> None: | |
| if not self._check_rate_limit(): | |
| self._send_text("Too Many Requests", status_code=429) | |
| return | |
| _run_route_safely(self, handle_post) | |
| def log_message(self, format: str, *args: object) -> None: | |
| return | |
| def _run_route_safely(handler: Any, route_fn: Any) -> None: | |
| """Run a route function, converting any unexpected error into a clean 500 | |
| instead of leaving the client with a hung/half-written connection.""" | |
| try: | |
| route_fn(handler) | |
| except Exception as error: # last line of defense for every request | |
| try: | |
| handler._logger.exception("Unhandled error in request handler: %s", error) | |
| handler._send_json({"ok": False, "error": "internal_error"}, status_code=500) | |
| except Exception: | |
| pass # response already partially written; nothing safe to do | |
| def run_server() -> None: | |
| port = int(os.environ.get("PORT", "8080")) | |
| ThreadingHTTPServer(("0.0.0.0", port), HealthHandler).serve_forever() | |