fix(backend): API 文本输入加 max_length + 请求体大小中间件(CR-H9)

This commit is contained in:
Yaojia Wang
2026-07-08 10:53:44 +02:00
parent 26f10d023d
commit e1a24a2c2c
13 changed files with 223 additions and 30 deletions

View File

@@ -0,0 +1,109 @@
"""CR-H9用户请求文本字段 max_length 上界 + 应用级请求体大小中间件(防 DoS/成本)。
- 单元:各请求 schema 的字符串字段在 cap 处可构造、cap+1 触发 ValidationError含样本列表长度
与列表项两处上界——最大 DoS 面)。
- 集成(无 pg超大请求体经中间件直接 413不入路由/DB
"""
from __future__ import annotations
import os
from dataclasses import dataclass, field
from typing import Any
import pytest
from cryptography.fernet import Fernet
from pydantic import BaseModel, ValidationError
# create_app() 于 import ww_api.main 时构建,需 CREDENTIAL_ENC_KEY——先于任何 main 导入置好。
os.environ.setdefault("CREDENTIAL_ENC_KEY", Fernet.generate_key().decode())
from ww_api.schemas.foreshadow import ForeshadowRegisterRequest
from ww_api.schemas.generation import CharacterGenerateRequest, WorldGenerateRequest
from ww_api.schemas.projects import (
AcceptRequest,
DraftSaveRequest,
ProjectCreateRequest,
ReviewRequest,
)
from ww_api.schemas.providers import (
ProviderCredentialInput,
TierRoutingInput,
)
from ww_api.schemas.providers import (
TestConnectionRequest as _TestConnectionRequest, # 别名避免 pytest 误采集 Test* 类
)
from ww_api.schemas.rules import RuleCreateRequest
from ww_api.schemas.style import RefineRequest, StyleLearnRequest
from ww_api.schemas.templates import TemplateCreateRequest
@dataclass(frozen=True)
class _StrCase:
"""一个字符串字段的上界用例cap 处可构造、cap+1 触发 ValidationError。"""
schema: type[BaseModel]
field_name: str
cap: int
case_id: str
base: dict[str, Any] = field(default_factory=dict) # 该 schema 其余必填字段
_STR_CASES: list[_StrCase] = [
_StrCase(ProviderCredentialInput, "api_key", 512, "api_key<=512", {"provider": "p"}),
_StrCase(ProviderCredentialInput, "provider", 64, "provider<=64", {"api_key": "k"}),
_StrCase(TierRoutingInput, "model", 128, "model<=128", {"tier": "writer", "provider": "p"}),
_StrCase(
TierRoutingInput, "provider", 64, "routing-provider<=64", {"tier": "writer", "model": "m"}
),
_StrCase(_TestConnectionRequest, "provider", 64, "test-provider<=64"),
_StrCase(WorldGenerateRequest, "brief", 10_000, "world-brief<=10000"),
_StrCase(CharacterGenerateRequest, "brief", 10_000, "char-brief<=10000"),
_StrCase(ProjectCreateRequest, "title", 200, "title<=200"),
_StrCase(AcceptRequest, "final_text", 200_000, "final_text<=200000"),
_StrCase(DraftSaveRequest, "text", 200_000, "draft-text<=200000"),
_StrCase(ReviewRequest, "draft", 200_000, "review-draft<=200000"),
_StrCase(RefineRequest, "segment", 20_000, "segment<=20000"),
_StrCase(ForeshadowRegisterRequest, "code", 100, "code<=100", {"title": "t"}),
_StrCase(ForeshadowRegisterRequest, "title", 500, "foreshadow-title<=500", {"code": "c"}),
_StrCase(RuleCreateRequest, "content", 10_000, "rule-content<=10000", {"level": "project"}),
_StrCase(TemplateCreateRequest, "title", 200, "template-title<=200", {"body": "b"}),
_StrCase(TemplateCreateRequest, "body", 20_000, "template-body<=20000", {"title": "t"}),
]
@pytest.mark.parametrize("case", _STR_CASES, ids=[c.case_id for c in _STR_CASES])
def test_request_str_fields_reject_over_max_length(case: _StrCase) -> None:
# Arrange / Act — 长度 == cap构造成功。
case.schema(**{**case.base, case.field_name: "x" * case.cap})
# Assert — 长度 == cap+1超界即 ValidationError。
with pytest.raises(ValidationError):
case.schema(**{**case.base, case.field_name: "x" * (case.cap + 1)})
def test_style_learn_samples_item_and_list_caps() -> None:
# 单条样本 == 20_000 字符 OK20_001 → ValidationError最大 DoS 面:每条正文都设上界)。
StyleLearnRequest(samples=["x" * 20_000])
with pytest.raises(ValidationError):
StyleLearnRequest(samples=["x" * 20_001])
# 列表 == 20 条 OK21 条 → ValidationError同时约束列表长度
StyleLearnRequest(samples=["s"] * 20)
with pytest.raises(ValidationError):
StyleLearnRequest(samples=["s"] * 21)
def test_oversized_request_body_rejected_413() -> None:
"""超大请求体经 body-size 中间件直接 413不入路由/DB
不以 context manager 进入 TestClient → 不跑 lifespan/seed → 不需 Postgres。中间件在
路由前拦截,故请求根本不触达任何 DB。
"""
from fastapi.testclient import TestClient
from ww_api.main import create_app
from ww_api.middleware import MAX_BODY_BYTES
client = TestClient(create_app())
resp = client.post("/projects", content=b"x" * (MAX_BODY_BYTES + 1))
assert resp.status_code == 413
assert resp.json()["error"]["code"] == "PAYLOAD_TOO_LARGE"