#!/usr/bin/env python3 """本地静态服务器 + 用户认证 + 按用户存档API（SQLite） POST /api/register {user, pass, captcha} → 注册 POST /api/login {user, pass} → {token} GET /api/me (Authorization) → {user} GET /api/save (Authorization) → 存档 POST /api/save (Authorization, body) → 写入存档 POST /api/reset (Authorization) → 删除存档 GET /api/captcha → {id, img(base64)} 简单数字验证码 """ import http.server, socketserver, os, json, sqlite3, time, hashlib, secrets, base64, io, random, struct, zlib BASE = os.path.dirname(os.path.abspath(__file__)) DB = os.path.join(BASE, 'data', 'game_save.db') os.makedirs(os.path.dirname(DB), exist_ok=True) def db(): conn = sqlite3.connect(DB) conn.execute('CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT UNIQUE, pass TEXT, created REAL)') conn.execute('CREATE TABLE IF NOT EXISTS sessions (token TEXT PRIMARY KEY, user_id INTEGER, created REAL)') conn.execute('CREATE TABLE IF NOT EXISTS saves (user_id INTEGER, id INTEGER, data TEXT, ts REAL, PRIMARY KEY(user_id, id))') return conn def hash_pass(name, pwd): return hashlib.sha256(f'{name}:{pwd}'.encode()).hexdigest() def make_token(): return secrets.token_hex(32) def get_user_from_auth(handler): auth = handler.headers.get('Authorization', '') if not auth.startswith('Bearer '): return None token = auth[7:] conn = db() row = conn.execute('SELECT user_id FROM sessions WHERE token=?', (token,)).fetchone() conn.close() return row[0] if row else None # ---------- 简单PNG验证码生成（纯Python，无依赖） ---------- def make_captcha(): """SVG扭曲字符验证码：4位随机字符+噪点+干扰线+旋转""" # 排除易混淆字符 chars = 'ABCDEFGHJKLMNPQRSTUVWXYZ23456789' code = ''.join(random.choices(chars, k=4)) cid = secrets.token_hex(16) CAPTCHAS[cid] = {'answer': code.upper(), 'ts': time.time()} W, H = 160, 50 svg = f'' # SVG → data URI svg_b64 = base64.b64encode(svg.encode()).decode() img_data = f'data:image/svg+xml;base64,{svg_b64}' return cid, img_data CAPTCHAS = {} class Handler(http.server.SimpleHTTPRequestHandler): def _json(self, obj, code=200): body = json.dumps(obj).encode() self.send_response(code) self.send_header('Content-Type', 'application/json') self.send_header('Cache-Control', 'no-store') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) def _read_body(self): n = int(self.headers.get('Content-Length', 0)) return json.loads(self.rfile.read(n)) if n else {} def do_GET(self): path = self.path.split('?')[0] if path == '/api/captcha': cid, img_data = make_captcha() self._json({'id': cid, 'img': img_data}) elif path == '/api/me': uid = get_user_from_auth(self) if not uid: self._json({'error': '未登录'}, 401); return conn = db() row = conn.execute('SELECT name FROM users WHERE id=?', (uid,)).fetchone() conn.close() self._json({'user': row[0] if row else None}) elif path == '/api/save': # sendBeacon 用 query 参数传 token qs = self.path.split('?')[1] if '?' in self.path else '' qtoken = [x.split('=')[1] for x in qs.split('&') if x.startswith('token=')] uid = get_user_from_auth(self) or (db().execute( 'SELECT user_id FROM sessions WHERE token=?', (qtoken[0] if qtoken else '',) ).fetchone() or [None])[0] if not uid: self._json({'error': '未登录'}, 401); return conn = db() row = conn.execute('SELECT data FROM saves WHERE user_id=? AND id=1', (uid,)).fetchone() conn.close() self._json(json.loads(row[0]) if row else None) else: super().do_GET() def do_POST(self): path = self.path.split('?')[0] if path == '/api/register': d = self._read_body() name, pwd, cap_id, cap_ans = d.get('user','').strip(), d.get('pass',''), d.get('captchaId',''), d.get('captchaAns','') if not name or not pwd: self._json({'error': '用户名和密码不能为空'}, 400); return if len(name) < 2 or len(name) > 20: self._json({'error': '用户名需要2~20个字符'}, 400); return if len(pwd) < 4: self._json({'error': '密码至少4位'}, 400); return # 注册无需验证码 conn = db() try: conn.execute('INSERT INTO users (name, pass, created) VALUES (?,?,?)', (name, hash_pass(name, pwd), time.time())) conn.commit() uid = conn.execute('SELECT id FROM users WHERE name=?', (name,)).fetchone()[0] token = make_token() conn.execute('INSERT INTO sessions VALUES (?,?,?)', (token, uid, time.time())) conn.commit() self._json({'ok': True, 'token': token, 'user': name}) except sqlite3.IntegrityError: self._json({'error': '用户名已被注册'}, 409) finally: conn.close() elif path == '/api/login': d = self._read_body() name, pwd = d.get('user','').strip(), d.get('pass','') # 登录也需验证码 cap = CAPTCHAS.pop(d.get('captchaId',''), None) if not cap or time.time() - cap['ts'] > 300: self._json({'error': '验证码已过期，请刷新'}, 400); return if str(cap['answer']).upper() != str(d.get('captchaAns','')).strip().upper(): self._json({'error': '验证码错误'}, 400); return conn = db() row = conn.execute('SELECT id, pass FROM users WHERE name=?', (name,)).fetchone() conn.close() if not row or row[1] != hash_pass(name, pwd): self._json({'error': '用户名或密码错误'}, 401); return token = make_token() conn = db() conn.execute('INSERT INTO sessions VALUES (?,?,?)', (token, row[0], time.time())) conn.commit(); conn.close() self._json({'ok': True, 'token': token, 'user': name}) elif path == '/api/logout': uid = get_user_from_auth(self) if uid: auth = self.headers.get('Authorization', '')[7:] conn = db() conn.execute('DELETE FROM sessions WHERE token=?', (auth,)) conn.commit(); conn.close() self._json({'ok': True}) elif path == '/api/save': # sendBeacon 用 query 参数传 token qs = self.path.split('?')[1] if '?' in self.path else '' qtoken = [x.split('=')[1] for x in qs.split('&') if x.startswith('token=')] uid = get_user_from_auth(self) or (db().execute( 'SELECT user_id FROM sessions WHERE token=?', (qtoken[0] if qtoken else '',) ).fetchone() or [None])[0] if not uid: self._json({'error': '未登录'}, 401); return body = self.rfile.read(int(self.headers.get('Content-Length', 0))).decode() conn = db() conn.execute('INSERT OR REPLACE INTO saves (user_id, id, data, ts) VALUES (?, 1, ?, ?)', (uid, body, time.time())) conn.commit(); conn.close() self._json({'ok': True}) elif path == '/api/reset': uid = get_user_from_auth(self) if not uid: self._json({'error': '未登录'}, 401); return conn = db() conn.execute('DELETE FROM saves WHERE user_id=? AND id=1', (uid,)) conn.commit(); conn.close() self._json({'ok': True}) else: self.send_error(404) def do_OPTIONS(self): self.send_response(204) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET,POST,OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type,Authorization') self.end_headers() def end_headers(self): if not self.path.startswith('/api/'): self.send_header('Cache-Control', 'no-store, must-revalidate') super().end_headers() os.chdir(BASE) socketserver.TCPServer.allow_reuse_address = True with socketserver.ThreadingTCPServer(('0.0.0.0', 8791), Handler) as httpd: print('serving at http://127.0.0.1:8791 (auth + user saves + captcha)') httpd.serve_forever()