chenxin
7c57d27494
- 迁移 18：avatar_model（live2d/vrm/gltf、validated/rejected、校验报告、来源许可、当前命令指针）+ avatar_command（十类命令检查约束、(avatar,sequence) 唯一、undone 标记） - GLB/glTF 解析器（纯 Python）：GLB magic/版本/chunk 结构校验、JSON 场景图引用一致性（scene→node→mesh→material 逐层索引越界检测）、VRM 0.x/1.0 扩展检测与 meta 提取 - 资源上限（恶意模型文件防线）：文件 200MB、节点 10 万、网格 1 万、纹理 2 千、材质 5 千、动画 1 千——超限入库 rejected 并保留报告 - Cubism model3.json 校验：Version/FileReferences 必需、moc3 引用、纹理列表、表情/动作/HitArea 计数 - 上传 MIME 白名单扩展 model/ 与 application/json - 命令历史：十类命令（param_map/expression/physics/motion/material/texture/bone/attachment/lighting/camera）追加（sequence 单调）、撤销（标记 undone 非破坏）、rejected 模型不接受编辑（422 前置校验） - 测试 111 项全绿（+9）：合法 GLB/VRM 检测/损坏文件/引用越界/超限/Cubism 正反例/命令历史与撤销幂等/rejected 拒编
358 lines
13 KiB
Python
358 lines
13 KiB
Python
"""Avatar 域：Live2D/VRM/glTF 导入校验与非破坏命令历史（WP-09）。
|
||
|
||
校验器纯 Python 实现，输入为文件字节；全部按不可信输入处理
|
||
（恶意模型文件防线：magic/结构/引用/资源计数硬上限）。
|
||
"""
|
||
|
||
import json
|
||
import struct
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
from fastapi import HTTPException, status
|
||
|
||
# 资源硬上限（契约第 7 节：限额由配置发布但 Schema 有硬上限；此处取工程默认）
|
||
MAX_FILE_BYTES = 200 * 1024 * 1024
|
||
MAX_NODES = 100_000
|
||
MAX_MESHES = 10_000
|
||
MAX_TEXTURES = 2_000
|
||
MAX_MATERIALS = 5_000
|
||
MAX_ANIMATIONS = 1_000
|
||
|
||
GLB_MAGIC = 0x46546C67 # 'glTF'
|
||
CHUNK_TYPE_JSON = 0x4E4F534A
|
||
CHUNK_TYPE_BIN = 0x004E4942
|
||
|
||
|
||
@dataclass
|
||
class ValidationReport:
|
||
format: str = ""
|
||
valid: bool = False
|
||
errors: list[str] = field(default_factory=list)
|
||
stats: dict[str, int] = field(default_factory=dict)
|
||
vrm_version: str | None = None
|
||
vrm_meta: dict[str, Any] = field(default_factory=dict)
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
out: dict[str, Any] = {
|
||
"format": self.format, "valid": self.valid, "errors": self.errors,
|
||
"stats": self.stats,
|
||
}
|
||
if self.vrm_version:
|
||
out["vrm_version"] = self.vrm_version
|
||
if self.vrm_meta:
|
||
out["vrm_meta"] = self.vrm_meta
|
||
return out
|
||
|
||
|
||
def _limit_check(report: ValidationReport, gltf: dict[str, Any]) -> bool:
|
||
stats = {
|
||
"nodes": len(gltf.get("nodes", [])),
|
||
"meshes": len(gltf.get("meshes", [])),
|
||
"textures": len(gltf.get("textures", [])),
|
||
"materials": len(gltf.get("materials", [])),
|
||
"animations": len(gltf.get("animations", [])),
|
||
}
|
||
report.stats = stats
|
||
limits = {
|
||
"nodes": MAX_NODES, "meshes": MAX_MESHES, "textures": MAX_TEXTURES,
|
||
"materials": MAX_MATERIALS, "animations": MAX_ANIMATIONS,
|
||
}
|
||
ok = True
|
||
for key, cap in limits.items():
|
||
if stats[key] > cap:
|
||
report.errors.append(f"{key} 数 {stats[key]} 超过上限 {cap}")
|
||
ok = False
|
||
return ok
|
||
|
||
|
||
def _check_references(report: ValidationReport, gltf: dict[str, Any]) -> bool:
|
||
"""场景图引用一致性：node.mesh/mesh.material/material 指向有效索引。"""
|
||
meshes = len(gltf.get("meshes", []))
|
||
materials = len(gltf.get("materials", []))
|
||
scenes = gltf.get("scenes", [])
|
||
if not scenes:
|
||
report.errors.append("缺少 scenes")
|
||
return False
|
||
for i, scene in enumerate(scenes):
|
||
for node_idx in scene.get("nodes", []):
|
||
if node_idx >= len(gltf.get("nodes", [])):
|
||
report.errors.append(f"scene[{i}] 引用不存在的 node {node_idx}")
|
||
return False
|
||
for i, node in enumerate(gltf.get("nodes", [])):
|
||
mesh_idx = node.get("mesh")
|
||
if mesh_idx is not None and mesh_idx >= meshes:
|
||
report.errors.append(f"node[{i}] 引用不存在的 mesh {mesh_idx}")
|
||
return False
|
||
for i, mesh in enumerate(gltf.get("meshes", [])):
|
||
for prim in mesh.get("primitives", []):
|
||
mat_idx = prim.get("material")
|
||
if mat_idx is not None and mat_idx >= materials:
|
||
report.errors.append(f"mesh[{i}] 引用不存在的 material {mat_idx}")
|
||
return False
|
||
return True
|
||
|
||
|
||
def validate_glb(data: bytes) -> ValidationReport:
|
||
"""GLB 二进制格式校验：magic、版本、chunk 结构、JSON 场景图。"""
|
||
report = ValidationReport(format="gltf")
|
||
if len(data) > MAX_FILE_BYTES:
|
||
report.errors.append(f"文件超过 {MAX_FILE_BYTES} 字节上限")
|
||
return report
|
||
if len(data) < 12:
|
||
report.errors.append("文件不足 GLB 头部长度（12 字节）")
|
||
return report
|
||
magic, version, length = struct.unpack_from("<III", data, 0)
|
||
if magic != GLB_MAGIC:
|
||
report.errors.append("magic 不符：不是 GLB 文件")
|
||
return report
|
||
if version != 2:
|
||
report.errors.append(f"glTF 容器版本 {version}，仅支持 2")
|
||
return report
|
||
if length != len(data):
|
||
report.errors.append(f"声明长度 {length} 与实际 {len(data)} 不符")
|
||
return report
|
||
# JSON chunk
|
||
if len(data) < 20:
|
||
report.errors.append("缺少 JSON chunk 头")
|
||
return report
|
||
json_len, json_type = struct.unpack_from("<II", data, 12)
|
||
if json_type != CHUNK_TYPE_JSON:
|
||
report.errors.append("首 chunk 不是 JSON 类型")
|
||
return report
|
||
json_bytes = data[20 : 20 + json_len]
|
||
if len(json_bytes) < json_len:
|
||
report.errors.append("JSON chunk 截断")
|
||
return report
|
||
try:
|
||
gltf = json.loads(json_bytes.rstrip(b"\x00 \t\r\n"))
|
||
except json.JSONDecodeError as exc:
|
||
report.errors.append(f"JSON 解析失败：{exc}")
|
||
return report
|
||
# VRM 扩展检测
|
||
extensions_used = gltf.get("extensionsUsed", [])
|
||
extensions = gltf.get("extensions", {})
|
||
if "VRMC_vrm" in extensions_used or "VRMC_vrm" in extensions:
|
||
report.format = "vrm"
|
||
report.vrm_version = "1.0"
|
||
report.vrm_meta = dict(extensions.get("VRMC_vrm", {}).get("meta", {}))
|
||
elif "VRM" in extensions_used or "VRM" in extensions:
|
||
report.format = "vrm"
|
||
report.vrm_version = "0.x"
|
||
report.vrm_meta = dict(extensions.get("VRM", {}).get("meta", {}))
|
||
# 结构与上限
|
||
ok = _check_references(report, gltf) and _limit_check(report, gltf)
|
||
report.valid = ok and not report.errors
|
||
return report
|
||
|
||
|
||
def validate_gltf_json(data: bytes) -> ValidationReport:
|
||
"""纯 .gltf（JSON）校验。"""
|
||
report = ValidationReport(format="gltf")
|
||
if len(data) > MAX_FILE_BYTES:
|
||
report.errors.append(f"文件超过 {MAX_FILE_BYTES} 字节上限")
|
||
return report
|
||
try:
|
||
gltf = json.loads(data)
|
||
except json.JSONDecodeError as exc:
|
||
report.errors.append(f"JSON 解析失败：{exc}")
|
||
return report
|
||
if not isinstance(gltf, dict) or "asset" not in gltf:
|
||
report.errors.append("缺少 asset 段（glTF 2.0 必需）")
|
||
return report
|
||
ok = _check_references(report, gltf) and _limit_check(report, gltf)
|
||
report.valid = ok and not report.errors
|
||
return report
|
||
|
||
|
||
# Cubism model3.json 必需字段与引用文件类别
|
||
_CUBISM_REQUIRED = {"Version", "FileReferences"}
|
||
_CUBISM_FILE_KEYS = {" moc3": "moc", ".model.json": "physics", }
|
||
|
||
|
||
def validate_cubism(data: bytes) -> ValidationReport:
|
||
"""Cubism model3.json 校验：必需字段、 moc3 引用、组引用一致性。"""
|
||
report = ValidationReport(format="live2d")
|
||
if len(data) > MAX_FILE_BYTES:
|
||
report.errors.append(f"文件超过 {MAX_FILE_BYTES} 字节上限")
|
||
return report
|
||
try:
|
||
model = json.loads(data)
|
||
except json.JSONDecodeError as exc:
|
||
report.errors.append(f"JSON 解析失败：{exc}")
|
||
return report
|
||
if not isinstance(model, dict):
|
||
report.errors.append("根不是对象")
|
||
return report
|
||
for key in _CUBISM_REQUIRED:
|
||
if key not in model:
|
||
report.errors.append(f"缺少必需字段 {key}")
|
||
refs = model.get("FileReferences", {})
|
||
if not isinstance(refs, dict):
|
||
report.errors.append("FileReferences 不是对象")
|
||
refs = {}
|
||
moc = refs.get("moc")
|
||
if not moc or not str(moc).endswith(".moc3"):
|
||
report.errors.append("缺少 moc3 模型文件引用")
|
||
textures = refs.get("Textures", [])
|
||
if not isinstance(textures, list) or not textures:
|
||
report.errors.append("缺少纹理引用列表")
|
||
# 参数区间引用（Expressions/Physics/Motions 仅做存在性结构检查）
|
||
stats = {
|
||
"textures": len(textures) if isinstance(textures, list) else 0,
|
||
"expressions": len(refs.get("Expressions", []) or []),
|
||
"motions": sum(
|
||
len(v) if isinstance(v, list) else 0
|
||
for v in (refs.get("Motions", {}) or {}).values()
|
||
),
|
||
"hit_areas": len(model.get("HitAreas", []) or []),
|
||
"parameters": len(model.get("Parameters", []) or []),
|
||
}
|
||
report.stats = stats
|
||
if stats["textures"] > MAX_TEXTURES:
|
||
report.errors.append(f"纹理数超过上限 {MAX_TEXTURES}")
|
||
report.valid = not report.errors
|
||
return report
|
||
|
||
|
||
def detect_and_validate(data: bytes, declared_format: str) -> ValidationReport:
|
||
"""格式探测 + 校验入口：magic 优先，声明辅助；二进制非 GLB 一律 magic 拒绝。"""
|
||
if data[:4] == b"glTF":
|
||
report = validate_glb(data)
|
||
elif declared_format == "live2d":
|
||
report = validate_cubism(data)
|
||
elif data[:1] not in (b"{", b"["):
|
||
# 声明 gltf 但内容是二进制且无 GLB magic：magic 不符
|
||
report = ValidationReport(format="gltf")
|
||
report.errors.append("magic 不符：不是 GLB/glTF 文件")
|
||
else:
|
||
report = validate_gltf_json(data)
|
||
if report.format != declared_format and declared_format != "":
|
||
report.errors.append(
|
||
f"声明格式 {declared_format} 与检测结果 {report.format} 不符"
|
||
)
|
||
report.valid = False
|
||
return report
|
||
|
||
|
||
async def import_avatar(
|
||
db: Any, domain_id: str, name: str, declared_format: str, file_id: str,
|
||
trace_id: str | None,
|
||
) -> Any:
|
||
"""导入 Avatar 模型：文件须 ready；校验失败入库 rejected（保留报告供排查）。"""
|
||
from app.media.service import require_file
|
||
from app.models import AuditEvent, AvatarModel
|
||
|
||
asset = await require_file(db, domain_id, file_id)
|
||
if asset.status != "ready":
|
||
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "文件未就绪")
|
||
if declared_format not in {"live2d", "vrm", "gltf"}:
|
||
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "format 取值不合法")
|
||
from pathlib import Path
|
||
|
||
from app.media.service import STORAGE_ROOT
|
||
|
||
path = Path(STORAGE_ROOT) / asset.storage_key
|
||
if not path.exists():
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "存储对象缺失")
|
||
data = path.read_bytes()
|
||
report = detect_and_validate(data, declared_format)
|
||
model = AvatarModel(
|
||
domain_id=domain_id,
|
||
name=name if name.strip() else asset.filename,
|
||
format=declared_format,
|
||
file_id=file_id,
|
||
status="validated" if report.valid else "rejected",
|
||
validation_report=report.to_dict(),
|
||
)
|
||
db.add(model)
|
||
await db.flush()
|
||
db.add(
|
||
AuditEvent(
|
||
actor_type="user", actor_id=domain_id, action="avatar.imported",
|
||
aggregate_type="avatar_model", aggregate_id=model.id,
|
||
payload={"valid": report.valid, "errors": report.errors[:10]},
|
||
trace_id=trace_id,
|
||
)
|
||
)
|
||
await db.flush()
|
||
return model
|
||
|
||
|
||
async def append_command(
|
||
db: Any, domain_id: str, avatar_id: str, command_type: str, payload: dict[str, Any],
|
||
trace_id: str | None,
|
||
) -> Any:
|
||
"""追加非破坏编辑命令：sequence 单调递增；rejected 模型不接受编辑。"""
|
||
from sqlalchemy import select
|
||
|
||
from app.models import AuditEvent, AvatarCommand, AvatarModel
|
||
|
||
model = await db.get(AvatarModel, avatar_id)
|
||
if model is None or model.domain_id != domain_id:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "模型不存在")
|
||
if model.status != "validated":
|
||
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "仅已验证模型可编辑")
|
||
valid_types = {
|
||
"param_map", "expression", "physics", "motion", "material", "texture",
|
||
"bone", "attachment", "lighting", "camera",
|
||
}
|
||
if command_type not in valid_types:
|
||
raise HTTPException(
|
||
status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||
f"command_type 取值不合法：{command_type}",
|
||
)
|
||
last = (
|
||
await db.execute(
|
||
select(AvatarCommand.sequence)
|
||
.where(AvatarCommand.avatar_id == avatar_id)
|
||
.order_by(AvatarCommand.sequence.desc())
|
||
.limit(1)
|
||
)
|
||
).scalar_one_or_none()
|
||
command = AvatarCommand(
|
||
avatar_id=avatar_id,
|
||
sequence=(last + 1) if last is not None else 1,
|
||
command_type=command_type,
|
||
payload=payload,
|
||
)
|
||
db.add(command)
|
||
await db.flush()
|
||
model.current_command_id = command.id
|
||
db.add(
|
||
AuditEvent(
|
||
actor_type="user", actor_id=domain_id, action="avatar.command_appended",
|
||
aggregate_type="avatar_model", aggregate_id=avatar_id,
|
||
payload={"sequence": command.sequence, "type": command_type},
|
||
trace_id=trace_id,
|
||
)
|
||
)
|
||
await db.flush()
|
||
return command
|
||
|
||
|
||
async def undo_command(
|
||
db: Any, domain_id: str, avatar_id: str, command_id: str, trace_id: str | None,
|
||
) -> Any:
|
||
"""撤销最近命令（标记 undone，不物理删除——非破坏历史）。"""
|
||
from app.models import AuditEvent, AvatarCommand, AvatarModel
|
||
|
||
model = await db.get(AvatarModel, avatar_id)
|
||
if model is None or model.domain_id != domain_id:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "模型不存在")
|
||
command = await db.get(AvatarCommand, command_id)
|
||
if command is None or command.avatar_id != avatar_id:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "命令不存在")
|
||
if command.undone:
|
||
return command # 幂等
|
||
command.undone = True
|
||
db.add(
|
||
AuditEvent(
|
||
actor_type="user", actor_id=domain_id, action="avatar.command_undone",
|
||
aggregate_type="avatar_model", aggregate_id=avatar_id,
|
||
payload={"sequence": command.sequence}, trace_id=trace_id,
|
||
)
|
||
)
|
||
await db.flush()
|
||
return command
|