fix: 校验加固(嵌套敏感键/naive时间防御)
- _depth_ok 与敏感键扫描合并为单次递归 _walk_props,dict 所有层级键均过 _SENSITIVE(此前嵌套键不扫描,utm 白名单内可夹带敏感键) - occurred_at 为 naive datetime 时返回 invalid_occurred_at(原为 TypeError 500) - 删除未使用的 timedelta 导入 - 深度超限即停止下探,10 万层嵌套不爆栈 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -1,6 +1,6 @@
|
|||||||
import json
|
import json
|
||||||
import re
|
import re
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime
|
||||||
from urllib.parse import parse_qsl, urlencode, urlparse
|
from urllib.parse import parse_qsl, urlencode, urlparse
|
||||||
from django.utils import timezone
|
from django.utils import timezone
|
||||||
from tracker import constants as C
|
from tracker import constants as C
|
||||||
@@ -15,14 +15,28 @@ def sanitize_url(url: str) -> str:
|
|||||||
if not _URL_DROP_KEY.fullmatch(k)]
|
if not _URL_DROP_KEY.fullmatch(k)]
|
||||||
return p._replace(query=urlencode(kept)).geturl()
|
return p._replace(query=urlencode(kept)).geturl()
|
||||||
|
|
||||||
def _depth_ok(v, depth=0) -> bool:
|
def _walk_props(v, depth=0):
|
||||||
|
"""单次递归同时做深度检查与敏感键扫描(dict 所有层级的键都过 _SENSITIVE)。
|
||||||
|
返回 (depth_ok, sensitive_hit)。深度超限即停止下探,防恶意深嵌套递归爆栈。"""
|
||||||
if depth > C.MAX_PROP_DEPTH:
|
if depth > C.MAX_PROP_DEPTH:
|
||||||
return False
|
return False, False
|
||||||
if isinstance(v, dict):
|
if isinstance(v, dict):
|
||||||
return all(_depth_ok(x, depth + 1) for x in v.values())
|
depth_ok, sens = True, False
|
||||||
|
for k, x in v.items():
|
||||||
|
if _SENSITIVE.search(str(k)):
|
||||||
|
sens = True
|
||||||
|
d, s = _walk_props(x, depth + 1)
|
||||||
|
depth_ok = depth_ok and d
|
||||||
|
sens = sens or s
|
||||||
|
return depth_ok, sens
|
||||||
if isinstance(v, list):
|
if isinstance(v, list):
|
||||||
return all(_depth_ok(x, depth + 1) for x in v)
|
depth_ok, sens = True, False
|
||||||
return True
|
for x in v:
|
||||||
|
d, s = _walk_props(x, depth + 1)
|
||||||
|
depth_ok = depth_ok and d
|
||||||
|
sens = sens or s
|
||||||
|
return depth_ok, sens
|
||||||
|
return True, False
|
||||||
|
|
||||||
def validate_event(ev) -> str | None:
|
def validate_event(ev) -> str | None:
|
||||||
"""返回 None 表示通过;否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。"""
|
"""返回 None 表示通过;否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。"""
|
||||||
@@ -39,20 +53,22 @@ def validate_event(ev) -> str | None:
|
|||||||
occ = ev.get("occurred_at")
|
occ = ev.get("occurred_at")
|
||||||
if not isinstance(occ, datetime):
|
if not isinstance(occ, datetime):
|
||||||
return "invalid_occurred_at"
|
return "invalid_occurred_at"
|
||||||
|
if timezone.is_naive(occ):
|
||||||
|
return "invalid_occurred_at"
|
||||||
skew = (timezone.now() - occ).total_seconds()
|
skew = (timezone.now() - occ).total_seconds()
|
||||||
if abs(skew) > C.CLOCK_SKEW_REJECT_SEC:
|
if abs(skew) > C.CLOCK_SKEW_REJECT_SEC:
|
||||||
return "occurred_at_out_of_window"
|
return "occurred_at_out_of_window"
|
||||||
props = ev.get("properties") or {}
|
props = ev.get("properties") or {}
|
||||||
if not isinstance(props, dict) or len(props) > C.MAX_PROP_KEYS:
|
if not isinstance(props, dict) or len(props) > C.MAX_PROP_KEYS:
|
||||||
return "properties_invalid"
|
return "properties_invalid"
|
||||||
# 敏感键扫描必须先于属性白名单:白名单内无任何键命中敏感模式,
|
# 敏感键扫描(含嵌套层级)先于属性白名单:白名单内无任何键命中敏感模式,
|
||||||
# 若放在白名单之后,sensitive_field 将不可达(见 test_sensitive_property_rejected)。
|
# 若放在白名单之后,sensitive_field 将不可达(见 test_sensitive_property_rejected)。
|
||||||
for k in props:
|
depth_ok, sensitive_hit = _walk_props(props)
|
||||||
if _SENSITIVE.search(str(k)):
|
if sensitive_hit:
|
||||||
return "sensitive_field"
|
return "sensitive_field"
|
||||||
if not set(props) <= C.EVENT_PROP_WHITELIST.get(ev["event_name"], set()):
|
if not set(props) <= C.EVENT_PROP_WHITELIST.get(ev["event_name"], set()):
|
||||||
return "property_not_allowed"
|
return "property_not_allowed"
|
||||||
if not _depth_ok(props):
|
if not depth_ok:
|
||||||
return "properties_depth"
|
return "properties_depth"
|
||||||
if len(json.dumps(props, ensure_ascii=False, default=str)) > C.MAX_PROP_JSON_BYTES:
|
if len(json.dumps(props, ensure_ascii=False, default=str)) > C.MAX_PROP_JSON_BYTES:
|
||||||
return "properties_too_large"
|
return "properties_too_large"
|
||||||
|
|||||||
@@ -40,3 +40,10 @@ def test_sanitize_url_drops_sensitive_params():
|
|||||||
u = sanitize_url("/p?email=a@b.com&token=x&id=3&utm_source=google&monkey=1")
|
u = sanitize_url("/p?email=a@b.com&token=x&id=3&utm_source=google&monkey=1")
|
||||||
assert "email=" not in u and "token=" not in u and "utm_source=google" in u
|
assert "email=" not in u and "token=" not in u and "utm_source=google" in u
|
||||||
assert "monkey=1" in u # fullmatch:含 key 子串的正常参数不误伤
|
assert "monkey=1" in u # fullmatch:含 key 子串的正常参数不误伤
|
||||||
|
|
||||||
|
def test_nested_sensitive_key_rejected():
|
||||||
|
r = validate_event(_ev(properties={"utm": {"utm_a": 1, "user_password": "x"}}))
|
||||||
|
assert "sensitive" in r # utm 在 page_view 白名单内,嵌套敏感键仍须命中
|
||||||
|
|
||||||
|
def test_naive_datetime_rejected():
|
||||||
|
assert validate_event(_ev(occurred_at=NOW.replace(tzinfo=None))) == "invalid_occurred_at"
|
||||||
|
|||||||
Reference in New Issue
Block a user