Files
meida_backend/tests/unit/test_log_redaction.py
xsl f46ecc5eb0 fix(llm): price by call time and stop paying for thinking tokens
Two defects that only surface against the real DeepSeek API, both of which
break the cost model rather than the code:

deepseek-v4-flash enables thinking mode by default. The same single-turn
question took 13.5s with it on and 1.8s with it off, and the thinking tokens
are billed at the output rate. That misses NFR-PERF-02b (15s per turn) and
inflates every conversational call. SceneConfig now carries an explicit
`thinking` flag, all four V1 scenes set it to false, and the provider always
sends the parameter rather than relying on the upstream default. Reasoning
tokens are recorded separately in Usage.tokens_reasoning so the cost of ever
turning it back on is visible.

DeepSeek moves to peak/off-peak pricing on 2026-08-17, with peak (Beijing
09:00-12:00 and 14:00-18:00) at twice the off-peak rate. Reports are generated
mostly during peak hours, so a single blended rate would leave ob_ai_call --
the only input to V2 pricing -- off by nearly a factor of two.
PeakAwareRateTable resolves the rate at call time; cost is computed and stored
then, never recomputed later at whatever the current rate happens to be.

Also here:
- Replace the live API key in the redaction test with a synthetic one. The
  test that proves we redact secrets was itself a secret leak waiting for its
  first commit.
- Correct LLM_COST_RATES_JSON, which held placeholder rates.
- Make ids.SnowflakeGenerator take an injectable clock so rollover and
  clock-rollback are testable without patching a slots class.
- Rename exceptions to the -Error suffix (N818) and clear the remaining ruff,
  mypy and import-linter findings, including two dead lines: a duplicate
  pytestmark assignment and a no-op assert on a None return.

Gates: ruff, mypy (96 files incl. tests), 4/4 import contracts,
detect-secrets, 166 unit + 3 integration tests, migration up/down round-trip.
2026-08-14 00:54:40 +08:00

78 lines
2.7 KiB
Python
Executable File

"""Log redaction (NFR-SEC-03, 架构文档 §11.2).
日志中禁止出现手机号、UnionID、令牌、图片地址与诊断结论。
"""
from __future__ import annotations
from packages.shared.logging import REDACTED, make_redaction_processor
def redact(event: dict[str, object], extra: tuple[str, ...] = ()) -> dict[str, object]:
return dict(make_redaction_processor(extra)(None, "info", dict(event)))
def test_sensitive_top_level_keys_are_masked() -> None:
out = redact(
{
"event": "login",
"phone": "13800138000",
"union_id": "u-123",
"access_token": "abc.def.ghi",
"avatar_key": "user/1/avatar.png",
"user_id": 42,
}
)
assert out["phone"] == REDACTED
assert out["union_id"] == REDACTED
assert out["access_token"] == REDACTED
assert out["avatar_key"] == REDACTED
assert out["user_id"] == 42 # internal id is not sensitive on its own
def test_nested_structures_are_redacted() -> None:
out = redact({"event": "x", "payload": {"inner": {"phone": "13800138000", "ok": 1}}})
payload = out["payload"]
assert isinstance(payload, dict)
assert payload["inner"]["phone"] == REDACTED
assert payload["inner"]["ok"] == 1
def test_phone_numbers_inside_free_text_are_masked() -> None:
out = redact({"event": "sms", "detail": "验证码已发送至 13800138000"})
assert "13800138000" not in str(out["detail"])
assert "138****8000" in str(out["detail"])
# Synthetic key. Never paste a live key here: this file is committed, so a real
# key would land in git history permanently on the first commit.
FAKE_API_KEY = "sk-" + "0" * 8 + "deadbeef" + "0" * 16
def test_api_keys_in_free_text_are_masked() -> None:
out = redact({"event": "cfg", "detail": f"using {FAKE_API_KEY}"})
assert FAKE_API_KEY not in str(out["detail"])
assert REDACTED in str(out["detail"])
def test_bearer_tokens_in_free_text_are_masked() -> None:
out = redact({"event": "req", "detail": "Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.x.y"})
assert "eyJhbGciOiJIUzI1NiJ9" not in str(out["detail"])
def test_diagnosis_output_is_treated_as_sensitive() -> None:
out = redact({"event": "report", "dimensions": {"color_season": "浅暖柔和型"}})
assert out["dimensions"] == REDACTED
def test_extra_keys_can_be_configured() -> None:
out = redact({"event": "x", "custom_field": "secret"}, extra=("custom_field",))
assert out["custom_field"] == REDACTED
def test_lists_are_traversed() -> None:
out = redact({"event": "x", "items": [{"phone": "13800138000"}]})
items = out["items"]
assert isinstance(items, list)
assert items[0]["phone"] == REDACTED