浅念
a43c654337
- POST /api/platform/tenants/{slug}/brand/avatar（require_tenant）：
multipart 收文件，魔数判定类型（\x89PNG/\xff\xd8\xff/RIFF....WEBP，
优先于扩展名——改后缀骗不过，存盘扩展名取自内容）、≤2MB（413，
upload.size 预检免拷贝）、服务端 uuid4 重命名（原始文件名不参与
路径），落 data/uploads/<slug>/（目录懒建，gitignore 已忽略）
- 走 Task 6 配置版本管线（复用 config_routes 内部 helper）：读当前
快照 → brand 组只替换 avatar 字段 → validate_group →
write_config_version(comment=头像上传) → registry_reload 热生效
- /api/public/avatar 三级解析链：uploads → seed 资产 → 静态兜底；
Phase 1 文件名穿越防线（拒 / \ . ..）扩到 uploads 段
- registry.uploads_root：自 data_dir.parent 派生（生产=<repo>/data/
uploads；测试注入 tmp data_dir 时自动落 tmp 内，隔离免污染）
- 新依赖 python-multipart（multipart 表单解析）
Co-Authored-By: Claude Code <noreply@anthropic.com>
236 lines
10 KiB
Python
236 lines
10 KiB
Python
"""头像上传与解析链(Phase 2 Task 7)。
|
|
|
|
POST /api/platform/tenants/{slug}/brand/avatar(multipart,require_tenant):
|
|
- 魔数校验优先于扩展名——改后缀骗不过(.png 名的 JPEG 按内容存 .jpg;
|
|
RIFF 但非 WEBP 拒;纯文本拒);
|
|
- ≤2MB(超出 413);服务端 uuid 重命名,原始文件名不参与路径;
|
|
- 走配置版本管线(快照 brand 组改 avatar 字段 → validate_group →
|
|
write_config_version(comment=头像上传)→ registry_reload);
|
|
- /api/public/avatar 三级解析链: data/uploads/<slug>/<name> →
|
|
seed_profiles/<slug>/assets/<name> → static/kefu-small.png;文件名的
|
|
穿越防线(PHase 1)扩到 uploads 段。
|
|
|
|
夹具 db_source_app/super_client(conftest);助手复用 test_tenant_admin_api。
|
|
上传落盘位置 = registry.uploads_root(DB 源 tmp 布局下 = tmp_path/data/
|
|
uploads,隔离免污染;生产行为见 registry.uploads_root 注释)。
|
|
"""
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from tests.conftest import SEED_DIR
|
|
from tests.test_tenant_admin_api import (
|
|
_auth,
|
|
_bearer,
|
|
_create,
|
|
_tenant_admin_token,
|
|
)
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
TENANTS = "/api/platform/tenants"
|
|
LOGIN = "/api/platform/auth/login"
|
|
|
|
# 最小"合法头"测试字节——魔数校验只看头几个字节,后续填充唯一标记
|
|
PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"avatar-body-UNIQUE-42"
|
|
JPG_BYTES = b"\xff\xd8\xff" + b"jpeg-body"
|
|
WEBP_BYTES = b"RIFF" + b"\x24\x00\x00\x00" + b"WEBP" + b"VP8 webp-body"
|
|
UUID_EXT_RE = re.compile(r"^[0-9a-f]{32}\.(png|jpg|webp)$")
|
|
|
|
|
|
def _upload(client, slug, name, data, content_type="image/octet-stream",
|
|
headers=None):
|
|
return client.post(f"{TENANTS}/{slug}/brand/avatar",
|
|
headers=headers or _auth(client),
|
|
files={"file": (name, data, content_type)})
|
|
|
|
|
|
def _config(client, slug):
|
|
r = client.get(f"{TENANTS}/{slug}/config", headers=_auth(client))
|
|
assert r.status_code == 200, r.text
|
|
return r.json()
|
|
|
|
|
|
def _avatar(client, host):
|
|
return client.get("/api/public/avatar", headers={"host": host})
|
|
|
|
|
|
# ---- 上传 → 版本管线 → 热生效 ------------------------------------------------------
|
|
|
|
|
|
def test_upload_and_serve(super_client, db_source_app, tmp_path):
|
|
_create(super_client, slug="demo", domains=["demo.example.com"])
|
|
fallback = (REPO / "static" / "kefu-small.png").read_bytes()
|
|
|
|
# 基线:DB 源租户(无 seed 资产目录)→ 静态兜底
|
|
r = _avatar(super_client, "demo.example.com")
|
|
assert r.status_code == 200
|
|
assert r.content == fallback
|
|
|
|
r = _upload(super_client, "demo", "me.png", PNG_BYTES, "image/png")
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
# 服务端 uuid 重命名:uuid4 hex + 魔数判定的扩展名;原始文件名不出现
|
|
assert UUID_EXT_RE.fullmatch(body["avatar"]), body["avatar"]
|
|
assert body["avatar"].endswith(".png")
|
|
assert "me.png" not in body["avatar"]
|
|
|
|
# 落盘:data/uploads/<slug>/<uuid>.<ext>(目录懒建),字节原样
|
|
stored = tmp_path / "data" / "uploads" / "demo" / body["avatar"]
|
|
assert stored.is_file()
|
|
assert stored.read_bytes() == PNG_BYTES
|
|
|
|
# 配置版本管线:版本 +1,brand 组只改 avatar 字段(其余字段原样保留),
|
|
# 版本历史 comment 记"头像上传"
|
|
cfg = _config(super_client, "demo")
|
|
assert cfg["version"] == 2
|
|
assert cfg["config"]["brand"]["avatar"] == body["avatar"]
|
|
assert cfg["config"]["brand"]["title"] == "HESUPIPE Support" # jiancai 档案值
|
|
versions = super_client.get(f"{TENANTS}/demo/config/versions",
|
|
headers=_auth(super_client)).json()["items"]
|
|
assert versions[-1]["comment"] == "头像上传"
|
|
|
|
# 热生效:registry_reload 后 /api/public/avatar(该租户 Host)返回新图字节
|
|
r = _avatar(super_client, "demo.example.com")
|
|
assert r.status_code == 200
|
|
assert r.headers["content-type"].startswith("image/")
|
|
assert r.content == PNG_BYTES
|
|
assert r.content != fallback
|
|
|
|
|
|
def test_magic_beats_extension(super_client, tmp_path):
|
|
"""魔数优先于扩展名:内容是 JPEG 的 .png 名文件按内容收为 .jpg;
|
|
webp 魔数(RIFF....WEBP)同理。"""
|
|
_create(super_client, slug="demo")
|
|
r = _upload(super_client, "demo", "sneaky.png", JPG_BYTES, "image/png")
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["avatar"].endswith(".jpg")
|
|
|
|
r = _upload(super_client, "demo", "logo.webp", WEBP_BYTES, "image/webp")
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["avatar"].endswith(".webp")
|
|
|
|
|
|
# ---- 拒收面:类型 / 大小 ------------------------------------------------------------
|
|
|
|
|
|
def test_reject_bad_type_and_oversize(super_client, tmp_path):
|
|
_create(super_client, slug="demo")
|
|
# .txt(内容也非图)→ 400
|
|
r = _upload(super_client, "demo", "a.txt", b"hello plain text", "text/plain")
|
|
assert r.status_code == 400
|
|
# .png 后缀但内容非 png(魔数不符)→ 400,改后缀骗不过
|
|
r = _upload(super_client, "demo", "fake.png", b"definitely not an image",
|
|
"image/png")
|
|
assert r.status_code == 400
|
|
# RIFF 头但非 WEBP(如 wav)→ 400
|
|
r = _upload(super_client, "demo", "a.wav",
|
|
b"RIFF\x24\x00\x00\x00WAVEfmt ", "audio/wav")
|
|
assert r.status_code == 400
|
|
# 3MB(合法 png 头)→ 413
|
|
big = b"\x89PNG\r\n\x1a\n" + b"x" * (3 * 1024 * 1024)
|
|
r = _upload(super_client, "demo", "big.png", big, "image/png")
|
|
assert r.status_code == 413
|
|
# 缺 file 字段 → 400(不是 422/500)
|
|
r = super_client.post(f"{TENANTS}/demo/brand/avatar",
|
|
headers=_auth(super_client), data={"x": "1"})
|
|
assert r.status_code == 400
|
|
# 全部拒收:版本不涨、盘上无文件
|
|
assert _config(super_client, "demo")["version"] == 1
|
|
uploads_dir = tmp_path / "data" / "uploads" / "demo"
|
|
assert not uploads_dir.exists() or not any(uploads_dir.iterdir())
|
|
# 恰在上限内(2MB 整)→ 通过
|
|
ok = b"\x89PNG\r\n\x1a\n" + b"y" * (2 * 1024 * 1024 - 8)
|
|
assert len(ok) == 2 * 1024 * 1024
|
|
r = _upload(super_client, "demo", "edge.png", ok, "image/png")
|
|
assert r.status_code == 200, r.text
|
|
|
|
|
|
# ---- 穿越三防 ---------------------------------------------------------------------
|
|
|
|
|
|
def test_traversal_safe(super_client, db_source_app, tmp_path):
|
|
_create(super_client, slug="demo", domains=["demo.example.com"])
|
|
# 防 1:上传文件名带 ../ 不生效——服务端 uuid 重命名,路径成分不落盘
|
|
r = _upload(super_client, "demo", "../../evil.png", PNG_BYTES, "image/png")
|
|
assert r.status_code == 200, r.text
|
|
avatar = r.json()["avatar"]
|
|
assert UUID_EXT_RE.fullmatch(avatar), avatar
|
|
assert (tmp_path / "data" / "uploads" / "demo" / avatar).is_file()
|
|
assert not (tmp_path / "data" / "uploads" / "evil.png").exists()
|
|
assert not (tmp_path / "evil.png").exists()
|
|
assert not (tmp_path / "data" / "evil.png").exists()
|
|
|
|
# 防 2:解析链 uploads 段拒绝路径字符——brand.avatar 指到 uploads 目录外
|
|
# 的真实文件(Windows 反斜杠语义下 ..\secret.txt 会穿出 uploads/demo/),
|
|
# 命中防线即回退静态兜底,绝不 serve 穿越目标
|
|
secret = tmp_path / "data" / "uploads" / "secret.txt"
|
|
secret.write_bytes(b"TOPSECRET-never-serve")
|
|
r = super_client.put(f"{TENANTS}/demo/config/brand",
|
|
headers=_auth(super_client),
|
|
json={"avatar": "..\\secret.txt"})
|
|
assert r.status_code == 200, r.text
|
|
r = _avatar(super_client, "demo.example.com")
|
|
assert r.status_code == 200
|
|
assert b"TOPSECRET" not in r.content
|
|
assert r.content == (REPO / "static" / "kefu-small.png").read_bytes()
|
|
|
|
# 防 3:正斜杠形态(PurePosixPath 剥成基名,单成分不可能越出 uploads/<slug>/)
|
|
r = super_client.put(f"{TENANTS}/demo/config/brand",
|
|
headers=_auth(super_client),
|
|
json={"avatar": "../../secret.txt"})
|
|
assert r.status_code == 200, r.text
|
|
r = _avatar(super_client, "demo.example.com")
|
|
assert r.status_code == 200
|
|
assert b"TOPSECRET" not in r.content
|
|
|
|
|
|
# ---- 鉴权边界(require_tenant:401/409/403,本租户管理员放行,未知租户 404) ----------
|
|
|
|
|
|
def test_upload_auth_boundaries(super_client):
|
|
_create(super_client, slug="demo")
|
|
_create(super_client, slug="other")
|
|
url = f"{TENANTS}/demo/brand/avatar"
|
|
|
|
# 未认证 → 401
|
|
r = super_client.post(url, files={"file": ("a.png", PNG_BYTES, "image/png")})
|
|
assert r.status_code == 401
|
|
|
|
# 引导密码未改(登录即持 must_change 标志)→ 409
|
|
r = super_client.post(LOGIN, json={"username": "demo-admin",
|
|
"password": "demo-pass-1"})
|
|
fresh_token = r.json()["token"]
|
|
r = super_client.post(url, headers=_bearer(fresh_token),
|
|
files={"file": ("a.png", PNG_BYTES, "image/png")})
|
|
assert r.status_code == 409
|
|
|
|
# 其他租户管理员 → 403
|
|
other_token = _tenant_admin_token(super_client, "other-admin",
|
|
"other-pass-1")
|
|
r = super_client.post(url, headers=_bearer(other_token),
|
|
files={"file": ("a.png", PNG_BYTES, "image/png")})
|
|
assert r.status_code == 403
|
|
|
|
# 本租户管理员 → 200(require_tenant 放行租户全开)
|
|
demo_token = _tenant_admin_token(super_client, "demo-admin",
|
|
"demo-pass-1")
|
|
r = super_client.post(url, headers=_bearer(demo_token),
|
|
files={"file": ("a.png", PNG_BYTES, "image/png")})
|
|
assert r.status_code == 200, r.text
|
|
|
|
# 未知租户 → 404
|
|
r = _upload(super_client, "nope", "a.png", PNG_BYTES, "image/png")
|
|
assert r.status_code == 404
|
|
|
|
|
|
# ---- 解析链既有级行为回归(seed 段优先级在 uploads 之后) ----------------------------
|
|
|
|
|
|
def test_seed_assets_still_served_for_file_tenants(app_client):
|
|
"""文件源租户(jiancai)不上传时:uploads 段落空 → seed 资产照常服务
|
|
(uploads 段的加入不改变既有两级链的行为)。"""
|
|
r = app_client.get("/api/public/avatar")
|
|
assert r.status_code == 200
|
|
assert r.headers["content-type"].startswith("image/")
|
|
assert r.content == (SEED_DIR / "jiancai" / "assets" /
|
|
"kefu-small.png").read_bytes()
|