Spaces:
Running
Running
"""
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 "<empty>",
)
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,
)
"""
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)
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"