diff --git a/__main__.py b/__main__.py deleted file mode 100644 index 707ab47..0000000 --- a/__main__.py +++ /dev/null @@ -1,6 +0,0 @@ -import sys -from portablemc.cli import main - -if __name__ == "__main__": - sys.argv = ["portablemc"] + sys.argv[1:] - sys.exit(main()) diff --git a/main.py b/main.py new file mode 100644 index 0000000..c9bd6bc --- /dev/null +++ b/main.py @@ -0,0 +1,379 @@ +#!/usr/bin/env python3 +""" +Cross‑platform bootstrap script for embedded Python 3.14 (Windows only). +Attempts to download the native portablemc binary, falls back to pip. +Run with any Python (e.g., system 3.11) to set up and launch the launcher. +""" + +import os +import sys +import subprocess +import urllib.request +import zipfile +import tarfile +import shutil +import platform +import ssl +from pathlib import Path + +# --- Windows-only guard --- +if platform.system().lower() != 'windows': + print("❌ This bootstrap script currently only supports Windows.") + sys.exit(1) + +# --- Configuration --- +EMBEDDED_DIR = Path(__file__).parent / "python" +EMBEDDED_PYTHON = EMBEDDED_DIR / "python.exe" +PYTHON_VERSION = "3.14.3" +PYTHON_URL = f"https://www.python.org/ftp/python/{PYTHON_VERSION}/python-{PYTHON_VERSION}-embed-amd64.zip" +GET_PIP_URL = "https://bootstrap.pypa.io/get-pip.py" +BASE_PACKAGES = ["flask", "flask-socketio", "psutil", "ansi2html", "certifi"] +PORTABLEMC_BIN_DIR = Path(__file__).parent / "portablemc_bin" + +# --- OS and architecture detection (for portablemc binary) --- +SYSTEM = platform.system().lower() +MACHINE = platform.machine().lower() + +ARCH_MAP = { + 'x86_64': 'x86_64', + 'amd64': 'x86_64', + 'i686': 'i686', + 'i386': 'i686', + 'aarch64': 'aarch64', + 'arm64': 'aarch64', + 'armv7l': 'arm-gnueabihf', + 'arm': 'arm-gnueabihf', +} + +OS_MAP = { + 'windows': 'windows', + 'linux': 'linux', + 'darwin': 'macos', +} + +def get_portablemc_url(): + """Return the download URL for the native portablemc binary, or None.""" + os_name = OS_MAP.get(SYSTEM) + if not os_name: + print(f"⚠️ Unsupported OS: {SYSTEM}") + return None + arch = ARCH_MAP.get(MACHINE, 'x86_64') + if os_name == 'macos': + arch = 'aarch64' if arch == 'aarch64' else 'x86_64' + if os_name == 'linux' and arch not in ('arm-gnueabihf',): + arch += '-gnu' + base = "https://github.com/mindstorm38/portablemc/releases/download/v5.0.2/" + if os_name == 'windows': + ext = "zip" + filename = f"portablemc-5.0.2-{os_name}-{arch}-msvc.{ext}" + else: + ext = "tar.gz" + filename = f"portablemc-5.0.2-{os_name}-{arch}.{ext}" + return base + filename + +# --- Helper functions --- +def ensure_embedded_python(): + if EMBEDDED_PYTHON.exists(): + print("✅ Embedded Python already exists.") + return True + + print("📥 Downloading embedded Python...") + # Try with verification first + try: + urllib.request.urlretrieve(PYTHON_URL, "python-embed.zip") + except Exception as e: + print(f"⚠️ First download attempt failed: {e}") + print("📥 Retrying with SSL verification disabled (insecure)...") + try: + # Create an unverified SSL context + ssl_context = ssl._create_unverified_context() + with urllib.request.urlopen(PYTHON_URL, context=ssl_context) as response: + with open("python-embed.zip", 'wb') as out_file: + out_file.write(response.read()) + except Exception as e2: + print(f"❌ Failed to download Python even with unverified SSL: {e2}") + return False + + print("📦 Extracting...") + with zipfile.ZipFile("python-embed.zip", "r") as zip_ref: + zip_ref.extractall(EMBEDDED_DIR) + os.remove("python-embed.zip") + print("✅ Embedded Python ready.") + return True + +def fix_pth_file(): + pth_files = list(EMBEDDED_DIR.glob("*._pth")) + if not pth_files: + print("❌ No ._pth file found; cannot enable site-packages.") + return False + pth = pth_files[0] + with open(pth, "r") as f: + content = f.read() + if "import site" in content and "#import site" in content: + content = content.replace("#import site", "import site") + with open(pth, "w") as f: + f.write(content) + print("✅ Enabled site-packages in ._pth file.") + else: + print("ℹ️ site-packages already enabled.") + return True + +def test_embedded_python(): + """Test if the embedded Python executable can be run.""" + try: + result = subprocess.run([str(EMBEDDED_PYTHON), "--version"], + capture_output=True, text=True, timeout=5) + if result.returncode == 0: + print(f"✅ Embedded Python runs: {result.stdout.strip()}") + return True + else: + print(f"❌ Embedded Python test failed: {result.stderr}") + return False + except OSError as e: + # Common error: WinError 1260 (group policy block) + print(f"❌ Cannot run embedded Python (blocked by group policy?): {e}") + return False + except Exception as e: + print(f"❌ Embedded Python test error: {e}") + return False + +def download_get_pip(): + pip_script = EMBEDDED_DIR / "get-pip.py" + if pip_script.exists(): + print("✅ get-pip.py already present.") + return pip_script + + print("📥 Downloading get-pip.py (ignoring SSL cert for this request)...") + try: + # Create an unverified SSL context to bypass certificate errors + ssl_context = ssl._create_unverified_context() + with urllib.request.urlopen(GET_PIP_URL, context=ssl_context) as response: + with open(pip_script, 'wb') as out_file: + out_file.write(response.read()) + print("✅ get-pip.py downloaded successfully.") + except Exception as e: + print(f"❌ Failed to download get-pip.py: {e}") + return None + return pip_script + +def run_pip_command(args, isolated=True): + env = os.environ.copy() + if isolated: + env["PYTHONNOUSERSITE"] = "1" + env["PYTHONPATH"] = "" + cmd = [str(EMBEDDED_PYTHON)] + args + # SAFE: args are static or come from trusted BASE_PACKAGES + result = subprocess.run(cmd, env=env, capture_output=True, text=True) + if result.returncode != 0: + print(f"❌ Pip command failed: {' '.join(args)}") + print(result.stderr) + return False + print(result.stdout) + return True + +def install_pip(): + pip_script = download_get_pip() + if not pip_script: + return False + env = os.environ.copy() + env["PYTHONNOUSERSITE"] = "1" + env["PYTHONPATH"] = "" + cmd = [str(EMBEDDED_PYTHON), str(pip_script), + "--trusted-host=files.pythonhosted.org", + "--trusted-host=pypi.org"] + result = subprocess.run(cmd, env=env, capture_output=True, text=True) + if result.returncode != 0: + print("❌ Failed to install pip.") + print(result.stderr) + return False + print("✅ pip installed.") + return True + +def install_base_packages(): + print("📦 Installing base packages...") + if not run_pip_command(["-m", "pip", "install", "--upgrade", "pip"], isolated=True): + print("⚠️ Pip upgrade failed, continuing anyway.") + for pkg in BASE_PACKAGES: + print(f" Installing {pkg}...") + if not run_pip_command(["-m", "pip", "install", pkg], isolated=True): + print(f"❌ Failed to install {pkg}.") + return False + return True + +def get_certifi_path(): + """Return the path to certifi's CA bundle, or None if certifi not installed.""" + try: + result = subprocess.run( + [str(EMBEDDED_PYTHON), "-c", "import certifi; print(certifi.where())"], + capture_output=True, text=True, check=True, + env={"PYTHONNOUSERSITE": "1"} # ensure isolation + ) + path = result.stdout.strip() + if path and Path(path).exists(): + return path + except Exception: + pass + return None + +def download_portablemc_binary(): + url = get_portablemc_url() + if not url: + return False + print(f"📥 Downloading portablemc from {url}...") + archive_path = Path(__file__).parent / "portablemc_download" + try: + # Attempt with verification first + urllib.request.urlretrieve(url, archive_path) + except Exception as e: + print(f"⚠️ First download attempt failed: {e}") + print("📥 Retrying with SSL verification disabled (insecure)...") + try: + ssl_context = ssl._create_unverified_context() + with urllib.request.urlopen(url, context=ssl_context) as response: + with open(archive_path, 'wb') as out_file: + out_file.write(response.read()) + except Exception as e2: + print(f"❌ Failed to download even with unverified SSL: {e2}") + return False + + PORTABLEMC_BIN_DIR.mkdir(exist_ok=True) + try: + if url.endswith(".zip"): + with zipfile.ZipFile(archive_path, "r") as zip_ref: + zip_ref.extractall(PORTABLEMC_BIN_DIR) + else: + with tarfile.open(archive_path, "r:gz") as tar_ref: + tar_ref.extractall(PORTABLEMC_BIN_DIR) + except Exception as e: + print(f"❌ Failed to extract archive: {e}") + return False + finally: + archive_path.unlink(missing_ok=True) + # Move all extracted files to the root of PORTABLEMC_BIN_DIR + for item in PORTABLEMC_BIN_DIR.iterdir(): + if item.is_dir(): + for sub in item.iterdir(): + sub.rename(PORTABLEMC_BIN_DIR / sub.name) + item.rmdir() + print(f"✅ portablemc binary extracted to {PORTABLEMC_BIN_DIR}") + return True + +def test_portablemc(): + """Check if portablemc is available (binary or module). Returns 'binary' or 'module' or None.""" + # Try binary first + exe_name = "portablemc.exe" if SYSTEM == "windows" else "portablemc" + binary_path = PORTABLEMC_BIN_DIR / exe_name + if binary_path.exists(): + if SYSTEM != "windows": + binary_path.chmod(binary_path.stat().st_mode | 0o111) + env = os.environ.copy() + env["PATH"] = str(PORTABLEMC_BIN_DIR) + os.pathsep + env.get("PATH", "") + try: + result = subprocess.run([str(binary_path), "--help"], env=env, + capture_output=True, text=True, timeout=5) + if result.returncode == 0: + print("✅ portablemc binary works.") + return "binary" + except (subprocess.TimeoutExpired, FileNotFoundError): + print("⏱️ portablemc binary check timed out or not found, trying module.") + + # Try module + env = os.environ.copy() + env["PYTHONNOUSERSITE"] = "1" + env["PYTHONPATH"] = "" + cmd = [str(EMBEDDED_PYTHON), "-m", "portablemc", "--help"] + try: + result = subprocess.run(cmd, env=env, capture_output=True, text=True, timeout=5) + if result.returncode == 0: + print("✅ portablemc module works.") + return "module" + except subprocess.TimeoutExpired: + print("⏱️ portablemc module check timed out, assuming not available.") + return None + +def ensure_portablemc(): + """Make portablemc available – try binary, fallback to pip. Returns method string or None.""" + method = test_portablemc() + if method: + return method + if download_portablemc_binary(): + method = test_portablemc() + if method: + return method + print("⚠️ Binary download failed, falling back to pip.") + print("📦 Installing portablemc via pip...") + if run_pip_command(["-m", "pip", "install", "portablemc"], isolated=True): + method = test_portablemc() + if method: + return method + return None + +def launch_launcher(method): + launcher_script = Path(__file__).parent / "portablemc.py" + if not launcher_script.exists(): + print("❌ portablemc.py not found in the same directory.") + return False + + env = os.environ.copy() + paths = [str(EMBEDDED_DIR), str(EMBEDDED_DIR / "Scripts"), str(PORTABLEMC_BIN_DIR)] + env["PATH"] = os.pathsep.join(paths) + os.pathsep + env.get("PATH", "") + env["__compat_layer"] = "runasinvoker" + env["PYTHONHOME"] = str(EMBEDDED_DIR) + env["CLICOLOR_FORCE"] = "1" + env["PYTHONNOUSERSITE"] = "1" + env["PYTHONPATH"] = "" + env["PORTABLEMC_METHOD"] = method + + cert_path = get_certifi_path() + if cert_path: + env["SSL_CERT_FILE"] = cert_path + env["REQUESTS_CA_BUNDLE"] = cert_path + + cmd = [str(EMBEDDED_PYTHON), str(launcher_script)] + print(f"🚀 Launching: {' '.join(cmd)}") + try: + subprocess.run(cmd, env=env, check=True) + except subprocess.CalledProcessError as e: + print(f"❌ Launcher exited with error: {e}") + return False + except KeyboardInterrupt: + print("⏹️ Interrupted by user.") + return True + +def main(): + print("=== Embedded Python Bootstrap (Windows only) ===") + if not ensure_embedded_python(): + sys.exit(1) + if not fix_pth_file(): + sys.exit(1) + if not test_embedded_python(): + print("❌ Embedded Python cannot be executed. Please check group policy or run without embedded isolation.") + print(" You may try to run portablemc.py directly with your system Python after installing dependencies:") + print(" pip install flask flask-socketio psutil ansi2html portablemc") + sys.exit(1) + + # Check pip + env_check = os.environ.copy() + env_check["PYTHONNOUSERSITE"] = "1" + env_check["PYTHONPATH"] = "" + pip_check = subprocess.run([str(EMBEDDED_PYTHON), "-m", "pip", "--version"], + env=env_check, capture_output=True, text=True) + if pip_check.returncode != 0: + print("📦 pip not found, installing...") + if not install_pip(): + sys.exit(1) + else: + print(f"✅ pip already installed: {pip_check.stdout.strip()}") + + if not install_base_packages(): + sys.exit(1) + + method = ensure_portablemc() + if not method: + sys.exit(1) + + print("✅ Setup complete. Launching portablemc.py...") + launch_launcher(method) + +if __name__ == "__main__": + main() diff --git a/launch.py b/portablemc.py similarity index 88% rename from launch.py rename to portablemc.py index 5d1d725..035b83c 100644 --- a/launch.py +++ b/portablemc.py @@ -10,7 +10,7 @@ import shutil import threading import logging -import psutil +import traceback import os import json import importlib.util @@ -18,14 +18,12 @@ app = Flask(__name__) socketio = SocketIO(app, cors_allowed_origins="*") - def escape_html(s): """Escape only &, <, > for safe innerHTML – leaves quotes and apostrophes untouched.""" - return s.replace("&", "&").replace("<", "<").replace(">", ">") - + return s.replace('&', '&').replace('<', '<').replace('>', '>') # --- CONFIGURATION --- -VALID_USERNAME_REGEX = re.compile(r"^[a-zA-Z0-9_]{3,16}$") +VALID_USERNAME_REGEX = re.compile(r'^[a-zA-Z0-9_]{3,16}$') FORBIDDEN_LIST = ["CubeUniform840", "Admin", "Owner"] PASS_KEY = "1234" SERVER_IP = "77.103.184.72" @@ -41,20 +39,9 @@ def escape_html(s): clients_lock = threading.Lock() logging.basicConfig(level=logging.INFO) -# Fallbacks when certain libraries are restricted/corrupted/disabled -try: - from ansi2html import Ansi2HTMLConverter - - ansi_converter = Ansi2HTMLConverter(dark_bg=True, inline=True) - use_server_conversion = True -except ImportError: - # ansi2html not installed – fallback to raw ANSI - use_server_conversion = False - logging.warning("ansi2html not installed; client will handle ANSI conversion.") - +# Optional psutil for process tree killing try: import psutil - PSUTIL_AVAILABLE = True except ImportError: psutil = None @@ -64,6 +51,15 @@ def escape_html(s): "Install psutil for full cleanup." ) +# Fallbacks when certain libraries are restricted/corrupted/disabled +try: + from ansi2html import Ansi2HTMLConverter + ansi_converter = Ansi2HTMLConverter(dark_bg=True, inline=True) + use_server_conversion = True +except ImportError: + # ansi2html not installed – fallback to raw ANSI + use_server_conversion = False + logging.warning("ansi2html not installed; client will handle ANSI conversion.") # --- HTML TEMPLATE (with 4-space indented JavaScript) --- HTML_TEMPLATE = r""" @@ -914,45 +910,36 @@ def escape_html(s): """ # --- ROUTES --- - - -@socketio.on("connect") +@socketio.on('connect') def handle_connect(): global connected_clients with clients_lock: connected_clients += 1 client_id = request.sid - print(f"Client connected: {client_id} (Total: {connected_clients})") + print(f'Client connected: {client_id} (Total: {connected_clients})') + emit('status', {'core': 'online', 'minecraft': 'checking'}) - # Send initial status - emit("status", {"core": "online", "minecraft": "checking"}) - - -@socketio.on("disconnect") +@socketio.on('disconnect') def handle_disconnect(): global connected_clients with clients_lock: connected_clients -= 1 client_id = request.sid - print(f"Client disconnected: {client_id} (Total: {connected_clients})") - + print(f'Client disconnected: {client_id} (Total: {connected_clients})') -@socketio.on("ping_minecraft") +@socketio.on('ping_minecraft') def handle_ping_minecraft(): - """Check Minecraft server status on demand""" try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(1.5) s.connect((SERVER_IP, 25565)) s.close() - emit("minecraft_status", {"online": True}) + emit('minecraft_status', {'online': True}) except Exception: - emit("minecraft_status", {"online": False}) - + emit('minecraft_status', {'online': False}) @app.route("/ping") def ping(): - """Checks if the Minecraft server is online.""" try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(1.5) @@ -960,10 +947,8 @@ def ping(): s.close() return jsonify(online=True) except Exception: - # Always return 200 OK with online=false return jsonify(online=False), 200 - @app.route("/stream") def stream(): # --- PORTABLEMC AVAILABILITY CHECK --- @@ -973,17 +958,15 @@ def is_portablemc_available(): return importlib.util.find_spec("portablemc") is not None if not is_portablemc_available(): - def error_gen(): lines = [ "\x1b[91m[!] PORTABLEMC NOT FOUND\x1b[0m", - "\x1b[93mPlease install it via 'pip install portablemc'.\x1b[0m", + "\x1b[93mPlease install it via 'pip install portablemc'.\x1b[0m" ] for line in lines: - payload = json.dumps({"type": "ansi", "content": line}) + payload = json.dumps({'type': 'ansi', 'content': line}) yield f"data: {payload}\n\n" yield "data: CLOSE\n\n" - return Response(error_gen(), mimetype="text/event-stream") # --- GET USER INPUT --- @@ -992,44 +975,31 @@ def error_gen(): # --- VALIDATIONS (JSON wrapped) --- if not user: - def error_gen(): - payload = json.dumps( - {"type": "ansi", "content": "\x1b[91m[!] USERNAME REQUIRED\x1b[0m"} - ) + payload = json.dumps({'type': 'ansi', 'content': "\x1b[91m[!] USERNAME REQUIRED\x1b[0m"}) yield f"data: {payload}\n\n" yield "data: CLOSE\n\n" - return Response(error_gen(), mimetype="text/event-stream") if not VALID_USERNAME_REGEX.match(user): - def error_gen(): lines = [ "\x1b[91m[!] INVALID USERNAME\x1b[0m", - "\x1b[93mUsername must be 3-16 characters and only letters, numbers, or underscore.\x1b[0m", + "\x1b[93mUsername must be 3-16 characters and only letters, numbers, or underscore.\x1b[0m" ] for line in lines: - payload = json.dumps({"type": "ansi", "content": line}) + payload = json.dumps({'type': 'ansi', 'content': line}) yield f"data: {payload}\n\n" yield "data: CLOSE\n\n" - return Response(error_gen(), mimetype="text/event-stream") user_lower = user.lower() forbidden_lower = [name.lower() for name in FORBIDDEN_LIST] if user_lower in forbidden_lower and password != PASS_KEY: - def error_gen(): - payload = json.dumps( - { - "type": "ansi", - "content": "\x1b[91m[!] ACCESS DENIED – INVALID SECURE_KEY\x1b[0m", - } - ) + payload = json.dumps({'type': 'ansi', 'content': "\x1b[91m[!] ACCESS DENIED – INVALID SECURE_KEY\x1b[0m"}) yield f"data: {payload}\n\n" yield "data: CLOSE\n\n" - return Response(error_gen(), mimetype="text/event-stream") # --- PREVENT MULTIPLE LAUNCHES (thread-safe) --- @@ -1038,35 +1008,67 @@ def error_gen(): lines = [ "\x1b[91m[!] CORE BUSY\x1b[0m", "\x1b[93mAnother Minecraft instance is already running.\x1b[0m", - "\x1b[90mPlease close the game before launching again.\x1b[0m", + "\x1b[90mPlease close the game before launching again.\x1b[0m" ] - def error_gen(): for line in lines: - payload = json.dumps({"type": "ansi", "content": line}) + payload = json.dumps({'type': 'ansi', 'content': line}) yield f"data: {payload}\n\n" yield "data: CLOSE\n\n" - return Response(error_gen(), mimetype="text/event-stream") if user in active_processes: del active_processes[user] - # --- BUILD COMMAND --- - if shutil.which("portablemc"): + # --- BUILD COMMAND (with binary vs module adaptation) --- + using_binary = os.environ.get("PORTABLEMC_METHOD") == "binary" + + # Launcher command – tied directly to the method, not PATH + if using_binary: launcher_cmd = ["portablemc"] else: launcher_cmd = [sys.executable, "-m", "portablemc"] - global_args = ["--main-dir", ".", "--timeout", "60", "--output", "human-color"] - start_args = ["--server", SERVER_IP, "--jvm-args", JVM_OPTS, "fabric:", "-u", user] + # Global arguments (before 'start') + global_args = ["--main-dir", "."] + if not using_binary: + # Python module supports these extras + global_args.extend(["--timeout", "60", "--output", "human-color"]) + + # Start‑specific arguments + start_args = [] + if using_binary: + # Binary uses --join-server + start_args.extend(["--join-server", SERVER_IP]) + else: + # Python module uses --server + start_args.extend(["--server", SERVER_IP]) + + # JVM arguments + if using_binary: + # Binary expects --jvm-arg=OPT1,OPT2,OPT3 + jvm_opts_combined = ",".join(JVM_OPTS.split()) + start_args.append(f"--jvm-arg={jvm_opts_combined}") + else: + # Python module accepts a single --jvm-args string + start_args.extend(["--jvm-args", JVM_OPTS]) + + # Version (fabric:) + start_args.append("fabric:") + + # Username + start_args.extend(["-u", user]) - # Custom Java path + # Custom Java path (supported by both) java_exe = "java.exe" if os.name == "nt" else "java" java_bin = Path.cwd() / "jvm" / "java-runtime-delta" / "bin" / java_exe if java_bin.exists(): - start_args.insert(0, "--jvm") - start_args.insert(1, str(java_bin)) + # Insert --jvm and its path right after the server arguments + # Current start_args: [server args, jvm args, version, -u user] + # We want to insert before the version, so after the server args. + # Server args are two elements, so index 2. + start_args[2:2] = ["--jvm", str(java_bin)] + # Full command cmd = launcher_cmd + global_args + ["start"] + start_args # --- DISCONNECT DETECTION --- @@ -1077,28 +1079,20 @@ def generate(): process = None got_generator_exit = False try: - # --- ATOMIC PROCESS CREATION AND TRACKING --- with processes_lock: - # Double-check that no process for this user is still running if user in active_processes and active_processes[user].poll() is None: - busy_payload = json.dumps( - {"type": "ansi", "content": "\x1b[91m[!] CORE BUSY\x1b[0m"} - ) + busy_payload = json.dumps({'type': 'ansi', 'content': '\x1b[91m[!] CORE BUSY\x1b[0m'}) yield f"data: {busy_payload}\n\n" yield "data: CLOSE\n\n" return - environment = os.environ.copy() - environment["__compat_layer"] = "runasinvoker" - startupinfo = None - if os.name == "nt": + if os.name == 'nt': startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW startupinfo.wShowWindow = subprocess.SW_HIDE # SAFE: shell=False, all arguments are separate list elements. - # The 'user' input is passed as a single argument, never executed. process = subprocess.Popen( cmd, stdout=subprocess.PIPE, @@ -1109,8 +1103,7 @@ def generate(): errors="replace", bufsize=1, creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0, - startupinfo=startupinfo, - env=environment, + startupinfo=startupinfo ) active_processes[user] = process logging.info(f"Started process for {user} with PID {process.pid}") @@ -1122,12 +1115,7 @@ def generate(): while True: if closed_event.is_set(): - disc_msg = json.dumps( - { - "type": "ansi", - "content": "\x1b[91m[SYSTEM] CONNECTION CLOSED\x1b[0m", - } - ) + disc_msg = json.dumps({'type': 'ansi', 'content': "\x1b[91m[SYSTEM] CONNECTION CLOSED\x1b[0m"}) try: yield f"data: {disc_msg}\n\n" except (BrokenPipeError, OSError, GeneratorExit): @@ -1140,7 +1128,7 @@ def generate(): break continue - raw_line = line.rstrip("\n") + raw_line = line.rstrip('\n') now = time.perf_counter() if not ok_reached and "[ OK ]" in raw_line: @@ -1151,23 +1139,18 @@ def generate(): if use_server_conversion and ansi_converter: try: safe_line = escape_html(raw_line) - html_line = ansi_converter.convert( - safe_line, full=False - ).strip() - payload = json.dumps({"type": "html", "content": html_line}) + html_line = ansi_converter.convert(safe_line, full=False).strip() + payload = json.dumps({'type': 'html', 'content': html_line}) except Exception as e: logging.error(f"Conversion failed: {e}, sending raw ANSI") - payload = json.dumps({"type": "ansi", "content": raw_line}) + payload = json.dumps({'type': 'ansi', 'content': raw_line}) else: - payload = json.dumps({"type": "ansi", "content": raw_line}) + payload = json.dumps({'type': 'ansi', 'content': raw_line}) try: if progress_match and not ok_reached: current_file = progress_match.group(1) - if ( - current_file != last_progress - and (now - last_send_time) > update_interval - ): + if current_file != last_progress and (now - last_send_time) > update_interval: yield f"data: {payload}\n\n" last_progress = current_file last_send_time = now @@ -1181,21 +1164,14 @@ def generate(): except FileNotFoundError as e: if not closed_event.is_set(): try: - payload = json.dumps( - { - "type": "ansi", - "content": f"\x1b[91m[SYSTEM] Launcher not found: {str(e)}\x1b[0m", - } - ) + payload = json.dumps({'type': 'ansi', 'content': f"\x1b[91m[SYSTEM] Launcher not found: {str(e)}\x1b[0m"}) yield f"data: {payload}\n\n" except Exception: pass except Exception as e: if not closed_event.is_set(): try: - payload = json.dumps( - {"type": "ansi", "content": f"[SYSTEM ERROR] {str(e)}"} - ) + payload = json.dumps({'type': 'ansi', 'content': f"[SYSTEM ERROR] {str(e)}"}) yield f"data: {payload}\n\n" except Exception: pass @@ -1210,18 +1186,8 @@ def generate(): if not got_generator_exit: try: - ended_msg = json.dumps( - { - "type": "ansi", - "content": "\x1b[90m[SYSTEM] SESSION ENDED\x1b[0m", - } - ) - tip_msg = json.dumps( - { - "type": "ansi", - "content": "\x1b[34m[TIP] Click the console to return to login.\x1b[0m", - } - ) + ended_msg = json.dumps({'type': 'ansi', 'content': "\x1b[90m[SYSTEM] SESSION ENDED\x1b[0m"}) + tip_msg = json.dumps({'type': 'ansi', 'content': "\x1b[34m[TIP] Click the console to return to login.\x1b[0m"}) yield f"data: {ended_msg}\n\n" yield f"data: {tip_msg}\n\n" yield "data: CLOSE\n\n" @@ -1234,18 +1200,15 @@ def generate(): response.call_on_close(closed_event.set) return response - @app.route("/") def home(): return render_template_string(HTML_TEMPLATE, forbidden_list=FORBIDDEN_LIST) - def kill_process_tree(proc): """Kill a process and all its children. Falls back to simple kill if psutil missing.""" if proc.poll() is not None: logging.debug(f"Process {proc.pid} already dead") return - if PSUTIL_AVAILABLE: try: parent = psutil.Process(proc.pid) @@ -1260,7 +1223,6 @@ def kill_process_tree(proc): except psutil.NoSuchProcess: logging.debug(f"Process {proc.pid} already gone") else: - # Fallback: terminate the main process only try: proc.terminate() proc.wait(timeout=2) @@ -1268,7 +1230,6 @@ def kill_process_tree(proc): proc.kill() logging.info(f"Terminated process PID {proc.pid} (psutil unavailable)") - def graceful_shutdown(sig, frame): logging.info("SHUTTING DOWN CORE...") with processes_lock: @@ -1276,13 +1237,10 @@ def graceful_shutdown(sig, frame): kill_process_tree(proc) sys.exit(0) - -# Set signal handler for SIGINT (Ctrl+C) signal.signal(signal.SIGINT, graceful_shutdown) if __name__ == "__main__": try: socketio.run(app, port=5000, debug=False, allow_unsafe_werkzeug=True) except KeyboardInterrupt: - # Fallback if signal handler doesn't catch it graceful_shutdown(None, None)