""" Telearn Arcade Hub — FastAPI server. Serves the project catalog, static assets, and WebSocket bridge that streams headless Pygame/OpenCV student projects to the browser. """ from __future__ import annotations import argparse import asyncio import hashlib import json import logging import os import signal import socket import shutil import stat import subprocess import sys import tempfile import time import secrets import re from collections import deque from pathlib import Path from typing import Any, Optional from urllib.parse import urlsplit from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from classroom_service import router as classroom_router from admin_service import router as admin_router # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- BASE_DIR = os.path.dirname(os.path.abspath(__file__)) PROJECTS_DIR = os.path.join(BASE_DIR, "projects") TEMPLATES_DIR = os.path.join(BASE_DIR, "templates") STATIC_DIR = os.path.join(BASE_DIR, "static") MODELS_DIR = os.path.join(BASE_DIR, "models") MAX_CONCURRENT_RUNNERS = int(os.environ.get("MAX_CONCURRENT_RUNNERS", "3")) SESSION_TIMEOUT_SEC = int(os.environ.get("SESSION_TIMEOUT_SEC", "900")) # 15 min IDLE_TIMEOUT_SEC = max(1, int(os.environ.get("IDLE_TIMEOUT_SEC", "720"))) # 12 min IDLE_WARNING_SEC = min( IDLE_TIMEOUT_SEC, max(0, int(os.environ.get("IDLE_WARNING_SEC", "600"))) ) # warn after 10 min idle HEARTBEAT_TIMEOUT_SEC = max( 30, int(os.environ.get("HEARTBEAT_TIMEOUT_SEC", "75")) ) PROJECT_CACHE_TTL_SEC = float(os.environ.get("PROJECT_CACHE_TTL_SEC", "10")) RUNNER_CONNECT_TIMEOUT_SEC = float(os.environ.get("RUNNER_CONNECT_TIMEOUT_SEC", "8")) RUNNER_TMP_DIR = os.environ.get("RUNNER_TMP_DIR", "/tmp/telearn_runner") RUNNER_MEMORY_LIMIT_MB = max(512, int(os.environ.get("RUNNER_MEMORY_LIMIT_MB", "3072"))) RUNNER_CPU_LIMIT_SEC = max(30, int(os.environ.get("RUNNER_CPU_LIMIT_SEC", "900"))) RUNNER_TMP_LIMIT_MB = max(64, int(os.environ.get("RUNNER_TMP_LIMIT_MB", "256"))) RUNNER_PROJECT_LIMIT_MB = max(8, int(os.environ.get("RUNNER_PROJECT_LIMIT_MB", "128"))) RUNNER_PROJECT_FILE_LIMIT = max(20, int(os.environ.get("RUNNER_PROJECT_FILE_LIMIT", "2000"))) RUNNER_STARTS_PER_IP_PER_HOUR = max(1, int(os.environ.get("RUNNER_STARTS_PER_IP_PER_HOUR", "120"))) RUNNER_START_COOLDOWN_SEC = max(0.5, float(os.environ.get("RUNNER_START_COOLDOWN_SEC", "0.75"))) ENABLE_HSTS = os.environ.get("ENABLE_HSTS", "false").lower() in {"1", "true", "yes"} HSTS_MAX_AGE_SEC = max(0, int(os.environ.get("HSTS_MAX_AGE_SEC", "31536000"))) BWRAP_PATH = shutil.which("bwrap") PRLIMIT_PATH = shutil.which("prlimit") RUNNER_ISOLATION_MODE = os.environ.get("RUNNER_ISOLATION_MODE", "sandbox").lower() os.makedirs(PROJECTS_DIR, exist_ok=True) os.makedirs(TEMPLATES_DIR, exist_ok=True) os.makedirs(STATIC_DIR, exist_ok=True) os.makedirs(MODELS_DIR, exist_ok=True) os.makedirs(RUNNER_TMP_DIR, mode=0o700, exist_ok=True) os.chmod(RUNNER_TMP_DIR, 0o700) logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s [server] %(message)s", datefmt="%Y-%m-%dT%H:%M:%S", ) log = logging.getLogger("telearn.server") app = FastAPI(title="Student Pygame/OpenCV Arcade Hub") templates = Jinja2Templates(directory=TEMPLATES_DIR) app.mount("/classroom-assets", StaticFiles(directory=os.path.join(BASE_DIR, "classroom_static")), name="classroom-assets") app.include_router(classroom_router) app.include_router(admin_router) @app.middleware("http") async def browser_runtime_headers(request: Request, call_next): """Set browser isolation and response security policies.""" nonce = secrets.token_urlsafe(18) request.scope.setdefault("state", {})["csp_nonce"] = nonce response = await call_next(request) response.headers["Cross-Origin-Opener-Policy"] = "same-origin" response.headers["Cross-Origin-Embedder-Policy"] = "credentialless" response.headers["X-Content-Type-Options"] = "nosniff" response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" response.headers["Permissions-Policy"] = "camera=(self), microphone=(), geolocation=(), payment=(), usb=()" response.headers["Content-Security-Policy"] = ( "default-src 'self'; base-uri 'self'; object-src 'none'; " f"script-src 'self' 'nonce-{nonce}' 'wasm-unsafe-eval'; " "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; " "img-src 'self' data: blob: https://api.qrserver.com; " "font-src 'self' data: https://fonts.gstatic.com; " "connect-src 'self' ws: wss:; " "worker-src 'self' blob:; " "frame-src 'self' https://appetize.io https://*.appetize.io https://makecode.com https://*.makecode.com; " "frame-ancestors 'self' https://huggingface.co; form-action 'self'" ) # Student-authored web projects need JavaScript, but must remain an # opaque-origin sandbox even when opened directly instead of in the player. if request.url.path.startswith("/static_projects/"): response.headers["Content-Security-Policy"] = ( "sandbox allow-scripts allow-forms allow-pointer-lock; " "default-src 'self' data: blob: https:; " "script-src 'self' 'unsafe-inline' 'unsafe-eval' https:; " "style-src 'self' 'unsafe-inline' https:; " "img-src 'self' data: blob: https:; font-src 'self' data: https:; " "connect-src 'self' https: wss:; media-src 'self' data: blob: https:; " "object-src 'none'; base-uri 'self'; form-action 'self'; frame-ancestors 'self'" ) if HSTS_MAX_AGE_SEC and (ENABLE_HSTS or request.url.scheme == "https"): response.headers["Strict-Transport-Security"] = f"max-age={HSTS_MAX_AGE_SEC}" return response # Concurrency + live session metrics runner_semaphore = asyncio.Semaphore(MAX_CONCURRENT_RUNNERS) active_sessions: dict[str, dict[str, Any]] = {} _sessions_lock = asyncio.Lock() _metrics = { "sessions_started": 0, "sessions_rejected": 0, "sessions_timed_out": 0, "sessions_idle_timed_out": 0, "sessions_heartbeat_timed_out": 0, "sessions_failed_connect": 0, } # Project tree cache _project_cache: dict[str, Any] = {"tree": None, "built_at": 0.0, "signature": None} _runner_start_windows: dict[str, deque] = {} def allow_runner_start(remote_ip: str) -> bool: """Limit public server-side Python starts per client IP across sessions.""" current = time.monotonic() window = _runner_start_windows.get(remote_ip) if window is None: window = deque() _runner_start_windows[remote_ip] = window while window and current - window[0] >= 3600: window.popleft() if window and current - window[-1] < RUNNER_START_COOLDOWN_SEC: return False if len(window) >= RUNNER_STARTS_PER_IP_PER_HOUR: return False window.append(current) if len(_runner_start_windows) > 4096: for key, old_window in list(_runner_start_windows.items()): while old_window and current - old_window[0] >= 3600: old_window.popleft() if not old_window: _runner_start_windows.pop(key, None) if len(_runner_start_windows) <= 3072: break return True def _session_idle_state( idle_for_sec: float, warning_sec: float = IDLE_WARNING_SEC, timeout_sec: float = IDLE_TIMEOUT_SEC, ) -> str: """Return active, warning, or timeout for a session's idle duration.""" if idle_for_sec >= timeout_sec: return "timeout" if warning_sec < timeout_sec and idle_for_sec >= warning_sec: return "warning" return "active" # --------------------------------------------------------------------------- # Project scanning (cached) # --------------------------------------------------------------------------- def _dir_signature(root: str) -> str: """Cheap signature: path + mtime + name for dirs up to depth 3.""" parts: list[str] = [] if not os.path.isdir(root): return "" try: for class_name in sorted(os.listdir(root)): class_path = os.path.join(root, class_name) if not os.path.isdir(class_path) or class_name.startswith("."): continue try: parts.append(f"{class_name}:{os.path.getmtime(class_path)}") except OSError: parts.append(class_name) for student_name in sorted(os.listdir(class_path)): student_path = os.path.join(class_path, student_name) if not os.path.isdir(student_path) or student_name.startswith("."): continue try: parts.append(f"{class_name}/{student_name}:{os.path.getmtime(student_path)}") except OSError: parts.append(f"{class_name}/{student_name}") for project_name in sorted(os.listdir(student_path)): project_path = os.path.join(student_path, project_name) if not os.path.isdir(project_path) or project_name.startswith("."): continue try: parts.append( f"{class_name}/{student_name}/{project_name}:{os.path.getmtime(project_path)}" ) except OSError: parts.append(f"{class_name}/{student_name}/{project_name}") except OSError: return "" return "|".join(parts) def _scan_projects_uncached() -> dict: """ Scan projects/ into: { class_name: { student_name: [ {name, type, url}, ... ] } } Priority: python > html > apk > external """ tree: dict = {} if not os.path.exists(PROJECTS_DIR): return tree for class_name in sorted(os.listdir(PROJECTS_DIR)): class_path = os.path.join(PROJECTS_DIR, class_name) if not os.path.isdir(class_path) or class_name.startswith("."): continue tree[class_name] = {} for student_name in sorted(os.listdir(class_path)): student_path = os.path.join(class_path, student_name) if not os.path.isdir(student_path) or student_name.startswith("."): continue projects = [] for project_name in sorted(os.listdir(student_path)): project_dir = os.path.join(student_path, project_name) if not os.path.isdir(project_dir) or project_name.startswith("."): continue main_py = os.path.join(project_dir, "main.py") index_html = os.path.join(project_dir, "index.html") link_txt = os.path.join(project_dir, "link.txt") apk_files = [] try: apk_files = [ f for f in os.listdir(project_dir) if f.endswith(".apk") and not f.startswith(".") ] except OSError: pass if os.path.exists(main_py): projects.append({"name": project_name, "type": "python", "url": ""}) elif os.path.exists(index_html): projects.append({"name": project_name, "type": "html", "url": ""}) elif apk_files: projects.append({"name": project_name, "type": "apk", "url": apk_files[0]}) elif os.path.exists(link_txt): try: with open(link_txt, "r", encoding="utf-8") as f: url = f.read().strip() if _safe_external_url(url): projects.append({"name": project_name, "type": "external", "url": url}) except OSError: pass if projects: for project in projects: project["display_name"] = _project_display_name(class_name, student_name, project["name"]) tree[class_name][student_name] = projects if not tree[class_name]: del tree[class_name] return tree def scan_projects(force: bool = False) -> dict: """Return cached project tree; refresh when TTL expires or directory signature changes.""" now = time.time() sig = _dir_signature(PROJECTS_DIR) cache_age = now - _project_cache["built_at"] if ( not force and _project_cache["tree"] is not None and cache_age < PROJECT_CACHE_TTL_SEC and _project_cache["signature"] == sig ): return _project_cache["tree"] tree = _scan_projects_uncached() _project_cache["tree"] = tree _project_cache["built_at"] = now _project_cache["signature"] = sig return tree def _project_display_name(class_name: str, student_name: str, project_name: str) -> str: if class_name == "DEMO" and student_name == "展示用學生": return re.sub(r"^\d+_", "", project_name) return project_name def get_id(name: str) -> str: """Stable 10-char hex hash for URL tokens.""" return hashlib.sha256(name.encode("utf-8")).hexdigest()[:10] def decode_class_student( class_name_or_id: str, student_name_or_id: Optional[str] = None ) -> tuple[Optional[str], Optional[str]]: """Resolve hashed or raw class/student identifiers to folder names.""" all_projects = scan_projects() if not student_name_or_id: for c in all_projects: for s in all_projects[c]: if get_id(f"{c}/{s}") == class_name_or_id: return c, s if "/" in class_name_or_id: parts = class_name_or_id.split("/", 1) c, s = parts[0], parts[1] if c in all_projects and s in all_projects[c]: return c, s real_class = None for c in all_projects: if get_id(c) == class_name_or_id: real_class = c break if not real_class and class_name_or_id in all_projects: real_class = class_name_or_id return real_class, None real_class = None for c in all_projects: if get_id(c) == class_name_or_id: real_class = c break if not real_class and class_name_or_id in all_projects: real_class = class_name_or_id real_student = None if real_class and real_class in all_projects: for s in all_projects[real_class]: if get_id(f"{real_class}/{s}") == student_name_or_id: real_student = s break if not real_student and student_name_or_id in all_projects[real_class]: real_student = student_name_or_id return real_class, real_student def _safe_project_path(*parts: str) -> Optional[str]: """Join under PROJECTS_DIR; return None if path escapes the root.""" file_path = os.path.join(PROJECTS_DIR, *parts) abs_path = os.path.abspath(file_path) root = os.path.abspath(PROJECTS_DIR) if not abs_path.startswith(root + os.sep) and abs_path != root: return None return abs_path def _resolve_python_project(class_name: str, student_name: str, project_name: str) -> str: """Resolve only catalogued Python projects beneath their student's folder.""" student_projects = scan_projects().get(class_name, {}).get(student_name, []) allowed = any(item["name"] == project_name and item["type"] == "python" for item in student_projects) student_root = os.path.realpath(os.path.join(PROJECTS_DIR, class_name, student_name)) project_root = os.path.realpath(os.path.join(student_root, project_name)) if not allowed or os.path.commonpath((student_root, project_root)) != student_root: raise FileNotFoundError("project is not a listed Python project") main_file = os.path.join(project_root, "main.py") if not os.path.isfile(main_file): raise FileNotFoundError("project has no main.py") return project_root def _safe_external_url(value: str) -> str: """Only expose absolute HTTPS project links to browser navigation/iframes.""" value = value.strip() try: parsed = urlsplit(value) if parsed.scheme == "https" and parsed.hostname and not parsed.username and not parsed.password: return value except ValueError: pass return "" def _stage_runner_project(source_root: str) -> tuple[str, str]: """Copy regular project files into an ephemeral, per-run writable directory.""" session_root = tempfile.mkdtemp(prefix="run-", dir=RUNNER_TMP_DIR) os.chmod(session_root, 0o700) workspace = os.path.join(session_root, "project") os.mkdir(workspace, 0o700) total_bytes = 0 total_files = 0 ignored_dirs = {".git", ".venv", "venv", "__pycache__", "node_modules"} for current, dirs, files in os.walk(source_root, followlinks=False): dirs[:] = [name for name in dirs if name not in ignored_dirs and not os.path.islink(os.path.join(current, name))] relative = os.path.relpath(current, source_root) target_dir = workspace if relative == "." else os.path.join(workspace, relative) if target_dir != workspace: os.makedirs(target_dir, mode=0o700, exist_ok=True) for name in files: source = os.path.join(current, name) try: info = os.lstat(source) except OSError: continue if not stat.S_ISREG(info.st_mode): continue lowered = name.casefold() if lowered == ".env" or lowered.startswith(".env.") or lowered in { "credentials.json", "service-account.json", "secrets.json", "token.txt", }: continue total_files += 1 total_bytes += info.st_size if total_files > RUNNER_PROJECT_FILE_LIMIT: shutil.rmtree(session_root, ignore_errors=True) raise ValueError(f"專案檔案數超過上限（{RUNNER_PROJECT_FILE_LIMIT} 個）") if total_bytes > RUNNER_PROJECT_LIMIT_MB * 1024 * 1024: shutil.rmtree(session_root, ignore_errors=True) raise ValueError(f"專案資料超過執行上限（{RUNNER_PROJECT_LIMIT_MB} MB）") destination = os.path.join(target_dir, name) shutil.copyfile(source, destination) os.chmod(destination, 0o600) os.chmod(target_dir, 0o700) if not os.path.isfile(os.path.join(workspace, "main.py")): shutil.rmtree(session_root, ignore_errors=True) raise FileNotFoundError("專案沒有可執行的 main.py") return session_root, workspace def _runner_sandbox_command(workspace: str, socket_path: str) -> list[str]: """Build a fail-closed bubblewrap command with no network or app secrets.""" if sys.platform != "linux" or not BWRAP_PATH or not PRLIMIT_PATH: raise RuntimeError("伺服器缺少 Linux bubblewrap/prlimit，已拒絕未隔離執行。") bwrap = [ BWRAP_PATH, "--die-with-parent", "--new-session", "--unshare-user", "--unshare-pid", "--unshare-ipc", "--unshare-net", "--unshare-uts", "--unshare-cgroup-try", "--disable-userns", ] for system_path in ("/usr", "/lib", "/lib64", "/bin", "/sbin", "/etc", "/opt"): if os.path.exists(system_path): bwrap.extend(("--ro-bind", system_path, system_path)) bwrap.extend(( "--dev", "/dev", "--proc", "/proc", "--size", str(RUNNER_TMP_LIMIT_MB * 1024 * 1024), "--tmpfs", "/tmp", "--dir", "/workspace", "--dir", "/workspace/project", "--bind", workspace, "/workspace/project", "--ro-bind", os.path.join(BASE_DIR, "runner.py"), "/runner.py", "--ro-bind", MODELS_DIR, "/models", "--ro-bind", os.path.dirname(socket_path), "/run-ipc", "--clearenv", "--setenv", "PATH", "/usr/local/bin:/usr/bin:/bin", "--setenv", "HOME", "/tmp", "--setenv", "LANG", "C.UTF-8", "--setenv", "PYTHONNOUSERSITE", "1", "--setenv", "PYTHONDONTWRITEBYTECODE", "1", "--setenv", "SDL_VIDEODRIVER", "dummy", "--setenv", "SDL_AUDIODRIVER", "dummy", "--chdir", "/workspace/project", "--", sys.executable, "/runner.py", "--project_path", "/workspace/project/main.py", "--socket-path", "/run-ipc/runner.sock", "--models_dir", "/models", )) return _runner_with_resource_limits(bwrap) def _runner_with_resource_limits(command: list[str]) -> list[str]: if not PRLIMIT_PATH: raise RuntimeError("伺服器缺少 prlimit，無法套用執行資源限制。") memory_bytes = RUNNER_MEMORY_LIMIT_MB * 1024 * 1024 file_bytes = 128 * 1024 * 1024 return [ PRLIMIT_PATH, f"--cpu={RUNNER_CPU_LIMIT_SEC}:{RUNNER_CPU_LIMIT_SEC}", f"--as={memory_bytes}:{memory_bytes}", f"--fsize={file_bytes}:{file_bytes}", "--nproc=64:64", "--nofile=128:128", "--core=0:0", "--", *command, ] def _runner_command(workspace: str, socket_path: str) -> list[str]: if RUNNER_ISOLATION_MODE == "sandbox": return _runner_sandbox_command(workspace, socket_path) if RUNNER_ISOLATION_MODE != "process" or sys.platform != "linux": raise RuntimeError("不支援的 Python 執行模式。") return _runner_with_resource_limits([ sys.executable, os.path.join(BASE_DIR, "runner.py"), "--project_path", os.path.join(workspace, "main.py"), "--socket-path", socket_path, "--models_dir", MODELS_DIR, ]) def _runner_environment(workspace: str) -> dict[str, str]: """Do not pass application tokens or admin credentials to child processes.""" return { "PATH": "/usr/local/bin:/usr/bin:/bin", "HOME": workspace, "LANG": "C.UTF-8", "PYTHONNOUSERSITE": "1", "PYTHONDONTWRITEBYTECODE": "1", "PYTHONUNBUFFERED": "1", "SDL_VIDEODRIVER": "dummy", "SDL_AUDIODRIVER": "dummy", "OPENBLAS_NUM_THREADS": "1", "OMP_NUM_THREADS": "1", } # --------------------------------------------------------------------------- # Static / logo # --------------------------------------------------------------------------- if os.path.isdir(STATIC_DIR): app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") @app.get("/images/logo-light.png") @app.get("/logo.png") async def get_logo(): return FileResponse(os.path.join(TEMPLATES_DIR, "logo-light.png")) # Controlled project file serving only (no raw StaticFiles mount on projects/) @app.get("/static_projects/{student_token}/{project_name}/{path:path}") async def serve_static_project_file(student_token: str, project_name: str, path: str): real_class, real_student = decode_class_student(student_token) if not real_class or not real_student: # Legacy: /static_projects/{class_id}/{student_id}/{project}/{path} parts = path.split("/", 1) if len(parts) == 2: actual_project_name, actual_path = parts[0], parts[1] real_class, real_student = decode_class_student(student_token, project_name) if real_class and real_student: file_path = _safe_project_path( real_class, real_student, actual_project_name, actual_path ) if file_path and os.path.isfile(file_path): return FileResponse(file_path) return HTMLResponse("Not Found", status_code=404) file_path = _safe_project_path(real_class, real_student, project_name, path) if not file_path or not os.path.isfile(file_path): return HTMLResponse("File Not Found", status_code=404) return FileResponse(file_path) # --------------------------------------------------------------------------- # HTTP routes # --------------------------------------------------------------------------- @app.get("/health") async def health(): return { "status": "ok", "active_sessions": len(active_sessions), "max_concurrent_runners": MAX_CONCURRENT_RUNNERS, "session_timeout_sec": SESSION_TIMEOUT_SEC, "idle_warning_sec": IDLE_WARNING_SEC, "idle_timeout_sec": IDLE_TIMEOUT_SEC, "heartbeat_timeout_sec": HEARTBEAT_TIMEOUT_SEC, } @app.get("/api/metrics") async def metrics(): return { "active_sessions": len(active_sessions), "max_concurrent_runners": MAX_CONCURRENT_RUNNERS, "sessions": list(active_sessions.values()), **_metrics, } @app.get("/", response_class=HTMLResponse) async def read_index(request: Request): return templates.TemplateResponse( request=request, name="index.html", context={"restrict_class": "", "restrict_student": ""}, ) @app.get("/class/{class_name}", response_class=HTMLResponse) async def read_class(request: Request, class_name: str): real_class, _ = decode_class_student(class_name) if not real_class: return HTMLResponse("Class not found", status_code=404) return templates.TemplateResponse( request=request, name="index.html", context={"restrict_class": real_class, "restrict_student": ""}, ) @app.get("/student/{student_token}", response_class=HTMLResponse) async def read_student(request: Request, student_token: str): real_class, real_student = decode_class_student(student_token) if not real_class or not real_student: return HTMLResponse("Student not found", status_code=404) return templates.TemplateResponse( request=request, name="index.html", context={ "restrict_class": real_class, "restrict_student": real_student, }, ) @app.get("/student/{class_name}/{student_name}", response_class=HTMLResponse) async def read_student_old(request: Request, class_name: str, student_name: str): real_class, real_student = decode_class_student(class_name, student_name) if not real_class or not real_student: return HTMLResponse("Student not found", status_code=404) student_token = get_id(f"{real_class}/{real_student}") return RedirectResponse(url=f"/student/{student_token}", status_code=307) @app.get("/api/projects", response_class=JSONResponse) async def get_projects(class_name: str = None, student_name: str = None): all_projects = scan_projects() class_ids = {c: get_id(c) for c in all_projects} student_ids = { f"{c}/{s}": get_id(f"{c}/{s}") for c in all_projects for s in all_projects[c] } if class_name: real_class, real_student = decode_class_student(class_name, student_name) if real_class in all_projects: if real_student: if real_student in all_projects[real_class]: filtered = {real_class: {real_student: all_projects[real_class][real_student]}} else: filtered = {real_class: {}} else: filtered = {real_class: all_projects[real_class]} else: filtered = {} else: filtered = all_projects return {"projects": filtered, "class_ids": class_ids, "student_ids": student_ids} @app.get("/run/{student_token}/{project_name}", response_class=HTMLResponse) async def run_project_page(request: Request, student_token: str, project_name: str): real_class, real_student = decode_class_student(student_token) if not real_class or not real_student: return HTMLResponse("Project not found", status_code=404) proj_dir = os.path.join(PROJECTS_DIR, real_class, real_student, project_name) proj_type = "unknown" external_url = "" apk_name = "" apk_size = "" appetize_key = "" view_only = request.query_params.get("view") == "1" apk_files = [] if os.path.isdir(proj_dir): try: apk_files = [ f for f in os.listdir(proj_dir) if f.endswith(".apk") and not f.startswith(".") ] except OSError: pass if os.path.exists(os.path.join(proj_dir, "main.py")): proj_type = "python" elif os.path.exists(os.path.join(proj_dir, "index.html")): proj_type = "html" elif apk_files: proj_type = "apk" apk_name = apk_files[0] try: size_bytes = os.path.getsize(os.path.join(proj_dir, apk_name)) if size_bytes < 1024 * 1024: apk_size = f"{size_bytes / 1024:.1f} KB" else: apk_size = f"{size_bytes / (1024 * 1024):.1f} MB" except OSError: pass appetize_path = os.path.join(proj_dir, "appetize.txt") if os.path.exists(appetize_path): try: with open(appetize_path, "r", encoding="utf-8") as f: candidate_key = f.read().strip() if re.fullmatch(r"[A-Za-z0-9_-]{1,256}", candidate_key): appetize_key = candidate_key except OSError: pass elif os.path.exists(os.path.join(proj_dir, "link.txt")): try: with open(os.path.join(proj_dir, "link.txt"), "r", encoding="utf-8") as f: external_url = _safe_external_url(f.read()) if external_url: proj_type = "external" except OSError: pass return templates.TemplateResponse( request=request, name="player.html", context={ "class_name": real_class, "student_name": real_student, "student_token": student_token, "project_name": project_name, "project_display_name": _project_display_name(real_class, real_student, project_name), "project_type": proj_type, "external_url": external_url, "apk_name": apk_name, "apk_size": apk_size, "appetize_key": appetize_key, "view_only": view_only, "session_timeout_sec": SESSION_TIMEOUT_SEC, "idle_warning_sec": IDLE_WARNING_SEC, "idle_timeout_sec": IDLE_TIMEOUT_SEC, "heartbeat_interval_sec": min(20, max(10, HEARTBEAT_TIMEOUT_SEC // 3)), "max_concurrent_runners": MAX_CONCURRENT_RUNNERS, }, ) @app.get("/run/{class_name}/{student_name}/{project_name}", response_class=HTMLResponse) async def run_project_page_old( request: Request, class_name: str, student_name: str, project_name: str ): real_class, real_student = decode_class_student(class_name, student_name) if not real_class or not real_student: return HTMLResponse("Project not found", status_code=404) student_token = get_id(f"{real_class}/{real_student}") qs = f"?{request.url.query}" if request.url.query else "" return RedirectResponse( url=f"/run/{student_token}/{project_name}{qs}", status_code=307 ) # --------------------------------------------------------------------------- # Process helpers # --------------------------------------------------------------------------- def _kill_process_group(process: asyncio.subprocess.Process) -> None: """Terminate the entire process group (runner + children).""" if process.returncode is not None: return pid = process.pid try: if sys.platform == "win32": process.terminate() else: try: os.killpg(pid, signal.SIGTERM) except (ProcessLookupError, PermissionError, OSError): process.terminate() except ProcessLookupError: pass async def _cleanup_process(process: asyncio.subprocess.Process) -> None: if process is None or process.returncode is not None: return log.info("Cleaning up process group pid=%s", process.pid) _kill_process_group(process) try: await asyncio.wait_for(process.wait(), timeout=2.0) except asyncio.TimeoutError: try: if sys.platform == "win32": process.kill() else: try: os.killpg(process.pid, signal.SIGKILL) except (ProcessLookupError, PermissionError, OSError): process.kill() except ProcessLookupError: pass try: await process.wait() except Exception: pass async def _capture_bounded_stderr( stream: asyncio.StreamReader, captured: bytearray, limit: int = 4096 ) -> None: """Drain child stderr without allowing an unbounded buffer or blocked pipe.""" while True: chunk = await stream.read(1024) if not chunk: return remaining = limit - len(captured) if remaining > 0: captured.extend(chunk[:remaining]) # --------------------------------------------------------------------------- # WebSocket play session # --------------------------------------------------------------------------- async def play_project_ws_internal( websocket: WebSocket, real_class: str, real_student: str, project_name: str ): if not real_class or not real_student: await websocket.accept() await websocket.send_json( {"type": "log", "data": "錯誤: 找不到該專案的執行路徑！\n"} ) await websocket.close() return try: source_root = _resolve_python_project(real_class, real_student, project_name) except (OSError, ValueError): await websocket.accept() await websocket.send_json( {"type": "log", "data": "錯誤: 找不到該專案的 main.py 檔案！\n"} ) await websocket.close() return # Concurrency gate: reject immediately if all runner slots are busy acquired = False try: await asyncio.wait_for(runner_semaphore.acquire(), timeout=0.05) acquired = True except asyncio.TimeoutError: _metrics["sessions_rejected"] += 1 await websocket.accept() await websocket.send_json( { "type": "log", "data": ( f"目前展示名額已滿（最多同時 {MAX_CONCURRENT_RUNNERS} 個專案）。" "請稍後再試。\n" ), } ) await websocket.send_json({"type": "status", "data": "busy"}) await websocket.close(code=1013) return remote_ip = websocket.client.host if websocket.client else "unknown" if not allow_runner_start(remote_ip): runner_semaphore.release() acquired = False _metrics["sessions_rejected"] += 1 await websocket.accept() await websocket.send_json({"type": "log", "data": "這個網路來源的專案啟動次數已達上限，請稍後再試。\n"}) await websocket.send_json({"type": "status", "data": "rate_limited"}) await websocket.close(code=4429, reason="runner start rate limited") return await websocket.accept() session_id = f"{real_class}/{real_student}/{project_name}/{time.time():.0f}" started_at = time.time() started_monotonic = time.monotonic() last_activity_at = started_monotonic last_client_seen_at = started_monotonic async with _sessions_lock: active_sessions[session_id] = { "class": real_class, "student": real_student, "project": project_name, "started_at": started_at, "last_activity_at": started_at, } _metrics["sessions_started"] += 1 process = None process_stderr_task = None process_stderr = bytearray() ipc_server = None runner_writer = None session_root = None try: connected_event = asyncio.Event() runner_reader = None async def handle_runner_conn(reader, writer): nonlocal runner_reader, runner_writer runner_reader = reader runner_writer = writer connected_event.set() try: session_root, workspace = _stage_runner_project(source_root) ipc_dir = os.path.join(session_root, "ipc") os.mkdir(ipc_dir, 0o700) socket_path = os.path.join(ipc_dir, "runner.sock") cmd = _runner_command(workspace, socket_path) ipc_server = await asyncio.start_unix_server( handle_runner_conn, path=socket_path ) os.chmod(socket_path, 0o600) except (OSError, RuntimeError, ValueError) as exc: log.warning("Unable to prepare runner: %s", exc) await websocket.send_json({"type": "log", "data": f"專案執行器無法啟動，已停止執行：{exc}\n"}) await websocket.send_json({"type": "status", "data": "error"}) return popen_kwargs: dict[str, Any] = { "stdout": subprocess.DEVNULL, "stderr": subprocess.PIPE, "cwd": workspace, "env": _runner_environment(workspace), } if sys.platform != "win32": popen_kwargs["start_new_session"] = True process = await asyncio.create_subprocess_exec(*cmd, **popen_kwargs) if process.stderr is not None: process_stderr_task = asyncio.create_task( _capture_bounded_stderr(process.stderr, process_stderr) ) log.info("Started runner mode=%s pid=%s session=%s", RUNNER_ISOLATION_MODE, process.pid, session_id) try: await asyncio.wait_for( connected_event.wait(), timeout=RUNNER_CONNECT_TIMEOUT_SEC ) log.info("Runner pid=%s connected via private IPC", process.pid) await websocket.send_json( {"type": "status", "data": "running"} ) await websocket.send_json( { "type": "log", "data": f"[系統] 專案執行器已就緒（最長 {SESSION_TIMEOUT_SEC // 60} 分鐘）。\n", } ) except asyncio.TimeoutError: _metrics["sessions_failed_connect"] += 1 if process.returncode is None: try: await asyncio.wait_for(process.wait(), timeout=0.2) except asyncio.TimeoutError: pass if process.returncode is not None and process_stderr_task is not None: await asyncio.gather(process_stderr_task, return_exceptions=True) stderr_detail = process_stderr.decode("utf-8", errors="replace").strip() log.error( "Runner pid=%s IPC connect timeout; returncode=%s; stderr=%s", process.pid, process.returncode, stderr_detail or "", ) await websocket.send_json( {"type": "log", "data": ( "安全執行環境無法連線（可能是主機未啟用 Linux namespace）；已拒絕未隔離執行。\n" if RUNNER_ISOLATION_MODE == "sandbox" else "專案執行器啟動失敗，請稍後重試或請老師查看伺服器日誌。\n" )} ) await websocket.send_json({"type": "status", "data": "error"}) return async def tcp_to_ws(): try: while True: type_bytes = await runner_reader.readexactly(4) msg_type = type_bytes.decode("utf-8") len_bytes = await runner_reader.readexactly(4) length = int.from_bytes(len_bytes, byteorder="big") payload = await runner_reader.readexactly(length) if msg_type == "FRME": await websocket.send_bytes(payload) elif msg_type == "LOGS": log_text = payload.decode("utf-8", errors="ignore") await websocket.send_json({"type": "log", "data": log_text}) elif msg_type == "EXIT": log.info("Runner pid=%s sent EXIT", process.pid) await websocket.send_json({"type": "status", "data": "stopped"}) break except asyncio.IncompleteReadError: log.info("Runner pid=%s TCP incomplete read", process.pid) except WebSocketDisconnect: pass except Exception as e: log.exception("tcp_to_ws error: %s", e) async def ws_to_tcp(): nonlocal last_activity_at, last_client_seen_at try: while True: data = await websocket.receive() now = time.monotonic() last_client_seen_at = now if "text" in data: message = json.loads(data["text"]) mtype = message.get("type") if mtype == "heartbeat": continue if mtype == "activity": last_activity_at = now active_sessions[session_id]["last_activity_at"] = time.time() elif mtype in ("keydown", "keyup"): last_activity_at = now active_sessions[session_id]["last_activity_at"] = time.time() payload = json.dumps(message).encode("utf-8") runner_writer.write( b"KEYS" + len(payload).to_bytes(4, "big") + payload ) await runner_writer.drain() elif mtype in ("mousemove", "mousedown", "mouseup"): last_activity_at = now active_sessions[session_id]["last_activity_at"] = time.time() payload = json.dumps(message).encode("utf-8") runner_writer.write( b"MOUS" + len(payload).to_bytes(4, "big") + payload ) await runner_writer.drain() elif mtype == "cv2key": last_activity_at = now active_sessions[session_id]["last_activity_at"] = time.time() payload = json.dumps( {"key_code": message.get("keyCode")} ).encode("utf-8") runner_writer.write( b"CV2K" + len(payload).to_bytes(4, "big") + payload ) await runner_writer.drain() elif mtype == "terminal_input": last_activity_at = now active_sessions[session_id]["last_activity_at"] = time.time() text = message.get("data", "") if not text.endswith("\n"): text += "\n" payload = text.encode("utf-8") runner_writer.write( b"INPT" + len(payload).to_bytes(4, "big") + payload ) await runner_writer.drain() elif mtype == "pygame_quit": last_activity_at = now runner_writer.write(b"QUIT" + (0).to_bytes(4, "big") + b"") await runner_writer.drain() elif "bytes" in data: payload = data["bytes"] runner_writer.write( b"CAMF" + len(payload).to_bytes(4, "big") + payload ) await runner_writer.drain() except WebSocketDisconnect: log.info("Browser WebSocket disconnected session=%s", session_id) except Exception as e: log.exception("ws_to_tcp error: %s", e) async def session_watchdog(): warning_sent = False while True: await asyncio.sleep(1) now = time.monotonic() idle_for = now - last_activity_at session_age = now - started_monotonic heartbeat_age = now - last_client_seen_at idle_state = _session_idle_state(idle_for) if session_age >= SESSION_TIMEOUT_SEC: _metrics["sessions_timed_out"] += 1 log.warning("Session hard timeout session=%s", session_id) await websocket.send_json( { "type": "log", "data": f"\n[系統] 已達最長執行時間（{SESSION_TIMEOUT_SEC // 60} 分鐘），自動結束。\n", } ) await websocket.send_json({"type": "status", "data": "timeout"}) break if heartbeat_age >= HEARTBEAT_TIMEOUT_SEC: _metrics["sessions_heartbeat_timed_out"] += 1 log.warning("Session heartbeat timeout session=%s", session_id) await websocket.send_json( { "type": "log", "data": "\n[系統] 瀏覽器已離線，自動釋放展示名額。\n", } ) await websocket.send_json( {"type": "status", "data": "connection_timeout"} ) break if idle_state == "timeout": _metrics["sessions_idle_timed_out"] += 1 log.info("Session idle timeout session=%s", session_id) await websocket.send_json( { "type": "log", "data": f"\n[系統] 已閒置 {IDLE_TIMEOUT_SEC // 60} 分鐘，自動結束並釋放展示名額。\n", } ) await websocket.send_json( {"type": "status", "data": "idle_timeout"} ) break if idle_state == "warning" and not warning_sent: warning_sent = True await websocket.send_json( { "type": "idle_warning", "remaining_sec": max( 0, int(IDLE_TIMEOUT_SEC - idle_for) ), } ) elif idle_state == "active" and warning_sent: warning_sent = False await websocket.send_json({"type": "idle_resumed"}) try: await websocket.close(code=1000) except Exception: pass tcp_task = asyncio.create_task(tcp_to_ws()) ws_task = asyncio.create_task(ws_to_tcp()) watchdog = asyncio.create_task(session_watchdog()) try: _, pending = await asyncio.wait( {tcp_task, ws_task, watchdog}, return_when=asyncio.FIRST_COMPLETED, ) for task in pending: task.cancel() await asyncio.gather(*pending, return_exceptions=True) finally: for task in (tcp_task, ws_task, watchdog): if not task.done(): task.cancel() await asyncio.gather( tcp_task, ws_task, watchdog, return_exceptions=True ) finally: if process is not None: await _cleanup_process(process) if process_stderr_task is not None and not process_stderr_task.done(): try: await asyncio.wait_for(process_stderr_task, timeout=1.0) except asyncio.TimeoutError: process_stderr_task.cancel() await asyncio.gather(process_stderr_task, return_exceptions=True) if runner_writer is not None: try: runner_writer.close() await runner_writer.wait_closed() except Exception: pass if ipc_server is not None: ipc_server.close() await ipc_server.wait_closed() if session_root is not None: shutil.rmtree(session_root, ignore_errors=True) try: await websocket.close() except Exception: pass async with _sessions_lock: active_sessions.pop(session_id, None) if acquired: runner_semaphore.release() log.info( "Session ended session=%s duration=%.1fs", session_id, time.time() - started_at, ) @app.websocket("/ws/play/{student_token}/{project_name}") async def play_project_ws(websocket: WebSocket, student_token: str, project_name: str): real_class, real_student = decode_class_student(student_token) await play_project_ws_internal(websocket, real_class, real_student, project_name) @app.websocket("/ws/play/{class_name}/{student_name}/{project_name}") async def play_project_ws_old( websocket: WebSocket, class_name: str, student_name: str, project_name: str ): real_class, real_student = decode_class_student(class_name, student_name) await play_project_ws_internal(websocket, real_class, real_student, project_name) # --------------------------------------------------------------------------- # Entrypoint # --------------------------------------------------------------------------- if __name__ == "__main__": import uvicorn parser = argparse.ArgumentParser(description="Telearn Arcade Hub") parser.add_argument("--host", default="127.0.0.1", help="Server host") parser.add_argument("--port", type=int, default=8000, help="Server port") parser.add_argument( "--reload", action="store_true", default=os.environ.get("DEV_RELOAD", "").lower() in ("1", "true", "yes"), help="Enable auto-reload (dev only)", ) args = parser.parse_args() uvicorn.run( "server:app", host=args.host, port=args.port, reload=args.reload, )