diff --git a/app/config/settings.py b/app/config/settings.py index b96c208..6fb1bfe 100644 --- a/app/config/settings.py +++ b/app/config/settings.py @@ -1,5 +1,7 @@ """环境配置(从 .env 读取,见 .env.example)。""" +from decimal import Decimal + from pydantic_settings import BaseSettings, SettingsConfigDict @@ -29,5 +31,17 @@ class Settings(BaseSettings): deepseek_api_key: str = "" deepseek_base_url: str = "https://api.deepseek.com" + # ===== Risk 阈值(默认值=冻结规则 · docs/PRD/附-风控规则表.md)===== + risk_assessment_valid_days: int = 365 + risk_large_amount: Decimal = Decimal("500000") + risk_daily_total: Decimal = Decimal("500000") + risk_freq_count: int = 3 + risk_probe_window_minutes: int = 5 + risk_probe_count: int = 3 + risk_probe_amount: Decimal = Decimal("400000") + risk_small_amount: Decimal = Decimal("10000") + risk_small_count: int = 3 + risk_aml_default_threshold: Decimal = Decimal("0.85") + settings = Settings() diff --git a/app/utils/desensitize.py b/app/utils/desensitize.py new file mode 100644 index 0000000..676fbed --- /dev/null +++ b/app/utils/desensitize.py @@ -0,0 +1,43 @@ +"""数据脱敏纯函数(DESENS-001~004 · docs/PRD/附-风控规则表.md §4)。 + +原则:API 出参 / 预警 payload / LLM 上下文传入前必须已脱敏; +customer_id 为系统内部代理键,不脱敏(DESENS 规则明确); +DESENS-005(资产金额展示)归前端组件,后端不实现。 +""" + +from __future__ import annotations + + +def mask_id_card(value: str | None) -> str: + """DESENS-001 身份证号:保留前 3 后 4,中间 ****。""" + return _mask_middle(value, head=3, tail=4) + + +def mask_phone(value: str | None) -> str: + """DESENS-002 手机号:保留前 3 后 4,中间 ****。""" + return _mask_middle(value, head=3, tail=4) + + +def mask_name(value: str | None) -> str: + """DESENS-003 姓名:保留姓氏,其余 **。""" + if not value: + return "" + return value[0] + "**" + + +def mask_bank_card(value: str | None) -> str: + """DESENS-004 银行卡号:保留后 4 位,其余 ****。""" + if not value: + return "" + if len(value) <= 4: + return "*" * len(value) + return "*" * (len(value) - 4) + value[-4:] + + +def _mask_middle(value: str | None, head: int, tail: int) -> str: + """通用中段掩码;长度不足 head+tail 时整体掩码(宁多勿漏)。""" + if not value: + return "" + if len(value) <= head + tail: + return "*" * len(value) + return value[:head] + "****" + value[-tail:] diff --git a/app/utils/trace.py b/app/utils/trace.py new file mode 100644 index 0000000..3dd2569 --- /dev/null +++ b/app/utils/trace.py @@ -0,0 +1,26 @@ +"""trace_id 全链路贯通(contextvars · 架构 §5.1)。 + +约束(编码前必读): +- 中间件必须在 call_next 之前 set(),禁止在 endpoint/service 内重新 set(); +- 后台任务(create_task)须显式 contextvars.copy_context(); +- 同步 def 路由跑线程池时 context 经 anyio 传播,由 B8 trace 一致性断言实测兜底。 +""" + +from __future__ import annotations + +from contextvars import ContextVar +from uuid import uuid4 + +_trace_id: ContextVar[str] = ContextVar("trace_id", default="") + + +def new_trace(trace_id: str | None = None) -> str: + """生成并绑定新 trace_id;透传外部 X-Trace-Id 时传入该值。""" + tid = trace_id or f"trc-{uuid4().hex[:16]}" + _trace_id.set(tid) + return tid + + +def current_trace() -> str: + """读取当前 trace_id;未初始化时返回空串(调用方应兜底生成)。""" + return _trace_id.get() diff --git a/requirements.txt b/requirements.txt index 83d473f..8cec466 100644 --- a/requirements.txt +++ b/requirements.txt @@ -21,3 +21,6 @@ python-jose[cryptography]>=3.3.0 # Utils python-multipart>=0.0.18 + +# Testing +pytest>=8.3.0 diff --git a/tests/test_desensitize.py b/tests/test_desensitize.py new file mode 100644 index 0000000..c70480a --- /dev/null +++ b/tests/test_desensitize.py @@ -0,0 +1,61 @@ +"""desensitize.py 单测(A1 · DESENS-001~004,005 归前端)。""" + +from app.utils.desensitize import mask_bank_card, mask_id_card, mask_name, mask_phone + + +class TestMaskIdCard: + def test_normal_18(self): + assert mask_id_card("110101199003071234") == "110****1234" + + def test_short_overmasked(self): + # 长度不足 head+tail:整体掩码(宁多勿漏) + assert mask_id_card("123456") == "******" + + def test_none_and_empty(self): + assert mask_id_card(None) == "" + assert mask_id_card("") == "" + + +class TestMaskPhone: + def test_normal_11(self): + assert mask_phone("13812349527") == "138****9527" + + def test_boundary_7(self): + # 恰好 head+tail 长度 → 整体掩码 + assert mask_phone("1234567") == "*******" + + def test_boundary_8(self): + # 8 = 3+4+1,可保留头尾 + assert mask_phone("12345678") == "123****5678" + + +class TestMaskName: + def test_keeps_surname(self): + assert mask_name("张三丰") == "张**" + assert mask_name("李四") == "李**" + + def test_single_char(self): + # 单字名:姓氏 + **(规则如此,不特判) + assert mask_name("王") == "王**" + + def test_none_and_empty(self): + assert mask_name(None) == "" + assert mask_name("") == "" + + +class TestMaskBankCard: + def test_normal_16(self): + assert mask_bank_card("6222021234561234") == "************1234" + + def test_keep_last4_only(self): + masked = mask_bank_card("6222021234561234") + assert masked[-4:] == "1234" + assert set(masked[:-4]) == {"*"} + + + def test_short_overmasked(self): + assert mask_bank_card("123") == "***" + + def test_none_and_empty(self): + assert mask_bank_card(None) == "" + assert mask_bank_card("") == "" diff --git a/tests/test_settings.py b/tests/test_settings.py new file mode 100644 index 0000000..08dde41 --- /dev/null +++ b/tests/test_settings.py @@ -0,0 +1,25 @@ +"""settings risk_* 配置单测(A1):默认值=冻结规则。""" + +from decimal import Decimal + +from app.config.settings import Settings + + +def test_risk_defaults_match_frozen_rules(): + s = Settings(_env_file=None) # 不读 .env,纯默认值 + assert s.risk_assessment_valid_days == 365 + assert s.risk_large_amount == Decimal("500000") + assert s.risk_daily_total == Decimal("500000") + assert s.risk_freq_count == 3 + assert s.risk_probe_window_minutes == 5 + assert s.risk_probe_count == 3 + assert s.risk_probe_amount == Decimal("400000") + assert s.risk_small_amount == Decimal("10000") + assert s.risk_small_count == 3 + assert s.risk_aml_default_threshold == Decimal("0.85") + + +def test_amounts_are_decimal(): + s = Settings(_env_file=None) + assert isinstance(s.risk_large_amount, Decimal) + assert isinstance(s.risk_aml_default_threshold, Decimal) diff --git a/tests/test_trace.py b/tests/test_trace.py new file mode 100644 index 0000000..9462638 --- /dev/null +++ b/tests/test_trace.py @@ -0,0 +1,21 @@ +"""trace.py 单测(A1)。""" + +from app.utils.trace import current_trace, new_trace + + +def test_new_trace_generates_and_binds(): + tid = new_trace() + assert tid.startswith("trc-") + assert len(tid) == 20 # trc- + 16 hex + assert current_trace() == tid + + +def test_new_trace_accepts_external_id(): + tid = new_trace("trc-external-0001") + assert current_trace() == "trc-external-0001" + assert tid == "trc-external-0001" + + +def test_uniqueness(): + ids = {new_trace() for _ in range(100)} + assert len(ids) == 100