"""Minimal private-account server for the posture research tool. Run: python3 app_server.py --host 127.0.0.1 --port 8501 For deployment, place behind HTTPS reverse proxy and set POSTURE_SECRET. """ from __future__ import annotations import argparse import base64 import hashlib import hmac import json import os import secrets import sqlite3 import time from http import HTTPStatus from http.cookies import SimpleCookie from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path ROOT = Path(__file__).parent.resolve() DATA = ROOT / "data" DB = DATA / "accounts.db" SECRET = os.environ.get("POSTURE_SECRET", "replace-this-before-public-deployment").encode() SESSIONS: dict[str, tuple[str, bool, float]] = {} def password_hash(password: str, salt: bytes | None = None) -> str: salt = salt or secrets.token_bytes(16) digest = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 200_000) return base64.b64encode(salt + digest).decode() def verify_password(password: str, stored: str) -> bool: data = base64.b64decode(stored.encode()); salt, expected = data[:16], data[16:] actual = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 200_000) return hmac.compare_digest(actual, expected) def init_db() -> None: DATA.mkdir(exist_ok=True) with sqlite3.connect(DB) as conn: conn.execute("CREATE TABLE IF NOT EXISTS users (username TEXT PRIMARY KEY, password_hash TEXT NOT NULL, is_admin INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL)") if not conn.execute("SELECT 1 FROM users WHERE username='admin'").fetchone(): conn.execute("INSERT INTO users VALUES (?, ?, 1, ?)", ("admin", password_hash("123456"), int(time.time()))) def user_for(handler: SimpleHTTPRequestHandler): cookie = SimpleCookie(handler.headers.get("Cookie")); token = cookie.get("posture_session") if not token or token.value not in SESSIONS: return None username, admin, expires = SESSIONS[token.value] if expires < time.time(): del SESSIONS[token.value]; return None return username, admin class Handler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(ROOT), **kwargs) def log_message(self, format, *args): pass def end_headers(self): if self.path.startswith("/assets/mediapipe/"): self.send_header("Cache-Control", "public, max-age=31536000, immutable") elif self.path.startswith("/assets/"): self.send_header("Cache-Control", "public, max-age=86400") super().end_headers() def json(self, status: int, data: dict, cookie: str | None = None): body = json.dumps(data, ensure_ascii=False).encode() self.send_response(status); self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))); self.send_header("Cache-Control", "no-store") if cookie: self.send_header("Set-Cookie", cookie) self.end_headers(); self.wfile.write(body) def payload(self): length = int(self.headers.get("Content-Length", 0)); return json.loads(self.rfile.read(length) or b"{}") def do_GET(self): # Assessment labels and account data are private server-side records. if self.path == "/data" or self.path.startswith("/data/"): return self.json(404, {"error": "未找到资源"}) if self.path == "/api/me": user = user_for(self) return self.json(200, {"authenticated": bool(user), "username": user[0] if user else None, "admin": bool(user and user[1])}) if self.path == "/api/users": user = user_for(self) if not user or not user[1]: return self.json(403, {"error":"仅管理员可查看账号"}) with sqlite3.connect(DB) as conn: rows = conn.execute("SELECT username,is_admin,created_at FROM users ORDER BY created_at").fetchall() return self.json(200, {"users":[{"username":r[0],"admin":bool(r[1]),"created_at":r[2]} for r in rows]}) return super().do_GET() def do_POST(self): try: data = self.payload() except Exception: return self.json(400, {"error":"无效请求"}) if self.path == "/api/login": username = str(data.get("username", "")).strip(); password = str(data.get("password", "")) with sqlite3.connect(DB) as conn: row = conn.execute("SELECT password_hash,is_admin FROM users WHERE username=?", (username,)).fetchone() if not row or not verify_password(password, row[0]): return self.json(401, {"error":"账号或密码错误"}) token=secrets.token_urlsafe(32); SESSIONS[token]=(username, bool(row[1]), time.time()+8*3600) return self.json(200, {"username":username,"admin":bool(row[1])}, f"posture_session={token}; HttpOnly; SameSite=Lax; Path=/; Max-Age=28800") if self.path == "/api/logout": cookie=SimpleCookie(self.headers.get("Cookie")); token=cookie.get("posture_session") if token: SESSIONS.pop(token.value, None) return self.json(200, {"ok":True}, "posture_session=; HttpOnly; SameSite=Lax; Path=/; Max-Age=0") if self.path == "/api/change-password": user = user_for(self) if not user: return self.json(401, {"error":"请先登录"}) old_password=str(data.get("old_password", "")); new_password=str(data.get("new_password", "")) if len(new_password) < 8: return self.json(400, {"error":"新密码至少 8 位"}) with sqlite3.connect(DB) as conn: row=conn.execute("SELECT password_hash FROM users WHERE username=?", (user[0],)).fetchone() if not row or not verify_password(old_password, row[0]): return self.json(401, {"error":"当前密码错误"}) conn.execute("UPDATE users SET password_hash=? WHERE username=?", (password_hash(new_password), user[0])) return self.json(200, {"ok":True}) if self.path == "/api/users": user=user_for(self) if not user or not user[1]: return self.json(403, {"error":"仅管理员可创建账号"}) username=str(data.get("username", "")).strip(); password=str(data.get("password", "")) if not (3 <= len(username) <= 32 and username.replace("_", "").isalnum() and len(password) >= 6): return self.json(400,{"error":"账号需为 3–32 位字母、数字或下划线；密码至少 6 位"}) try: with sqlite3.connect(DB) as conn: conn.execute("INSERT INTO users VALUES (?, ?, 0, ?)",(username,password_hash(password),int(time.time()))) except sqlite3.IntegrityError: return self.json(409,{"error":"该账号已存在"}) return self.json(201,{"ok":True,"username":username}) return self.json(404,{"error":"未找到接口"}) if __name__ == "__main__": parser=argparse.ArgumentParser(); parser.add_argument("--host",default="127.0.0.1"); parser.add_argument("--port",type=int,default=8501); args=parser.parse_args() init_db(); print(f"Posture server listening on http://{args.host}:{args.port}") ThreadingHTTPServer((args.host,args.port), Handler).serve_forever()