# -*- coding: utf-8 -*- """ ResLib License 申请系统后端 (FastAPI) 存储：data/requests/.txt（DokuWiki 式纯文本） 公开：申请 / 状态查询 / 下载 license 管理：密码登录 -> 审批 -> 生成 res_center.lic """ import os import re import secrets import time from datetime import datetime from pathlib import Path from fastapi import FastAPI, Form, HTTPException, Request, Response from fastapi.responses import FileResponse, PlainTextResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field from license import generate_lic, normalize_mac BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" REQ_DIR = DATA_DIR / "requests" LIC_DIR = DATA_DIR / "licenses" PKG_DIR = DATA_DIR / "packages" STATIC_DIR = BASE_DIR / "static" def load_admin_password() -> str: env = os.environ.get("RESLIB_ADMIN_PASSWORD", "").strip() if env: return env env_file = BASE_DIR / ".env" if env_file.is_file(): for line in env_file.read_text(encoding="utf-8").splitlines(): line = line.strip() if line and not line.startswith("#") and "=" in line: k, _, v = line.partition("=") if k.strip() == "RESLIB_ADMIN_PASSWORD": return v.strip() return "reslib-admin" ADMIN_PASSWORD = load_admin_password() SESSION_FILE = DATA_DIR / "session.txt" STATUS_PENDING = "pending" STATUS_APPROVED = "approved" STATUS_DENIED = "denied" for d in (REQ_DIR, LIC_DIR, PKG_DIR): d.mkdir(parents=True, exist_ok=True) app = FastAPI(title="ResLib License System", docs_url=None, redoc_url=None) # --------------------------------------------------------------------------- # 工具 # --------------------------------------------------------------------------- def new_request_id() -> str: stamp = datetime.now().strftime("%Y%m%d-%H%M%S") return f"{stamp}-{secrets.token_hex(3).upper()}" def request_path(req_id: str) -> Path: if not re.fullmatch(r"\d{8}-\d{6}-[0-9A-F]{6}", req_id): raise HTTPException(400, "无效的申请编号") return REQ_DIR / f"{req_id}.txt" def read_request(req_id: str) -> dict: path = request_path(req_id) if not path.is_file(): raise HTTPException(404, "申请不存在") data = {} for line in path.read_text(encoding="utf-8").splitlines(): if ":" in line: k, _, v = line.partition(":") data[k.strip()] = v.strip() return data def write_request(data: dict): path = request_path(data["id"]) content = "\n".join(f"{k}: {data.get(k, '')}" for k in ( "id", "created", "status", "mac", "company", "contact", "email", "creo", "seats", "notes", "reason", )) path.write_text(content + "\n", encoding="utf-8") def save_lic(req_id: str, mac: str) -> Path: lic_path = LIC_DIR / f"{req_id}.lic" lic_path.write_text(generate_lic(mac) + "\n", encoding="ascii") return lic_path def new_session_token() -> str: token = secrets.token_hex(24) SESSION_FILE.write_text(token, encoding="utf-8") return token def get_session_token() -> str: if SESSION_FILE.is_file(): return SESSION_FILE.read_text(encoding="utf-8").strip() return "" def is_admin(request: Request) -> bool: cookie = request.cookies.get("reslib_admin") return bool(cookie) and secrets.compare_digest(cookie, get_session_token()) def require_admin(request: Request): if not is_admin(request): raise HTTPException(401, "未登录或会话已失效") # --------------------------------------------------------------------------- # 公开 API # --------------------------------------------------------------------------- class ApplyForm(BaseModel): mac: str = Field(..., min_length=1) company: str = Field(..., min_length=1) contact: str = Field(..., min_length=1) email: str = Field(..., min_length=1) creo: str = "" seats: str = "1" notes: str = "" @app.post("/api/apply") def api_apply(form: ApplyForm): try: mac = normalize_mac(form.mac) except ValueError as e: raise HTTPException(400, str(e)) if "@" not in form.email or "." not in form.email.split("@")[-1]: raise HTTPException(400, "邮箱格式不正确") req_id = new_request_id() data = { "id": req_id, "created": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "status": STATUS_PENDING, "mac": mac, "company": form.company.strip(), "contact": form.contact.strip(), "email": form.email.strip(), "creo": form.creo.strip(), "seats": form.seats.strip(), "notes": form.notes.strip(), "reason": "", } write_request(data) return {"id": req_id, "status": STATUS_PENDING} @app.get("/api/status/{req_id}") def api_status(req_id: str): data = read_request(req_id) return { "id": data.get("id"), "status": data.get("status"), "created": data.get("created"), "reason": data.get("reason", ""), } @app.get("/api/license/{req_id}") def api_license(req_id: str): data = read_request(req_id) if data.get("status") != STATUS_APPROVED: raise HTTPException(403, "申请尚未通过审批") lic_path = LIC_DIR / f"{req_id}.lic" if not lic_path.is_file(): save_lic(req_id, data["mac"]) lic_path = LIC_DIR / f"{req_id}.lic" return FileResponse( lic_path, filename="res_center.lic", media_type="application/octet-stream", ) # --------------------------------------------------------------------------- # 资源包（仅审核通过的用户可获取） # --------------------------------------------------------------------------- @app.get("/api/package/{req_id}") def api_package_list(req_id: str): data = read_request(req_id) if data.get("status") != STATUS_APPROVED: raise HTTPException(403, "资源包仅对审核通过的用户提供，请先完成 License 审批") files = sorted(p.name for p in PKG_DIR.glob("*") if p.is_file()) if not files: return {"files": []} return {"files": files} @app.get("/api/package/download/{req_id}/{filename}") def api_package_download(req_id: str, filename: str): data = read_request(req_id) if data.get("status") != STATUS_APPROVED: raise HTTPException(403, "资源包仅对审核通过的用户提供") name = os.path.basename(filename) path = PKG_DIR / name if not path.is_file(): raise HTTPException(404, "资源包不存在") return FileResponse(path, filename=name, media_type="application/octet-stream") # --------------------------------------------------------------------------- # 管理 API # --------------------------------------------------------------------------- @app.post("/api/admin/login") def admin_login(response: Response, password: str = Form(...)): if not secrets.compare_digest(password, ADMIN_PASSWORD): raise HTTPException(401, "密码错误") token = new_session_token() response.set_cookie("reslib_admin", token, httponly=True, samesite="lax", max_age=86400 * 7) return {"ok": True} @app.post("/api/admin/logout") def admin_logout(request: Request, response: Response): response.delete_cookie("reslib_admin") return {"ok": True} @app.get("/api/admin/list") def admin_list(request: Request): require_admin(request) items = [] for path in sorted(REQ_DIR.glob("*.txt"), reverse=True): items.append(read_request(path.stem)) return items @app.post("/api/admin/approve/{req_id}") def admin_approve(req_id: str, request: Request): require_admin(request) data = read_request(req_id) if data.get("status") == STATUS_APPROVED: return {"ok": True, "id": req_id} save_lic(req_id, data["mac"]) data["status"] = STATUS_APPROVED data["reason"] = "" write_request(data) return {"ok": True, "id": req_id} @app.post("/api/admin/deny/{req_id}") def admin_deny(req_id: str, request: Request, reason: str = Form("")): require_admin(request) data = read_request(req_id) data["status"] = STATUS_DENIED data["reason"] = reason.strip()[:200] write_request(data) return {"ok": True, "id": req_id} # --------------------------------------------------------------------------- # 页面与静态资源 # --------------------------------------------------------------------------- @app.get("/") def index(): return FileResponse(STATIC_DIR / "home.html") @app.get("/about/") def about_page(): return FileResponse(STATIC_DIR / "about.html") @app.get("/neural/") def neural_page(): return FileResponse(STATIC_DIR / "neural.html") @app.get("/links.html") def links_page(): return FileResponse(STATIC_DIR / "links.html") @app.get("/reslib/") def reslib_home(): return FileResponse(STATIC_DIR / "index.html") @app.get("/reslib/apply") def reslib_apply(): return FileResponse(STATIC_DIR / "apply.html") @app.get("/reslib/status") def reslib_status(): return FileResponse(STATIC_DIR / "status.html") @app.get("/reslib/package") def reslib_package(): return FileResponse(STATIC_DIR / "package.html") @app.get("/admin") def admin_page(request: Request): if not is_admin(request): return RedirectResponse("/admin/login", status_code=302) return FileResponse(STATIC_DIR / "admin.html") @app.get("/admin/login") def admin_login_page(): return FileResponse(STATIC_DIR / "admin_login.html") @app.get("/admin/logout") def admin_logout_page(request: Request): return RedirectResponse("/admin/login", status_code=302) app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") app.mount("/blog", StaticFiles(directory=STATIC_DIR / "blog", html=True, check_dir=False), name="blog")