146 lines
7.2 KiB
Python
146 lines
7.2 KiB
Python
"""Synthetic message fixtures: no real contact or customer content."""
|
|
import unittest
|
|
|
|
from archive_content_parser import parse_message_content
|
|
|
|
|
|
def varint(number):
|
|
result = bytearray()
|
|
while number > 127:
|
|
result.append((number & 127) | 128)
|
|
number >>= 7
|
|
result.append(number)
|
|
return bytes(result)
|
|
|
|
|
|
def field(number, value):
|
|
if isinstance(value, int):
|
|
return varint(number << 3) + varint(value)
|
|
if isinstance(value, str):
|
|
value = value.encode('utf-8')
|
|
return varint((number << 3) | 2) + varint(len(value)) + value
|
|
|
|
|
|
def text_message(text):
|
|
return field(1, field(1, 0) + field(2, field(1, text)))
|
|
|
|
|
|
class MessagePlaintextTests(unittest.TestCase):
|
|
def assert_decoded(self, value, expected, content_type=2):
|
|
self.assertEqual(parse_message_content(value, content_type),
|
|
{'text': expected, 'status': 'decoded', 'kind': 'text'})
|
|
|
|
def test_native_short_text_is_decoded_without_loss(self):
|
|
for text in ('1', '0', '14.1', '好', '?', '??', 'OK', 'abc中文',
|
|
'3盒', '👍', '👩\u200d⚕️', 'line 1\nline 2'):
|
|
for content_type in (0, 1, 2, '文本', 'text'):
|
|
with self.subTest(text=text, content_type=content_type):
|
|
self.assert_decoded(text_message(text), text, content_type)
|
|
|
|
def test_repeated_native_text_segments_preserve_order(self):
|
|
self.assert_decoded(text_message('合成正文') + text_message('👍'), '合成正文👍')
|
|
self.assert_decoded(text_message('hello ') + text_message('world'), 'hello world')
|
|
self.assertEqual(parse_message_content(text_message('ok') + field(1, b'\xff'), 2)['status'], 'unsupported')
|
|
|
|
def test_flat_text_envelope_is_decoded(self):
|
|
self.assert_decoded(field(1, '13字符abcdefghij'), '13字符abcdefghij')
|
|
self.assert_decoded(field(1, 'abcdefghijklm'), 'abcdefghijklm')
|
|
|
|
def test_utf8_text_and_literal_hex_remain_text(self):
|
|
for value in ('1', '你好', 'OK', '1234567890abcdef', 'https://example.test/a',
|
|
'aGVsbG8=', '👩\u200d⚕️'):
|
|
with self.subTest(value=value):
|
|
self.assert_decoded(value, value)
|
|
self.assert_decoded(value.encode(), value)
|
|
|
|
def test_memoryview_and_bytearray_supported(self):
|
|
self.assert_decoded(memoryview(text_message('1')), '1')
|
|
self.assert_decoded(bytearray(text_message('OK')), 'OK')
|
|
|
|
def test_no_arbitrary_protobuf_string_extraction(self):
|
|
raw = field(50, 'this is metadata not the message')
|
|
self.assertEqual(parse_message_content(raw, 2)['status'], 'unsupported')
|
|
|
|
def test_malformed_text_never_becomes_hex(self):
|
|
for raw in (b'\xff\x00', b'\x0a\x06a', field(1, b'\x08\x00\x12\xff'),
|
|
text_message('hello') + b'\x00', b'\x00', b''):
|
|
with self.subTest(raw=raw):
|
|
parsed = parse_message_content(raw, 2)
|
|
self.assertEqual(parsed['text'], '[文本暂无法解析]')
|
|
self.assertEqual(parsed['status'], 'unsupported')
|
|
|
|
def test_size_type_and_unicode_bounds(self):
|
|
for raw in (b'a' * (128 * 1024 + 1), '\ud800', None, 123, {'content': 'hello'}):
|
|
self.assertEqual(parse_message_content(raw, 2)['status'], 'unsupported')
|
|
nested = field(1, 'hello')
|
|
for _ in range(20):
|
|
nested = field(1, nested)
|
|
self.assertEqual(parse_message_content(nested, 2)['status'], 'unsupported')
|
|
|
|
def test_opaque_reference_is_not_guessed_as_text(self):
|
|
token = '*1*' + 'AbCd0123456789+/' * 50
|
|
self.assertEqual(parse_message_content(token, 2)['status'], 'unsupported')
|
|
self.assertEqual(parse_message_content(text_message(token), 2)['status'], 'unsupported')
|
|
|
|
def test_moments_use_caption_not_longer_media_reference(self):
|
|
token = '*1*' + 'AbCd0123456789+/' * 50
|
|
raw = field(1, 2) + field(2, '分享朋友圈') + field(4,
|
|
field(5, '合成朋友圈正文') + field(6, field(2, token)) + field(19, 1))
|
|
self.assertEqual(parse_message_content(raw, 529),
|
|
{'text': '[朋友圈] 合成朋友圈正文', 'status': 'summary', 'kind': 'moments'})
|
|
|
|
def test_file_only_exposes_filename_and_size(self):
|
|
token = '*1*' + 'AbCd0123456789+/' * 50
|
|
raw = field(1, token) + field(2, r'C:\private\report.pdf') + field(4, 1024) + field(10, 'A' * 32)
|
|
parsed = parse_message_content(raw, 20)
|
|
self.assertEqual(parsed['kind'], 'file')
|
|
self.assertIn('report.pdf', parsed['text'])
|
|
self.assertIn('1.0 KB', parsed['text'])
|
|
for private in (token, 'C:', 'private', 'A' * 32):
|
|
self.assertNotIn(private, parsed['text'])
|
|
|
|
def test_media_has_typed_summary_instead_of_paths(self):
|
|
raw = field(1, '/storage/emulated/0/private/file.silk') + field(2, '*1*' + 'A' * 500)
|
|
for content_type, expected in ((3, '[图片]'), (14, '[图片]'), (123, '[图片]'),
|
|
(4, '[语音待转文字]'), (16, '[语音待转文字]'),
|
|
(8, '[视频]'), (5, '[表情]'), (111, '[文件]')):
|
|
self.assertEqual(parse_message_content(raw, content_type)['text'], expected)
|
|
|
|
def test_mini_program_summary_excludes_references(self):
|
|
card = (field(1, 'gh_opaque') + field(2, 'wx123') + field(3, '/pages/private?token=SECRET')
|
|
+ field(6, 'https://example.test/token') + field(7, '示例程序') + field(10, '示例应用'))
|
|
parsed = parse_message_content(field(1, card), 78)
|
|
self.assertEqual(parsed['text'], '[小程序] 示例程序\n应用：示例应用')
|
|
self.assertNotIn('SECRET', parsed['text'])
|
|
|
|
def test_link_title_excludes_url(self):
|
|
raw = field(1, field(1, '合成标题') + field(2, 'https://example.test/token')) + field(2, 0)
|
|
self.assertEqual(parse_message_content(raw, 31)['text'], '[链接] 合成标题')
|
|
|
|
def test_contact_status_system_notice_is_retained(self):
|
|
notice = '对方开启了朋友验证，请先发送朋友验证请求'
|
|
self.assertEqual(parse_message_content(notice.encode(), 1011)['text'], notice)
|
|
|
|
def test_card_does_not_return_json_or_metadata(self):
|
|
raw = field(1, '合成通知') + field(2, '{"private_token": "SECRET"}')
|
|
self.assertEqual(parse_message_content(raw, 573)['text'], '[应用消息] 合成通知')
|
|
|
|
def test_extra_content_never_becomes_customer_message(self):
|
|
parsed = parse_message_content(b'\xff', 2, extra_content=text_message('metadata SECRET'))
|
|
self.assertNotIn('SECRET', parsed['text'])
|
|
self.assertEqual(parsed['status'], 'unsupported')
|
|
|
|
def test_summary_fields_reject_paths_and_opaque_values(self):
|
|
for secret in ('*1*' + 'A' * 500, '/storage/emulated/0/private',
|
|
'https://example.test/SECRET', '{"secret":"SECRET"}'):
|
|
raw = field(4, field(5, secret))
|
|
self.assertEqual(parse_message_content(raw, 529)['text'], '[朋友圈]')
|
|
|
|
def test_unknown_type_cannot_surface_binary_metadata(self):
|
|
parsed = parse_message_content(field(1, 'SECRET'), 999999)
|
|
self.assertEqual(parsed, {'text': '[暂不支持的消息类型]', 'status': 'unsupported', 'kind': 'unknown'})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|