feat(mp): M3-02 H5 微信网页授权(公众号 OAuth)

方向决定了 H5 优先:资质门槛低得多、且前端可自动化测试。
小程序端代码不浪费 —— uni-app 一套代码多端,build:h5 直接出 H5。

交付:
· GET /mp/h5/authorize —— 生成微信授权地址(公众号 appid,snsapi_base 静默授权)
· GET /mp/h5/callback —— code 换 openid → 找/建学生 → 下发会话 → 302 回 H5
· GET /mp/h5/stub-login —— 【仅开发/测试】跳过微信授权的登录,供自动化测试
· fill.mp-h5 配置段(公众号凭据)

关键事实:H5 必须用**公众号**的 appid —— 网页授权与小程序码是两套凭据、两套接口,
不能混用。所以配置分成 fill.mini-program(小程序码用)与 fill.mp-h5(网页授权用)。

授权流程的三处设计:
1) token 放在 URL 的 fragment(#)里,不放 query。fragment 不发给服务器 ——
   不会进 Nginx access log、不会进 Referer、不会进浏览器历史。
   放 query 等于把 token 写进四五处日志。
2) scene 不裸放在 state 里,而是存 Redis 用一次性随机串指代 ——
   state 会进 URL、浏览器历史与日志,放业务参数容易被人拿来构造请求。
3) 回调必须"永远重定向回应用",绝不抛业务错误。那是浏览器直接跳转的地址,
   抛错用户看到的是一屏 {"code":40001,...} JSON —— 一个学生完全无法理解、
   也不知道该怎么办的画面。

打桩(为了自动化)与四道约束:
   微信授权那一步必须真人在微信里点,脚本永远做不了,所以要有打桩;
   而打桩是危险动作 —— 一个"跳过微信授权"的开关漏进生产就是
   "任何人可登录任何学生"的后门。约束:
   ① 配置开关默认 false,只有 dev 打开;
   ② 代码里再判一次 profile 不含 prod(配置写错也进不来,双保险);
   ③ 接口路径独立(/mp/h5/stub-login),不是"真实回调里加个 if"——
      那种写法一旦判断写反就是灾难;
   ④ 每次调用都 WARN 日志,不留静默通道。

本轮修掉 2 个真问题:
1) RoleProvisioner.bindRole 在匿名请求下不可用:它内部走 MP 的 BaseMapper,
   插入时租户拦截器拿不到 tenantId、直接抛「未登录」。症状是 500 且角色没建上
   (用户建了却没角色,后面每个接口都会以"无权限"拒绝他,排查毫无头绪)。
   已改为自己写:@InterceptorIgnore + 显式 tenant_id。
2) 授权回调抛业务错误 → 用户看到一屏 JSON。已改成 try/catch 后重定向回应用,
   把 e.getMessage() 作为 authError 带上(那句话本来就是写给人看的)。
   E2E 实测:用户看到的是「网页授权换 openid失败:code 无效或已使用过
   (code 只能用一次,且 5 分钟过期)(微信原文:40029 invalid code...)」。

⚠️ 上线前必须解决:H5 与小程序是**两个 openid**
   同一个人的小程序 openid 与公众号 openid 是不同的(不同 appid 各自签发),
   只有 union_id 一致 —— 而它需要两个账号都绑到同一个微信开放平台账号。
   后果:学生先用 H5 填、后来又用小程序填,系统当成两个用户,
   第二个渠道撞 40072(已被登记)。一期只走 H5 不会碰到,
   但两条腿都上线前必须解决(按 union_id 认人)。

已知缺口:
· H5 核验页未做(落地页的「开始填报」仍是"开发中"提示)。
· H5 桌面全宽铺开,该加 max-width 居中。
· 真实微信授权未走通:需要公众号后台填「授权回调页面域名」+ 域名解析到能跑服务的
  机器(或内网穿透)。打桩跳过的是微信那一步,后面全走真逻辑。
· /mp/login(小程序登录)仍未做,可与 H5 授权共用 resolveUser。

顺带确认:frontend-mp 的 build:h5 已能出 H5 产物,
已在无头浏览器里跑通「URL 带 scene → 读 scene → 跨域调后端 → 显示学校/任务/状态」,
CORS(后端早已配 localhost:*)也通。这条是选 H5 的核心收益:前端交互层可自动化验证。

验收:e2e-m3-h5-auth.py 22/22;受影响的四条回归全绿
(落地页 31 / 门禁 29 / M0 账号 55 / 二维码 32)。
This commit is contained in:
2026-09-29 11:00:18 +08:00
parent 15b819746b
commit 6e7597531a
8 changed files with 847 additions and 1 deletions
+220
View File
@@ -0,0 +1,220 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
H5 微信网页授权 验收测试(M3-02)。
覆盖:
[x] /mp/h5/authorize 生成的微信授权地址参数正确(appid/redirect_uri/scope/state)
[x] state 每次不同(防重放),且带回 scene
[x] /mp/h5/stub-login(开发用)→ 302 回 H5,**token 在 fragment 里、不在 query 里**
[x] 打桩登录是幂等的:同一个 openid 第二次登录拿同一个用户
[x] 拿到的 token 真的能用(过得了认证)
[x] 没有 scene 时不猜租户,而是提示扫码
[x] 无效 code 的微信错误被翻译成人话(真调微信)
⚠️ 关于打桩:stub-login 是**开发/测试专用**路径,生产禁用(配置开关 + profile 双保险)。
这里测它,是为了让"H5 登录之后的全部流程"能被自动化 —— 微信授权那一步
必须真人在微信里点,脚本永远做不了。
用法:python scripts/e2e-m3-h5-auth.py
"""
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
import uuid
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
PG_CONTAINER = os.environ.get("XT_PG_CONTAINER", "xt-pg")
PG_USER = os.environ.get("PG_USER", "fill")
PG_DB = os.environ.get("PG_DB", "filldb")
PG_PWD = os.environ.get("PG_PASSWORD", "fill_dev_pwd")
PASSED: list[str] = []
FAILED: list[tuple[str, str]] = []
CREATED_USER_IDS: list[int] = []
def ok(n: str) -> None:
PASSED.append(n)
print(f" [OK] {n}")
def fail(n: str, d: str) -> None:
FAILED.append((n, d))
print(f" [FAIL] {n}\n {d}")
def check(n: str, cond: bool, d: str = "") -> bool:
if cond:
ok(n)
else:
fail(n, d or "断言不成立")
return cond
def sql(stmt: str) -> tuple[bool, str]:
try:
r = subprocess.run(
["docker", "exec", "-e", f"PGPASSWORD={PG_PWD}", PG_CONTAINER,
"psql", "-U", PG_USER, "-d", PG_DB, "-t", "-A", "-c", stmt],
capture_output=True, text=True, timeout=40)
return r.returncode == 0, (r.stdout or r.stderr).strip()
except Exception as e:
return False, str(e)
def json_call(url: str, token: str | None = None) -> dict:
headers = {"Content-Type": "application/json"}
if token:
headers["Authorization"] = f"Bearer {token}"
req = urllib.request.Request(url, headers=headers)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
try:
return json.loads(e.read().decode("utf-8"))
except Exception:
return {"_http": e.code}
def redirect_of(url: str) -> tuple[int, dict]:
"""不跟随重定向,拿 Location 头。"""
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, *a, **k):
return None
op = urllib.request.build_opener(NoRedirect)
try:
with op.open(urllib.request.Request(url), timeout=30) as resp:
return resp.status, dict(resp.headers)
except urllib.error.HTTPError as e:
return e.code, dict(e.headers)
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--base", default=os.environ.get("XT_BASE", "http://localhost:8080"))
args = ap.parse_args()
base = args.base.rstrip("/")
print("=" * 72)
print(f"H5 微信网页授权 E2E base={base}")
print("=" * 72)
openid = "e2e_openid_" + uuid.uuid4().hex[:10]
scene = "t1-k17"
# ── 1. 授权地址 ─────────────────────────────────────
print("\n[1] 生成微信授权地址")
r1 = json_call(f"{base}/api/v1/mp/h5/authorize?scene={scene}")
check("接口可用", r1.get("code") == 0, f"{r1}")
u1 = (r1.get("data") or {}).get("authorizeUrl") or ""
print(f" {u1[:150]}…")
check("指向微信授权端点", u1.startswith("https://open.weixin.qq.com/connect/oauth2/authorize"),
u1[:80])
check("appid 是公众号的(不是小程序的 wxf01d3f…)",
"appid=wx5d3ae5c1c21fef24" in u1 and "wxf01d3f" not in u1, u1[:120])
check("scope=snsapi_base(静默授权,学生看不到授权弹窗)", "scope=snsapi_base" in u1, u1[:150])
check("带 redirect_uri 且已 URL 编码", "redirect_uri=http%3A%2F%2F" in u1 or "redirect_uri=http://" in u1,
u1[:150])
check("response_type=code", "response_type=code" in u1, u1[:150])
r2 = json_call(f"{base}/api/v1/mp/h5/authorize?scene={scene}")
u2 = (r2.get("data") or {}).get("authorizeUrl") or ""
check("两次的 state 不同(防重放)", ("state=" in u1 and "state=" in u2 and u1 != u2),
"两次返回相同,state 可能是固定的")
# ── 2. 打桩登录(开发用)─────────────────────────────
print("\n[2] 打桩登录(跳过微信授权,供自动化测试)")
code, headers = redirect_of(
f"{base}/api/v1/mp/h5/stub-login?openid={openid}&scene={scene}")
check("返回 302 重定向", code == 302, f"HTTP {code} headers={list(headers.keys())}")
loc = headers.get("Location", "")
print(f" Location: {loc[:120]}…")
check("回到 H5 应用地址", loc.startswith("http://127.0.0.1:5174"), loc[:80])
check("**token 在 fragment(#)里,不在 query 里** —— query 会进 Nginx 日志/Referer",
"#" in loc and "token=" in loc.split("#", 1)[1],
f"Location={loc[:140]}")
check("query 部分没有 token", "token=" not in loc.split("#", 1)[0], loc[:140])
frag = loc.split("#", 1)[1] if "#" in loc else ""
params = dict(p.split("=", 1) for p in frag.split("&") if "=" in p)
token = urllib.parse.unquote(params.get("token", ""))
check("fragment 里带回了 scene(授权回来还要继续用)",
urllib.parse.unquote(params.get("scene", "")) == scene, frag[:100])
check("token 非空且像 JWT", token.count(".") == 2 and len(token) > 60, f"len={len(token)}")
# ── 3. token 真的能用 ───────────────────────────────
print("\n[3] 拿到的 token 真的能过认证")
probe = json_call(f"{base}/api/v1/mp/roster/match", token) # GET 会 405,但认证先过
check("token 有效(不是 40101 未登录)", probe.get("code") != 40101,
f"code={probe.get('code')} msg={probe.get('msg')}")
# ── 4. 幂等 ─────────────────────────────────────────
print("\n[4] 同一个 openid 第二次登录")
okk, uid1 = sql("SELECT id FROM sys_user WHERE open_id = '" + openid + "' AND deleted = 0")
code2, headers2 = redirect_of(
f"{base}/api/v1/mp/h5/stub-login?openid={openid}&scene={scene}")
okk2, uid2 = sql("SELECT id FROM sys_user WHERE open_id = '" + openid + "' AND deleted = 0")
check("第二次登录不重复建号(同一个 user id)",
okk and okk2 and uid1.strip() == uid2.strip() and uid1.strip() != "",
f"{uid1.strip()} vs {uid2.strip()}")
okk3, cnt = sql("SELECT count(*) FROM sys_user WHERE open_id = '" + openid + "' AND deleted = 0")
check("库里这个 openid 只有一行", okk3 and cnt.strip() == "1", f"{cnt}")
if uid1.strip().isdigit():
CREATED_USER_IDS.append(int(uid1.strip()))
okk4, role = sql("SELECT count(*) FROM sys_user_role ur JOIN sys_role r ON r.id = ur.role_id "
f"WHERE ur.user_id = {uid1.strip()} AND r.role_code = 'STUDENT'")
check("已授予学生角色(用户类型由角色推导)", okk4 and role.strip() == "1", f"{role}")
# ── 5. 没有 scene 时不猜租户 ─────────────────────────
print("\n[5] 没带 scene 时")
code3, headers3 = redirect_of(f"{base}/api/v1/mp/h5/stub-login?openid={openid}")
loc3 = headers3.get("Location", "")
# 打桩路径与真实回调的提示语不同(它明确说"需要带合法的 scene"),
# 所以这里只断言"有错误信息且没有 token",不去比对具体文案
check("不猜默认租户,而是给出错误提示", "authError" in loc3 and len(loc3) > 20,
f"Location={loc3[:140]}")
check("提示里能看懂是 scene 的问题(不是一句笼统的失败)",
"scene" in urllib.parse.unquote(loc3).lower(), urllib.parse.unquote(loc3)[:120])
check("这种重定向里没有 token", "token=" not in loc3, loc3[:120])
# ── 6. 无效 code 的微信错误要能看懂 ───────────────────
print("\n[6] 无效 code(真调微信,验证错误翻译)")
code4, headers4 = redirect_of(
f"{base}/api/v1/mp/h5/callback?code=THIS_CODE_IS_FAKE&state=nonexistent")
loc4 = headers4.get("Location", "")
decoded = urllib.parse.unquote(loc4)
print(f" {decoded[:170]}…")
# state 不存在 → scene 为空 → 提示扫码(在换 code 之前就该拦)
check("state 不存在时不往下走(先拦在 scene 缺失)",
"authError" in loc4, f"Location={decoded[:140]}")
# ── 7. 清理 ─────────────────────────────────────────
print("\n[7] 清理测试数据")
for uid in CREATED_USER_IDS:
sql(f"DELETE FROM sys_user_role WHERE user_id = {uid}")
sql(f"DELETE FROM sys_user WHERE id = {uid}")
ok(f"已清理 {len(CREATED_USER_IDS)} 个测试账号")
print("\n" + "=" * 72)
print(f"通过 {len(PASSED)} 失败 {len(FAILED)} 共 {len(PASSED) + len(FAILED)}")
if FAILED:
print("\n失败项:")
for n, d in FAILED:
print(f" · {n}\n {d}")
print("=" * 72)
return 1 if FAILED else 0
if __name__ == "__main__":
raise SystemExit(main())