fix: 校验守门按字节+递归长度+非dict拒绝
- 体积守门改 UTF-8 字节数(原按字符数,"汉"*1500 约 4.5KB 可绕过 4096 上限)
- 字符串长度检查并入 _walk_props 递归,嵌套字符串同样受 MAX_STR_LEN 约束
(顺序:体积先行,故同一值既超长又超体积时报 properties_too_large)
- properties 传了但非 dict([]/""/0/None)不再被 or {} 静默放行,缺省键仍放行
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -16,27 +16,32 @@ def sanitize_url(url: str) -> str:
|
||||
return p._replace(query=urlencode(kept)).geturl()
|
||||
|
||||
def _walk_props(v, depth=0):
|
||||
"""单次递归同时做深度检查与敏感键扫描(dict 所有层级的键都过 _SENSITIVE)。
|
||||
返回 (depth_ok, sensitive_hit)。深度超限即停止下探,防恶意深嵌套递归爆栈。"""
|
||||
"""单次递归同时做深度检查、敏感键扫描与字符串超长检查。
|
||||
dict 所有层级的键都过 _SENSITIVE,所有层级的字符串都受 MAX_STR_LEN 约束。
|
||||
返回 (depth_ok, sensitive_hit, str_too_long)。深度超限即停止下探,防恶意深嵌套递归爆栈。"""
|
||||
if depth > C.MAX_PROP_DEPTH:
|
||||
return False, False
|
||||
return False, False, False
|
||||
if isinstance(v, dict):
|
||||
depth_ok, sens = True, False
|
||||
depth_ok, sens, too_long = True, False, False
|
||||
for k, x in v.items():
|
||||
if _SENSITIVE.search(str(k)):
|
||||
sens = True
|
||||
d, s = _walk_props(x, depth + 1)
|
||||
d, s, l = _walk_props(x, depth + 1)
|
||||
depth_ok = depth_ok and d
|
||||
sens = sens or s
|
||||
return depth_ok, sens
|
||||
too_long = too_long or l
|
||||
return depth_ok, sens, too_long
|
||||
if isinstance(v, list):
|
||||
depth_ok, sens = True, False
|
||||
depth_ok, sens, too_long = True, False, False
|
||||
for x in v:
|
||||
d, s = _walk_props(x, depth + 1)
|
||||
d, s, l = _walk_props(x, depth + 1)
|
||||
depth_ok = depth_ok and d
|
||||
sens = sens or s
|
||||
return depth_ok, sens
|
||||
return True, False
|
||||
too_long = too_long or l
|
||||
return depth_ok, sens, too_long
|
||||
if isinstance(v, str) and len(v) > C.MAX_STR_LEN:
|
||||
return True, False, True
|
||||
return True, False, False
|
||||
|
||||
def validate_event(ev) -> str | None:
|
||||
"""返回 None 表示通过;否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。"""
|
||||
@@ -58,23 +63,23 @@ def validate_event(ev) -> str | None:
|
||||
skew = (timezone.now() - occ).total_seconds()
|
||||
if abs(skew) > C.CLOCK_SKEW_REJECT_SEC:
|
||||
return "occurred_at_out_of_window"
|
||||
props = ev.get("properties") or {}
|
||||
props = ev.get("properties", {})
|
||||
if not isinstance(props, dict) or len(props) > C.MAX_PROP_KEYS:
|
||||
return "properties_invalid"
|
||||
return "properties_invalid" # 传了但非 dict([]/""/0/None)一律拒
|
||||
# 敏感键扫描(含嵌套层级)先于属性白名单:白名单内无任何键命中敏感模式,
|
||||
# 若放在白名单之后,sensitive_field 将不可达(见 test_sensitive_property_rejected)。
|
||||
depth_ok, sensitive_hit = _walk_props(props)
|
||||
depth_ok, sensitive_hit, str_too_long = _walk_props(props)
|
||||
if sensitive_hit:
|
||||
return "sensitive_field"
|
||||
if not set(props) <= C.EVENT_PROP_WHITELIST.get(ev["event_name"], set()):
|
||||
return "property_not_allowed"
|
||||
if not depth_ok:
|
||||
return "properties_depth"
|
||||
if len(json.dumps(props, ensure_ascii=False, default=str)) > C.MAX_PROP_JSON_BYTES:
|
||||
# 体积按 UTF-8 字节数守门(汉字多字节,按字符数会被绕过)
|
||||
if len(json.dumps(props, ensure_ascii=False, default=str).encode("utf-8")) > C.MAX_PROP_JSON_BYTES:
|
||||
return "properties_too_large"
|
||||
for v in props.values():
|
||||
if isinstance(v, str) and len(v) > C.MAX_STR_LEN:
|
||||
return "properties_too_long"
|
||||
if str_too_long:
|
||||
return "properties_too_long"
|
||||
if len(ev.get("path") or "") > C.MAX_PATH_LEN:
|
||||
ev["path"] = ev["path"][:C.MAX_PATH_LEN]
|
||||
if len(ev.get("title") or "") > C.MAX_TITLE_LEN:
|
||||
|
||||
@@ -47,3 +47,13 @@ def test_nested_sensitive_key_rejected():
|
||||
|
||||
def test_naive_datetime_rejected():
|
||||
assert validate_event(_ev(occurred_at=NOW.replace(tzinfo=None))) == "invalid_occurred_at"
|
||||
|
||||
def test_prop_size_gate_counts_bytes():
|
||||
r = validate_event(_ev(properties={"utm": {"a": "汉" * 1500}}))
|
||||
assert r == "properties_too_large" # 字符数约 2K 未超,UTF-8 约 4.5KB 超 4096,须按字节算
|
||||
|
||||
def test_nested_string_length_capped():
|
||||
assert validate_event(_ev(properties={"utm": {"a": "x" * 600}})) == "properties_too_long"
|
||||
|
||||
def test_falsy_non_dict_properties_rejected():
|
||||
assert validate_event(_ev(properties=[])) == "properties_invalid"
|
||||
|
||||
Reference in New Issue
Block a user