294 lines
9.4 KiB
Python
294 lines
9.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""归档消息中短文本 protobuf 的保守解析器。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import re
|
||
from typing import Any
|
||
|
||
|
||
_HEX_RE = re.compile(r"^[0-9a-fA-F]+$")
|
||
_TEXT_TYPES = {"text", "文本"}
|
||
_MAX_HEX_CHARS = 256 * 1024
|
||
|
||
|
||
def _read_varint(data: bytes, offset: int) -> tuple[int, int] | None:
|
||
value = 0
|
||
shift = 0
|
||
for index in range(10):
|
||
if offset >= len(data):
|
||
return None
|
||
current = data[offset]
|
||
offset += 1
|
||
value |= (current & 0x7F) << shift
|
||
if current < 0x80:
|
||
return value, offset
|
||
shift += 7
|
||
return None
|
||
|
||
|
||
def _readable_utf8(data: bytes) -> str:
|
||
if not data:
|
||
return ""
|
||
try:
|
||
text = data.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
return ""
|
||
if "\x00" in text or "\ufffd" in text:
|
||
return ""
|
||
if not all(character.isprintable() or character in "\r\n\t" for character in text):
|
||
return ""
|
||
return text.strip()
|
||
|
||
|
||
def _protobuf_texts(
|
||
data: bytes, *, depth: int = 0, max_depth: int = 8
|
||
) -> list[tuple[int, str]]:
|
||
if not data or depth > max_depth:
|
||
return []
|
||
offset = 0
|
||
result: list[tuple[int, str]] = []
|
||
parsed_fields = 0
|
||
while offset < len(data):
|
||
key_result = _read_varint(data, offset)
|
||
if key_result is None:
|
||
return []
|
||
key, offset = key_result
|
||
field_number = key >> 3
|
||
wire_type = key & 7
|
||
if field_number <= 0:
|
||
return []
|
||
parsed_fields += 1
|
||
if wire_type == 0:
|
||
value_result = _read_varint(data, offset)
|
||
if value_result is None:
|
||
return []
|
||
_, offset = value_result
|
||
elif wire_type == 1:
|
||
offset += 8
|
||
elif wire_type == 2:
|
||
length_result = _read_varint(data, offset)
|
||
if length_result is None:
|
||
return []
|
||
length, offset = length_result
|
||
end = offset + length
|
||
if length < 0 or end > len(data):
|
||
return []
|
||
value = data[offset:end]
|
||
offset = end
|
||
readable = _readable_utf8(value)
|
||
if readable:
|
||
result.append((depth, readable))
|
||
result.extend(_protobuf_texts(value, depth=depth + 1, max_depth=max_depth))
|
||
elif wire_type == 5:
|
||
offset += 4
|
||
else:
|
||
return []
|
||
if offset > len(data):
|
||
return []
|
||
return result if parsed_fields else []
|
||
|
||
|
||
def decode_hex_protobuf_text(value: Any, message_type: Any) -> str:
|
||
"""从十六进制 protobuf 中提取文本，只对明确的文本消息生效。"""
|
||
|
||
if str(message_type or "").strip().casefold() not in _TEXT_TYPES:
|
||
return ""
|
||
raw = str(value or "").strip()
|
||
if not raw or len(raw) > _MAX_HEX_CHARS:
|
||
return ""
|
||
decoded_parts: list[str] = []
|
||
for raw_part in raw.split("|"):
|
||
part = re.sub(r"\s+", "", raw_part)
|
||
if not part or len(part) % 2 or not _HEX_RE.fullmatch(part):
|
||
return ""
|
||
try:
|
||
data = bytes.fromhex(part)
|
||
except ValueError:
|
||
return ""
|
||
candidates = _protobuf_texts(data)
|
||
if not candidates:
|
||
return ""
|
||
deepest = max(depth for depth, _ in candidates)
|
||
texts = [text for depth, text in candidates if depth == deepest]
|
||
for text in texts:
|
||
if text and text not in decoded_parts:
|
||
decoded_parts.append(text)
|
||
return " | ".join(decoded_parts)
|
||
|
||
|
||
def _protobuf_fields(data: bytes) -> dict[int, list[tuple[int, Any]]]:
|
||
fields: dict[int, list[tuple[int, Any]]] = {}
|
||
offset = 0
|
||
while offset < len(data):
|
||
key_result = _read_varint(data, offset)
|
||
if key_result is None:
|
||
return {}
|
||
key, offset = key_result
|
||
field_number = key >> 3
|
||
wire_type = key & 7
|
||
if field_number <= 0:
|
||
return {}
|
||
if wire_type == 0:
|
||
value_result = _read_varint(data, offset)
|
||
if value_result is None:
|
||
return {}
|
||
value, offset = value_result
|
||
elif wire_type == 1:
|
||
end = offset + 8
|
||
if end > len(data):
|
||
return {}
|
||
value = data[offset:end]
|
||
offset = end
|
||
elif wire_type == 2:
|
||
length_result = _read_varint(data, offset)
|
||
if length_result is None:
|
||
return {}
|
||
length, offset = length_result
|
||
end = offset + length
|
||
if end > len(data):
|
||
return {}
|
||
value = data[offset:end]
|
||
offset = end
|
||
elif wire_type == 5:
|
||
end = offset + 4
|
||
if end > len(data):
|
||
return {}
|
||
value = data[offset:end]
|
||
offset = end
|
||
else:
|
||
return {}
|
||
fields.setdefault(field_number, []).append((wire_type, value))
|
||
return fields
|
||
|
||
|
||
def parse_file_message_metadata(value: Any, content_type: Any = None) -> dict[str, Any]:
|
||
"""解析企业微信 content_type=20 的文件卡片，不暴露远端令牌。"""
|
||
|
||
try:
|
||
if int(content_type) != 20:
|
||
return {}
|
||
except (TypeError, ValueError):
|
||
return {}
|
||
if isinstance(value, memoryview):
|
||
value = value.tobytes()
|
||
if not isinstance(value, bytes) or not value:
|
||
return {}
|
||
fields = _protobuf_fields(value)
|
||
filename_values = fields.get(2) or []
|
||
size_values = fields.get(4) or []
|
||
checksum_values = fields.get(10) or []
|
||
if not filename_values or not size_values:
|
||
return {}
|
||
filename = _readable_utf8(filename_values[0][1])
|
||
filename = re.split(r"[\\/]", filename)[-1].strip()[:512]
|
||
if not filename or "." not in filename or any(ord(char) < 32 for char in filename):
|
||
return {}
|
||
size_bytes = int(size_values[0][1]) if size_values[0][0] == 0 else 0
|
||
if size_bytes <= 0:
|
||
return {}
|
||
checksum = ""
|
||
if checksum_values and checksum_values[0][0] == 2:
|
||
checksum = _readable_utf8(checksum_values[0][1]).upper()
|
||
if not re.fullmatch(r"[0-9A-F]{32,64}", checksum):
|
||
checksum = ""
|
||
remote_values = fields.get(1) or []
|
||
reference_hash = ""
|
||
if remote_values and remote_values[0][0] == 2:
|
||
reference_hash = hashlib.sha256(remote_values[0][1]).hexdigest()
|
||
return {
|
||
"original_filename": filename,
|
||
"size_bytes": size_bytes,
|
||
"checksum": checksum,
|
||
"media_type": "file",
|
||
"source_reference_sha256": reference_hash,
|
||
}
|
||
|
||
|
||
def file_message_content(metadata: dict[str, Any], *, cached: bool) -> str:
|
||
filename = str(metadata.get("original_filename") or "未命名文件")
|
||
size = int(metadata.get("size_bytes") or 0)
|
||
if size >= 1024 * 1024 * 1024:
|
||
size_label = f"{size / (1024 * 1024 * 1024):.2f} GB"
|
||
elif size >= 1024 * 1024:
|
||
size_label = f"{size / (1024 * 1024):.2f} MB"
|
||
elif size >= 1024:
|
||
size_label = f"{size / 1024:.1f} KB"
|
||
else:
|
||
size_label = f"{size} B"
|
||
availability = "" if cached else "，源文件未缓存"
|
||
return f"[文件] {filename}（{size_label}{availability}）"
|
||
|
||
|
||
def _protobuf_text_field(
|
||
fields: dict[int, list[tuple[int, Any]]], field_number: int
|
||
) -> str:
|
||
values = fields.get(field_number) or []
|
||
if not values or values[0][0] != 2:
|
||
return ""
|
||
return _readable_utf8(values[0][1])
|
||
|
||
|
||
def parse_mini_program_metadata(
|
||
value: Any, content_type: Any = None
|
||
) -> dict[str, Any]:
|
||
"""解析企业微信 content_type=78 的小程序卡片。"""
|
||
|
||
try:
|
||
if int(content_type) != 78:
|
||
return {}
|
||
except (TypeError, ValueError):
|
||
return {}
|
||
if isinstance(value, memoryview):
|
||
value = value.tobytes()
|
||
if not isinstance(value, bytes) or not value:
|
||
return {}
|
||
top_fields = _protobuf_fields(value)
|
||
candidate: dict[int, list[tuple[int, Any]]] = {}
|
||
for values in top_fields.values():
|
||
for wire_type, nested_value in values:
|
||
if wire_type != 2 or not isinstance(nested_value, bytes):
|
||
continue
|
||
nested = _protobuf_fields(nested_value)
|
||
app_id = _protobuf_text_field(nested, 2)
|
||
page_path = _protobuf_text_field(nested, 3)
|
||
if app_id.startswith("wx") and page_path:
|
||
candidate = nested
|
||
break
|
||
if candidate:
|
||
break
|
||
if not candidate:
|
||
return {}
|
||
title = (
|
||
_protobuf_text_field(candidate, 7)
|
||
or _protobuf_text_field(candidate, 8)
|
||
or _protobuf_text_field(top_fields, 3)
|
||
)
|
||
app_name = _protobuf_text_field(candidate, 10)
|
||
page_path = _protobuf_text_field(candidate, 3)
|
||
if not title or not page_path:
|
||
return {}
|
||
return {
|
||
"title": title[:512],
|
||
"app_name": app_name[:512],
|
||
"app_id": _protobuf_text_field(candidate, 2)[:128],
|
||
"username": _protobuf_text_field(candidate, 1)[:255],
|
||
"page_path": page_path[:2048],
|
||
"cover_url": (
|
||
_protobuf_text_field(candidate, 6)
|
||
or _protobuf_text_field(top_fields, 2)
|
||
)[:4096],
|
||
}
|
||
|
||
|
||
def mini_program_content(metadata: dict[str, Any]) -> str:
|
||
lines = [f"[小程序] {str(metadata.get('title') or '未命名小程序')}"]
|
||
if metadata.get("app_name"):
|
||
lines.append(f"应用：{metadata['app_name']}")
|
||
if metadata.get("app_id"):
|
||
lines.append(f"AppID：{metadata['app_id']}")
|
||
if metadata.get("page_path"):
|
||
lines.append(f"页面：{metadata['page_path']}")
|
||
return "\n".join(lines)
|