tamakyi/tamabox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

安全: 修复凭据、遥测与账号数据清理风险

Browse Source
This commit is contained in:
tamakyi
2026-11-04 14:38:19 +08:00
parent c633789a92
commit 3e7719f200
19 changed files with 305 additions and 3378 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dev/__pycache__/devcheck-e2e.cpython-313.pyc
BIN
View File
Binary file not shown.
dev/devcheck-avatar.py
-346
View File
@@ -1,346 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""头像上传链路端到端验证（批次 A 分档 + B1 EXIF 归一）。
为什么走头像而不是提问图：
- 提问图那条路要经过内容审核/提问入库才能拿到 key，链路长且易受其它因素干扰；
- 头像强制 1:1 且 POST /user/profile/update 直接返回页面，便于立即取回 CDN URL。
验证两件事：
A 分档生效：上传 1600x1600 的大图，头像档 maxSide=512 -> 存盘应为 512x512
（改前是 1920，实测体积差 ~7 倍）
B1 EXIF Orientation=6 归一：上传「像素 512x256 + EXIF 顺时针 90」的方形图，
归一后仍应是 512x512（宽高互换但比例不变，故改用体积+尺寸双断言，
并额外验证竖长条图 256x512 + Orientation=6 -> 应变成 512x256）
用法：
python dev/devcheck-avatar.py [--base https://box-dev-api.337845818.best]
依赖：Pillow（仅用于本地生成测试图片）
清理：TamaBox ban --email <输出中的测试邮箱> --yes
"""
import argparse
import http.cookiejar
import io
import json
import re
import struct
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36")
RESULTS = []
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102
return None
class Resp:
def __init__(self, status, headers, body):
self.status = status
self.h = {k.lower(): v for k, v in headers.items()}
self.body = body
@property
def loc(self):
return self.h.get("location", "")
@property
def text(self):
return self.body.decode("utf-8", "replace")
class Client:
def __init__(self, base):
self.base = base
self.cj = http.cookiejar.CookieJar()
self.op = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(self.cj), NoRedirect())
def _send(self, req, timeout=120):
try:
with self.op.open(req, timeout=timeout) as r:
return Resp(r.status, dict(r.headers), r.read())
except urllib.error.HTTPError as e:
try:
body = e.read()
except Exception: # noqa: BLE001
body = b""
return Resp(e.code, dict(e.headers), body)
except Exception as e: # noqa: BLE001
return Resp(-1, {}, ("TRANSPORT_ERROR: %s" % e).encode())
def req(self, method, path, data=None, content_type=None, timeout=120):
url = path if path.startswith("http") else self.base + path
headers = {"User-Agent": UA, "Referer": self.base + path.split("?")[0]}
body = None
if data is not None:
body = data if isinstance(data, bytes) else urllib.parse.urlencode(data).encode()
headers["Content-Type"] = content_type or "application/x-www-form-urlencoded"
return self._send(urllib.request.Request(url, data=body, headers=headers, method=method),
timeout)
def get(self, p, **kw):
r = self.req("GET", p, **kw)
return r if r.status != -1 else self.req("GET", p, **kw)
def post(self, p, data=None, **kw):
return self.req("POST", p, data=data, **kw)
def get_binary(self, url):
return self._send(urllib.request.Request(
url, headers={"User-Agent": UA}, method="GET"), timeout=60)
def step(desc, ok, extra=""):
RESULTS.append((desc, ok, extra))
print(" [%s] %s%s" % ("PASS" if ok else "FAIL", desc, (" (%s)" % extra) if extra else ""))
return ok
def csrf_of(html):
m = re.search(r'name="_csrf"\s+value="([^"]+)"', html)
return m.group(1) if m else None
def captcha(c):
j = json.loads(c.get("/api/captcha/generate").text)
pid, q = j["data"]["id"], j["data"]["question"]
m = re.match(r"\s*(\d+)\s*([+\-×])\s*(\d+)\s*=", q)
a, op, b = int(m.group(1)), m.group(2), int(m.group(3))
return "%s:%d" % (pid, {"+": a + b, "-": a - b, "×": a * b}[op])
def multipart(fields, files):
boundary = "----tbav%s" % uuid.uuid4().hex
buf = io.BytesIO()
for k, v in fields.items():
buf.write(("--%s\r\n" % boundary).encode())
buf.write(('Content-Disposition: form-data; name="%s"\r\n\r\n' % k).encode())
buf.write(("%s\r\n" % v).encode())
for name, fn, ct, data in files:
buf.write(("--%s\r\n" % boundary).encode())
buf.write(('Content-Disposition: form-data; name="%s"; filename="%s"\r\n' % (name, fn)).encode())
buf.write(("Content-Type: %s\r\n\r\n" % ct).encode())
buf.write(data)
buf.write(b"\r\n")
buf.write(("--%s--\r\n" % boundary).encode())
return buf.getvalue(), "multipart/form-data; boundary=%s" % boundary
def sniff(d):
if len(d) >= 12 and d[:4] == b"RIFF" and d[8:12] == b"WEBP":
return "webp", webp_dims(d)
if d[:3] == b"\xff\xd8\xff":
return "jpeg", jpeg_dims(d)
if d[:8] == b"\x89PNG\r\n\x1a\n":
return "png", (struct.unpack(">II", d[16:24]) if len(d) >= 24 else None)
return "unknown", None
def webp_dims(d):
c = d[12:16]
if c == b"VP8 " and len(d) >= 30:
return (struct.unpack("<H", d[26:28])[0] & 0x3FFF,
struct.unpack("<H", d[28:30])[0] & 0x3FFF)
if c == b"VP8L" and len(d) >= 25:
b = struct.unpack("<I", d[21:25])[0]
return (b & 0x3FFF) + 1, ((b >> 14) & 0x3FFF) + 1
if c == b"VP8X" and len(d) >= 30:
return ((d[24] | d[25] << 8 | d[26] << 16) + 1,
(d[27] | d[28] << 8 | d[29] << 16) + 1)
return None
def jpeg_dims(d):
i = 2
while i + 9 < len(d):
if d[i] != 0xFF:
i += 1
continue
m = d[i + 1]
if m in (0xC0, 0xC1, 0xC2, 0xC3):
h, w = struct.unpack(">HH", d[i + 5:i + 9])
return (w, h)
if m in (0xD8, 0xD9) or 0xD0 <= m <= 0xD7:
i += 2
continue
i += 2 + struct.unpack(">H", d[i + 2:i + 4])[0]
return None
def make_avatar(w, h, orientation=None):
"""生成一张带明显方位特征的方形头像。"""
from PIL import Image, ImageDraw
im = Image.new("RGB", (w, h), (30, 120, 200))
d = ImageDraw.Draw(im)
# 顶部一条粗红条：旋转 90 度后它会跑到侧面，方向一眼可辨
d.rectangle([0, 0, w, max(2, h // 12)], fill=(230, 30, 30))
for i in range(0, w, max(8, w // 20)):
d.line([(i, h // 4), (i, h)], fill=(255, 255, 255), width=2)
buf = io.BytesIO()
if orientation:
exif = im.getexif()
exif[0x0112] = orientation
im.save(buf, "JPEG", quality=90, exif=exif)
else:
im.save(buf, "JPEG", quality=90)
return buf.getvalue()
def scrape_profile_fields(html):
"""从 profile 页抓取所有 input 的 name/value，保证回填必填项。
UpdateProfile 的必填字段有十来个（昵称/介绍/5 个二维码颜色/点大小/透明度…），
漏一个就会因 `valid:"required"` 被 Bind 挡下、头像根本不会走到上传分支。
"""
fields = {}
for m in re.finditer(r'<input\b[^>]*>', html, re.I):
tag = m.group(0)
nm = re.search(r'name="([^"]+)"', tag, re.I)
if not nm:
continue
name = nm.group(1)
if name in ("avatar", "background", "backgroundimage", "_csrf", "submit"):
continue
typ = re.search(r'type="([^"]+)"', tag, re.I)
if typ and typ.group(1).lower() in ("checkbox", "radio"):
# 只保留勾选态的，避免把「本想取消」的项又勾上
if not re.search(r'\bchecked\b', tag, re.I):
continue
val = re.search(r'value="([^"]*)"', tag, re.I)
fields[name] = val.group(1) if val else ""
# select 走第一个非空 option
for m in re.finditer(r'<select\b[^>]*name="([^"]+)"[^>]*>(.*?)</select>', html, re.I | re.S):
name, body = m.group(1), m.group(2)
opts = re.findall(r'<option\b([^>]*)>', body, re.I)
chosen = ""
for o in opts:
vm = re.search(r'value="([^"]*)"', o)
v = vm.group(1) if vm else ""
if re.search(r'\bselected\b', o, re.I):
chosen = v
break
if not chosen:
chosen = v
fields[name] = chosen
# textarea
for m in re.finditer(r'<textarea\b[^>]*name="([^"]+)"[^>]*>(.*?)</textarea>', html, re.I | re.S):
fields[m.group(1)] = re.sub(r"<[^>]+>", "", m.group(2)).strip()
return fields
def upload_avatar(c, data, fname="avatar.jpg"):
page = c.get("/user/profile")
if page.status != 200:
return None, "profile HTTP %s" % page.status
fields = scrape_profile_fields(page.text)
fields["_csrf"] = csrf_of(page.text)
body, ct = multipart(fields, [("avatar", fname, "image/jpeg", data)])
res = c.post("/user/profile/update", data=body, content_type=ct)
# 成功时是 302 跳回 /user/profile，头像 URL 在跳转后的页面里
if res.status == 302 and res.loc:
res = c.get(res.loc if res.loc.startswith("/") else "/user/profile")
if res.status != 200:
return None, "update 后页面 HTTP %s" % res.status
# 从页面里取头像 URL：<img src="https://r2.../picture/..."> 或 og:image
urls = re.findall(r'https://r2\.[^"\'\s<>]+/picture/[0-9]+/\d{2}/\d{2}/[0-9a-f]+', res.text)
if not urls:
m = re.search(r'(https://r2\.[^"\'\s<>]+/picture/[^"\'\s<>]+)', res.text)
urls = [m.group(1)] if m else []
if urls:
return urls[-1], None
# 没匹配到：落盘返回页并打印可疑片段，便于定位是「上传被拒」还是「正则没覆盖」
dump = "C:/Users/Tamakyi Gekka/AppData/Local/Temp/tbav_profile.html"
with open(dump, "wb") as f:
f.write(res.body)
alerts = re.findall(r'class="uk-alert[^"]*"[^>]*>(.*?)</div>', res.text, re.S)
hint = " | ".join(re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", a)).strip()[:80]
for a in alerts[:3])
return None, "未匹配到 picture URL（已存 %s）%s%s" % (dump, (" 提示:" + hint) if hint else "",
(" HTTP %s" % res.status))
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--base", default="https://box-dev-api.337845818.best")
args = ap.parse_args()
base = args.base.rstrip("/")
pwd = "TbAv!2026"
ts = time.strftime("%m%d%H%M%S")
email, domain = "tbav%s@example.com" % ts, "tbav%s" % ts
print("== 账号：%s / %s ==" % (email, domain))
# 方形 1600x1600：验证 A 分档（应存成 512x512）
square = make_avatar(1600, 1600)
# EXIF Orientation=6（顺时针 90）：像素存成横的 1536x512，归一后显示为竖的 512x1536，
# 再经头像档限边 512 -> 512x512。
# 注意必须用「像素横躺」的样本：头像有 1:1 比例校验，若像素本身就是竖的会被先拒掉，
# 那样就测不到 EXIF 分支了。
exif_shot = make_avatar(1536, 512, orientation=6)
print(" 本地样本 square 1600x1600%7d bytes" % len(square))
print(" 本地样本 exifshot 1536x512 (EXIF=6)%d bytes" % len(exif_shot))
c = Client(base)
r = c.get("/register")
reg = c.post("/register", {
"email": email, "domain": domain, "name": "TBAv%s" % ts,
"password": pwd, "repeat_password": pwd,
"g-recaptcha-response": captcha(c), "language": "zh-CN",
"_csrf": csrf_of(r.text),
})
if not step("注册", reg.status == 302, "HTTP %s -> %s" % (reg.status, reg.loc)):
return 1
c = Client(base)
r = c.get("/login")
lg = c.post("/login", {"email": email, "password": pwd,
"g-recaptcha-response": captcha(c), "_csrf": csrf_of(r.text)})
if not step("登录", lg.status == 302, "HTTP %s -> %s" % (lg.status, lg.loc)):
return 1
# ---- A 分档 ----
url, err = upload_avatar(c, square)
if not step("上传方形大头像（1600x1600）", url is not None, err or (url or "")):
return 1
img = c.get_binary(url)
st, data = img.status, img.body
if not step("拉回头像", st == 200 and bool(data), "HTTP %s" % st):
return 1
kind, dims = sniff(data)
print(" 存储结果：%s %s %d bytes" % (kind, dims, len(data)))
step("A 分档生效：大图被缩到头像档 512（改前是 1920）",
dims == (512, 512), "实际 %s" % (dims,))
# ---- B1 EXIF ----
# 关键：两次上传的 URL 必须不同，否则说明第二次没生效（回退到了旧头像）。
url2, err2 = upload_avatar(c, exif_shot)
if not step("上传 EXIF 横躺头像（1536x512, Orientation=6）", bool(url2), err2 or (url2 or "")):
return 0
if url2 == url:
step("B1 EXIF 归一（第二次上传未生效，URL 未变）", False,
"两次上传返回同一 URL，无法区分；请检查是否被 1:1 校验拒绝")
return 0
img2 = c.get_binary(url2)
kind2, dims2 = sniff(img2.body)
print(" 存储结果：%s %s %d bytes" % (kind2, dims2, len(img2.body)))
# 归一后：像素 1536x512 -> 旋转成 512x1536 -> 再限边 512 -> 512x512
step("B1 EXIF Orientation=6 归一（1536x512 -> 512x1536 -> 512x512）",
dims2 == (512, 512), "实际 %s" % (dims2,))
# 体积对照：未归一时 1536x512 的 WebP 明显更大
step("B1 归一后体积已按 512 档压过（应与方形样本同量级）",
len(img2.body) < 6000, "%d bytes" % len(img2.body))
print("\n== 结果：%d/%d 通过 ==" % (sum(1 for _, ok, _ in RESULTS if ok), len(RESULTS)))
print("清理：TamaBox ban --email %s --yes" % email)
return 0 if all(ok for _, ok, _ in RESULTS) else 1
if __name__ == "__main__":
sys.exit(main())
dev/devcheck-e2e.py
-360
View File
@@ -1,360 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""TamaBox 部署后冒烟巡检（端到端）。
用途：对已部署环境跑一遍核心链路，用于发版/重新部署后的回归。
覆盖：注册 -> 登录 -> 个人页/提问列表/投票面板 -> 投票全链路（建票发布/公开页/匿名提交/
结果导出计数/草稿/表单型）-> 匿名提问 -> 停用
-> /deactivated 成功页 -> **同邮箱重注册必须被明确拒绝并提示联系站长**
-> 公开页只读巡检 -> /deactivated 四语言渲染。
语义约定（2026-09-15 起）：停用 = 清空内容 + 保留用户行作「身份占位」，
故邮箱/域名继续被占用，重注册应得到「账号已停用，请找站长」的提示而不是 302 或 500。
用法：
python dev/devcheck-e2e.py [--base https://box-dev-api.337845818.best]
注意：
- 会在目标环境**真实创建账号并发送提问**（使用 tbe2e<时间戳>@example.com 之类的可识别测试邮箱），
仅在测试/开发环境运行。
- 每次完整运行都会**留下一个已停用的身份占位**（邮箱/域名被占用）。清理方式：
`TamaBox ban --email <该邮箱> --yes`（连行物理删除）或 `TamaBox restore --email <该邮箱> --yes`（恢复账号）。
- 依赖数学验证码（conf.Recaptcha.CaptchaType == "math"）；若站点改用 recaptcha/turnstile 则无法自动通过。
"""
import argparse
import http.cookiejar
import json
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36")
RESULTS = []
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102
return None
class Resp:
def __init__(self, status, headers, body):
self.status = status
self.h = {k.lower(): v for k, v in headers.items()}
self.body = body
@property
def loc(self):
return self.h.get("location", "")
def alert(self):
m = re.search(r'class="uk-alert[^"]*"[^>]*>(.*?)</div>', self.body, re.S)
return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", m.group(0))).strip()[:200] if m else ""
class Client:
def __init__(self, base):
self.base = base
self.cj = http.cookiejar.CookieJar()
self.op = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(self.cj), NoRedirect())
def req(self, method, path, data=None, timeout=90):
url = path if path.startswith("http") else self.base + path
headers = {"User-Agent": UA, "Referer": self.base + path.split("?")[0]}
body = urllib.parse.urlencode(data).encode() if data is not None else None
if body:
headers["Content-Type"] = "application/x-www-form-urlencoded"
req = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with self.op.open(req, timeout=timeout) as r:
return Resp(r.status, dict(r.headers), r.read().decode("utf-8", "replace"))
except urllib.error.HTTPError as e:
# 读取错误响应体本身也可能超时（尤其 302 无体/远端慢），不能让它把巡检整轮打断
try:
body = e.read().decode("utf-8", "replace")
except Exception: # noqa: BLE001
body = ""
return Resp(e.code, dict(e.headers), body)
except Exception as e: # noqa: BLE001
return Resp(-1, {}, "TRANSPORT_ERROR: %s: %s" % (type(e).__name__, e))
def get(self, p, **kw):
# 只读 GET 重试一次：远端偶发慢请求会产生 TRANSPORT_ERROR，避免巡检出现假失败。
r = self.req("GET", p, **kw)
if r.status == -1:
r = self.req("GET", p, **kw)
return r
def post(self, p, data=None, **kw):
return self.req("POST", p, data=data, **kw)
def csrf_of(html):
m = re.search(r'name="_csrf"\s+value="([^"]+)"', html)
return m.group(1) if m else None
def captcha(c):
"""取一道数学验证码并算出答案，返回 <id>:<answer>。"""
j = json.loads(c.get("/api/captcha/generate").body)
pid, q = j["data"]["id"], j["data"]["question"]
m = re.match(r"\s*(\d+)\s*([+\-×])\s*(\d+)\s*=", q)
a, op, b = int(m.group(1)), m.group(2), int(m.group(3))
ans = {"+": a + b, "-": a - b, "×": a * b}[op]
return "%s:%d" % (pid, ans)
def step(desc, ok, extra=""):
RESULTS.append((desc, ok, extra))
print(" [%s] %s%s" % ("PASS" if ok else "FAIL", desc, (" (%s)" % extra) if extra else ""))
# 停用提示的跨语言特征串（站点默认语言可能是 ja，断言不能只看中文）
DEACTIVATED_HINTS = ("已停用", "has been deactivated", "は停止されています")
INTERNAL_ERROR_HINTS = ("サーバー内部エラー", "Internal Server Error", "服务器内部错误")
def is_deactivated_rejected(r):
"""停用后的重注册应被明确拒绝：HTTP 200 + 提示「账号已停用」，且不得是 500 兜底。"""
if r.status != 200:
return False
if any(h in r.body for h in INTERNAL_ERROR_HINTS):
return False
return any(h in r.body for h in DEACTIVATED_HINTS)
def check_public_pages(base):
"""公开页只读巡检：期望全部 200（/user/votes 之类的登录页除外）。"""
paths = ["/", "/explore", "/register", "/login", "/forgot-password",
"/sponsor", "/change-logs", "/about-me", "/robots.txt",
"/sitemap.xml", "/llms.txt", "/llms-full.txt", "/favicon.ico",
"/deactivated", "/api/captcha/generate", "/api/v1/site-stats"]
c = Client(base)
bad = []
for p in paths:
r = c.get(p, timeout=40)
if r.status != 200:
bad.append("%s=%s" % (p, r.status))
step("公开页只读巡检（%d 个端点）" % len(paths), not bad, ", ".join(bad))
def create_vote(c, title, vtype="single", options=None, fields=None, action="publish"):
"""提交一次投票创建（POST /user/votes）。
后端对创建动作一律按 draft=true 解析（草稿可随时暂存、跳过审核），
action=publish 时再走 validateForPublish + 内容审核；因此这里两种动作都可直接测。
"""
page = c.get("/user/votes/new")
data = {
"title": title, "description": "", "visibility": "public",
"voter_scope": "anyone", "show_result": "after_vote",
"type": vtype, "max_choices": "1", "allow_custom": "",
"start_at": "", "end_at": "", "quota_limit": "0",
"options": json.dumps(options or [], ensure_ascii=False),
"form_fields": json.dumps(fields or [], ensure_ascii=False),
"action": action, "_csrf": csrf_of(page.body),
}
return c.post("/user/votes", data)
def check_user_assets(owner):
"""登录态的接口/导出，以及停用页的「身份占位」说明。
停用页说明用**结构断言**（.uk-text-muted 段落非空），不写死文案 —— 站点语言随会话变化。
"""
r = owner.get("/api/v1/user")
step("登录态 /api/v1/user 返回 JSON", r.status == 200 and r.body.lstrip().startswith("{"),
"HTTP %s" % r.status)
step("登录态 /explore", owner.get("/explore").status == 200)
csrf = csrf_of(owner.get("/user/profile").body)
r = owner.post("/user/profile/export-json", {"_csrf": csrf})
step("导出 JSON", r.status == 200 and r.body.lstrip().startswith(("{", "[")),
"HTTP %s" % r.status)
r = owner.post("/user/profile/export", {"_csrf": csrf})
step("导出 Excel", r.status == 200, "HTTP %s %s" % (r.status, r.h.get("content-type", "")))
r = owner.get("/user/profile/deactivate")
note = re.search(r'class="uk-text-small uk-text-muted">(.*?)</span>', r.body, re.S)
step("停用页含身份占位说明", r.status == 200 and bool(note) and note.group(1).strip() != "")
def check_vote_flow(owner, base):
"""投票全链路：建票发布 -> 管理列表 -> 公开页 -> 匿名提交 -> 导出计数 -> 草稿 -> 表单型。"""
ts = time.strftime("%m%d%H%M%S")
title = "巡检单选投票%s" % ts
r = create_vote(owner, title, "single", options=["选项甲", "选项乙", "选项丙"])
step("创建单选投票并发布", r.status == 302 and "questions" in r.loc,
"HTTP %s -> %s %s" % (r.status, r.loc, r.alert()))
q = owner.get("/user/questions")
step("投票出现在管理列表", title in q.body)
mt = re.search(r"copyShareURL\('([^']+)'", q.body)
mv = re.search(r"/user/votes/(\d+)/edit", q.body)
token, vid = (mt.group(1) if mt else None), (mv.group(1) if mv else None)
step("取得分享 token / 投票 ID", bool(token and vid), "%s / %s" % (token, vid))
if not (token and vid):
return
anon = Client(base)
pv = anon.get("/vote/%s" % token)
step("公开投票页渲染", pv.status == 200 and title in pv.body, "HTTP %s" % pv.status)
mo = re.search(r'name="option_id" value="(\d+)"', pv.body)
step("公开页含可选项", bool(mo))
if not mo:
return
sub = anon.post("/vote/%s/submit" % token, {
"option_id": mo.group(1), "g-recaptcha-response": captcha(anon),
"custom_option": "", "_csrf": csrf_of(pv.body),
})
step("匿名提交投票", sub.status == 302, "HTTP %s -> %s %s" % (sub.status, sub.loc, sub.alert()))
# 导出仅作只读校验：单选导出的 CSV 每行是 <选项,票数>，末行为 <总参与,N>。
# 只认「总参与」这一行，避免把各选项票数一起求和导致 2 倍计数。
exp = owner.get("/user/votes/%s/export" % vid)
mtot = re.search(r"总参与,(\d+)", exp.body)
total = int(mtot.group(1)) if mtot else -1
step("导出显示总参与 = 1", exp.status == 200 and total == 1,
"HTTP %s 总参与=%s" % (exp.status, total))
dr = create_vote(owner, "巡检草稿%s" % ts, "single", options=["甲", "乙"], action="save")
step("保存草稿跳编辑页(?saved=1)",
dr.status == 302 and "/user/votes/" in dr.loc and "saved=1" in dr.loc,
"HTTP %s -> %s" % (dr.status, dr.loc))
fr = create_vote(owner, "巡检表单投票%s" % ts, "form", fields=[
{"field_type": "text", "label": "昵称", "required": True, "max_length": 50},
{"field_type": "textarea", "label": "建议", "required": False, "max_length": 200},
])
step("创建表单型投票并发布", fr.status == 302 and "questions" in fr.loc,
"HTTP %s -> %s %s" % (fr.status, fr.loc, fr.alert()))
def check_deactivated_i18n(base):
"""停用成功页四语言渲染（用语言无关的 ratio:3 标记 + 标题非空）。"""
bad = []
for lang in ["zh-CN", "zh-TW", "en", "ja"]:
c = Client(base)
c.get("/i18n/%s?to=/deactivated" % lang, timeout=40) # 302，语言写入 session
r = c.get("/deactivated", timeout=40) # 再请求一次才渲染新语言
title = re.search(r'uk-card-body uk-text-center".*?<h3[^>]*>([^<]*)</h3>',
r.body, re.S)
if not title or not title.group(1).strip():
bad.append(lang)
step("停用成功页四语言渲染", not bad, "缺失: " + ",".join(bad) if bad else "")
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--base", default="https://box-dev-api.337845818.best")
ap.add_argument("--reuse-email", default="",
help="只跑一项：用该既有邮箱重新注册（用于复测停用后能否重注册）")
args = ap.parse_args()
base = args.base.rstrip("/")
pwd = "TbCheck!2026"
if args.reuse_email:
domain = args.reuse_email.split("@")[0]
c = Client(base)
r = c.get("/register")
r = c.post("/register", {
"email": args.reuse_email, "domain": domain, "name": "TBCheck",
"password": pwd, "repeat_password": pwd,
"g-recaptcha-response": captcha(c), "language": "zh-CN", "agree_rules": "1",
"_csrf": csrf_of(r.body),
})
step("以既有邮箱重新注册（已停用则应被明确拒绝并提示找站长）", is_deactivated_rejected(r),
"HTTP %s %s" % (r.status, r.alert()))
return 0 if all(x[1] for x in RESULTS) else 1
ts = time.strftime("%m%d%H%M%S")
email, domain = "tbe2e%s@example.com" % ts, "tbe2e%s" % ts
print("== 新账号：%s / %s ==" % (email, domain))
pub = Client(base)
r = pub.get("/register")
reg = pub.post("/register", {
"email": email, "domain": domain, "name": "TB%s" % ts,
"password": pwd, "repeat_password": pwd,
"g-recaptcha-response": captcha(pub), "language": "zh-CN", "agree_rules": "1",
"_csrf": csrf_of(r.body),
})
step("注册", reg.status == 302, "HTTP %s -> %s %s" % (reg.status, reg.loc, reg.alert()))
owner = Client(base)
r = owner.get("/login")
r = owner.post("/login", {
"email": email, "password": pwd,
"g-recaptcha-response": captcha(owner), "_csrf": csrf_of(r.body),
})
step("登录", r.status == 302, "HTTP %s -> %s" % (r.status, r.loc))
step("登录态 /user/profile", owner.get("/user/profile").status == 200)
step("登录态 /user/questions", owner.get("/user/questions").status == 200)
r = owner.get("/user/votes")
step("/user/votes 302 -> /user/questions#votes",
r.status == 302 and "questions" in r.loc, "HTTP %s -> %s" % (r.status, r.loc))
step("新建投票页", owner.get("/user/votes/new").status == 200)
check_user_assets(owner)
check_vote_flow(owner, base)
# 匿名提问：无邮箱时成功后直接 200 渲染提问成功页（不重定向），故只断言提问详情可达
r = pub.get("/_/%s" % domain)
step("打开提问箱", r.status == 200)
r = pub.post("/_/%s" % domain, {
"content": "自动化巡检测试提问（可忽略）",
"g-recaptcha-response": captcha(pub),
"receive_reply_email": "", "private": "", "ask_password": "",
"start_chat_val": "", "_csrf": csrf_of(r.body),
})
step("提交提问", r.status in (200, 302), "HTTP %s -> %s" % (r.status, r.loc))
r = owner.get("/user/profile/deactivate")
step("打开停用页", r.status == 200)
r = owner.post("/user/profile/deactivate", {"_csrf": csrf_of(r.body)})
step("停用后跳 /deactivated", r.status == 302 and "deactivated" in r.loc,
"HTTP %s -> %s" % (r.status, r.loc))
r = owner.get("/deactivated")
step("停用成功页渲染", r.status == 200 and "ratio: 3" in r.body, "HTTP %s" % r.status)
pub2 = Client(base)
r = pub2.get("/register")
r = pub2.post("/register", {
"email": email, "domain": domain, "name": "TB%sb" % ts,
"password": pwd, "repeat_password": pwd,
"g-recaptcha-response": captcha(pub2), "language": "zh-CN", "agree_rules": "1",
"_csrf": csrf_of(r.body),
})
step("停用后同邮箱重注册被明确拒绝并提示找站长（核心回归）", is_deactivated_rejected(r),
"HTTP %s %s" % (r.status, r.alert()))
check_public_pages(base)
check_deactivated_i18n(base)
print("\n===== 汇总 =====")
for d, ok, ex in RESULTS:
print("%s %s%s" % ("PASS" if ok else "FAIL", d, (" (%s)" % ex) if ex else ""))
n = sum(1 for x in RESULTS if not x[1])
print("\n共 %d 项，失败 %d 项" % (len(RESULTS), n))
return 1 if n else 0
if __name__ == "__main__":
try:
sys.exit(main())
except KeyboardInterrupt:
raise
except Exception as e: # noqa: BLE001
# 远端抖动等意外中断时，仍把已完成项的结论打出来，避免只剩一段 traceback
print("\n巡检意外中断：%s: %s" % (type(e).__name__, e))
for d, ok, ex in RESULTS:
print("%s %s%s" % ("PASS" if ok else "FAIL", d, (" (%s)" % ex) if ex else ""))
sys.exit(2)
dev/devcheck-image.py
-370
View File
@@ -1,370 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""图片上传链路端到端验证（批次 A / B1 / B2）。
针对已部署环境验证三件事：
B1 EXIF Orientation=6 的竖拍 JPEG 转换后必须变成竖图（核心验收点）
B2 动画 GIF 必须被转成 .webp（cwebp 可用性 + 提问路径与头像路径一致性）
A 转换后不比原图小时必须回退原图（用一张极小线框图验证）
做法：注册账号 -> 匿名提问（multipart 同时带三张图）-> 解析成功页里的 CDN URL
-> 逐张拉回字节，解析真实容器格式与尺寸。
用法：
python dev/devcheck-image.py [--base https://box-dev-api.337845818.best]
依赖：Pillow（仅用于在本地生成测试图片，不参与被测逻辑）。
注意：会在目标环境真实创建账号并发送带图提问（测试邮箱前缀 tbimg<ts>@）。
清理：TamaBox ban --email <该邮箱> --yes
"""
import argparse
import http.cookiejar
import io
import json
import os
import tempfile
import re
import struct
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36")
RESULTS = []
CREATED_EMAIL = ""
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102
return None
class Resp:
def __init__(self, status, headers, body):
self.status = status
self.h = {k.lower(): v for k, v in headers.items()}
self.body = body
@property
def loc(self):
return self.h.get("location", "")
@property
def text(self):
return self.body.decode("utf-8", "replace")
class Client:
def __init__(self, base):
self.base = base
self.cj = http.cookiejar.CookieJar()
self.op = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(self.cj), NoRedirect())
def _send(self, req, timeout=120):
try:
with self.op.open(req, timeout=timeout) as r:
return Resp(r.status, dict(r.headers), r.read())
except urllib.error.HTTPError as e:
try:
body = e.read()
except Exception: # noqa: BLE001
body = b""
return Resp(e.code, dict(e.headers), body)
except Exception as e: # noqa: BLE001
return Resp(-1, {}, ("TRANSPORT_ERROR: %s" % e).encode())
def req(self, method, path, data=None, content_type=None, timeout=120):
url = path if path.startswith("http") else self.base + path
headers = {"User-Agent": UA, "Referer": self.base + path.split("?")[0]}
body = None
if data is not None:
body = data if isinstance(data, bytes) else urllib.parse.urlencode(data).encode()
headers["Content-Type"] = content_type or "application/x-www-form-urlencoded"
return self._send(urllib.request.Request(url, data=body, headers=headers, method=method),
timeout)
def get(self, p, **kw):
r = self.req("GET", p, **kw)
return r if r.status != -1 else self.req("GET", p, **kw)
def post(self, p, data=None, **kw):
return self.req("POST", p, data=data, **kw)
def get_binary(self, url):
return self._send(urllib.request.Request(
url, headers={"User-Agent": UA}, method="GET"), timeout=60)
def step(desc, ok, extra=""):
RESULTS.append((desc, ok, extra))
print(" [%s] %s%s" % ("PASS" if ok else "FAIL", desc, (" (%s)" % extra) if extra else ""))
return ok
def csrf_of(html):
m = re.search(r'name="_csrf"\s+value="([^"]+)"', html)
return m.group(1) if m else None
def captcha(c):
j = json.loads(c.get("/api/captcha/generate").text)
pid, q = j["data"]["id"], j["data"]["question"]
m = re.match(r"\s*(\d+)\s*([+\-×])\s*(\d+)\s*=", q)
a, op, b = int(m.group(1)), m.group(2), int(m.group(3))
return "%s:%d" % (pid, {"+": a + b, "-": a - b, "×": a * b}[op])
def multipart(fields, files):
boundary = "----tbimg%s" % uuid.uuid4().hex
buf = io.BytesIO()
for k, v in fields.items():
buf.write(("--%s\r\n" % boundary).encode())
buf.write(('Content-Disposition: form-data; name="%s"\r\n\r\n' % k).encode())
buf.write(("%s\r\n" % v).encode())
for name, fn, ct, data in files:
buf.write(("--%s\r\n" % boundary).encode())
buf.write(('Content-Disposition: form-data; name="%s"; filename="%s"\r\n' % (name, fn)).encode())
buf.write(("Content-Type: %s\r\n\r\n" % ct).encode())
buf.write(data)
buf.write(b"\r\n")
buf.write(("--%s--\r\n" % boundary).encode())
return buf.getvalue(), "multipart/form-data; boundary=%s" % boundary
# ---------- 容器格式解析（纯 stdlib） ----------
def sniff(data):
"""返回 (kind, (w,h))。kind in {webp, jpeg, gif, png, unknown}"""
if len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP":
return "webp", webp_dims(data)
if data[:3] == b"\xff\xd8\xff":
return "jpeg", jpeg_dims(data)
if data[:3] == b"GIF":
return "gif", gif_dims(data)
if data[:8] == b"\x89PNG\r\n\x1a\n":
return "png", png_dims(data)
return "unknown", None
def webp_dims(d):
c = d[12:16]
if c == b"VP8 " and len(d) >= 30:
return (struct.unpack("<H", d[26:28])[0] & 0x3FFF,
struct.unpack("<H", d[28:30])[0] & 0x3FFF)
if c == b"VP8L" and len(d) >= 25:
b = struct.unpack("<I", d[21:25])[0]
return (b & 0x3FFF) + 1, ((b >> 14) & 0x3FFF) + 1
if c == b"VP8X" and len(d) >= 30:
return ((d[24] | d[25] << 8 | d[26] << 16) + 1,
(d[27] | d[28] << 8 | d[29] << 16) + 1)
return None
def jpeg_dims(d):
i = 2
while i + 9 < len(d):
if d[i] != 0xFF:
i += 1
continue
m = d[i + 1]
if m in (0xC0, 0xC1, 0xC2, 0xC3):
h, w = struct.unpack(">HH", d[i + 5:i + 9])
return (w, h)
if m in (0xD8, 0xD9) or 0xD0 <= m <= 0xD7:
i += 2
continue
i += 2 + struct.unpack(">H", d[i + 2:i + 4])[0]
return None
def gif_dims(d):
return struct.unpack("<HH", d[6:10]) if len(d) >= 10 else None
def png_dims(d):
return struct.unpack(">II", d[16:24]) if len(d) >= 24 else None
# ---------- 测试图片生成（Pillow 仅此一处使用） ----------
def make_images():
from PIL import Image, ImageDraw
import io as _io
# B1：EXIF Orientation=6。像素存 240x120，正确归一后应显示 120x240。
base = Image.new("RGB", (240, 120))
dr = ImageDraw.Draw(base)
for i in range(0, 240, 20):
dr.line([(i, 0), (i, 120)], fill=(200, 30, 30), width=3)
dr.rectangle([0, 0, 239, 119], outline=(0, 0, 255), width=4)
exif = base.getexif()
exif[0x0112] = 6
buf = _io.BytesIO()
base.save(buf, "JPEG", quality=88, exif=exif)
portrait = buf.getvalue()
# B2：6 帧动画 GIF。加大到 640x480，确保转换后仍小于原图（否则会走回退）。
frames = []
pal_frames = []
for f in range(6):
im = Image.new("RGB", (640, 480), (240, 240, 240))
d = ImageDraw.Draw(im)
d.ellipse([40 + f * 60, 120, 200 + f * 60, 280], fill=(220, 40, 40))
d.rectangle([300, 60 + f * 20, 600, 200], fill=(30, 90, 220))
pal_frames.append(im.convert("P", palette=Image.ADAPTIVE, colors=64))
gbuf = _io.BytesIO()
pal_frames[0].save(gbuf, "GIF", save_all=True, append_images=pal_frames[1:],
duration=120, loop=0)
animated = gbuf.getvalue()
# A：极小线框图（200x200），WebP 有损编码后反而比 PNG 大 -> 应回退原图。
wire = Image.new("RGB", (200, 200), (255, 255, 255))
d = ImageDraw.Draw(wire)
for i in range(0, 200, 20):
d.line([(i, 0), (i, 199)], fill=(0, 0, 0))
d.line([(0, i), (199, i)], fill=(0, 0, 0))
wbuf = _io.BytesIO()
wire.save(wbuf, "PNG")
wireframe = wbuf.getvalue()
return {"portrait": portrait, "anim": animated, "wire": wireframe}
def main():
global CREATED_EMAIL
ap = argparse.ArgumentParser()
ap.add_argument("--base", default="https://box-dev-api.337845818.best")
args = ap.parse_args()
base = args.base.rstrip("/")
pwd = "TbImg!2026"
ts = time.strftime("%m%d%H%M%S")
email, domain = "tbimg%s@example.com" % ts, "tbimg%s" % ts
CREATED_EMAIL = email
print("== 账号：%s / %s ==" % (email, domain))
imgs = make_images()
for k, v in imgs.items():
print(" 本地样本 %-9s %7d bytes" % (k, len(v)))
c = Client(base)
r = c.get("/register")
reg = c.post("/register", {
"email": email, "domain": domain, "name": "TBImg%s" % ts,
"password": pwd, "repeat_password": pwd,
"g-recaptcha-response": captcha(c), "language": "zh-CN",
"_csrf": csrf_of(r.text),
})
if not step("注册", reg.status == 302, "HTTP %s -> %s" % (reg.status, reg.loc)):
return 1
c = Client(base)
r = c.get("/login")
lg = c.post("/login", {"email": email, "password": pwd,
"g-recaptcha-response": captcha(c), "_csrf": csrf_of(r.text)})
if not step("登录", lg.status == 302, "HTTP %s -> %s" % (lg.status, lg.loc)):
return 1
page = c.get("/_/%s" % domain)
if not step("打开提问箱", page.status == 200, "HTTP %s" % page.status):
return 1
# route/question/page.go 只取 f.Images[0]（单张上限），
# 故每张样本各发一条提问，逐张验证。
fetched = []
for key, (fname, ctype) in (("portrait", ("portrait.jpg", "image/jpeg")),
("anim", ("anim.gif", "image/gif")),
("wire", ("wire.png", "image/png"))):
box = c.get("/_/%s" % domain)
body, ct = multipart({
"content": "图片上传链路验证 %s（可忽略）" % key,
"g-recaptcha-response": captcha(c),
"receive_reply_email": "", "private": "", "ask_password": "",
"start_chat_val": "", "_csrf": csrf_of(box.text),
}, [("images", fname, ctype, imgs[key])])
res = c.post("/_/%s" % domain, data=body, content_type=ct)
if not step("提交带图提问 [%s]" % key, res.status in (200, 302),
"HTTP %s -> %s" % (res.status, res.loc)):
continue
# 成功页不渲染图片（templates/question/success.html 无图片区），
# 故从成功页取出 question ID，再访问详情页取图片 key。
qid = re.search(r'/_/%s/(\d+)' % re.escape(domain), res.text)
if not qid:
step(" └ 取提问 ID [%s]" % key, False, "成功页未给出 /_/<domain>/<id>")
continue
detail = c.get("/_/%s/%s" % (domain, qid.group(1)))
# 提问图的 key 不带 picture/ 前缀（是 "<年>/<月>/<纳秒>.<ext>"），
# 模板拼成 https://<CDN>/<key>，故按 CDN host + 数字开头路径来匹配。
urls = sorted(set(re.findall(
r'https://r2\.[^"\'\s<>]+/\d{4}/\d{1,2}/\d+\.\w+', detail.text)))
if not urls:
step(" └ 详情页回显图片 [%s]" % key, False,
"HTTP %s，未找到 picture URL" % detail.status)
continue
step(" └ 详情页回显图片 [%s]" % key, True, "%d 个" % len(urls))
for u in urls:
st, data = c.get_binary(u)
if st != 200 or not data:
step(" └ 拉回图片 [%s]" % key, False, "HTTP %s" % st)
continue
kind, dims = sniff(data)
fetched.append((key, kind, dims, len(data)))
print(" %-14s %-7s %-12s %7d bytes" % (u.rsplit("/", 1)[-1], kind, dims, len(data)))
print("\n== 断言 ==")
# B1
portrait = [f for f in fetched if f[0] == "portrait" and f[1] in ("webp", "jpeg")
and f[2] == (120, 240)]
land = [f for f in fetched if f[0] == "portrait" and f[1] in ("webp", "jpeg")
and f[2] == (240, 120)]
if portrait:
step("B1 EXIF Orientation=6 已归一（240x120 -> 120x240）", True,
"%s, %d bytes" % (portrait[0][1], portrait[0][3]))
elif land:
step("B1 EXIF Orientation=6 已归一", False, "仍是 240x120，说明未应用 EXIF 旋转")
else:
step("B1 EXIF 方向归一", False, "未匹配到 240x120 / 120x240 的图片，实际: %s" %
[(f[1], f[2]) for f in fetched])
# B2：GIF 样本必须变成 webp；关键是按 key 归属判定，而不是看整体有没有 webp
gif_res = [f for f in fetched if f[0] == "anim"]
if not gif_res:
step("B2 动画 GIF 转 WebP", False, "未取到 anim 样本结果")
elif gif_res[0][1] == "webp":
step("B2 动画 GIF 已转为 WebP", True,
"%dx%d, %d bytes（原 GIF %d bytes，-%.0f%%）" % (
gif_res[0][2][0], gif_res[0][2][1], gif_res[0][3], len(imgs["anim"]),
(1 - gif_res[0][3] / len(imgs["anim"])) * 100))
elif gif_res[0][1] == "gif":
step("B2 动画 GIF 已转为 WebP", False,
"仍是原始 GIF %d bytes -> cwebp 不可用或转换失败" % gif_res[0][3])
else:
step("B2 动画 GIF 转 WebP", False, "得到 %s（预期 webp）" % gif_res[0][1])
# A：线框图应回退原图（PNG），而非转成更大的 WebP
wire = [f for f in fetched if f[0] == "wire"]
if not wire:
step("A 线框图兜底回退", False, "未取到 wire 样本结果")
elif wire[0][1] == "png":
step("A 转换无收益时回退原图（线框图保持 PNG）", True,
"%d bytes（原 PNG %d bytes）" % (wire[0][3], len(imgs["wire"])))
else:
step("A 线框图兜底回退", False,
"被转成了 %s（%d bytes）——本应回退原图" % (wire[0][1], wire[0][3]))
print("\n== 结果：%d/%d 通过 ==" % (sum(1 for _, ok, _ in RESULTS if ok), len(RESULTS)))
print("清理：TamaBox ban --email %s --yes" % email)
return 0 if all(ok for _, ok, _ in RESULTS) else 1
if __name__ == "__main__":
sys.exit(main())
dev/devcheck-upload.py
-419
View File
@@ -1,419 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""上传链路端到端验证（A 分档 / B1 EXIF 归一 + EXIF 比例校验 / B3 单次解码）。
为什么主用例走「背景」而不是头像
--------------------------------
头像要求 1:1，是**对称**比例：EXIF 归一（5/6/7/8 交换宽高）后比例不变。
所以无论修没修比例校验的 EXIF 缺口，头像路径的表现完全一样 —— **测不出缺口**。
背景要求 16:9，非对称，且用户真实场景就是「手机竖拍一张照当背景」：
像素 1080x1920（竖，比例 0.5625）+ EXIF Orientation=6（顺时针 90）
-> 用户肉眼看到的是 1920x1080 横图，比例 1.7778，符合 16:9
-> 修复前：只看原始像素 0.5625 vs 1.7778，偏差 1.215 > 0.01 → **被误拒**
-> 修复后：先交换宽高得 1920x1080，偏差 0 → 放行
头像用例保留，但只用来回归 A 分档（不再承担 EXIF 比例校验的验证职责）。
用法：
python dev/devcheck-upload.py [--base https://box-dev-api.337845818.best]
依赖：Pillow（仅用于本地生成测试图片）
清理：TamaBox ban --email <输出中的测试邮箱> --yes
"""
import argparse
import http.cookiejar
import io
import json
import re
import struct
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from PIL import Image, ImageOps
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36")
RESULTS = []
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102
return None
class Resp:
def __init__(self, status, headers, body):
self.status = status
self.h = {k.lower(): v for k, v in headers.items()}
self.body = body
@property
def loc(self):
return self.h.get("location", "")
@property
def text(self):
return self.body.decode("utf-8", "replace")
class Client:
def __init__(self, base):
self.base = base
self.cj = http.cookiejar.CookieJar()
self.op = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(self.cj), NoRedirect())
def _send(self, req, timeout=180):
try:
with self.op.open(req, timeout=timeout) as r:
return Resp(r.status, dict(r.headers), r.read())
except urllib.error.HTTPError as e:
try:
body = e.read()
except Exception: # noqa: BLE001
body = b""
return Resp(e.code, dict(e.headers), body)
except Exception as e: # noqa: BLE001
return Resp(-1, {}, ("TRANSPORT_ERROR: %s" % e).encode())
def req(self, method, path, data=None, content_type=None, timeout=180):
url = path if path.startswith("http") else self.base + path
headers = {"User-Agent": UA, "Referer": self.base + path.split("?")[0]}
body = None
if data is not None:
body = data if isinstance(data, bytes) else urllib.parse.urlencode(data).encode()
headers["Content-Type"] = content_type or "application/x-www-form-urlencoded"
return self._send(urllib.request.Request(url, data=body, headers=headers, method=method),
timeout)
def get(self, p, **kw):
r = self.req("GET", p, **kw)
return r if r.status != -1 else self.req("GET", p, **kw)
def post(self, p, data=None, **kw):
return self.req("POST", p, data=data, **kw)
def get_binary(self, url):
return self._send(urllib.request.Request(
url, headers={"User-Agent": UA}, method="GET"), timeout=60)
def step(desc, ok, extra=""):
RESULTS.append((desc, ok, extra))
print(" [%s] %s%s" % ("PASS" if ok else "FAIL", desc, (" (%s)" % extra) if extra else ""))
return ok
def redness_bbox(im):
"""返回红色标记条的包围盒 (x0, y0, x1, y1)。
用来做「方向是否真的被旋转」的判据：尺寸对不代表方向对，必须看特征位置。
有损 WebP 会让红色略有偏差，故阈值放宽并按 4px 步长采样。
"""
W, H = im.size
px = im.load()
xs, ys = [], []
for y in range(0, H, 4):
for x in range(0, W, 4):
r, g, b = px[x, y]
if r > 150 and g < 90 and b < 90:
xs.append(x)
ys.append(y)
return (min(xs), min(ys), max(xs), max(ys)) if xs else None
def csrf_of(html):
m = re.search(r'name="_csrf"\s+value="([^"]+)"', html)
return m.group(1) if m else None
def captcha(c):
j = json.loads(c.get("/api/captcha/generate").text)
pid, q = j["data"]["id"], j["data"]["question"]
m = re.match(r"\s*(\d+)\s*([+\-×])\s*(\d+)\s*=", q)
a, op, b = int(m.group(1)), m.group(2), int(m.group(3))
return "%s:%d" % (pid, {"+": a + b, "-": a - b, "×": a * b}[op])
def multipart(fields, files):
boundary = "----tbuv%s" % uuid.uuid4().hex
buf = io.BytesIO()
for k, v in fields.items():
buf.write(("--%s\r\n" % boundary).encode())
buf.write(('Content-Disposition: form-data; name="%s"\r\n\r\n' % k).encode())
buf.write(("%s\r\n" % v).encode())
for name, fn, ct, data in files:
buf.write(("--%s\r\n" % boundary).encode())
buf.write(('Content-Disposition: form-data; name="%s"; filename="%s"\r\n' % (name, fn)).encode())
buf.write(("Content-Type: %s\r\n\r\n" % ct).encode())
buf.write(data)
buf.write(b"\r\n")
buf.write(("--%s--\r\n" % boundary).encode())
return buf.getvalue(), "multipart/form-data; boundary=%s" % boundary
def sniff(d):
if len(d) >= 12 and d[:4] == b"RIFF" and d[8:12] == b"WEBP":
return "webp", webp_dims(d)
if d[:3] == b"\xff\xd8\xff":
return "jpeg", jpeg_dims(d)
if d[:8] == b"\x89PNG\r\n\x1a\n":
return "png", (struct.unpack(">II", d[16:24]) if len(d) >= 24 else None)
return "unknown", None
def webp_dims(d):
c = d[12:16]
if c == b"VP8 " and len(d) >= 30:
return (struct.unpack("<H", d[26:28])[0] & 0x3FFF,
struct.unpack("<H", d[28:30])[0] & 0x3FFF)
if c == b"VP8L" and len(d) >= 25:
b = struct.unpack("<I", d[21:25])[0]
return (b & 0x3FFF) + 1, ((b >> 14) & 0x3FFF) + 1
if c == b"VP8X" and len(d) >= 30:
return ((d[24] | d[25] << 8 | d[26] << 16) + 1,
(d[27] | d[28] << 8 | d[29] << 16) + 1)
return None
def jpeg_dims(d):
i = 2
while i + 9 < len(d):
if d[i] != 0xFF:
i += 1
continue
m = d[i + 1]
if m in (0xC0, 0xC1, 0xC2, 0xC3):
h, w = struct.unpack(">HH", d[i + 5:i + 9])
return (w, h)
if m in (0xD8, 0xD9) or 0xD0 <= m <= 0xD7:
i += 2
continue
i += 2 + struct.unpack(">H", d[i + 2:i + 4])[0]
return None
def make_img(w, h, orientation=None, portrait=False):
"""生成带方位特征的图片。
portrait=True 时在**左侧**画一条粗红色竖条：EXIF 顺时针 90 后它会跑到顶部，
用于像素级确认方向确实被归一了（而非仅尺寸数字对）。
"""
from PIL import ImageDraw
im = Image.new("RGB", (w, h), (30, 120, 200))
d = ImageDraw.Draw(im)
if portrait:
d.rectangle([0, 0, max(2, w // 12), h], fill=(230, 30, 30))
d.text((max(4, w // 12 + 6), h // 2), "TOP-IS-LEFT", fill=(255, 255, 255))
else:
d.rectangle([0, 0, w, max(2, h // 12)], fill=(230, 30, 30))
d.text((w // 4, max(4, h // 12 + 4)), "TOP-IS-UP", fill=(255, 255, 255))
for i in range(0, w, max(8, w // 20)):
d.line([(i, h // 4), (i, h)], fill=(255, 255, 255), width=2)
buf = io.BytesIO()
if orientation:
exif = im.getexif()
exif[0x0112] = orientation
im.save(buf, "JPEG", quality=90, exif=exif)
else:
im.save(buf, "JPEG", quality=90)
return buf.getvalue()
def scrape_profile_fields(html):
"""从 profile 页抓取所有 input 的 name/value，保证回填必填项。
UpdateProfile 的必填字段有十来个（昵称/介绍/5 个二维码颜色/点大小/透明度…），
漏一个就会因 `valid:"required"` 被Bind 挡下、图片根本不会走到上传分支。
"""
fields = {}
for m in re.finditer(r'<input\b[^>]*>', html, re.I):
tag = m.group(0)
nm = re.search(r'name="([^"]+)"', tag, re.I)
if not nm:
continue
name = nm.group(1)
if name in ("avatar", "background", "backgroundimage", "_csrf", "submit"):
continue
typ = re.search(r'type="([^"]+)"', tag, re.I)
if typ and typ.group(1).lower() in ("checkbox", "radio"):
if not re.search(r'\bchecked\b', tag, re.I):
continue
val = re.search(r'value="([^"]*)"', tag, re.I)
fields[name] = val.group(1) if val else ""
for m in re.finditer(r'<select\b[^>]*name="([^"]+)"[^>]*>(.*?)</select>', html, re.I | re.S):
name, body = m.group(1), m.group(2)
opts = re.findall(r'<option\b([^>]*)>', body, re.I)
chosen = ""
for o in opts:
vm = re.search(r'value="([^"]*)"', o, re.I)
v = vm.group(1) if vm else ""
if re.search(r'\bselected\b', o, re.I):
chosen = v
break
if not chosen:
chosen = v
fields[name] = chosen
for m in re.finditer(r'<textarea\b[^>]*name="([^"]+)"[^>]*>(.*?)</textarea>', html, re.I | re.S):
fields[m.group(1)] = re.sub(r"<[^>]+>", "", m.group(2)).strip()
return fields
def upload(c, field, data, fname="img.jpg"):
"""上传图片到 profile 表单的指定字段（avatar / background）。
返回 (图片CDN URL 或 None, 错误描述或 None)。
"""
page = c.get("/user/profile")
if page.status != 200:
return None, "profile HTTP %s" % page.status
fields = scrape_profile_fields(page.text)
fields["_csrf"] = csrf_of(page.text)
body, ct = multipart(fields, [(field, fname, "image/jpeg", data)])
res = c.post("/user/profile/update", data=body, content_type=ct)
if res.status == 302 and res.loc:
res = c.get(res.loc if res.loc.startswith("/") else "/user/profile")
if res.status != 200:
return None, "update 后页面 HTTP %s" % res.status
if field == "background":
m = re.search(r'id="bg-prev"[^>]*src="([^"]+)"', res.text) or \
re.search(r'src="([^"]+)"[^>]*id="bg-prev"', res.text)
url = m.group(1) if m else None
else:
urls = re.findall(r'https://r2\.[^"\'\s<>]+/picture/[0-9]+/\d{2}/\d{2}/[0-9a-f]+', res.text)
if not urls:
m = re.search(r'(https://r2\.[^"\'\s<>]+/picture/[^"\'\s<>]+)', res.text)
urls = [m.group(1)] if m else []
url = urls[-1] if urls else None
if url:
return url, None
dump = "C:/Users/Tamakyi Gekka/AppData/Local/Temp/tbuv_profile.html"
with open(dump, "wb") as f:
f.write(res.body)
alerts = re.findall(r'class="uk-alert[^"]*"[^>]*>(.*?)</div>', res.text, re.S)
hint = " | ".join(re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", a)).strip()[:80]
for a in alerts[:3])
return None, "未取到图片 URL（已存 %s）%s%s" % (dump, (" 提示:" + hint) if hint else "",
(" HTTP %s" % res.status))
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--base", default="https://box-dev-api.337845818.best")
args = ap.parse_args()
base = args.base.rstrip("/")
pwd = "TbUv!2026"
ts = time.strftime("%m%d%H%M%S")
email, domain = "tbuv%s@example.com" % ts, "tbuv%s" % ts
print("== 账号：%s / %s ==" % (email, domain))
# ---- 本地样本 ----
square = make_img(1600, 1600) # A 分档回归：头像
bg_ok = make_img(1920, 1080) # 背景 16:9，无 EXIF（正向对照）
bg_ori6 = make_img(1080, 1920, orientation=6, portrait=True) # 核心用例
bg_bad = make_img(1080, 1920) # 无 EXIF 的竖图：应被拒（16:9 校验仍生效）
print(" 样本 avatar 1600x1600 %7d B" % len(square))
print(" 样本 bg-ok 1920x1080 %7d B" % len(bg_ok))
print(" 样本 bg-ori6 1080x1920 (EXIF=6) %7d B" % len(bg_ori6))
print(" 样本 bg-bad 1080x1920 (无 EXIF) %7d B" % len(bg_bad))
c = Client(base)
r = c.get("/register")
reg = c.post("/register", {
"email": email, "domain": domain, "name": "TBUv%s" % ts,
"password": pwd, "repeat_password": pwd,
"g-recaptcha-response": captcha(c), "language": "zh-CN",
"_csrf": csrf_of(r.text),
})
if not step("注册", reg.status == 302, "HTTP %s -> %s" % (reg.status, reg.loc)):
return 1
c = Client(base)
r = c.get("/login")
lg = c.post("/login", {"email": email, "password": pwd,
"g-recaptcha-response": captcha(c), "_csrf": csrf_of(r.text)})
if not step("登录", lg.status == 302, "HTTP %s -> %s" % (lg.status, lg.loc)):
return 1
# ================= A 分档（头像回归） =================
print("\n-- A 分档（头像 512/q78）--")
url, err = upload(c, "avatar", square)
if not step("上传方形大头像（1600x1600）", url is not None, err or (url or "")):
return 1
img = c.get_binary(url)
if not step("拉回头像", img.status == 200 and bool(img.body), "HTTP %s" % img.status):
return 1
kind, dims = sniff(img.body)
print(" 存储结果：%s %s %d bytes" % (kind, dims, len(img.body)))
step("A 分档生效：大图被缩到头像档 512", dims == (512, 512), "实际 %s" % (dims,))
# ================= B1 背景正向对照（无 EXIF） =================
print("\n-- 背景 16:9 正向对照（无 EXIF）--")
url_bg, err = upload(c, "background", bg_ok)
if not step("上传 16:9 背景（1920x1080，无 EXIF）", url_bg is not None, err or (url_bg or "")):
return 1
r1 = c.get_binary(url_bg)
k1, d1 = sniff(r1.body)
print(" 存储结果：%s %s %d bytes" % (k1, d1, len(r1.body)))
step("背景档 1920 且比例正确", k1 == "webp" and d1 == (1920, 1080), "实际 %s %s" % (k1, d1))
# ================= B1 核心：EXIF Orientation=6 的竖拍背景 =================
print("\n-- B1 核心：EXIF 比例校验缺口（手机竖拍当背景）--")
print(" 像素 1080x1920 + EXIF=6 ->肉眼看到 1920x1080 (16:9)")
print(" 修复前：按像素 0.5625 判比例 -> 偏差 1.215 >> 0.01 -> 被误拒")
url_o, err = upload(c, "background", bg_ori6)
if not step("上传 EXIF 竖拍背景（1080x1920, Orientation=6）", url_o is not None,
err or (url_o or "")):
print(" >>> 若此处 FAIL，说明 EXIF 比例校验缺口仍未修复")
return 1
if url_o == url_bg:
step("EXIF 背景确实换了新图", False, "两次上传返回同一 URL（第二次被拒、沿用旧值）")
return 1
r2 = c.get_binary(url_o)
k2, d2 = sniff(r2.body)
print(" 存储结果：%s %s %d bytes" % (k2, d2, len(r2.body)))
step("B1 EXIF 归一后按16:9 存盘（1920x1080）", k2 == "webp" and d2 == (1920, 1080),
"实际 %s %s" % (k2, d2))
step("B1 归一后限边生效（长边 1920）",
d2 is not None and max(d2) == 1920, "实际 %s" % (d2,))
# 尺寸对了不代表方向对了 —— 与 Pillow exif_transpose 作像素级黄金标准比对：
# 定位红色条的包围盒，两边必须一致（原图红条在左侧，顺时针 90 后应跑到顶部）。
ref = ImageOps.exif_transpose(Image.open(io.BytesIO(bg_ori6))).convert("RGB")
got = Image.open(io.BytesIO(r2.body)).convert("RGB")
if ref.size == got.size:
bb_ref, bb_got = redness_bbox(ref), redness_bbox(got)
step("B1 像素级方向与 Pillow exif_transpose 一致（红条包围盒）",
bb_ref is not None and bb_ref == bb_got,
"黄金标准 %s vs 存盘 %s" % (bb_ref, bb_got))
else:
step("B1 像素级方向与 Pillow exif_transpose 一致（红条包围盒）", False,
"尺寸不同：黄金标准 %s vs 存盘 %s" % (ref.size, got.size))
# ================= 反向用例：EXIF 修复不能放松比例校验 =================
print("\n-- 反向用例：无 EXIF 的竖图仍应被拒（确认没把校验放得过宽）--")
url_bad, err_bad = upload(c, "background", bg_bad)
# 期望：上传被拒 -> bg-prev 仍是上一步的 ori6 图（url_o）
if url_bad == url_o:
step("无 EXIF 的 9:16 竖图被 16:9 校验拒绝", True, "bg-prev 保持 ori6 那张")
elif url_bad is None:
step("无 EXIF 的 9:16 竖图被 16:9 校验拒绝", True, "上传被拒（%s）" % (err_bad or "")[:60])
else:
step("无 EXIF 的 9:16 竖图被 16:9 校验拒绝", False,
"竟被接受并存成 %s" % (url_bad,))
print("\n== 结果：%d/%d 通过 ==" % (sum(1 for _, ok, _ in RESULTS if ok), len(RESULTS)))
for d, ok, e in RESULTS:
if not ok:
print(" FAIL: %s %s" % (d, e))
print("清理：TamaBox ban --email %s --yes" % email)
return 0 if all(ok for _, ok, _ in RESULTS) else 1
if __name__ == "__main__":
sys.exit(main())