diff --git a/app/core/risk_cursor.py b/app/core/risk_cursor.py index 0481de7..6a28883 100644 --- a/app/core/risk_cursor.py +++ b/app/core/risk_cursor.py @@ -1,22 +1,66 @@ -"""风控列表游标；对外是不透明字符串，内部保存页偏移。""" +"""风控列表游标；对外是不透明字符串，内部保存页偏移与**绑定指纹**。 + +`docs/05` §3.8 要求「游标是不透明字符串，绑定用户、查询条件、排序字段和方向， +客户端不得解析或修改」；§3.6 的错误表把「**与过滤条件不符**」明确列为 `INVALID_CURSOR` +的触发条件之一（§16 同样写「分页游标绑定过滤条件」）。 + +原先这里只存 `{"offset": n}`，什么都没绑定，于是： + +- 任何拿到游标的人都能拿它去翻**别人的**结果集（越权）； +- 改了筛选条件还能继续用旧游标，而 offset 分页在结果集变化时本来就会跳行/重复， + 两者叠加会**静默返回错位的数据**。 + +现在游标里额外存一个**绑定指纹**：由「用户 + 查询条件 + 排序」压成。服务端解码时按当前 +请求重算指纹并比对，不一致就报 `INVALID_CURSOR`。 + +指纹用 SHA-256（无需密钥）：它要防的是"无意复用"，不是恶意伪造。 +真正的防篡改需要 HMAC 与密钥管理，属于后续加固项 —— 这里不假装做到了。 +""" import base64 +import hashlib import json +from collections.abc import Mapping from typing import Any from app.core.errors import InvalidCursorError -__all__ = ["decode_offset_cursor", "encode_offset_cursor"] +__all__ = ["cursor_binding", "decode_offset_cursor", "encode_offset_cursor"] + +_BINDING_KEY = "b" -def encode_offset_cursor(offset: int) -> str: +def cursor_binding(*, user_id: str, filters: Mapping[str, Any], order_by: str = "") -> str: + """把「谁、按什么条件、按什么顺序」压成一个稳定指纹。 + + `sort_keys=True` 让字典顺序不影响结果；`default=str` 兜住日期与 Decimal 之类不可直接 + 序列化的筛选值（它们同样应当参与比对）。 + """ + material = json.dumps( + { + "user_id": str(user_id), + "filters": {str(key): value for key, value in filters.items()}, + "order_by": order_by, + }, + sort_keys=True, + ensure_ascii=False, + default=str, + separators=(",", ":"), + ) + return hashlib.sha256(material.encode("utf-8")).hexdigest()[:32] + + +def encode_offset_cursor(offset: int, *, binding: str) -> str: if offset < 0: raise ValueError("offset must be non-negative") - payload = json.dumps({"offset": offset}, separators=(",", ":")).encode("utf-8") + payload = json.dumps( + {"offset": offset, _BINDING_KEY: binding}, separators=(",", ":") + ).encode("utf-8") return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=") -def decode_offset_cursor(raw: str | None) -> int: +def decode_offset_cursor(raw: str | None, *, binding: str) -> int: + """解码游标并校验绑定指纹；任何不一致都按非法游标处理（400 INVALID_CURSOR）。""" if raw is None or not raw.strip(): return 0 value = raw.strip() @@ -26,8 +70,12 @@ def decode_offset_cursor(raw: str | None) -> int: decoded: Any = json.loads(payload.decode("utf-8")) except (ValueError, UnicodeDecodeError, json.JSONDecodeError) as error: raise InvalidCursorError("cursor 非法或已过期") from error - if not isinstance(decoded, dict) or set(decoded) != {"offset"}: + if not isinstance(decoded, dict) or set(decoded) != {"offset", _BINDING_KEY}: raise InvalidCursorError("cursor 非法或已过期") + if decoded[_BINDING_KEY] != binding: + # 换了用户、改了筛选条件或排序：旧游标对新查询没有意义。继续用会静默返回错位的 + # 数据 —— 这正是 docs/05 把"与过滤条件不符"列为 INVALID_CURSOR 的原因。 + raise InvalidCursorError("cursor 与当前查询条件不符") offset = decoded["offset"] if not isinstance(offset, int) or isinstance(offset, bool) or offset < 0: raise InvalidCursorError("cursor 非法或已过期") diff --git a/app/service/risk_notification_service.py b/app/service/risk_notification_service.py index bb4128c..fe326bd 100644 --- a/app/service/risk_notification_service.py +++ b/app/service/risk_notification_service.py @@ -31,6 +31,7 @@ class RiskNotificationService: query: RiskNotificationPageQuery, ) -> dict[str, Any]: await AuthorizationService.require(context, "risk:alert:read") + binding = RiskQueryService._binding(context, query, "/notifications") page = await RiskRepository( self.session, scope=scope_from_context(context), @@ -41,10 +42,10 @@ class RiskNotificationService: end_time=query.end_time, page=PageRequest( limit=query.limit, - offset=decode_offset_cursor(query.cursor), + offset=decode_offset_cursor(query.cursor, binding=binding), ), ) - return RiskQueryService._page(page) + return RiskQueryService._page(page, binding=binding) def create_in_app( self, diff --git a/app/service/risk_query_service.py b/app/service/risk_query_service.py index 87f319f..b72bff7 100644 --- a/app/service/risk_query_service.py +++ b/app/service/risk_query_service.py @@ -12,7 +12,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.api.schemas.risk import RiskAlertPageQuery, RiskEvidencePageQuery from app.core.contracts import RequestContext from app.core.errors import GenericResourceNotFoundError -from app.core.risk_cursor import decode_offset_cursor, encode_offset_cursor +from app.core.risk_cursor import cursor_binding, decode_offset_cursor, encode_offset_cursor from app.core.timeutil import from_local from app.repository.fund_query_repository import CustomerScope, PageRequest from app.repository.risk_repository import RiskRepository @@ -68,6 +68,7 @@ class RiskQueryService: query: RiskAlertPageQuery, ) -> dict[str, Any]: await AuthorizationService.require(context, "risk:alert:read") + binding = self._binding(context, query) page = await self._repository(context).list_alerts( keyword=query.keyword, customer_no=query.customer_no, @@ -80,9 +81,12 @@ class RiskQueryService: # 两条路径口径必须一致，否则同一个筛选条件在界面与对话里查出不同结果。 start_time=from_local(query.start_time) if query.start_time else None, end_time=from_local(query.end_time) if query.end_time else None, - page=PageRequest(limit=query.limit, offset=decode_offset_cursor(query.cursor)), + page=PageRequest( + limit=query.limit, + offset=decode_offset_cursor(query.cursor, binding=binding), + ), ) - return self._page(page) + return self._page(page, binding=binding) async def get_alert_detail( self, @@ -102,9 +106,10 @@ class RiskQueryService: query: RiskEvidencePageQuery, ) -> dict[str, Any]: await AuthorizationService.require(context, "risk:alert:read") + binding = self._binding(context, query, source) page_request = PageRequest( limit=query.limit, - offset=decode_offset_cursor(query.cursor), + offset=decode_offset_cursor(query.cursor, binding=binding), ) repository = self._repository(context) if source == "customers": @@ -140,7 +145,7 @@ class RiskQueryService: ) elif source == "alerts": alert_page = await repository.list_alerts(keyword=query.keyword, page=page_request) - return self._page(alert_page) + return self._page(alert_page, binding=binding) elif source == "notifications": page = await repository.list_notifications( keyword=query.keyword, @@ -151,19 +156,40 @@ class RiskQueryService: ) else: raise GenericResourceNotFoundError("证据类型不存在") - return self._page(page) + return self._page(page, binding=binding) def _repository(self, context: RequestContext) -> RiskRepository: if self.repository is not None: return self.repository return RiskRepository(self.session, scope=scope_from_context(context)) + @staticmethod + def _binding(context: RequestContext, query: Any, extra: Any = None) -> str: + """列表查询的游标绑定指纹：用户 + 查询条件（**不含分页参数**）。 + + `docs/05` §3.8 要求游标绑定「用户、查询条件、排序字段和方向」。刻意排除 + `limit` 与 `cursor`：它们是分页参数、不是"查什么"；若算进指纹，客户翻页时改一下 + limit 就会让游标失效 —— 那不是安全要求，只是难用。 + + `model_dump(mode="json")` 保证取值可稳定序列化（datetime / Decimal 都会变成确定的 + 字符串），否则同一个查询两次算出的指纹可能不同。 + + `extra` 用于路径参数那种"不在 query 里但决定查哪张表"的条件（`/evidence/{source}`）， + 不带上它的话 customers 的游标可以直接拿去翻 products。 + """ + filters = query.model_dump(exclude={"limit", "cursor"}, mode="json") + if extra is not None: + filters["_path"] = extra + # data_scope 同样决定"能查到什么"：权限被收紧后，旧游标不该还能继续往下翻。 + filters["_scope"] = [context.data_scope, list(context.customer_ids or ())] + return cursor_binding(user_id=context.user_id, filters=filters) + @classmethod - def _page(cls, page: Any) -> dict[str, Any]: + def _page(cls, page: Any, *, binding: str) -> dict[str, Any]: return { "items": [cls._record(item) for item in page.items], "next_cursor": ( - encode_offset_cursor(page.next_offset) + encode_offset_cursor(page.next_offset, binding=binding) if page.next_offset is not None else None ), diff --git a/tests/unit/service/test_risk_query_service.py b/tests/unit/service/test_risk_query_service.py index d492d9a..8b29e60 100644 --- a/tests/unit/service/test_risk_query_service.py +++ b/tests/unit/service/test_risk_query_service.py @@ -7,7 +7,7 @@ import pytest from app.api.schemas.risk import RiskAlertPageQuery, RiskEvidencePageQuery from app.core.contracts import RequestContext from app.core.errors import GenericResourceNotFoundError, InvalidCursorError -from app.core.risk_cursor import decode_offset_cursor, encode_offset_cursor +from app.core.risk_cursor import cursor_binding, decode_offset_cursor, encode_offset_cursor from app.repository.fund_query_repository import CustomerScope, FundPage, FundRecord from app.service.risk_query_service import RiskQueryService, scope_from_context @@ -64,12 +64,15 @@ def context(**updates) -> RequestContext: def test_cursor_round_trip_and_invalid_values() -> None: - assert decode_offset_cursor(encode_offset_cursor(20)) == 20 - assert decode_offset_cursor(None) == 0 + binding = cursor_binding(user_id="990000002", filters={"keyword": None}) + assert ( + decode_offset_cursor(encode_offset_cursor(20, binding=binding), binding=binding) == 20 + ) + assert decode_offset_cursor(None, binding=binding) == 0 with pytest.raises(InvalidCursorError): - decode_offset_cursor("not-a-cursor") + decode_offset_cursor("not-a-cursor", binding=binding) with pytest.raises(InvalidCursorError): - decode_offset_cursor(encode_offset_cursor(1) + "x") + decode_offset_cursor(encode_offset_cursor(1, binding=binding) + "x", binding=binding) def test_query_schema_enforces_business_page_sizes() -> None: @@ -105,14 +108,55 @@ async def test_overview_maps_levels_and_serializes_high_priority() -> None: @pytest.mark.asyncio async def test_alert_page_returns_opaque_cursor() -> None: service = RiskQueryService(None, repository=FakeRepository()) + query = RiskAlertPageQuery(limit=5) - result = await service.list_alerts(context(), RiskAlertPageQuery(limit=5)) + result = await service.list_alerts(context(), query) assert result["has_more"] is True - assert decode_offset_cursor(result["next_cursor"]) == 5 + binding = RiskQueryService._binding(context(), query) + assert decode_offset_cursor(result["next_cursor"], binding=binding) == 5 assert result["items"][0]["rule_codes"] == ["RW-007"] +def test_cursor_is_bound_to_user_and_filters() -> None: + """§3.8：游标绑定用户与查询条件，换个人或换个筛选条件都必须失效。""" + query = RiskAlertPageQuery(limit=5) + raw = encode_offset_cursor(5, binding=RiskQueryService._binding(context(), query)) + + assert decode_offset_cursor(raw, binding=RiskQueryService._binding(context(), query)) == 5 + + with pytest.raises(InvalidCursorError): + decode_offset_cursor( + raw, + binding=RiskQueryService._binding(context(user_id="990000003"), query), + ) + with pytest.raises(InvalidCursorError): + other_filter = RiskAlertPageQuery(limit=5, keyword="张三") + decode_offset_cursor( + raw, + binding=RiskQueryService._binding(context(), other_filter), + ) + # data_scope 收紧后旧游标同样失效 + with pytest.raises(InvalidCursorError): + decode_offset_cursor( + raw, + binding=RiskQueryService._binding( + context(data_scope="own_customers", customer_ids=("9",)), query + ), + ) + + +def test_cursor_ignores_page_size_and_other_filters_do_not_collide() -> None: + """翻页时改 limit 不该让游标失效；不同证据类型之间不能互相串用游标。""" + base = RiskEvidencePageQuery(limit=10) + binding = RiskQueryService._binding(context(), base, "customers") + + again = RiskQueryService._binding(context(), RiskEvidencePageQuery(limit=10), "customers") + other = RiskQueryService._binding(context(), RiskEvidencePageQuery(limit=10), "products") + assert binding == again + assert binding != other + + @pytest.mark.asyncio async def test_missing_alert_detail_is_hidden() -> None: service = RiskQueryService(None, repository=FakeRepository())