Skip to content

fix(wecom): support pic_url and fix AES key base64 padding for image … #12390

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zogwei wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(wecom): support pic_url and fix AES key base64 padding for image … #12390

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 115 additions & 8 deletions gateway/platforms/wecom.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@
MessageEvent,
MessageType,
SendResult,
_looks_like_image,
cache_document_from_bytes,
cache_image_from_bytes,
)
Expand All @@ -72,6 +73,21 @@

DEFAULT_WS_URL = "wss://openws.work.weixin.qq.com"


def _decode_wecom_aes_key(aes_key: str) -> bytes:
"""Decode WeCom AES key (43-char base64 without padding -> 32 bytes)."""
payload = str(aes_key or "").strip()
if not payload:
raise ValueError("aes_key is empty")
# WeCom encoding_aeskey is 43 chars, missing the trailing '=' padding
# that standard base64 requires. Add it back.
padded = payload + "=" * ((4 - len(payload) % 4) % 4)
key = base64.b64decode(padded)
if len(key) != 32:
raise ValueError(f"Invalid WeCom AES key length: expected 32 bytes, got {len(key)}")
return key


APP_CMD_SUBSCRIBE = "aibot_subscribe"
APP_CMD_CALLBACK = "aibot_msg_callback"
APP_CMD_LEGACY_CALLBACK = "aibot_callback"
Expand Down Expand Up @@ -729,8 +745,9 @@ async def _cache_media(self, kind: str, media: Dict[str, Any]) -> Optional[Tuple
filename = str(media.get("filename") or media.get("name") or "wecom_file")
return cache_document_from_bytes(raw, filename), mimetypes.guess_type(filename)[0] or "application/octet-stream"

url = str(media.get("url") or "").strip()
url = str(media.get("url") or media.get("pic_url") or "").strip()

@teknium1 teknium1 Jul 12, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This fallback only helps if _extract_media() has already passed a media dictionary to _cache_media(). The PR does not add a reference for a top-level AI Bot pic_url, so that reported payload shape still never reaches this line. Construct the image reference in _extract_media() and cover it with a regression test.

if not url:
logger.debug("[%s] No url/pic_url found in inbound %s media", self.name, kind)
return None

try:
Expand All @@ -741,10 +758,19 @@ async def _cache_media(self, kind: str, media: Dict[str, Any]) -> Optional[Tuple

aes_key = str(media.get("aeskey") or "").strip()
if aes_key:
decrypted = None
try:
raw = self._decrypt_file_bytes(raw, aes_key)
decrypted = self._decrypt_file_bytes(raw, aes_key)
except Exception as exc:
logger.debug("[%s] Failed to decrypt %s from %s: %s", self.name, kind, url, exc)
logger.debug("[%s] Primary decrypt failed for %s, trying variants: %s", self.name, kind, exc)

if decrypted is None:
decrypted = self._try_decrypt_variants(raw, aes_key, kind)

if decrypted is not None:

@teknium1 teknium1 Jul 12, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_try_decrypt_variants() can return its first score-zero candidate, so this accepts plaintext that was not positively validated. Avoid heuristic AES-mode fallback or require a deterministic format/padding validation before caching the result.

raw = decrypted
else:
logger.debug("[%s] All decryption variants failed for %s", self.name, kind)
return None

content_type = str(headers.get("content-type") or "").split(";", 1)[0].strip() or "application/octet-stream"
Expand Down Expand Up @@ -997,12 +1023,8 @@ def _raise_for_wecom_error(cls, response: Dict[str, Any], operation: str) -> Non
def _decrypt_file_bytes(encrypted_data: bytes, aes_key: str) -> bytes:
if not encrypted_data:
raise ValueError("encrypted_data is empty")
if not aes_key:
raise ValueError("aes_key is required")

key = base64.b64decode(aes_key)
if len(key) != 32:
raise ValueError(f"Invalid WeCom AES key length: expected 32 bytes, got {len(key)}")
key = _decode_wecom_aes_key(aes_key)

try:
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
Expand All @@ -1021,6 +1043,91 @@ def _decrypt_file_bytes(encrypted_data: bytes, aes_key: str) -> bytes:

return decrypted[:-pad_len]

@staticmethod
def _try_decrypt_variants(encrypted_data: bytes, aes_key: str, kind: str = "media") -> Optional[bytes]:
"""Try multiple AES decryption strategies and return the best result."""
if not encrypted_data or not aes_key:
return None

try:
key = _decode_wecom_aes_key(aes_key)
except Exception:
return None

try:
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
except ImportError:
return None

def _decrypt_cbc(data: bytes, key_bytes: bytes, iv: bytes) -> bytes:
cipher = Cipher(algorithms.AES(key_bytes), modes.CBC(iv))
decryptor = cipher.decryptor()
decrypted = decryptor.update(data) + decryptor.finalize()
pad_len = decrypted[-1]
if 1 <= pad_len <= 32 and decrypted.endswith(bytes([pad_len]) * pad_len):
return decrypted[:-pad_len]
return decrypted

def _decrypt_ecb(data: bytes, key_bytes: bytes) -> bytes:
cipher = Cipher(algorithms.AES(key_bytes), modes.ECB())
decryptor = cipher.decryptor()
decrypted = decryptor.update(data) + decryptor.finalize()
pad_len = decrypted[-1]
if 1 <= pad_len <= 32 and decrypted.endswith(bytes([pad_len]) * pad_len):
return decrypted[:-pad_len]
return decrypted

variants = []

if len(key) == 32:
variants.append(("AES-256-CBC(iv=key[:16])", lambda d: _decrypt_cbc(d, key, key[:16])))
variants.append(("AES-256-ECB", lambda d: _decrypt_ecb(d, key)))
variants.append(("AES-128-CBC(iv=key[:16])", lambda d: _decrypt_cbc(d, key[:16], key[:16])))
variants.append(("AES-128-ECB", lambda d: _decrypt_ecb(d, key[:16])))
elif len(key) == 16:
variants.append(("AES-128-CBC(iv=key)", lambda d: _decrypt_cbc(d, key, key)))
variants.append(("AES-128-ECB", lambda d: _decrypt_ecb(d, key)))

# Weixin-style: key may be hex-encoded inside base64
if len(key) == 32:
try:
hex_text = key.decode("ascii", errors="ignore")
if hex_text and all(ch in "0123456789abcdefABCDEF" for ch in hex_text):
hex_bytes = bytes.fromhex(hex_text)
if len(hex_bytes) == 16:
variants.append(("AES-128-CBC(hex, iv=hex)", lambda d: _decrypt_cbc(d, hex_bytes, hex_bytes)))
variants.append(("AES-128-ECB(hex)", lambda d: _decrypt_ecb(d, hex_bytes)))
except Exception:
pass

best_result = None
best_score = -1

for name, decrypt_fn in variants:
try:
decrypted = decrypt_fn(encrypted_data)
score = 0
if _looks_like_image(decrypted):
score = 100
elif len(decrypted) > 100:
# Heuristic: avoid error pages / XML
head = decrypted[:200].lower()
if b"<html" not in head and b"<?xml" not in head and b"error" not in head:
score = 1

if score > best_score:
best_score = score
best_result = decrypted
logger.debug(
"[%s] Decryption success with %s for %s (score=%d, header=%s)",
"WeCom", name, kind, score, decrypted[:8].hex(),
)
except Exception as exc:
logger.debug("Decryption failed with %s for %s: %s", name, kind, exc)
continue

return best_result

async def _download_remote_bytes(
self,
url: str,
Expand Down