114 lines
5.0 KiB
Python
114 lines
5.0 KiB
Python
"""Parse native MCP replies and expose a sanitized, historical trial result.
|
|
|
|
This module never opens a process, starts a helper, or sends a network request.
|
|
A successful HTTP/MCP transport is not proof of business access or message delivery.
|
|
"""
|
|
from __future__ import annotations
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import time
|
|
|
|
MAX_REPLY_BYTES = 262144
|
|
GRAY_ERROR = -90020005
|
|
TRIAL_FILENAME = 'wecom-native-trial.json'
|
|
|
|
|
|
class NativeReplyError(RuntimeError):
|
|
def __init__(self, code):
|
|
self.code = code
|
|
super().__init__('当前账号未进入接口开放范围' if code == GRAY_ERROR else '原生接口响应未通过验证')
|
|
|
|
|
|
def parse_native_reply(response):
|
|
"""Validate the actual transport and every JSON envelope, including data strings."""
|
|
if (not isinstance(response, dict) or response.get('httpRecognized') is not True
|
|
or response.get('statusLine') not in ('HTTP/1.1 200 OK', 'HTTP/1.0 200 OK')
|
|
or response.get('stage') != 5 or response.get('winsockError') != 0):
|
|
raise NativeReplyError('transport_failed')
|
|
body = response.get('body')
|
|
if not isinstance(body, str) or len(body.encode('utf-8')) > MAX_REPLY_BYTES:
|
|
raise NativeReplyError('invalid_response')
|
|
try:
|
|
node = json.loads(body)
|
|
except (ValueError, RecursionError):
|
|
raise NativeReplyError('invalid_json') from None
|
|
if (not isinstance(node, dict) or node.get('jsonrpc') != '2.0'
|
|
or type(node.get('id')) is not int or node['id'] != 1
|
|
or ('result' in node) == ('error' in node)):
|
|
raise NativeReplyError('invalid_envelope')
|
|
for _ in range(12):
|
|
if isinstance(node, dict):
|
|
if 'error' in node:
|
|
error = node['error']
|
|
code = error.get('code') if isinstance(error, dict) else None
|
|
raise NativeReplyError(code if type(code) is int else 'business_error')
|
|
if node.get('isError') is True:
|
|
raise NativeReplyError('tool_error')
|
|
if 'jsonrpc' in node:
|
|
node = node['result']; continue
|
|
if 'content' in node:
|
|
content = node['content']
|
|
if (not isinstance(content, list) or len(content) != 1
|
|
or not isinstance(content[0], dict) or content[0].get('type') != 'text'
|
|
or not isinstance(content[0].get('text'), str)):
|
|
raise NativeReplyError('invalid_tool_content')
|
|
node = content[0]['text']; continue
|
|
if 'data' in node:
|
|
node = node['data']; continue
|
|
if type(node.get('errcode')) is int and node['errcode'] != 0:
|
|
raise NativeReplyError(node['errcode'])
|
|
return node
|
|
if isinstance(node, str):
|
|
try:
|
|
decoded = json.loads(node)
|
|
except (ValueError, RecursionError):
|
|
if node.lstrip().startswith(('{', '[')):
|
|
raise NativeReplyError('invalid_nested_json') from None
|
|
return node
|
|
if decoded == node:
|
|
return node
|
|
node = decoded; continue
|
|
return node
|
|
raise NativeReplyError('envelope_depth_exceeded')
|
|
|
|
|
|
def trial_path():
|
|
from runtime_paths import application_data_dir
|
|
return application_data_dir() / 'logs' / TRIAL_FILENAME
|
|
|
|
|
|
def load_native_trial(*, path=None, client_versions=()):
|
|
"""Read history only; ignore arbitrary fields and never promote send readiness."""
|
|
path = Path(path) if path is not None else trial_path()
|
|
try:
|
|
with path.open('r', encoding='utf-8') as handle:
|
|
raw = handle.read(16385)
|
|
if len(raw) > 16384:
|
|
return {}
|
|
value = json.loads(raw)
|
|
if not isinstance(value, dict) or value.get('schemaVersion') != 1:
|
|
return {}
|
|
stamp = value.get('checkedAt')
|
|
version = value.get('clientVersion')
|
|
if (type(stamp) not in (float, int) or not math.isfinite(stamp)
|
|
or not 0 < stamp <= time.time() + 300 or not isinstance(version, str)
|
|
or len(version) > 64 or value.get('scope') != 'historical_isolated_trial'):
|
|
return {}
|
|
checks = value.get('businessChecks')
|
|
if not isinstance(checks, dict):
|
|
return {}
|
|
denied = all(checks.get(name) == GRAY_ERROR for name in ('identity', 'chat_list'))
|
|
return {
|
|
'scope': 'historical_isolated_trial', 'checkedAt': stamp,
|
|
'clientVersion': version, 'sameClientVersion': version in client_versions,
|
|
'transportVerified': value.get('transportVerified') is True,
|
|
'toolsListVerified': value.get('toolsListVerified') is True,
|
|
'businessStatus': 'not_enabled' if denied else 'unverified',
|
|
'businessErrorCode': GRAY_ERROR if denied else None,
|
|
'businessErrorMessage': 'user not in gray' if denied else '',
|
|
'canSend': False, 'loginVerified': False, 'sendVerified': False,
|
|
'receiptVerified': False,
|
|
}
|
|
except (OSError, ValueError, TypeError, RecursionError):
|
|
return {} |