#!/usr/bin/env python3 """ Linux 学习平台 Server（知识导向版） - 提供课程结构、练习判题、沙盒执行 - 课程模型：module -> lesson -> exercise """ from __future__ import annotations import hashlib import http.server import json import os import urllib.parse from typing import Any from sandbox import LinuxSandbox USERS = { "admin": hashlib.sha256(b"safe_linux_2026").hexdigest(), } TASKS_FILE = os.path.join(os.path.dirname(__file__), "COURSE_TASKS.json") HTML_FILE = os.path.join(os.path.dirname(__file__), "index.html") SANDBOX = LinuxSandbox() def load_course() -> dict[str, Any]: try: with open(TASKS_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception as e: print(f"Error loading course: {e}") return {"meta": {}, "modules": []} COURSE = load_course() def flatten_exercises() -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for module in COURSE.get("modules", []): for lesson in module.get("lessons", []): for exercise in lesson.get("exercises", []): item = dict(exercise) item["module_id"] = module.get("id") item["module_title"] = module.get("title") item["lesson_id"] = lesson.get("id") item["lesson_title"] = lesson.get("title") item["lesson_goal"] = lesson.get("goal") item["lesson_command"] = lesson.get("command") rows.append(item) return rows def find_exercise(ex_id: str) -> dict[str, Any] | None: for item in flatten_exercises(): if item.get("id") == ex_id: return item return None class LinuxLearningHandler(http.server.BaseHTTPRequestHandler): def log_message(self, format, *args): pass def send_json(self, data: Any, status=200): raw = json.dumps(data, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) def send_file(self, path: str, content_type: str): with open(path, "rb") as f: body = f.read() self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def check_auth(self, auth_header: str, token: str) -> bool: if self.client_address[0] == "127.0.0.1": return True if token == "safe_linux_2026": return True if auth_header.startswith("Bearer ") and auth_header[7:] == "safe_linux_2026": return True return False def do_GET(self): parsed = urllib.parse.urlparse(self.path) path = parsed.path if path not in ["/", "/api/login", "/api/logout", "/api/course", "/api/health"]: auth_header = self.headers.get("Authorization", "") token = self.headers.get("X-Token", "") if not self.check_auth(auth_header, token) and "xiaoxiaoluohao.indevs.in" in self.headers.get("Host", ""): self.send_json({"error": "Authentication required"}, 401) return if path == "/": self.send_file(HTML_FILE, "text/html; charset=utf-8") return if path == "/api/health": self.send_json({"ok": True, "cwd": SANDBOX.cwd, "user": SANDBOX.user, "modules": len(COURSE.get("modules", []))}) return if path == "/api/course": self.send_json(COURSE) return if path == "/api/run": query = urllib.parse.parse_qs(parsed.query) cmd = query.get("cmd", [""])[0] if not cmd: self.send_json({"error": "No command provided"}, 400) return self.send_json(SANDBOX.execute(cmd)) return if path == "/api/check": query = urllib.parse.parse_qs(parsed.query) ex_id = query.get("exercise_id", [""])[0] cmd = query.get("last_cmd", [""])[0] output = query.get("output", [""])[0] if not ex_id: self.send_json({"error": "exercise_id required"}, 400) return exercise = find_exercise(ex_id) if not exercise: self.send_json({"error": "Exercise not found"}, 404) return state = { "cmd": cmd, "output": output, "cwd": SANDBOX.cwd, "exists": SANDBOX.exists, "is_executable": SANDBOX.is_executable, } success, reason = self.evaluate_exercise(exercise, state) self.send_json({ "exercise_id": ex_id, "success": success, "message": exercise.get("success_msg", "✅ 练习通过") if success else reason, "hint": exercise.get("hint"), "lesson_title": exercise.get("lesson_title"), "module_title": exercise.get("module_title"), "next_suggestion": self.build_next_suggestion(ex_id), }) return self.send_response(404) self.end_headers() self.wfile.write(b"Not Found") def do_POST(self): parsed = urllib.parse.urlparse(self.path) path = parsed.path content_length = int(self.headers.get("Content-Length", 0)) raw = self.rfile.read(content_length).decode() if content_length else "{}" if path == "/api/login": try: data = json.loads(raw) except Exception: self.send_json({"success": False, "error": "Invalid JSON"}, 400) return username = data.get("username", "") password = data.get("password", "") if username in USERS and hashlib.sha256(password.encode()).hexdigest() == USERS[username]: self.send_json({"success": True, "token": "safe_linux_2026", "message": "✅ 登录成功！"}) return self.send_json({"success": False, "error": "❌ 用户名或密码错误"}, 401) return if path == "/api/logout": self.send_json({"success": True, "message": "👋 已退出登录"}) return if path == "/api/reset": SANDBOX.reset() self.send_json({"success": True, "message": "♻️ 沙盒环境已重置"}) return self.send_response(404) self.end_headers() self.wfile.write(b"Not Found") def evaluate_exercise(self, exercise: dict[str, Any], state: dict[str, Any]) -> tuple[bool, str]: ex_type = exercise.get("type") if ex_type in {"understanding", "scenario"}: return False, "📝 这是理解类练习，请先阅读讲解并思考答案。" cmd = state["cmd"].strip() if exercise.get("solution"): for sol in exercise["solution"]: if cmd == sol.strip(): return True, "" success_test = exercise.get("success_test") if not success_test: return False, "❌ 暂未命中练习要求" try: ok = bool(eval(success_test, {"__builtins__": {}}, state)) return ok, "❌ 结果还没达到练习要求，再试一次" except Exception: return False, "❌ 当前命令没有通过判定，建议对照示例重新尝试" def build_next_suggestion(self, current_ex_id: str) -> str | None: rows = flatten_exercises() for i, item in enumerate(rows): if item.get("id") == current_ex_id and i + 1 < len(rows): nxt = rows[i + 1] return f"继续下一练：{nxt.get('title') or nxt.get('question') or nxt.get('id')}" return None if __name__ == "__main__": port = 8084 print(f"🐧 Linux 学习平台启动中... http://127.0.0.1:{port}") print("📚 线上地址: https://linux.xiaoxiaoluohao.indevs.in") http.server.ThreadingHTTPServer(("127.0.0.1", port), LinuxLearningHandler).serve_forever()