tamakyi
9c961b3c05
素材位置改成两选一（本地随二进制 / 自行填 Endpoint 的 S3 兼容存储）， R2、MinIO、B2、OSS-COS 的 S3 端点都只差 Endpoint/Region/路径风格。 - serve.py：手写 SigV4（PUT/GET/DELETE/HEAD），新增 /api/s3/test 与 /api/s3/upload。测试连接依次验连通、写权限、读回一致、匿名可读， 失败把 S3 的 <Code> 翻成可执行的建议；上传带 immutable 缓存头。 - index.html：新增「素材托管」面板（模式切换 + 7 个字段 + 测试/上传按钮）， 上传集合与导出共用 usedAssetIds()；成功后自动回填「素材前缀」。 密钥只进 sessionStorage 与内存，不进草稿/工程 json/导出 zip。 - 修：usedAssetIds 返回 Set，误用 length/map 导致「还没有素材」。 - 修：字段填完 700ms 内刷新会丢配置 → 失焦与模式切换时立即落盘。 - 自检：check_sigv4.py 与 aws-sdk-go-v2 signer 逐字节对拍（6/6）， selftest_upload.py 起假 S3 跑通测试连接→上传→读回（16 项）， mock_s3.py 供手动点面板用。
115 lines
4.6 KiB
Python
115 lines
4.6 KiB
Python
"""上传链路自测：起一个假 S3，跑一遍「测试连接 → 上传 → 读回」。
|
||
|
||
覆盖 serve.py 里真正容易错、又不能靠对拍发现的部分：
|
||
· 桶级 HEAD 与对象级 PUT/GET/DELETE 的路径拼接（path style / 虚拟主机）
|
||
· 前缀拼接、Content-Type、immutable 缓存头
|
||
· 4xx 时能否把 S3 的 <Code>/<Message> 还原成人话（含 hint）
|
||
签名数学由 check_sigv4.py 与官方 SDK 对拍保证，这里不重复。
|
||
|
||
用法（项目根）：
|
||
python dev/theme-builder/selftest_upload.py
|
||
"""
|
||
import base64
|
||
import os
|
||
import sys
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, HERE)
|
||
|
||
import serve # noqa: E402
|
||
from mock_s3 import BUCKET, MockS3 # noqa: E402
|
||
|
||
|
||
def spin():
|
||
import mock_s3
|
||
return mock_s3.spin()
|
||
|
||
|
||
FAILS = []
|
||
|
||
|
||
def check(desc, cond, extra=""):
|
||
print(("OK " if cond else "FAIL ") + desc + ((" ← " + extra) if extra and not cond else ""))
|
||
if not cond:
|
||
FAILS.append(desc)
|
||
|
||
|
||
def main():
|
||
srv, endpoint = spin()
|
||
base = dict(endpoint=endpoint, bucket=BUCKET, prefix="themes", region="us-east-1",
|
||
accessKey="AKID", secretKey="SECRET")
|
||
|
||
# 1) 测试连接：4 步应全过，探针最后被删掉
|
||
cfg = serve.S3Config(dict(base))
|
||
res = serve.test_connection(cfg)
|
||
check("测试连接返回 ok", res.get("ok") is True)
|
||
check("测试连接 4 步全过", all(s["ok"] for s in res["steps"]), str(res))
|
||
check("探针已清理", not any(k.startswith("themes/.tb-probe") for k in MockS3.store),
|
||
str(list(MockS3.store)))
|
||
|
||
# 2) 上传：前缀 / 内容 / 头都要对
|
||
f1 = b"\x52\x49\x46\x46fakewebp1"
|
||
f2 = b"\x52\x49\x46\x46fakewebp2"
|
||
up = serve.upload_files(cfg, [
|
||
{"key": "shiroko/asset1.webp", "contentType": "image/webp",
|
||
"data": base64.b64encode(f1).decode()},
|
||
{"key": "shiroko/asset2.webp", "contentType": "image/webp",
|
||
"data": base64.b64encode(f2).decode()},
|
||
])
|
||
check("上传全部成功", up["ok"] and up["total"] == 2 and up["failed"] == 0, str(up))
|
||
check("前缀拼进对象键", "themes/shiroko/asset1.webp" in MockS3.store, str(list(MockS3.store)))
|
||
stored, ctype, cache = MockS3.store.get("themes/shiroko/asset1.webp", (b"", None, None))
|
||
check("内容一致", stored == f1)
|
||
check("Content-Type = image/webp", ctype == "image/webp", str(ctype))
|
||
check("Cache-Control = immutable", "immutable" in (cache or ""), str(cache))
|
||
|
||
# 3) 错误密钥：HEAD 拿不到响应体（规范如此），只能按状态码归因；
|
||
# 真正能读到 S3 错误码的是带 body 的 PUT，所以用上传路径验证还原能力。
|
||
bad = serve.S3Config(dict(base, accessKey="BADKEY"))
|
||
try:
|
||
serve.test_connection(bad)
|
||
check("错误密钥应抛错", False)
|
||
except serve.S3Error as e:
|
||
check("HEAD 失败按状态码归因", e.status == 403 and e.code == "AccessDenied",
|
||
"%s %s" % (e.status, e.code))
|
||
|
||
badUp = serve.upload_files(bad, [{"key": "shiroko/x.webp", "contentType": "image/webp",
|
||
"data": base64.b64encode(b"x").decode()}])
|
||
codes = [f.get("code") for f in badUp["files"]]
|
||
check("PUT 失败能还原 SignatureDoesNotMatch", codes == ["SignatureDoesNotMatch"], str(codes))
|
||
check("该错误有中文提示", "签名不匹配" in serve.S3_HINTS.get("SignatureDoesNotMatch", ""))
|
||
|
||
# 4) 桶名写错：NoSuchBucket
|
||
wrong = serve.S3Config(dict(base, bucket="nope"))
|
||
try:
|
||
serve.test_connection(wrong)
|
||
check("错桶应抛错", False)
|
||
except serve.S3Error as e:
|
||
check("错桶 → NoSuchBucket", e.code == "NoSuchBucket", e.code)
|
||
|
||
# 5) 键里出现 .. 必须被拒（防止越权写到别的前缀）
|
||
try:
|
||
serve.full_key(cfg, "../etc/passwd")
|
||
check(".. 应被拒", False)
|
||
except ValueError:
|
||
check(".. 被拒绝", True)
|
||
|
||
# 6) 前缀留空 = 传桶根
|
||
plain = serve.S3Config(dict(base, prefix=""))
|
||
up2 = serve.upload_files(plain, [{"key": "root.webp", "contentType": "image/webp",
|
||
"data": base64.b64encode(b"x").decode()}])
|
||
check("空前缀传桶根", up2["ok"] and "root.webp" in MockS3.store, str(list(MockS3.store)))
|
||
|
||
# 7) 虚拟主机风格的路径（不带桶名）
|
||
vh = serve.S3Config(dict(base, pathStyle=False))
|
||
check("虚拟主机路径不含桶名", vh.object_path("themes/a.webp") == "/themes/a.webp", vh.object_path("x"))
|
||
check("虚拟主机 host 带桶名", vh.host.startswith(BUCKET + "."), vh.host)
|
||
|
||
srv.shutdown()
|
||
print("\n%d 项失败" % len(FAILS))
|
||
return 1 if FAILS else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|