restrict v1

This commit is contained in:
tamakyi
2026-10-18 21:14:59 +08:00
parent da7e76a7db
commit 6bd90779f8
23 changed files with 1754 additions and 1 deletions
+129
View File
@@ -0,0 +1,129 @@
#!/usr/bin/env bash
# 处置链路部署后自检(L1 停止分发 / L4 整页停用 / CLI 可用性)
#
# 为什么需要它:本机沙箱起不了 web(无 redis),L1/L4 的 HTTP 行为只能在服务器上验证。
# 这个脚本把「施加处置 → 观察 HTTP → 自动撤销」串起来,避免手工操作忘了解除。
#
# 用法:
# BASE_URL=https://box-dev.example.com DOMAIN=<测试箱主域名> TAMABOX=/path/to/NekoBox \
# bash dev/devcheck-restrict.sh # 只打印将要执行的命令(dry-run)
#
# BASE_URL=... DOMAIN=... TAMABOX=... RUN=1 \
# bash dev/devcheck-restrict.sh # 真正执行,结束时自动撤销
#
# 注意:
# - DOMAIN 必须是**测试箱主**,不要用真实用户;脚本会临时施加 L4/L1 再撤销。
# - 站点默认语言可能是 ja,页面文案断言按日文/英文各查一次,避免误判「没生效」。
# - 只依赖 CLI 与 curl,不需要 psql 凭据。
set -uo pipefail
BASE_URL=${BASE_URL:?need BASE_URL, e.g. https://box-dev.example.com}
DOMAIN=${DOMAIN:?need DOMAIN, a TEST box domain only}
TAMABOX=${TAMABOX:-./NekoBox}
RUN=${RUN:-0}
DAYS=${DAYS:-1}
pass=0
fail=0
restricted=0
say() { printf '%s\n' "$*"; }
check() {
local name=$1 want=$2 got=$3
if [ "$want" = "$got" ]; then
say " PASS $name (=$got)"
pass=$((pass + 1))
else
say " FAIL $name (want $want, got $got)"
fail=$((fail + 1))
fi
}
code() { curl -s -o /dev/null -w '%{http_code}' --max-time 10 "$1"; }
do_restrict() {
restricted=1
if [ "$RUN" = 1 ]; then
"$TAMABOX" restrict --domain "$DOMAIN" --level "$1" --days "$DAYS" --reason "deploy self-check" --yes
else
say " [dry-run] $TAMABOX restrict --domain $DOMAIN --level $1 --days $DAYS --reason 'deploy self-check' --yes"
fi
}
cleanup() {
if [ "$RUN" = 1 ] && [ "$restricted" = 1 ]; then
say ""
say "== 兜底撤销(无论中途是否失败)=="
"$TAMABOX" unrestrict --domain "$DOMAIN" || true
fi
}
trap cleanup EXIT
say "== 0. 二进制是否包含新命令 =="
if "$TAMABOX" --help 2>&1 | grep -q 'restrict'; then
say " PASS restrict 命令存在"
pass=$((pass + 1))
else
say " FAIL 二进制里没有 restrict —— 构建的是旧代码(注意入口是根目录 tamabox.go,或用 ./cmd/)"
fail=$((fail + 1))
say ""
say "结果: 无法继续,先修构建入口。"
exit 1
fi
say ""
say "== 1. 基线:处置前箱主页应可访问 =="
base=$(code "$BASE_URL/_/$DOMAIN")
check "box page before restrict" 200 "$base"
say ""
say "== 2. L4 整页停用 =="
do_restrict L4
sleep 1
check "box page under L4" 302 "$(code "$BASE_URL/_/$DOMAIN")"
check "restricted page" 200 "$(code "$BASE_URL/restricted")"
if [ "$RUN" = 1 ]; then
body=$(curl -sL --max-time 10 "$BASE_URL/_/$DOMAIN")
if printf '%s' "$body" | grep -qE '一時的に利用できません|temporarily unavailable|暫時無法存取|暂时不可访问'; then
say " PASS restricted page copy rendered"
pass=$((pass + 1))
else
say " FAIL restricted page copy not found (检查 i18n 与模板是否部署)"
fail=$((fail + 1))
fi
else
say " [dry-run] 跳过文案断言"
fi
say ""
say "== 3. L1 停止分发:页面恢复、分享卡被拦 =="
do_restrict L1
sleep 1
check "box page under L1" 200 "$(code "$BASE_URL/_/$DOMAIN")"
qid=$(curl -s --max-time 10 "$BASE_URL/_/$DOMAIN" | grep -o "/_/$DOMAIN/[0-9]\+" | head -1 | grep -o '[0-9]\+$')
if [ -n "${qid:-}" ]; then
check "share card under L1 (qid=$qid)" 302 "$(code "$BASE_URL/_/$DOMAIN/$qid/share-card")"
else
say " SKIP 该箱主没有已公开回答,无法验证分享卡(换一个有已答问题的测试箱主)"
fi
say ""
say "== 4. 撤销 =="
if [ "$RUN" = 1 ]; then
"$TAMABOX" unrestrict --domain "$DOMAIN"
restricted=0
check "box page after unrestrict" 200 "$(code "$BASE_URL/_/$DOMAIN")"
else
say " [dry-run] $TAMABOX unrestrict --domain $DOMAIN"
fi
say ""
say "== 结果: PASS=$pass FAIL=$fail =="
if [ "$fail" -gt 0 ]; then
say "有失败项,把上面输出 + 服务日志(journalctl / docker logs 里搜 'Failed to query user restriction')一起发我。"
exit 1
fi
say "全部通过。仍需人工手测两项:L2(回答应保存为私密)、L3(回答应被拒绝)。"
+419
View File
@@ -0,0 +1,419 @@
#!/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())