#!/usr/bin/env python3 # -*- coding: utf-8 -*- """xinquandashboard —— 密码保护的图文页面应用（零依赖，标准库 only）。路由（Caddy 通过 handle_path 剥掉 /xinquandashboard 前缀后转发至此）： GET / 单页 UI（HTML+JS 内联） GET /api/pages 页面列表 POST /api/pages 新建页面 body: {"title","content"} GET /api/pages/ 单页详情 PUT /api/pages/ 编辑页面 body: {"title","content"} DELETE /api/pages/ 删除页面 POST /api/upload 上传图片 body: {"data":"","name":"a.png","mime":"image/png"} GET /media/ 图片静态读取持久化：/data/pages.json（页面元数据）+ /data/uploads/（图片文件）。鉴权：由 Caddy basic_auth 在网关卡负责，本应用不鉴权。 """ import base64 import hashlib import hmac import json import os import re import threading import time import uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer DATA_DIR = os.environ.get("DATA_DIR", "/data") PAGES_FILE = os.path.join(DATA_DIR, "pages.json") UPLOADS_DIR = os.path.join(DATA_DIR, "uploads") PORT = int(os.environ.get("PORT", "8000")) _lock = threading.Lock() ALLOWED_IMG_EXT = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".gif": "image/gif", ".webp": "image/webp", ".svg": "image/svg+xml"} # ---- 应用内表单登录（替代 Caddy basic_auth，兼容微信内置浏览器）---- AUTH_USER = os.environ.get("XQ_USER", "xqzk") AUTH_PASS = os.environ.get("XQ_PASS", "66668888") AUTH_SECRET = os.environ.get("XQ_SECRET", "xinquan-session-secret-change-me") SESSION_DAYS = 7 RE_AUTH = re.compile(r"__AUTH__") def _sign_token(user, exp): payload = f"{user}|{exp}".encode() sig = hmac.new(AUTH_SECRET.encode(), payload, hashlib.sha256).hexdigest() return base64.urlsafe_b64encode(payload + b"|" + sig.encode()).decode() def _verify_token(token): try: raw = base64.urlsafe_b64decode(token.encode()).decode() user, exp, sig = raw.split("|") if int(exp) < int(time.time()): return None expect = hmac.new(AUTH_SECRET.encode(), f"{user}|{exp}".encode(), hashlib.sha256).hexdigest() return user if hmac.compare_digest(expect, sig) else None except Exception: return None def _ensure_dirs(): os.makedirs(DATA_DIR, exist_ok=True) os.makedirs(UPLOADS_DIR, exist_ok=True) if not os.path.exists(PAGES_FILE): _write_pages({}) def _read_pages(): if not os.path.exists(PAGES_FILE): return {} try: with open(PAGES_FILE, "r", encoding="utf-8") as f: data = json.load(f) return data if isinstance(data, dict) else {} except Exception: return {} def _write_pages(pages): tmp = PAGES_FILE + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(pages, f, ensure_ascii=False, indent=2) os.replace(tmp, PAGES_FILE) def _summary(content, n=160): text = re.sub(r"[\n\r]+", " ", content or "") text = re.sub(r"[\s　]+", " ", text).strip() return text[:n] + ("…" if len(text) > n else "") # ------------------------- 页面与 UI ------------------------- BOOTSTRAP = "https://cdn.jsdelivr.net/npm/marked@12/marked.min.js" INDEX_HTML = r""" Xinquan Dashboard

Xinquan Dashboard

""" # ------------------------- 服务器 ------------------------- _PREFIXES = ("/xinquandashboard",) # 即使未剥前缀也能自适应 class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" # ---- 基础 ---- def _send(self, code, body: bytes, ct="application/json; charset=utf-8", extra=None): self.send_response(code) self.send_header("Content-Type", ct) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") if extra: for k, v in extra.items(): self.send_header(k, v) self.end_headers() self.wfile.write(body) def _json(self, obj, code=200): self._send(code, json.dumps(obj, ensure_ascii=False).encode("utf-8")) def _read_body(self): n = int(self.headers.get("Content-Length") or 0) return self.rfile.read(n) if n else b"" def _read_json(self): raw = self._read_body() try: return json.loads(raw.decode("utf-8")) except Exception: return None # ---- 路由 ---- def _split(self): path = self.path.split("?", 1)[0] for pre in _PREFIXES: if path.startswith(pre): path = path[len(pre):] or "/" break if not path.startswith("/"): path = "/" + path return path # ---- 会话 ---- def _user(self): cookies = self.headers.get("Cookie") or "" tok = None for pair in cookies.split(";"): if not pair: continue kv = pair.strip().split("=", 1) if len(kv) == 2 and kv[0] == "xq_auth": tok = kv[1] return _verify_token(tok) if tok else None def _secure_cookie(self): return (self.headers.get("X-Forwarded-Proto", "") == "https") def _unauthed(self): """返回 True 表示未登录（调用方应返回 401）。""" return True def _login_ok(self, user): # 生成 7 天会话 exp = int(time.time()) + SESSION_DAYS * 86400 tok = _sign_token(user, exp) cookie = f"xq_auth={tok}; Path=/; HttpOnly; SameSite=Lax; Max-Age={SESSION_DAYS*86400}" if self._secure_cookie(): cookie += "; Secure" self._send(200, b'{"ok":true}', extra={"Set-Cookie": cookie}) def _login_fail(self): self._json({"error": "用户名或密码错误"}, 401) def _auth_denied(self): if self.path.startswith("/media"): return self._send(401, b"unauthorized", "text/plain; charset=utf-8") return self._json({"error": "unauthorized"}, 401) # ---- 路由 ---- def _split(self): path = self.path.split("?", 1)[0] for pre in _PREFIXES: if path.startswith(pre): path = path[len(pre):] or "/" break if not path.startswith("/"): path = "/" + path return path def do_GET(self): path = self._split() if path == "/health": return self._json({"ok": True}) user = self._user() if path.startswith("/api/"): if not user: return self._auth_denied() if path.startswith("/api/pages/"): pid = path[len("/api/pages/"):] return self._get_page(pid) if pid else self._json({"error": "missing id"}, 400) if path == "/api/pages": return self._list_pages() return self._json({"error": "not found"}, 404) if path.startswith("/media/"): if not user: return self._auth_denied() return self._media(path[len("/media/"):]) # 其余(/, /login, SPA 内部路由)：总是渲染外壳；未登录则显示登录表单 return self._send(200, self._index_html(bool(user)), "text/html; charset=utf-8") def _index_html(self, authed=True): flag = "true" if authed else "false" html = INDEX_HTML.replace("__MARKED__", BOOTSTRAP) # 注入登录态，避免未授权时闪出内容 result = RE_AUTH.sub(flag, html, count=1) return result.encode("utf-8") def do_POST(self): path = self._split() if path == "/login": data = self._read_json() if data and data.get("username") == AUTH_USER and \ data.get("password") == AUTH_PASS: return self._login_ok(AUTH_USER) return self._login_fail() if path == "/logout": return self._logout() if not self._user(): return self._auth_denied() if path == "/api/pages": return self._create_page() if path == "/api/upload": return self._upload() self._json({"error": "not found"}, 404) def _logout(self): cookie = "xq_auth=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0" self._send(200, b'{"ok":true}', extra={"Set-Cookie": cookie}) def do_PUT(self): if not self._user(): return self._auth_denied() path = self._split() if path.startswith("/api/pages/"): pid = path[len("/api/pages/"):] if pid: return self._update_page(pid) self._json({"error": "not found"}, 404) def do_DELETE(self): if not self._user(): return self._auth_denied() path = self._split() if path.startswith("/api/pages/"): pid = path[len("/api/pages/"):] if pid: return self._delete_page(pid) self._json({"error": "not found"}, 404) def log_message(self, *a): # 静音默认日志 pass # ---- 页面 API ---- def _list_pages(self): with _lock: pages = _read_pages() out = [{"id": pid, "title": p.get("title", ""), "summary": _summary(p.get("content", "")), "updated": p.get("updated")} for pid, p in pages.items()] out.sort(key=lambda x: x["updated"] or "", reverse=True) self._json(out) def _get_page(self, pid): with _lock: pages = _read_pages() p = pages.get(pid) if p is None: return self._json({"error": "page not found"}, 404) self._json({"id": pid, **p}) def _create_page(self): data = self._read_json() if not data or not data.get("title", "").strip(): return self._json({"error": "title required"}, 400) with _lock: pages = _read_pages() pid = uuid.uuid4().hex[:12] now = time.strftime("%Y-%m-%dT%H:%M:%S") pages[pid] = {"title": data["title"].strip(), "content": data.get("content", ""), "created": now, "updated": now} _write_pages(pages) self._json({"id": pid}, 201) def _update_page(self, pid): data = self._read_json() if not data: return self._json({"error": "bad body"}, 400) with _lock: pages = _read_pages() if pid not in pages: return self._json({"error": "page not found"}, 404) if data.get("title", "").strip(): pages[pid]["title"] = data["title"].strip() if "content" in data: pages[pid]["content"] = data["content"] or "" pages[pid]["updated"] = time.strftime("%Y-%m-%dT%H:%M:%S") _write_pages(pages) self._json({"id": pid}) def _delete_page(self, pid): with _lock: pages = _read_pages() if pid not in pages: return self._json({"error": "page not found"}, 404) del pages[pid] _write_pages(pages) self._json({"ok": True}) # ---- 图片 ---- def _upload(self): data = self._read_json() if not data or not data.get("data"): return self._json({"error": "no image data"}, 400) name = (data.get("name") or "img").lower() ext = os.path.splitext(name)[1].replace(".", "") if ext: full = "." + ext if full not in ALLOWED_IMG_EXT: # 依据 mime 兜底 mime = data.get("mime", "") full = next((e for e, m in ALLOWED_IMG_EXT.items() if m == mime), ".png") else: mime = data.get("mime", "") full = next((e for e, m in ALLOWED_IMG_EXT.items() if m == mime), ".png") try: raw = base64.b64decode(data["data"]) except Exception: return self._json({"error": "bad base64"}, 400) if not raw or len(raw) > 20 * 1024 * 1024: return self._json({"error": "image empty or too large (>20MB)"}, 413) fname = uuid.uuid4().hex + full with open(os.path.join(UPLOADS_DIR, fname), "wb") as f: f.write(raw) self._json({"url": f"media/{fname}", "name": name}) def _media(self, name): if "/" in name or "\\" in name or not name: return self._json({"error": "bad media"}, 400) p = os.path.join(UPLOADS_DIR, name) if not os.path.isfile(p): return self._send(404, b"not found", "text/plain") ext = os.path.splitext(name)[1].lower() ct = ALLOWED_IMG_EXT.get(ext, "application/octet-stream") with open(p, "rb") as f: body = f.read() self._send(200, body, ct, {"Cache-Control": "public, max-age=86400"}) def main(): _ensure_dirs() srv = ThreadingHTTPServer(("0.0.0.0", PORT), Handler) print(f"xinquandashboard running on 0.0.0.0:{PORT}", flush=True) srv.serve_forever() if __name__ == "__main__": main()