diff --git a/app/api/controllers/agent_runs.py b/app/api/controllers/agent_runs.py index 37b519d..330e2b2 100644 --- a/app/api/controllers/agent_runs.py +++ b/app/api/controllers/agent_runs.py @@ -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, diff --git a/app/api/controllers/risk.py b/app/api/controllers/risk.py index be2ddcf..0cc2da1 100644 --- a/app/api/controllers/risk.py +++ b/app/api/controllers/risk.py @@ -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" diff --git a/app/api/dependencies/negotiation.py b/app/api/dependencies/negotiation.py new file mode 100644 index 0000000..857f936 --- /dev/null +++ b/app/api/dependencies/negotiation.py @@ -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 diff --git a/app/service/risk_daily_report_service.py b/app/service/risk_daily_report_service.py index 6944c14..1861b4d 100644 --- a/app/service/risk_daily_report_service.py +++ b/app/service/risk_daily_report_service.py @@ -64,6 +64,15 @@ class RiskDailyReportService: suggestions, source = await self._suggestions(report) return await self._complete(report, suggestions, source) + async def authorize(self, context: RequestContext) -> None: + """SSE 端点的前置鉴权。 + + `stream()` 是 async generator，函数体直到**第一次迭代**才执行；若只把 `require` + 放在那里，403 只能在响应头已经发出之后抛出，客户端看到的是"200 + 半截流"。 + 因此调用方必须先 `await authorize(context)` 再构造 `StreamingResponse`。 + """ + await AuthorizationService.require(context, "risk:alert:read") + async def stream( self, context: RequestContext, diff --git a/docs/05-接口文档.md b/docs/05-接口文档.md index 5508c41..4a0998c 100644 --- a/docs/05-接口文档.md +++ b/docs/05-接口文档.md @@ -153,6 +153,7 @@ Run Query Service -> RunRepository/ConversationRepository -> JSON/SSE View | `403` | 角色、权限、适当性或数据范围拒绝 | | `404` | 资源不存在，或为防止越权枚举而隐藏资源 | | `409` | 幂等冲突、版本冲突或非法状态转换 | +| `413` | 请求体或上传文件超过大小限制 | | `422` | 已解析请求不满足字段或业务输入约束 | | `429` | 频率、并发或配额限制 | | `500` | 未分类内部错误 | @@ -787,10 +788,14 @@ Outbox 消费者按 `event_id` 幂等。失败事件保留并重试，超过阈 | 客服工单 | `/api/v1/customer-service/handover-tickets/**` | 客服业务文档 | 可生成摘要和转人工请求，不分配、接单、解决或关闭工单 | | 投顾方案 | `/api/v1/advisory-plans/**` | 投顾业务文档 | 只生成分析草案，不代替投顾审核发布 | | 场内模拟交易 | `/api/v1/sim-orders/**` | 交易业务文档 | 只读查询，不创建、确认或撤销委托 | -| 风控扫描 | `/api/v1/risk-scans/**` | 风控业务文档 | 可解释规则结果，不启动人工处置 | -| 风险预警 | `/api/v1/risk-alerts/**` | 风控业务文档 | 只读分析，不确认、升级或关闭预警 | +| 风控扫描 | `/api/v1/risk/**` | 风控业务文档 | 可解释规则结果，不启动人工处置 | +| 风险预警 | `/api/v1/risk/**` | 风控业务文档 | 只读分析，不确认、升级或关闭预警 | | 场外基金运营 | 不属于当前系统 | 独立运营系统 | 不读写场内交易表 | +风控模块落地时把扫描、预警、证据、通知和日报收在同一个 Controller 下，入口为 +`/api/v1/risk/**`（早期规划写作 `/risk-scans/**`、`/risk-alerts/**`，以本节的实际入口为准）。 +具体端点清单、权限与字段映射由风控业务文档登记，见第 19 节末尾。 + B 类业务写接口必须复用 JWT、响应信封、错误码、幂等、统一 `RequestContext`、事务和审计规则。Controller 只路由、校验、映射和调用 Service；禁止直接访问 Model。 ## 13. A 类和 B 类扩展 diff --git a/docs/风控业务演示文档/06-模块接口与字段映射.md b/docs/风控业务演示文档/06-模块接口与字段映射.md index f0039b0..a9f6be1 100644 --- a/docs/风控业务演示文档/06-模块接口与字段映射.md +++ b/docs/风控业务演示文档/06-模块接口与字段映射.md @@ -67,7 +67,7 @@ |---|---|---|---| | POST | `/daily-report` | 生成结构化日报 | `risk:alert:read` | | POST | `/daily-report/stream` | 流式生成日报 | `risk:alert:read` | -| POST | `/daily-report/mail` | 发送日报邮件 | 按主项目邮件策略执行 | +| POST | `/daily-report/mail` | 发送日报邮件 | `risk:report:mail` | ## Agent Run diff --git a/tests/unit/api/test_risk_controller.py b/tests/unit/api/test_risk_controller.py index facb517..69d9ea0 100644 --- a/tests/unit/api/test_risk_controller.py +++ b/tests/unit/api/test_risk_controller.py @@ -94,6 +94,9 @@ class StubRiskDailyReportService: def __init__(self, _session: Any) -> None: pass + async def authorize(self, _context: RequestContext) -> None: + """SSE 端点要求构造 `StreamingResponse` 之前先完成鉴权（docs/25 P3 #24）。""" + async def generate(self, _context: RequestContext, _report_time: Any) -> dict[str, Any]: return {"report_date": "2026-09-10", "content": "日报正文"} diff --git a/tests/unit/api/test_risk_stream_negotiation.py b/tests/unit/api/test_risk_stream_negotiation.py new file mode 100644 index 0000000..83a3962 --- /dev/null +++ b/tests/unit/api/test_risk_stream_negotiation.py @@ -0,0 +1,103 @@ +"""风控 SSE 端点的内容协商与鉴权时序（`docs/25` P3 #24）。 + +`POST /api/v1/risk/daily-report/stream` 此前既不校验 `Accept`，又把鉴权留在 +async generator 内部 —— 后者更隐蔽：`StreamingResponse` 已经返回、响应头已经发出， +`403` 只能变成"200 + 半截流"。所以这里同时断言两件事： + +1. 显式只接受 `application/json` → `406 SSE_NOT_ACCEPTABLE`，且响应体是统一 JSON 错误； +2. 无权限时即使 `Accept` 也非法，仍先得到 `403 AGENT_PERMISSION_DENIED`（顺序与 + `docs/05` §6.4 一致：鉴权先行，防止用状态码差异做探测）。 +""" + +from collections.abc import AsyncIterator +from typing import Any + +import pytest +from fastapi.testclient import TestClient + +from app.api.dependencies.auth import build_request_context +from app.api.dependencies.database import get_session +from app.core.contracts import RequestContext +from app.core.errors import AgentPermissionDeniedError +from app.main import create_app + +STREAM_PATH = "/api/v1/risk/daily-report/stream" + + +async def resolve_context() -> RequestContext: + return RequestContext( + user_id="1", + trace_id="trace-1", + permissions=("risk:alert:read",), + data_scope="all", + ) + + +class StubReportService: + """替身：`authorize` 通过、流立即收尾，避免测试连接挂住。""" + + def __init__(self, session: Any) -> None: + self.session = session + + async def authorize(self, _context: RequestContext) -> None: + return None + + async def stream( + self, + _context: RequestContext, + _now: Any, + ) -> AsyncIterator[dict[str, Any]]: + yield {"type": "done", "report": {}} + + +class DenyingReportService(StubReportService): + async def authorize(self, _context: RequestContext) -> None: + raise AgentPermissionDeniedError("缺少 risk:alert:read 权限") + + +def client_with( + monkeypatch: pytest.MonkeyPatch, + service_class: type[StubReportService], +) -> TestClient: + application = create_app() + application.dependency_overrides[build_request_context] = resolve_context + application.dependency_overrides[get_session] = lambda: None + monkeypatch.setattr("app.api.controllers.risk.RiskDailyReportService", service_class) + return TestClient(application) + + +@pytest.mark.parametrize("accept", [None, "", "*/*", "text/*", "text/event-stream"]) +def test_absent_or_wildcard_accept_is_allowed( + monkeypatch: pytest.MonkeyPatch, accept: str | None +) -> None: + headers = {} if accept is None else {"Accept": accept} + with client_with(monkeypatch, StubReportService) as client: + response = client.post(STREAM_PATH, json={}, headers=headers) + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/event-stream") + + +def test_json_only_accept_is_rejected_before_streaming( + monkeypatch: pytest.MonkeyPatch, +) -> None: + with client_with(monkeypatch, StubReportService) as client: + response = client.post( + STREAM_PATH, json={}, headers={"Accept": "application/json"} + ) + + assert response.status_code == 406 + assert response.json()["error"]["code"] == "SSE_NOT_ACCEPTABLE" + assert response.headers["content-type"].startswith("application/json") + + +def test_authorization_precedes_accept_negotiation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + with client_with(monkeypatch, DenyingReportService) as client: + response = client.post( + STREAM_PATH, json={}, headers={"Accept": "application/json"} + ) + + assert response.status_code == 403 + assert response.json()["error"]["code"] == "AGENT_PERMISSION_DENIED"