""" FreeExile PC Desktop Web Client Local Testing Server. Serves the dedicated full-screen PC client (client/web_pc) at root / on port 8089, while seamlessly routing shared assets, templates, and engine modules from client/webapp. """ import http.server import json import mimetypes import os from pathlib import Path import socket import socketserver import sys import urllib.parse if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") PROJECT_ROOT = Path(__file__).resolve().parent.parent PC_DIR = PROJECT_ROOT / "client" / "web_pc" WEBAPP_DIR = PROJECT_ROOT / "client" / "webapp" if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) try: from server.feedback.feedback_service import FeedbackService, SystemDiagnosticsDTO except ImportError: FeedbackService = None # type: ignore SystemDiagnosticsDTO = None # type: ignore try: from server.world.wilderness_map_generator import WildernessMapGenerator from server.world.map_binary_serializer import serialize_map_grid except ImportError: WildernessMapGenerator = None # type: ignore serialize_map_grid = None # type: ignore def get_local_ip() -> str: """Detects local LAN IPv4 address for local testing.""" s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] except Exception: ip = "127.0.0.1" finally: s.close() return ip class FreeExilePCHTTPOptionsMixin: """Handles CORS headers, options, and dev caching headers.""" def send_cors_headers(self) -> None: self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.send_header("Cache-Control", "no-cache, no-store, must-revalidate") class FreeExilePCRequestHandler(http.server.SimpleHTTPRequestHandler, FreeExilePCHTTPOptionsMixin): """Custom HTTP handler serving PC Desktop Web Client with shared asset routing.""" def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(PROJECT_ROOT), **kwargs) def end_headers(self): self.send_cors_headers() super().end_headers() def handle(self): try: super().handle() except (ConnectionResetError, BrokenPipeError, OSError): pass def finish(self): try: super().finish() except (ConnectionResetError, BrokenPipeError, OSError): pass def do_OPTIONS(self): self.send_response(200) self.end_headers() @staticmethod def _is_safe_child(base: Path, target: Path) -> bool: """Verifies that target resolves strictly within base directory and is a regular file.""" try: target_resolved = target.resolve() base_resolved = base.resolve() return ( (target_resolved.is_relative_to(base_resolved) if hasattr(target_resolved, "is_relative_to") else (target_resolved == base_resolved or base_resolved in target_resolved.parents)) and target_resolved.is_file() ) except (ValueError, RuntimeError, OSError): return False def _serve_file(self, file_path: Path) -> None: if not file_path.is_file(): self.send_error(404, f"File not found: {file_path.name}") return content_type, _ = mimetypes.guess_type(str(file_path)) if content_type is None: content_type = "application/octet-stream" if file_path.suffix == ".js": content_type = "application/javascript; charset=utf-8" elif file_path.suffix == ".css": content_type = "text/css; charset=utf-8" elif file_path.suffix == ".html": content_type = "text/html; charset=utf-8" try: with open(file_path, "rb") as f: data = f.read() self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) except (ConnectionResetError, BrokenPipeError): pass except OSError as e: if getattr(e, "winerror", None) == 10054: pass else: try: self.send_error(500, f"Error reading file: {e}") except Exception: pass except Exception as e: try: self.send_error(500, f"Error reading file: {e}") except Exception: pass def _handle_map_api(self, query: str) -> None: try: params = urllib.parse.parse_qs(query) zone_id = params.get("zone_id", ["zone_tang_kiem_nhai"])[0] seed = int(params.get("seed", [42])[0]) biome_param = params.get("biome", [None])[0] if WildernessMapGenerator and serialize_map_grid: gen = WildernessMapGenerator.for_zone(zone_id, biome_id=biome_param) grid_data = gen.generate(seed=seed, zone_id=zone_id) payload = serialize_map_grid(grid_data) self.send_response(200) self.send_header("Content-Type", "application/octet-stream") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) else: self.send_error(500, "WildernessMapGenerator not available") except Exception as e: err = json.dumps({"error": str(e)}).encode("utf-8") self.send_response(500) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(err))) self.end_headers() self.wfile.write(err) def _handle_feedback_get(self) -> None: try: records = [] if FeedbackService: svc = FeedbackService() feedbacks = svc.list_feedbacks(limit=20) records = [ { "feedback_id": f.feedback_id, "player_id": f.player_id, "player_name": f.player_name, "category": f.category, "title": f.title, "content": f.content, "status": f.status, "admin_response": f.admin_response, "created_at_ms": f.created_at_ms, } for f in feedbacks ] res = json.dumps({"feedbacks": records}, ensure_ascii=False).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(res))) self.end_headers() self.wfile.write(res) except Exception as e: err = json.dumps({"error": str(e)}).encode("utf-8") self.send_response(500) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(err))) self.end_headers() self.wfile.write(err) def _resolve_static_path(self, clean_path: str) -> Path | None: """Resolves URL paths across PC directory and WebApp directory with strict containment.""" if clean_path in ("", "index.html", "/"): index_p = PC_DIR / "index.html" return index_p if self._is_safe_child(PC_DIR, index_p) else None # URL-decode and normalize backslashes unquoted = urllib.parse.unquote(clean_path).replace("\\", "/") # Strict rejection of directory traversal and absolute drive tokens segments = [s for s in unquoted.split("/") if s] if ".." in segments or any(":" in s for s in segments): return None stripped = clean_path.lstrip("/") unquoted_stripped = unquoted.lstrip("/") # Asset fallback routing: map /web_pc/css/assets/, /web_pc/assets/, or /assets/ to WEBAPP_DIR / assets for prefix in ("web_pc/css/assets/", "web_pc/assets/", "assets/"): for s in (stripped, unquoted_stripped): if s.startswith(prefix): sub = s[len(prefix):] p = WEBAPP_DIR / "assets" / sub if self._is_safe_child(WEBAPP_DIR, p): return p if stripped.startswith("web_pc/"): sub = stripped[len("web_pc/"):] p = PC_DIR / sub if self._is_safe_child(PC_DIR, p): return p if stripped.startswith("webapp/"): sub = stripped[len("webapp/"):] p = WEBAPP_DIR / sub if self._is_safe_child(WEBAPP_DIR, p): return p # Check PC_DIR first (for PC-specific css, js, etc.) candidate_pc = PC_DIR / stripped if self._is_safe_child(PC_DIR, candidate_pc): return candidate_pc # Check WEBAPP_DIR for assets, js, css, templates candidate_webapp = WEBAPP_DIR / stripped if self._is_safe_child(WEBAPP_DIR, candidate_webapp): return candidate_webapp return None def do_GET(self): parsed = urllib.parse.urlparse(self.path) clean_path = parsed.path if clean_path.startswith("/api/map"): self._handle_map_api(parsed.query) return if clean_path.startswith("/api/feedback"): self._handle_feedback_get() return resolved = self._resolve_static_path(clean_path) if resolved and resolved.is_file(): self._serve_file(resolved) else: self.send_error(404, f"Path not found: {clean_path}") def do_POST(self): if self.path == "/api/feedback": self._handle_feedback_post() else: self.send_error(404, "Endpoint not found") def _handle_feedback_post(self) -> None: try: length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) data = json.loads(body.decode("utf-8")) diag_data = data.get("diagnostics", {}) diag = None if SystemDiagnosticsDTO: diag = SystemDiagnosticsDTO( client_platform=diag_data.get("client_platform", "PC"), os_version=diag_data.get("os_version", "Windows 11"), device_model=diag_data.get("device_model", "Desktop"), current_fps=float(diag_data.get("current_fps", 120.0)), current_ping_ms=int(diag_data.get("current_ping_ms", 12)), active_zone=diag_data.get("active_zone", "zone_tang_kiem_nhai"), player_coords=diag_data.get("player_coords", "0.0,0.0"), browser_ua=self.headers.get("User-Agent", "PC Browser"), ) success = True fb_id = "fb_pc_mock" msg = "Feedback received successfully" if FeedbackService: svc = FeedbackService() success, fb_id, msg = svc.submit_feedback( player_id=data.get("player_id", "pc_exile"), player_name=data.get("player_name", "PC Player"), category=data.get("category", "bug"), title=data.get("title", "PC Feedback"), content=data.get("content", ""), diagnostics=diag, ) res = json.dumps({"success": success, "feedback_id": fb_id, "message": msg}).encode("utf-8") self.send_response(200 if success else 400) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(res))) self.end_headers() self.wfile.write(res) except Exception as e: err = json.dumps({"success": False, "error": str(e)}).encode("utf-8") self.send_response(500) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(err))) self.end_headers() self.wfile.write(err) def start_server(port: int = 8089): """Starts the FreeExile PC Web Client HTTP server.""" local_ip = get_local_ip() class ThreadingHTTPServer(socketserver.ThreadingMixIn, socketserver.TCPServer): daemon_threads = True allow_reuse_address = True request_queue_size = 64 with ThreadingHTTPServer(("0.0.0.0", port), FreeExilePCRequestHandler) as httpd: print("==================================================================") print(f" FREEEXILE PC DESKTOP ARPG WEB CLIENT (PORT {port}) ") print("==================================================================") print(f" Local Host : http://localhost:{port}/") print(f" LAN Access : http://{local_ip}:{port}/") print(f" Serving : {PC_DIR}") print(" Press Ctrl+C to stop the server.") print("==================================================================") while True: try: httpd.serve_forever() except KeyboardInterrupt: print("\nShutting down PC web server...") httpd.shutdown() break except Exception as e: print(f"[!] Server transient warning: {e}", file=sys.stderr) if __name__ == "__main__": p = 8089 if len(sys.argv) > 1: if sys.argv[1] == "--port" and len(sys.argv) > 2: try: p = int(sys.argv[2]) except ValueError: p = 8089 elif "--port" in sys.argv: idx = sys.argv.index("--port") if idx + 1 < len(sys.argv): try: p = int(sys.argv[idx + 1]) except ValueError: p = 8089 elif sys.argv[1].startswith("--port="): try: p = int(sys.argv[1].split("=")[1]) except ValueError: p = 8089 else: try: p = int(sys.argv[1]) except ValueError: p = 8089 start_server(port=p)