From 87b0b419d671ecb7716bffc5d194045eadead47b Mon Sep 17 00:00:00 2001 From: Belugary <53219544+Belugary@users.noreply.github.com> Date: Sat, 23 May 2026 18:13:27 +0800 Subject: [PATCH] =?UTF-8?q?fix(export=5Fsns):=20=5Fparse=5Ftimeline=5Fxml?= =?UTF-8?q?=20=E5=85=BC=E5=AE=B9=204=20=E7=A7=8D=20content=20=E7=BC=96?= =?UTF-8?q?=E7=A0=81=20+=20=E8=80=81=20XML=20=E6=B8=85=E6=B4=97=20(#119)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 朋友圈 XML 解析跨版本兼容 + 顺手堵 XXE 绕过. 1. SnsTimeLine.content 跨版本会以 4 种编码出现: bytes (zstd 压缩, magic 28 B5 2F FD) / plain XML / hex / base64. 老逻辑直接 ET.fromstring 喂 raw 入参, 后三种 ParseError → 整条 row 静默丢失, 不报错不警告. 2. 老朋友圈 (2013-2017) XML 还含 ElementTree 拒绝的字符: URL 里的裸 &, 文本字段手打的 < >, 控制字符 (\x00-\x08). 一样导致 ParseError → 静默丢 row. 3. 安全修复: XXE / 长度检查老逻辑跑在 raw 入参, zstd / hex / base64 编码的恶意 DOCTYPE 能绕过. 新逻辑跑在 decoded payload 上, 拦得住. 修复: - _decode_sns_content_blob: 按 bytes / 已 XML / hex / base64 顺序检测, bytes 带 zstd magic 时先解压 - _sanitize_sns_pseudo_xml: 剥控制字符, CDATA 外裸 & 转义, text-only 节点内部裸 < > 转义 - _parse_timeline_xml: decode → 安全检查 (decoded payload) → sanitize → ET 测试: 17 case (DecodeContentBlobTests 10 + SanitizePseudoXmlTests 4 + SecurityAndLimitsTests 3), 全合成 XML 无 PII. --- export_sns.py | 120 ++++++++++++++++++- tests/test_export_sns_parse.py | 209 +++++++++++++++++++++++++++++++++ 2 files changed, 325 insertions(+), 4 deletions(-) create mode 100644 tests/test_export_sns_parse.py diff --git a/export_sns.py b/export_sns.py index fe28ab2..b3965a9 100644 --- a/export_sns.py +++ b/export_sns.py @@ -5,7 +5,10 @@ 汇总文件: //SNS/timeline.json 时间线: //SNS/timeline.html """ +import base64 +import binascii import bisect +import html import os import sys import json @@ -17,6 +20,8 @@ from datetime import datetime from urllib.request import urlopen, Request from urllib.error import URLError +import zstandard as zstd + if sys.platform == "win32": sys.stdout.reconfigure(encoding="utf-8", errors="replace") @@ -29,6 +34,106 @@ from decode_image import aligned_aes_block_size _SNS_XML_UNSAFE_RE = re.compile(r'（用户在 contentDesc 等里手打的尖括号） +# - 控制字符（\x00-\x08 等 XML 1.0 禁字符） +# CDATA 块内的 & < > 是合法的，不能动 —— 必须先把 CDATA 圈出来再清洗外面。 +_SNS_INVALID_CTRL_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]") +_SNS_CDATA_BLOCK_RE = re.compile(r"", re.DOTALL) +_SNS_BARE_AMP_RE = re.compile( + r"&(?!(amp|lt|gt|quot|apos|#\d+|#x[0-9a-fA-F]+);)" +) +_SNS_TEXT_ONLY_NODES = ( + "content", "title", "description", "nickname", "contentDesc", + "appname", "sourceName", "sourcename", "poiName", "displayName", + "feeddesc", +) +_SNS_TEXT_NODE_RE = re.compile( + r"(<(" + "|".join(_SNS_TEXT_ONLY_NODES) + r")\b[^>]*>)(.*?)()", + re.DOTALL, +) + + +def _decode_sns_content_blob(value): + """把 SnsTimeLine.content 列(任意编码形态)转成 UTF-8 XML 字符串。 + + bytes 优先尝试 zstd 解压；字符串按 plain XML / hex / base64 顺序检测。 + 无法识别时返回原值的 string 形态，让上层 ET.fromstring 自然 ParseError。 + None / 空值 → 空字符串。 + """ + if value is None: + return "" + + if isinstance(value, (bytes, bytearray)): + raw = bytes(value) + if raw.startswith(_SNS_ZSTD_MAGIC): + try: + raw = zstd.ZstdDecompressor().decompress(raw) + except Exception: + pass + return html.unescape(raw.decode("utf-8", errors="ignore").strip()) + + text = str(value).strip() + if not text: + return "" + if text.lstrip().startswith("<"): + return html.unescape(text) + + compact = "".join(text.split()) + if len(compact) >= 16 and len(compact) % 2 == 0 and _SNS_HEX_RE.match(compact): + try: + return _decode_sns_content_blob(bytes.fromhex(compact)) + except ValueError: + pass + if len(compact) >= 24 and len(compact) % 4 == 0 and _SNS_BASE64_RE.match(compact): + try: + return _decode_sns_content_blob(base64.b64decode(compact, validate=True)) + except (ValueError, binascii.Error): + pass + return html.unescape(text) + + +def _sanitize_sns_pseudo_xml(xml_text): + """修 WeChat 老朋友圈 XML 的非法字符，让 ElementTree 能解析。 + + CDATA 块内不动；块外把裸 & 转成 &。 + text-only 节点（content/title/description/...）内部的裸 < > 转义掉。 + 控制字符直接剥除。 + """ + s = _SNS_INVALID_CTRL_RE.sub("", xml_text) + parts = [] + last = 0 + for m in _SNS_CDATA_BLOCK_RE.finditer(s): + head = s[last:m.start()] + parts.append(_SNS_BARE_AMP_RE.sub("&", head)) + parts.append(m.group(0)) + last = m.end() + parts.append(_SNS_BARE_AMP_RE.sub("&", s[last:])) + out = "".join(parts) + + def _esc(m): + open_tag, _, text, close_tag = ( + m.group(1), m.group(2), m.group(3), m.group(4) + ) + return ( + open_tag + + text.replace("<", "<").replace(">", ">") + + close_tag + ) + + return _SNS_TEXT_NODE_RE.sub(_esc, out) + _cfg = load_config() DECRYPTED_DIR = _cfg["decrypted_dir"] SNS_DB_PATH = os.path.join(DECRYPTED_DIR, "sns", "sns.db") @@ -430,17 +535,24 @@ def _parse_media_list(timeline_obj): def _parse_timeline_xml(content_xml): - """解析 SnsTimeLine 的 Content XML，返回结构化数据""" + """解析 SnsTimeLine 的 Content XML，返回结构化数据。 + + content_xml 入参可能是 bytes / zstd 字节 / hex 串 / base64 串 / plain XML， + 先 decode 再 sanitize 老 XML 脏数据（裸 & < > 控制字符），最后才喂 ET。 + """ if not content_xml: return None - if len(content_xml) > _SNS_XML_MAX_LEN: + decoded = _decode_sns_content_blob(content_xml) + if not decoded: return None - if _SNS_XML_UNSAFE_RE.search(content_xml): + if len(decoded) > _SNS_XML_MAX_LEN: + return None + if _SNS_XML_UNSAFE_RE.search(decoded): # XXE 防护: 拒绝 DOCTYPE/ENTITY,避免恶意朋友圈 XML 通过 entity expansion # 或外部实体引用执行 SSRF/读取本地文件 return None try: - root = ET.fromstring(content_xml) + root = ET.fromstring(_sanitize_sns_pseudo_xml(decoded)) except ET.ParseError: return None diff --git a/tests/test_export_sns_parse.py b/tests/test_export_sns_parse.py new file mode 100644 index 0000000..11125c7 --- /dev/null +++ b/tests/test_export_sns_parse.py @@ -0,0 +1,209 @@ +"""Tests for `export_sns._parse_timeline_xml` content-blob robustness. + +Before this fix, `_parse_timeline_xml` called `ET.fromstring(content_xml)` +directly, assuming `SnsTimeLine.content` was always plain XML. In reality +the column ships in 4 encodings across WeChat versions / historical posts: + + 1. bytes (zstd-compressed or raw UTF-8) + 2. plain XML string + 3. hex string + 4. base64 string + +Plus 2013-2017 era posts contain pseudo-XML quirks that `ElementTree` +refuses: bare `&` in URLs, raw `<` / `>` inside user-typed text fields, +stray control characters. + +All of these previously caused `_parse_timeline_xml` to silently return +None → the row dropped from export with no diagnostic. These tests pin +the new decode + sanitize behavior. All fixtures are synthetic (no PII). +""" +import base64 +import unittest + +import zstandard as zstd + +import export_sns + + +_MINIMAL_XML = ( + '' + '' + 'fake-tid-1' + 'wxid_synthetic_user' + '1700000000' + 'hello world' + '2' + '' + '' +) + + +class DecodeContentBlobTests(unittest.TestCase): + """Detect 4 content encodings before XML parsing.""" + + def test_plain_xml_string_passthrough(self): + post = export_sns._parse_timeline_xml(_MINIMAL_XML) + self.assertIsNotNone(post) + self.assertEqual(post["id"], "fake-tid-1") + self.assertEqual(post["content_desc"], "hello world") + + def test_plain_xml_bytes_passthrough(self): + post = export_sns._parse_timeline_xml(_MINIMAL_XML.encode("utf-8")) + self.assertIsNotNone(post) + self.assertEqual(post["id"], "fake-tid-1") + + def test_zstd_compressed_bytes(self): + compressed = zstd.ZstdCompressor().compress(_MINIMAL_XML.encode("utf-8")) + # zstd magic 28 B5 2F FD must be at front + self.assertEqual(compressed[:4], b"\x28\xb5\x2f\xfd") + post = export_sns._parse_timeline_xml(compressed) + self.assertIsNotNone(post) + self.assertEqual(post["id"], "fake-tid-1") + + def test_hex_encoded_string(self): + hex_str = _MINIMAL_XML.encode("utf-8").hex() + post = export_sns._parse_timeline_xml(hex_str) + self.assertIsNotNone(post) + self.assertEqual(post["id"], "fake-tid-1") + + def test_base64_encoded_string(self): + b64_str = base64.b64encode(_MINIMAL_XML.encode("utf-8")).decode("ascii") + post = export_sns._parse_timeline_xml(b64_str) + self.assertIsNotNone(post) + self.assertEqual(post["id"], "fake-tid-1") + + def test_none_input(self): + self.assertIsNone(export_sns._parse_timeline_xml(None)) + + def test_empty_string_input(self): + self.assertIsNone(export_sns._parse_timeline_xml("")) + + def test_empty_bytes_input(self): + self.assertIsNone(export_sns._parse_timeline_xml(b"")) + + def test_short_hex_not_misdetected(self): + # Short hex-looking strings should NOT be eagerly decoded — + # they could be plain text that happens to be all hex chars. + # Length floor is 16, so anything shorter falls through to + # html.unescape path and ET.fromstring then ParseError → None. + post = export_sns._parse_timeline_xml("deadbeef") + self.assertIsNone(post) + + def test_garbage_input_returns_none_not_raises(self): + # Random non-XML, non-hex, non-base64 → None, no exception. + post = export_sns._parse_timeline_xml("this is not xml at all !!!") + self.assertIsNone(post) + + +class SanitizePseudoXmlTests(unittest.TestCase): + """Old WeChat posts contain XML 1.0 forbidden / unescaped chars.""" + + def test_bare_ampersand_in_url(self): + # 2013-2017 era: WeRead&Friends shipped as + # WeRead&Friends (bare &). Must escape. + xml = ( + '' + 'amp-1' + 'wxid_x' + '1700000000' + 'http://example.com/?a=1&b=2' + '3' + '' + ) + post = export_sns._parse_timeline_xml(xml) + self.assertIsNotNone(post) + self.assertEqual(post["content_desc"], "http://example.com/?a=1&b=2") + + def test_raw_angle_brackets_inside_text(self): + # User typed literal "<3" in the post body. Must escape to <3. + xml = ( + '' + 'ang-1' + 'wxid_x' + '1700000000' + 'love it <3 always' + '2' + '' + ) + post = export_sns._parse_timeline_xml(xml) + self.assertIsNotNone(post) + self.assertIn("<3", post["content_desc"]) + + def test_control_chars_stripped(self): + xml = ( + '' + 'ctrl-1' + 'wxid_x' + '1700000000' + 'hello\x01\x02world' + '2' + '' + ) + post = export_sns._parse_timeline_xml(xml) + self.assertIsNotNone(post) + self.assertEqual(post["content_desc"], "helloworld") + + def test_cdata_block_preserves_ampersand(self): + # & inside CDATA is legal, must NOT be re-escaped to &. + # Use a non-text-only node () so this exercises the + # CDATA-split logic in isolation; the text-only-node escape pass + # (contentDesc etc.) intentionally doesn't recurse into CDATA + # since real WeChat posts don't ship CDATA inside those nodes. + xml = ( + '' + 'cdata-1' + 'wxid_x' + '1700000000' + 'plain text' + '2' + '' + '' + ) + post = export_sns._parse_timeline_xml(xml) + self.assertIsNotNone(post) + # Post-parse, the CDATA value should be preserved as raw data. + # We verify via re-serialize since _parse_timeline_xml doesn't + # expose extraInfo, but a None return would mean ParseError. + self.assertEqual(post["content_desc"], "plain text") + + +class SecurityAndLimitsTests(unittest.TestCase): + """Existing XXE / length-cap guards must still fire after decode.""" + + def test_xxe_doctype_blocked(self): + xml = ( + ']>' + '' + 'xxe-1' + '1700000000' + '' + ) + self.assertIsNone(export_sns._parse_timeline_xml(xml)) + + def test_xxe_doctype_blocked_after_zstd_decode(self): + # XXE check must run on the DECODED payload — encoded DOCTYPE + # would otherwise sneak past a naive pre-decode regex. + evil = ( + ']>' + '' + 'xxe-zstd' + '1700000000' + '' + ) + compressed = zstd.ZstdCompressor().compress(evil.encode("utf-8")) + self.assertIsNone(export_sns._parse_timeline_xml(compressed)) + + def test_oversized_content_blocked(self): + big = ( + '' + 'big-1' + '1700000000' + '' + ('x' * 250_000) + '' + '2' + '' + ) + self.assertIsNone(export_sns._parse_timeline_xml(big)) + + +if __name__ == "__main__": + unittest.main()