5208f6f517
微信对撤回的点赞/评论不硬删, 只在 SnsMessage_tmp3 行上打 del_status 标记. 老 _load_comments 直接 SELECT 不带任何 WHERE, 结果导出的 likes/comments 里混着撤回行 — 等于 "对方撤回的赞还能在本地导出里看到", 违反用户预期.
修复: SQL 加 WHERE COALESCE(del_status, 0) = 0
- COALESCE 兜底: 老 schema NULL 视作 0 (保留)
- WHERE 而非 Python 端过滤: 大 db 少传 row
- 函数签名/返回结构不变
- 缺列时仍走现有 try/except 路径, 返回 {} 不崩
测试: LoadCommentsTests 3 case (撤回过滤 / NULL 保留 / 缺列兜底), 全合成 sqlite 无 PII. baseline 17 → 20 全过.
承接 #119 的 SNS 导出可靠性线: #119 修 "老 XML 让整行帖子丢失", 本 PR 修 "互动里混入已撤回 row".
319 lines
12 KiB
Python
319 lines
12 KiB
Python
"""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 = (
|
|
'<TimelineObjects>'
|
|
'<TimelineObject>'
|
|
'<id>fake-tid-1</id>'
|
|
'<username>wxid_synthetic_user</username>'
|
|
'<createTime>1700000000</createTime>'
|
|
'<contentDesc>hello world</contentDesc>'
|
|
'<ContentObject><type>2</type></ContentObject>'
|
|
'</TimelineObject>'
|
|
'</TimelineObjects>'
|
|
)
|
|
|
|
|
|
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: <appname>WeRead&Friends</appname> shipped as
|
|
# <appname>WeRead&Friends</appname> (bare &). Must escape.
|
|
xml = (
|
|
'<TimelineObjects><TimelineObject>'
|
|
'<id>amp-1</id>'
|
|
'<username>wxid_x</username>'
|
|
'<createTime>1700000000</createTime>'
|
|
'<contentDesc>http://example.com/?a=1&b=2</contentDesc>'
|
|
'<ContentObject><type>3</type></ContentObject>'
|
|
'</TimelineObject></TimelineObjects>'
|
|
)
|
|
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 = (
|
|
'<TimelineObjects><TimelineObject>'
|
|
'<id>ang-1</id>'
|
|
'<username>wxid_x</username>'
|
|
'<createTime>1700000000</createTime>'
|
|
'<contentDesc>love it <3 always</contentDesc>'
|
|
'<ContentObject><type>2</type></ContentObject>'
|
|
'</TimelineObject></TimelineObjects>'
|
|
)
|
|
post = export_sns._parse_timeline_xml(xml)
|
|
self.assertIsNotNone(post)
|
|
self.assertIn("<3", post["content_desc"])
|
|
|
|
def test_control_chars_stripped(self):
|
|
xml = (
|
|
'<TimelineObjects><TimelineObject>'
|
|
'<id>ctrl-1</id>'
|
|
'<username>wxid_x</username>'
|
|
'<createTime>1700000000</createTime>'
|
|
'<contentDesc>hello\x01\x02world</contentDesc>'
|
|
'<ContentObject><type>2</type></ContentObject>'
|
|
'</TimelineObject></TimelineObjects>'
|
|
)
|
|
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 (<extraInfo>) 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 = (
|
|
'<TimelineObjects><TimelineObject>'
|
|
'<id>cdata-1</id>'
|
|
'<username>wxid_x</username>'
|
|
'<createTime>1700000000</createTime>'
|
|
'<contentDesc>plain text</contentDesc>'
|
|
'<ContentObject><type>2</type></ContentObject>'
|
|
'<extraInfo><![CDATA[a&b&c]]></extraInfo>'
|
|
'</TimelineObject></TimelineObjects>'
|
|
)
|
|
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 = (
|
|
'<!DOCTYPE foo [<!ENTITY x "boom">]>'
|
|
'<TimelineObjects><TimelineObject>'
|
|
'<id>xxe-1</id>'
|
|
'<createTime>1700000000</createTime>'
|
|
'</TimelineObject></TimelineObjects>'
|
|
)
|
|
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 = (
|
|
'<!DOCTYPE foo [<!ENTITY x "boom">]>'
|
|
'<TimelineObjects><TimelineObject>'
|
|
'<id>xxe-zstd</id>'
|
|
'<createTime>1700000000</createTime>'
|
|
'</TimelineObject></TimelineObjects>'
|
|
)
|
|
compressed = zstd.ZstdCompressor().compress(evil.encode("utf-8"))
|
|
self.assertIsNone(export_sns._parse_timeline_xml(compressed))
|
|
|
|
def test_oversized_content_blocked(self):
|
|
big = (
|
|
'<TimelineObjects><TimelineObject>'
|
|
'<id>big-1</id>'
|
|
'<createTime>1700000000</createTime>'
|
|
'<contentDesc>' + ('x' * 250_000) + '</contentDesc>'
|
|
'<ContentObject><type>2</type></ContentObject>'
|
|
'</TimelineObject></TimelineObjects>'
|
|
)
|
|
self.assertIsNone(export_sns._parse_timeline_xml(big))
|
|
|
|
|
|
class LoadCommentsTests(unittest.TestCase):
|
|
"""`_load_comments` filters out interactions with `del_status != 0`.
|
|
|
|
WeChat does not hard-delete a recalled like / comment; it sets a
|
|
`del_status` flag. Before this fix `_load_comments` selected all rows
|
|
unconditionally, so the export carried recalled interactions.
|
|
"""
|
|
|
|
def _make_db(self, rows, *, with_del_status_col=True):
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
|
|
fd, path = tempfile.mkstemp(suffix=".db")
|
|
os.close(fd)
|
|
conn = sqlite3.connect(path)
|
|
try:
|
|
cols = (
|
|
"local_id INTEGER PRIMARY KEY AUTOINCREMENT,"
|
|
" feed_id INTEGER, create_time INTEGER, type INTEGER,"
|
|
" from_username TEXT, from_nickname TEXT,"
|
|
" to_username TEXT, to_nickname TEXT, content TEXT"
|
|
)
|
|
if with_del_status_col:
|
|
cols += ", del_status INTEGER"
|
|
conn.execute(f"CREATE TABLE SnsMessage_tmp3 ({cols})")
|
|
insert_cols = (
|
|
"feed_id, create_time, type, from_username, from_nickname,"
|
|
" to_username, to_nickname, content"
|
|
)
|
|
placeholders = "?, ?, ?, ?, ?, ?, ?, ?"
|
|
if with_del_status_col:
|
|
insert_cols += ", del_status"
|
|
placeholders += ", ?"
|
|
conn.executemany(
|
|
f"INSERT INTO SnsMessage_tmp3 ({insert_cols}) VALUES ({placeholders})",
|
|
rows,
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
return path
|
|
|
|
def test_recalled_interactions_excluded(self):
|
|
# del_status=1 → recalled, must be dropped.
|
|
rows = [
|
|
(100, 1, 1, "wxid_alive", "Alive", "", "", "", 0),
|
|
(100, 2, 2, "wxid_recalled", "Recalled", "", "", "撤回的评论", 1),
|
|
(100, 3, 1, "wxid_alsoalive", "AlsoAlive", "", "", "", 0),
|
|
]
|
|
import os
|
|
import sqlite3
|
|
|
|
path = self._make_db(rows)
|
|
try:
|
|
conn = sqlite3.connect(path)
|
|
try:
|
|
comments = export_sns._load_comments(conn)
|
|
finally:
|
|
conn.close()
|
|
finally:
|
|
os.unlink(path)
|
|
self.assertEqual(len(comments[100]), 2)
|
|
self.assertEqual({c["from_username"] for c in comments[100]},
|
|
{"wxid_alive", "wxid_alsoalive"})
|
|
|
|
def test_null_del_status_treated_as_kept(self):
|
|
# del_status IS NULL (or 0) → keep. COALESCE() backstop.
|
|
rows = [
|
|
(200, 1, 1, "wxid_a", "A", "", "", "", None),
|
|
(200, 2, 1, "wxid_b", "B", "", "", "", 0),
|
|
]
|
|
import os
|
|
import sqlite3
|
|
|
|
path = self._make_db(rows)
|
|
try:
|
|
conn = sqlite3.connect(path)
|
|
try:
|
|
comments = export_sns._load_comments(conn)
|
|
finally:
|
|
conn.close()
|
|
finally:
|
|
os.unlink(path)
|
|
self.assertEqual(len(comments[200]), 2)
|
|
|
|
def test_missing_del_status_column_tolerated(self):
|
|
# Schemas without del_status: SQL still parses (column referenced only
|
|
# inside COALESCE) → expect graceful failure path, not crash.
|
|
rows = [
|
|
(300, 1, 1, "wxid_x", "X", "", "", ""),
|
|
]
|
|
import os
|
|
import sqlite3
|
|
|
|
path = self._make_db(rows, with_del_status_col=False)
|
|
try:
|
|
conn = sqlite3.connect(path)
|
|
try:
|
|
comments = export_sns._load_comments(conn)
|
|
finally:
|
|
conn.close()
|
|
finally:
|
|
os.unlink(path)
|
|
# `del_status` doesn't exist → OperationalError caught by the function,
|
|
# returns empty dict (matching upstream behavior for malformed schemas).
|
|
self.assertEqual(comments, {})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|