129 lines
5.5 KiB
Python
129 lines
5.5 KiB
Python
|
"""操作日志入库前处理单元测试。
|
|||
|
|
|
|||
|
|
守的是入库前的**截断与脱敏**：
|
|||
|
|
|
|||
|
|
- 截断：各列超长在 MySQL 严格模式下抛 1406，而「审计失败不影响业务」的约定
|
|||
|
|
会把该异常吞掉 —— 整条日志静默消失；request_data 尤甚，中文一个字符
|
|||
|
|
占 3 字节，必须按 **UTF-8 字节边界**切，否则会切出乱码残字
|
|||
|
|
- 脱敏：密码是最容易从日志泄漏的凭据，request_data 入库前必经 scrub
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
from datetime import datetime
|
|||
|
|
|
|||
|
|
from app.models.operation_log import (
|
|||
|
|
ERROR_MESSAGE_MAX_LEN,
|
|||
|
|
REQUEST_DATA_MAX_BYTES,
|
|||
|
|
SUMMARY_MAX_LEN,
|
|||
|
|
)
|
|||
|
|
from app.repositories.operation_log_repository import _clip, _serialize_request_data
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestClip:
|
|||
|
|
"""按字符数截断（长度常量与数据库列宽均按字符定义）。"""
|
|||
|
|
|
|||
|
|
def test_超长截断至上限(self) -> None:
|
|||
|
|
result = _clip("x" * (SUMMARY_MAX_LEN + 50), SUMMARY_MAX_LEN)
|
|||
|
|
assert result is not None
|
|||
|
|
assert len(result) == SUMMARY_MAX_LEN
|
|||
|
|
|
|||
|
|
def test_未超长原样返回(self) -> None:
|
|||
|
|
assert _clip("正常长度", SUMMARY_MAX_LEN) == "正常长度"
|
|||
|
|
|
|||
|
|
def test_空值返回None(self) -> None:
|
|||
|
|
assert _clip(None, ERROR_MESSAGE_MAX_LEN) is None
|
|||
|
|
assert _clip("", ERROR_MESSAGE_MAX_LEN) is None
|
|||
|
|
|
|||
|
|
def test_中文按字符数截断不切字(self) -> None:
|
|||
|
|
"""str 切片天然按字符，不会切出半个汉字。"""
|
|||
|
|
result = _clip("中" * 100, 10)
|
|||
|
|
assert result == "中" * 10
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestSerializeRequestData:
|
|||
|
|
"""序列化：JSON 化 → 脱敏 → 字节边界截断。"""
|
|||
|
|
|
|||
|
|
def test_常规参数原样序列化(self) -> None:
|
|||
|
|
text = _serialize_request_data({"status": 0, "roles": ["user"]})
|
|||
|
|
assert text is not None
|
|||
|
|
assert json.loads(text) == {"status": 0, "roles": ["user"]}
|
|||
|
|
|
|||
|
|
def test_空值返回None(self) -> None:
|
|||
|
|
assert _serialize_request_data(None) is None
|
|||
|
|
assert _serialize_request_data({}) is None
|
|||
|
|
|
|||
|
|
def test_密码值被抹除而键名保留(self) -> None:
|
|||
|
|
"""键名保留便于审计阅读"提交过密码字段"，值必须抹掉。"""
|
|||
|
|
for payload in ({"password": "S3cret@2026"}, {"new_password": "S3cret@2026"}):
|
|||
|
|
text = _serialize_request_data(payload) or ""
|
|||
|
|
assert "S3cret@2026" not in text
|
|||
|
|
assert "***" in text
|
|||
|
|
|
|||
|
|
def test_手机号被脱敏(self) -> None:
|
|||
|
|
text = _serialize_request_data({"phone": "13812345678"}) or ""
|
|||
|
|
assert "13812345678" not in text
|
|||
|
|
assert "138****5678" in text
|
|||
|
|
|
|||
|
|
def test_超长中文按字节边界截断且无残字(self) -> None:
|
|||
|
|
"""75000 字节（25000 个汉字）必然超限：
|
|||
|
|
若按字符截断则字节数三倍超标；若按字节硬切可能切出半个字符。
|
|||
|
|
正确结果应是原始 JSON 串的**干净前缀**。"""
|
|||
|
|
payload = {"remark": "中" * 25000}
|
|||
|
|
text = _serialize_request_data(payload)
|
|||
|
|
assert text is not None
|
|||
|
|
assert len(text.encode("utf-8")) <= REQUEST_DATA_MAX_BYTES
|
|||
|
|
assert "<EFBFBD>" not in text # 无替换字符
|
|||
|
|
raw = json.dumps(payload, ensure_ascii=False)
|
|||
|
|
assert raw.startswith(text)
|
|||
|
|
|
|||
|
|
def test_未超长的中文完整保留(self) -> None:
|
|||
|
|
payload = {"remark": "中" * 100}
|
|||
|
|
text = _serialize_request_data(payload)
|
|||
|
|
assert text is not None
|
|||
|
|
assert json.loads(text) == payload
|
|||
|
|
|
|||
|
|
def test_不可JSON序列化的值降级为字符串(self) -> None:
|
|||
|
|
text = _serialize_request_data({"at": datetime(2026, 10, 10, 12, 0, 0)})
|
|||
|
|
assert text is not None
|
|||
|
|
assert "2026-10-10" in text
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestStructuredSanitization:
|
|||
|
|
"""结构化抹除 —— 按键名整值替换，不依赖文本正则。
|
|||
|
|
|
|||
|
|
回归背景：此前用文本正则脱敏，有三种逃逸形态 —— 值以空白开头时
|
|||
|
|
整体匹配失败（**完全不脱敏**）；值含空格/逗号时只抹首段；值含引号时
|
|||
|
|
既残留明文又破坏 JSON 结构。密码校验只查字节数、前后端都不 trim，
|
|||
|
|
这些形态都是正常可用密码。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
def test_前导空白密码被完全抹除(self) -> None:
|
|||
|
|
"""回归：此前 `" abc12345"` 原样明文入库。"""
|
|||
|
|
text = _serialize_request_data({"new_password": " abc12345"}) or ""
|
|||
|
|
assert "abc12345" not in text
|
|||
|
|
assert json.loads(text) == {"new_password": "***"}
|
|||
|
|
|
|||
|
|
def test_含空格逗号引号均整体抹除且JSON完整(self) -> None:
|
|||
|
|
for password in ("abc 12345", "abc,def", 'abc"def', "abc&def"):
|
|||
|
|
text = _serialize_request_data({"password": password}) or ""
|
|||
|
|
assert password not in text
|
|||
|
|
assert json.loads(text) == {"password": "***"}
|
|||
|
|
|
|||
|
|
def test_嵌套结构与键名大小写(self) -> None:
|
|||
|
|
text = (
|
|||
|
|
_serialize_request_data({"user": {"New_Password": "s3"}, "list": [{"TOKEN": "t"}]})
|
|||
|
|
or ""
|
|||
|
|
)
|
|||
|
|
assert "s3" not in text
|
|||
|
|
assert json.loads(text) == {"user": {"New_Password": "***"}, "list": [{"TOKEN": "***"}]}
|
|||
|
|
|
|||
|
|
def test_非敏感键不受影响(self) -> None:
|
|||
|
|
text = _serialize_request_data({"keyword": "a_b", "remark": "正常备注"}) or ""
|
|||
|
|
assert json.loads(text) == {"keyword": "a_b", "remark": "正常备注"}
|
|||
|
|
|
|||
|
|
def test_孤立代理字符不再抛错(self) -> None:
|
|||
|
|
"""回归：直接 encode('utf-8') 遇 \\ud800 抛 UnicodeEncodeError，
|
|||
|
|
该请求的审计记录会被兜底逻辑吞掉、静默丢失。"""
|
|||
|
|
text = _serialize_request_data({"username": "a\ud800b"})
|
|||
|
|
assert text is not None
|