风控 SSE 内容协商与鉴权时序;docs/05 补齐 413 与风控入口(docs/25 P3 #23 #24 #25)

#23:413 是上传超限的标准语义,前端文档(风控业务演示文档 17)也已按 413 做提示
映射,所以不把代码降成 422,而是在 docs/05 §3.5 状态码表补登 413 —— 契约以"补齐"
而不是"改动"的方式对齐。

#24:/api/v1/risk/daily-report/stream 此前既不校验 Accept,又把鉴权留在 async
generator 内部。后者更隐蔽:StreamingResponse 已经返回、响应头已经发出,403 只能
变成"200 + 半截流"。现在 controller 先 await service.authorize(context) 再判定
Accept,顺序与 §6.4 一致(鉴权先行,不用状态码差异做探测)。SSE 协商逻辑抽到
app/api/dependencies/negotiation.py,与 /agent-runs/{run_id}/events 共用同一口径,
避免同一种客户端在一个端点上 200、另一个端点上 406。

#25:复核后确认前半段不成立 —— §19 末尾写明业务域接口由各自业务文档登记,风控 15 条
端点已在 06-模块接口与字段映射.md 逐条登记。真问题是 §12 表里写的
/api/v1/risk-scans/**、/api/v1/risk-alerts/** 与实际实现 /api/v1/risk/** 不符,
按实际实现更新 §12 并加说明;顺带把风控文档里 /daily-report/mail 的权限从
"按主项目邮件策略执行"改为实际的 risk:report:mail。

新增 tests/unit/api/test_risk_stream_negotiation.py(7 例)。
This commit is contained in:
2026-09-11 14:08:55 +08:00
parent a572c09a5c
commit 790518114b
8 changed files with 183 additions and 39 deletions
+1 -34
View File
@@ -7,6 +7,7 @@ from starlette.responses import StreamingResponse
from app.api.dependencies.auth import build_request_context
from app.api.dependencies.database import get_session
from app.api.dependencies.negotiation import accepts_event_stream
from app.api.dependencies.rate_limit import enforce_rate_limit
from app.api.schemas.agent_runs import (
AgentRunAcceptedEnvelope,
@@ -25,40 +26,6 @@ from app.service.run_query_service import RunQueryService
router = APIRouter(prefix="/api/v1/agent-runs", tags=["agent-runs"],
dependencies=[Depends(enforce_rate_limit)])
SSE_MEDIA_TYPE = "text/event-stream"
def accepts_event_stream(accept: str | None) -> bool:
"""`Accept` 是否接受 `text/event-stream`(文档 §3.2:该头**非必填**)。
- 未携带(`None` 或空串)→ 放行:文档写明"默认 `application/json`;SSE 为
`text/event-stream`",即由接口自身决定响应类型,不是客户端错误;
- 携带 `text/event-stream`、`text/*` 或 `*/*` 且 `q != 0` → 放行;
- 显式携带但只接受其他类型(如 `application/json`)→ 拒绝,由调用方转
`406 SSE_NOT_ACCEPTABLE`(文档 §3.5/§6.4)。
只做"是否可接受"的判定,不参与内容协商排序:SSE 端点只有一种表示。
"""
if accept is None or not accept.strip():
return True
for entry in accept.split(","):
parts = entry.split(";")
media_type = parts[0].strip().lower()
if media_type not in {SSE_MEDIA_TYPE, "text/*", "*/*"}:
continue
quality = 1.0
for parameter in parts[1:]:
name, _, value = parameter.partition("=")
if name.strip().lower() == "q":
try:
quality = float(value.strip())
except ValueError:
quality = 0.0
if quality > 0:
return True
return False
@router.post(
"",
response_model=AgentRunAcceptedEnvelope,
+12 -2
View File
@@ -5,12 +5,13 @@ from collections.abc import AsyncIterator
from datetime import datetime, time
from typing import Any
from fastapi import APIRouter, Depends, File, Path, UploadFile
from fastapi import APIRouter, Depends, File, Path, Request, UploadFile
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.responses import StreamingResponse
from app.api.dependencies.auth import build_request_context
from app.api.dependencies.database import get_session
from app.api.dependencies.negotiation import accepts_event_stream
from app.api.dependencies.rate_limit import enforce_rate_limit
from app.api.schemas.risk import (
RiskAlertEscalationRequest,
@@ -24,6 +25,7 @@ from app.api.schemas.risk import (
RiskNotificationPageQuery,
)
from app.core.contracts import RequestContext
from app.core.errors import SseNotAcceptableError
from app.infrastructure.db import mysql_scan_lock
from app.service.risk_action_service import RiskActionService
from app.service.risk_daily_report_mail_service import RiskDailyReportMailService
@@ -191,6 +193,7 @@ async def generate_risk_daily_report(
@router.post("/daily-report/stream")
async def stream_risk_daily_report(
payload: RiskDailyReportGenerateRequest,
request: Request,
context: RequestContext = Depends(build_request_context), # noqa: B008
session: AsyncSession = Depends(get_session), # noqa: B008
) -> StreamingResponse:
@@ -199,9 +202,16 @@ async def stream_risk_daily_report(
if payload.report_date is not None
else None
)
service = RiskDailyReportService(session)
# 鉴权与内容协商都必须在返回 StreamingResponse **之前**完成:`stream()` 是 async
# generator,函数体到第一次迭代才执行,而那时响应头已经发出去了 —— 403/406 只能
# 变成"200 + 半截流"(docs/25 P3 #24)。顺序与 §6.4 一致:先鉴权,后 Accept。
await service.authorize(context)
if not accepts_event_stream(request.headers.get("Accept")):
raise SseNotAcceptableError("Accept 必须接受 text/event-stream")
async def events() -> AsyncIterator[str]:
async for event in RiskDailyReportService(session).stream(context, report_time):
async for event in service.stream(context, report_time):
event_type = str(event.get("type", "message"))
payload = json.dumps(event, ensure_ascii=False, default=str)
yield f"event: {event_type}\ndata: {payload}\n\n"
+47
View File
@@ -0,0 +1,47 @@
"""SSE 内容协商(`docs/05` §3.2、§3.5)。
`Accept` 在文档里是**非必填**头,所以判定规则是"客户端是否**显式拒绝**了
`text/event-stream`",而不是"客户端是否显式接受"。
放在这里而不是写在某个 Controller 里,是因为平台有两个 SSE 端点
(`/api/v1/agent-runs/{run_id}/events` 与 `/api/v1/risk/daily-report/stream`):口径
一旦分叉,同一份客户端代码就会在一个端点上拿到 200、在另一个端点上拿到 406。
"""
from __future__ import annotations
SSE_MEDIA_TYPE = "text/event-stream"
# `text/*` 与 `*/*` 都覆盖 `text/event-stream`,属于"可接受"。
ACCEPTABLE_SSE_TYPES = frozenset({SSE_MEDIA_TYPE, "text/*", "*/*"})
def accepts_event_stream(accept: str | None) -> bool:
"""`Accept` 是否接受 `text/event-stream`(文档 §3.2:该头**非必填**)。
- 未携带(`None` 或空串)→ 放行:文档写明"默认 `application/json`;SSE 为
`text/event-stream`",即由接口自身决定响应类型,不是客户端错误;
- 携带 `text/event-stream`、`text/*` 或 `*/*` 且 `q != 0` → 放行;
- 显式携带但只接受其他类型(如 `application/json`)→ 拒绝,由调用方转
`406 SSE_NOT_ACCEPTABLE`(文档 §3.5/§6.4)。
只做"是否可接受"的判定,不参与内容协商排序:SSE 端点只有一种表示。
"""
if accept is None or not accept.strip():
return True
for entry in accept.split(","):
parts = entry.split(";")
media_type = parts[0].strip().lower()
if media_type not in ACCEPTABLE_SSE_TYPES:
continue
quality = 1.0
for parameter in parts[1:]:
name, _, value = parameter.partition("=")
if name.strip().lower() == "q":
try:
quality = float(value.strip())
except ValueError:
quality = 0.0
if quality > 0:
return True
return False