diff --git a/.env.example b/.env.example index b03ed4e..c4758ab 100644 --- a/.env.example +++ b/.env.example @@ -30,3 +30,8 @@ SCENE_TTL_SECONDS=300 SESSION_TTL_HOURS=24 # 时间授权 usage_logs 节流窗口(秒) USAGE_LOG_THROTTLE_SECONDS=60 + +# 管理后台 /admin 的 HTTP Basic Auth 凭据 +# ADMIN_PASSWORD 留空时 /admin 返回 503(不鉴权放行),务必填一个长随机串 +ADMIN_USER=admin +ADMIN_PASSWORD= diff --git a/CODEBUDDY.md b/CODEBUDDY.md index 1c885af..05f0543 100644 --- a/CODEBUDDY.md +++ b/CODEBUDDY.md @@ -4,7 +4,7 @@ This file provides guidance to CodeBuddy Code when working with code in this rep ## 项目概述 -微信公众号扫码授权服务,基于 FastAPI,为 Windows MFC 桌面程序提供微信扫码授权与使用扣减。已实现微信服务器验证与消息/事件接收(`/wechat`)、扫码授权(`/auth`)、使用扣减(`/usage`)以及首次关注赠送 7 天免费授权。MySQL 已接入(aiomysql 连接池,5 张表);Redis 仅在配置层声明,尚未使用。 +微信公众号扫码授权服务,基于 FastAPI,为 Windows MFC 桌面程序提供微信扫码授权与使用扣减。已实现微信服务器验证与消息/事件接收(`/wechat`)、扫码授权(`/auth`)、使用扣减(`/usage`)、只读管理后台(`/admin`)以及首次关注赠送 7 天免费授权。MySQL 已接入(aiomysql 连接池,5 张表);Redis 仅在配置层声明,尚未使用。 ## 常用命令 @@ -38,6 +38,7 @@ wechat_api.py # 微信开放接口:access_token 内存缓存 + 临时二维 wechat.py # FastAPI app 本体:lifespan + /wechat 路由 + 签名校验 + XML 解析/构造 auth.py # /auth/* 路由 + 扫码授权业务逻辑(含 pending 激活) usage.py # /usage/consume 路由:会话校验与使用扣减 +admin.py # /admin 只读管理后台:Basic Auth + 服务端渲染 HTML run_local.py # 开发启动入口(reload=True) run_server.py # 生产启动入口(reload=False, workers=1) sql/schema.sql # 建表脚本(5 张表,含索引与外键) @@ -45,7 +46,7 @@ sql/schema.sql # 建表脚本(5 张表,含索引与外键) - **配置**:所有敏感值经 `config.py` 从 `.env` 读取,`.env` 已被 `.gitignore` 排除。`.env.example` 是字段模板。新增配置项需同时更新这两处。 - **应用入口**:两个启动脚本均以 `"wechat:app"` 字符串形式加载 `wechat.py` 中的 `app`,因此模块名/对象名不可随意重命名。`lifespan` 在启动时初始化 MySQL 连接池——**MySQL 不可达或库表不存在时应用会直接启动失败**。 -- **路由挂载**:`wechat.py` 通过 `include_router` 挂载 `auth.router` 与 `usage.router`。新增 MFC 侧接口应新建独立模块的 router,而不是塞进 `wechat.py`。 +- **路由挂载**:`wechat.py` 通过 `include_router` 挂载 `auth.router`、`usage.router` 与 `admin.router`。新增 MFC 侧接口应新建独立模块的 router,而不是塞进 `wechat.py`。 - **微信交互协议**: - 所有请求先经 `verify_signature()`(token+timestamp+nonce 字典序拼接后 SHA1 比对)校验,失败返回 403。 - GET 校验通过后原样返回 `echostr`。 @@ -63,6 +64,17 @@ sql/schema.sql # 建表脚本(5 张表,含索引与外键) - **usage_logs**:积分授权每次调用都写(计费凭证);时间授权按 `(user_id, device_id)` 在 `USAGE_LOG_THROTTLE_SECONDS`(默认 60 秒)窗口内节流。 - **失败语义**:`/usage/consume` 一律返回 HTTP 200,用 `{"ok": false, "reason": "expired | exhausted | invalid_token"}` 表达失败。 +## 管理后台(/admin) + +只读单页,用于查看用户与授权现状:概览统计、用户+当前授权、最近使用记录、会话令牌、扫码场景。 + +- **鉴权**:HTTP Basic Auth,凭据取自 `.env` 的 `ADMIN_USER` / `ADMIN_PASSWORD`,用 `secrets.compare_digest` 做定时安全比较。 +- **`ADMIN_PASSWORD` 为空时 `/admin` 返回 503 而非放行**——不要把它当成可选项,否则等于把全库用户数据公开。 +- **只提供 GET**,没有任何写操作(改授权/加积分/封号一律不做)。 +- **不引模板引擎**:HTML 由 f-string 拼装,所有入库字段经 `html.escape()`;CSS 内联,不依赖任何 CDN(服务器出网不可靠)。 +- 页脚会显示数据库名;页头显示数据生成时间。不做自动刷新。 +- `admin.py` 被 `wechat.py` import,因此**不要在 `admin.py` 里反向 import `wechat`**(循环依赖)。 + ## 数据库 5 张表(定义见 `sql/schema.sql`):`users`、`authorizations`(授权,type 分 time/points)、`auth_scenes`(扫码场景,300 秒一次性)、`usage_logs`(使用日志)、`sessions`(会话令牌)。 diff --git a/admin.py b/admin.py new file mode 100644 index 0000000..ab02d45 --- /dev/null +++ b/admin.py @@ -0,0 +1,382 @@ +""" +授权管理后台(/admin) + +单页、只读、服务端渲染:概览统计 + 用户与当前授权 + 最近使用记录 + 会话与扫码场景。 + +鉴权:HTTP Basic Auth,凭据来自 .env 的 ADMIN_USER / ADMIN_PASSWORD。 + ADMIN_PASSWORD 为空时整体返回 503(而不是放行)——避免「忘了配密码 = 谁都能进」。 + 只提供 GET,不提供任何写操作。 + +刻意不引入 Jinja2 / Bootstrap / 任何 CDN 资源: + - 服务器出网不可靠,页面不能依赖外部资源; + - 少一个依赖就少一份部署风险。 +所有来自数据库的文本都经 html.escape(),避免昵称之类的字段注入 HTML。 + +注意:本模块被 wechat.py import 并挂载,反向 import wechat 会造成循环依赖。 +""" + +import html +import logging +import secrets +from datetime import datetime + +import aiomysql +from fastapi import APIRouter, Depends, HTTPException, status +from fastapi.responses import HTMLResponse +from fastapi.security import HTTPBasic, HTTPBasicCredentials + +import db +from config import ADMIN_PASSWORD, ADMIN_USER, MYSQL_DB + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/admin", tags=["admin"]) + +# realm 必须是 ASCII:Starlette 按 latin-1 编码响应头,中文会直接抛异常 +_basic = HTTPBasic(realm="CadOcr Admin") + + +def require_admin(credentials: HTTPBasicCredentials = Depends(_basic)) -> None: + """校验 Basic Auth;不匹配一律 401 并带 WWW-Authenticate(浏览器才会弹密码框)""" + if not ADMIN_PASSWORD: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="管理后台未启用:请在 .env 里配置 ADMIN_PASSWORD", + ) + + # 比字节而不是比字符串:compare_digest 对含非 ASCII 的 str 会直接抛 TypeError + user_ok = secrets.compare_digest( + credentials.username.encode("utf-8"), ADMIN_USER.encode("utf-8") + ) + pass_ok = secrets.compare_digest( + credentials.password.encode("utf-8"), ADMIN_PASSWORD.encode("utf-8") + ) + if not (user_ok and pass_ok): + # 只记用户名,绝不记密码 + logger.warning("管理后台鉴权失败 user=%s", credentials.username) + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Unauthorized", + headers={"WWW-Authenticate": 'Basic realm="CadOcr Admin"'}, + ) + + +# --------------------------------------------------------------------------- +# 查询 +# --------------------------------------------------------------------------- + +OVERVIEW_SQL = """ +SELECT + (SELECT COUNT(*) FROM users) AS users_total, + (SELECT COUNT(*) FROM users WHERE has_claimed_free = 1) AS users_free, + (SELECT COUNT(*) FROM users WHERE created_at >= CURDATE()) AS users_today, + (SELECT COUNT(*) FROM authorizations WHERE status = 'active') AS auth_active, + (SELECT COUNT(*) FROM authorizations WHERE status = 'pending') AS auth_pending, + (SELECT COUNT(*) FROM authorizations WHERE status = 'expired') AS auth_expired, + (SELECT COUNT(*) FROM authorizations WHERE status = 'exhausted') AS auth_exhausted, + (SELECT COUNT(*) FROM authorizations WHERE type = 'time') AS auth_time, + (SELECT COUNT(*) FROM authorizations WHERE type = 'points') AS auth_points, + (SELECT COUNT(*) FROM usage_logs) AS usage_total, + (SELECT COUNT(*) FROM usage_logs WHERE used_at >= CURDATE()) AS usage_today, + (SELECT COUNT(*) FROM sessions WHERE expires_at > NOW()) AS sessions_valid +""" + +# 互斥原则保证同一用户最多一条 active,所以这个 LEFT JOIN 不会让用户行翻倍 +USERS_SQL = """ +SELECT + u.id, u.openid, u.nickname, u.created_at, u.last_seen_at, u.has_claimed_free, + a.type, a.source, a.end_at, a.remaining_points, a.total_points, + TIMESTAMPDIFF(SECOND, NOW(), a.end_at) AS remain_seconds, + (SELECT COUNT(*) FROM authorizations x WHERE x.user_id = u.id) AS auth_count +FROM users u +LEFT JOIN authorizations a ON a.user_id = u.id AND a.status = 'active' +ORDER BY u.id DESC +LIMIT 200 +""" + +USAGE_SQL = """ +SELECT l.id, l.user_id, u.openid, l.device_id, l.cost_type, l.cost_points, l.used_at +FROM usage_logs l +LEFT JOIN users u ON u.id = l.user_id +ORDER BY l.id DESC +LIMIT 50 +""" + +SESSIONS_SQL = """ +SELECT s.id, s.user_id, s.token, u.openid, s.device_id, s.created_at, s.expires_at, + (s.expires_at > NOW()) AS still_valid +FROM sessions s +LEFT JOIN users u ON u.id = s.user_id +ORDER BY s.id DESC +LIMIT 50 +""" + +SCENES_SQL = """ +SELECT id, scene_str, device_id, status, user_id, created_at, expires_at, authorized_at +FROM auth_scenes +ORDER BY id DESC +LIMIT 50 +""" + + +# --------------------------------------------------------------------------- +# 展示辅助 +# --------------------------------------------------------------------------- + +def _e(value) -> str: + """转义后用于 HTML 文本位置;None 显示为占位符""" + if value is None: + return "—" + return html.escape(str(value)) + + +def _dt(value) -> str: + """DATETIME → 字符串(aiomysql 返回的是 naive datetime,按原样显示即可)""" + if value is None: + return "—" + if isinstance(value, datetime): + return value.strftime("%Y-%m-%d %H:%M:%S") + return html.escape(str(value)) + + +def _token(value) -> str: + """会话令牌是活的凭据,页面只显示首尾便于比对,不给全量""" + if not value: + return "—" + s = str(value) + if len(s) <= 16: + return html.escape(s) + return html.escape(f"{s[:10]}…{s[-4:]}") + + +def _human_duration(seconds) -> str: + if seconds is None: + return "—" + seconds = int(seconds) + if seconds <= 0: + return "已过期" + days, rem = divmod(seconds, 86400) + hours, rem = divmod(rem, 3600) + minutes = rem // 60 + if days: + return f"{days} 天 {hours} 小时" + if hours: + return f"{hours} 小时 {minutes} 分" + return f"{minutes} 分" + + +def _tag(text: str, kind: str) -> str: + return f'{_e(text)}' + + +def _card(number, label: str) -> str: + return f'
{_e(number)}
{_e(label)}
' + + +def _table(headers: list[str], rows: list[list[str]], empty_text: str) -> str: + if not rows: + return f'
{_e(empty_text)}
' + head = "".join(f"{_e(h)}" for h in headers) + body = "".join("" + "".join(f"{c}" for c in r) + "" for r in rows) + return f"{head}{body}
" + + +def _render_overview(ov: dict) -> str: + cards = [ + _card(ov["users_total"], "用户总数"), + _card(ov["users_today"], "今日新增用户"), + _card(ov["users_free"], "已领免费授权"), + _card(ov["auth_active"], "生效中授权"), + _card(ov["auth_pending"], "待生效授权"), + _card(ov["auth_expired"], "已过期授权"), + _card(ov["auth_exhausted"], "已耗尽授权"), + _card(ov["auth_time"], "时间授权"), + _card(ov["auth_points"], "积分授权"), + _card(ov["usage_total"], "使用总次数"), + _card(ov["usage_today"], "今日使用次数"), + _card(ov["sessions_valid"], "有效会话"), + ] + return '
' + "".join(cards) + "
" + + +def _render_users(rows: list[dict]) -> str: + out = [] + for r in rows: + if r["type"] is None: + current = _tag("无可用授权", "none") + expire = "—" + else: + kind = "时间" if r["type"] == "time" else "积分" + source = {"free": "免费", "purchase": "购买", "admin": "后台"}.get(r["source"], r["source"]) + current = f"{_e(kind)} · {_e(source)} " + _tag("生效中", "active") + if r["type"] == "time": + expire = f"{_dt(r['end_at'])}
剩 {_e(_human_duration(r['remain_seconds']))}" + else: + expire = f"{_e(r['remaining_points'])} / {_e(r['total_points'])} 次" + + out.append([ + _e(r["id"]), + f'{_e(r["openid"])}', + _e(r["nickname"]), + _dt(r["created_at"]), + _dt(r["last_seen_at"]), + _tag("是", "active") if r["has_claimed_free"] else _tag("否", "expired"), + current, + expire, + _e(r["auth_count"]), + ]) + return _table( + ["ID", "openid", "昵称", "首次关注", "最近活跃", "已领免费", "当前授权", "到期 / 余量", "授权条数"], + out, + "还没有任何用户", + ) + + +def _render_usage(rows: list[dict]) -> str: + out = [] + for r in rows: + kind = "时间" if r["cost_type"] == "time" else "积分" + out.append([ + _e(r["id"]), + _e(r["user_id"]), + f'{_e(r["openid"])}', + f'{_e(r["device_id"])}', + _e(kind), + _e(r["cost_points"]), + _dt(r["used_at"]), + ]) + return _table( + ["ID", "用户ID", "openid", "设备", "计费类型", "扣减", "时间"], + out, + "还没有使用记录", + ) + + +def _render_sessions(rows: list[dict]) -> str: + out = [] + for r in rows: + out.append([ + _e(r["id"]), + _e(r["user_id"]), + f'{_e(r["openid"])}', + f'{_e(r["device_id"])}', + f'{_token(r["token"])}', + _dt(r["created_at"]), + _dt(r["expires_at"]), + _tag("有效", "active") if r["still_valid"] else _tag("已过期", "expired"), + ]) + return _table( + ["ID", "用户ID", "openid", "设备", "令牌", "签发时间", "过期时间", "状态"], + out, + "还没有会话", + ) + + +def _render_scenes(rows: list[dict]) -> str: + kind_map = {"pending": "pending", "scanned": "pending", "authorized": "active", "expired": "expired"} + out = [] + for r in rows: + out.append([ + _e(r["id"]), + f'{_e(r["scene_str"])}', + f'{_e(r["device_id"])}', + _tag(r["status"], kind_map.get(r["status"], "expired")), + _e(r["user_id"]), + _dt(r["created_at"]), + _dt(r["expires_at"]), + _dt(r["authorized_at"]), + ]) + return _table( + ["ID", "scene_str", "设备", "状态", "用户ID", "创建时间", "过期时间", "授权完成时间"], + out, + "还没有扫码场景", + ) + + +_CSS = """ +* { box-sizing: border-box; } +body { + font-family: -apple-system, "Segoe UI", "Microsoft YaHei", sans-serif; + margin: 0; padding: 24px 28px 48px; background: #f5f6f8; color: #1f2328; + font-size: 14px; line-height: 1.5; +} +h1 { font-size: 20px; margin: 0 0 2px; } +h2 { font-size: 15px; margin: 30px 0 8px; padding-bottom: 6px; border-bottom: 1px solid #e3e5e8; } +.meta { color: #6b7280; font-size: 12px; margin-bottom: 4px; } +.meta a { color: #2563eb; text-decoration: none; } +.meta a:hover { text-decoration: underline; } +.cards { display: flex; flex-wrap: wrap; gap: 10px; } +.card { background: #fff; border: 1px solid #e3e5e8; border-radius: 6px; padding: 10px 16px; min-width: 104px; } +.card .n { font-size: 22px; font-weight: 600; font-variant-numeric: tabular-nums; } +.card .l { font-size: 12px; color: #6b7280; margin-top: 2px; white-space: nowrap; } +table { border-collapse: collapse; width: 100%; background: #fff; font-size: 12.5px; } +th, td { border: 1px solid #e3e5e8; padding: 5px 9px; text-align: left; vertical-align: top; } +th { background: #f0f1f3; font-weight: 600; white-space: nowrap; } +code { font-family: Consolas, "Courier New", monospace; font-size: 12px; word-break: break-all; } +.dim { color: #9ca3af; } +.tag { display: inline-block; padding: 1px 7px; border-radius: 3px; font-size: 11px; white-space: nowrap; } +.tag.active { background: #dcfce7; color: #166534; } +.tag.pending { background: #fef9c3; color: #854d0e; } +.tag.expired, .tag.exhausted, .tag.cancelled { background: #f3f4f6; color: #6b7280; } +.tag.none { background: #fee2e2; color: #991b1b; } +.empty { color: #9ca3af; font-size: 13px; padding: 8px 2px; } +footer { margin-top: 36px; color: #9ca3af; font-size: 12px; } +""" + + +# --------------------------------------------------------------------------- +# 路由 +# --------------------------------------------------------------------------- + +@router.get("", response_class=HTMLResponse) +async def admin_page(_: None = Depends(require_admin)) -> HTMLResponse: + """管理页:一次拉齐四个区块的数据后整页渲染""" + async with db.acquire() as conn: + async with conn.cursor(aiomysql.DictCursor) as cur: + await cur.execute(OVERVIEW_SQL) + overview = await cur.fetchone() + + await cur.execute(USERS_SQL) + users = await cur.fetchall() + + await cur.execute(USAGE_SQL) + usage = await cur.fetchall() + + await cur.execute(SESSIONS_SQL) + sessions = await cur.fetchall() + + await cur.execute(SCENES_SQL) + scenes = await cur.fetchall() + + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + page = f""" + + + + +CadOcr 授权管理 + + + +

CadOcr 微信授权管理

+
数据生成时间 {_e(now)} · 刷新
+ +

概览

+{_render_overview(overview)} + +

用户与当前授权(最多 200 条,按 ID 倒序)

+{_render_users(users)} + +

最近使用记录(最多 50 条)

+{_render_usage(usage)} + +

会话令牌(最多 50 条)

+{_render_sessions(sessions)} + +

扫码场景(最多 50 条)

+{_render_scenes(scenes)} + + + +""" + return HTMLResponse(content=page) diff --git a/config.py b/config.py index 6c2af7b..f556afe 100644 --- a/config.py +++ b/config.py @@ -32,3 +32,8 @@ SCENE_TTL_SECONDS = int(os.getenv("SCENE_TTL_SECONDS", "300")) # 二维码/scen SESSION_TTL_HOURS = int(os.getenv("SESSION_TTL_HOURS", "24")) # session_token 有效期(小时) # 时间授权 usage_logs 节流窗口(秒):同一用户+设备在此窗口内只记一条 USAGE_LOG_THROTTLE_SECONDS = int(os.getenv("USAGE_LOG_THROTTLE_SECONDS", "60")) + +# 管理后台(/admin)的 HTTP Basic Auth 凭据。 +# ADMIN_PASSWORD 为空时 /admin 直接返回 503,绝不放行——避免「忘了配密码 = 谁都能进」。 +ADMIN_USER = os.getenv("ADMIN_USER", "admin") +ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "") diff --git a/wechat.py b/wechat.py index ce77280..f0576ce 100644 --- a/wechat.py +++ b/wechat.py @@ -4,7 +4,8 @@ GET /wechat - 微信服务器验证(签名校验 + 返回 echostr) POST /wechat - 接收微信推送的消息和事件(subscribe / SCAN 触发扫码授权) -授权接口在 auth.py、使用扣减接口在 usage.py 中定义,通过 include_router 挂载。 +授权接口在 auth.py、使用扣减接口在 usage.py、管理后台在 admin.py 中定义, +通过 include_router 挂载。 """ import hashlib @@ -16,6 +17,7 @@ from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException, Query, Request from fastapi.responses import PlainTextResponse +import admin import auth import db import usage @@ -41,6 +43,7 @@ async def lifespan(app: FastAPI): app = FastAPI(title="WeChat API", version="0.3.0", lifespan=lifespan) app.include_router(auth.router) app.include_router(usage.router) +app.include_router(admin.router) def verify_signature(signature: str, timestamp: str, nonce: str) -> bool: