本文最后更新于10 天前，其中的信息可能已经过时，如有错误请5Yqg5b6u5L+ha2Fud2s2NjY=
"""mtgsig 生成（微信小程序风控请求头），单文件实现，与 @mtfe/wx-jsguard 1.2 逐字节一致。
依赖
pip install pycryptodome # AES-128-CBC
Python 3.7+，无其它第三方依赖
快速上手
import mtgsig
s = mtgsig.MtgsigSigner.bootstrap("wx0000000000000000",
dfpid="") # 传 dfpid=纯离线；留空=自动注册一次
head, canon = s.sign("POST", url, body_dict, {"Content-Type": "application/json"})
requests.post(url, data=mtgsig.json_stringify(body_dict), headers={"mtgsig": head, ...})
要点
- 入签内容 = method + path + 排序后的 query + body 原文；**不含任何请求头**。
- body 必须按发送时一致的字节入签：dict 用 `json_stringify()`，str 原样。
- `bootstrap(dfpid=...)` 不产生任何网络请求；`dfpid=""` 会 POST 一次设备注册接口拿 dfpid
并回填 `server_time_diff`，注册失败直接抛异常（不做静默降级）。
- `a6` 载荷内嵌 dfpid，注册后会自动重建 `a6`，二者始终保持一致。
- 设备档案是常量 `DEVICE_PROFILE`（桌面微信端）；换端整块替换该常量或传 `profile=`。
公开 API
MtgsigSigner.bootstrap / .from_capture / .sign / .dfp_request / .dfp_report
calc_mtgsig(...) 无状态一次签名，参数全自己给
decode_mtgsig(head) / decode_a6(a6) 反解原料，用于核对与排错
json_stringify(obj) 与 JS JSON.stringify 等价的序列化
DEVICE_PROFILE / FPV / AES_KEY / ALPHABET 可调常量
算法对应关系（原始 JS 模块内行号，仅供核对）
ue():998-1058 z = ce(METHOD + " " + path + " " + sorted_query)，非 GET 非 form 时追加 body 字节
ue():1114-1120 l = BE4(ts) ；v = md5hex(ce(a6)+l) ；y = re(v[:15])，y[7]=(env^Gn(l))&0xff，
y += l，y += BE4(Gn(y)) -> siua 头 16 字节
ue():1121-1381 a5 = base64_custom( y ++ RC4(key=y, plain=json(qn)) ) KSA 比标准多 +31
ue():1382-1387 x = murmur3(z, ts)，_ = murmur3(ce(a5), ts)，
a4 = hex( BE4(x) ++ BE4(_) ++ LE4[x, _, x^o, x^_^o] )
ue():1397-1404 M = a1+a2+a3+a4+(_>>>0)+v+a7 -> md5 四字，j=(o<<3)|(o<<29)
B0^=j, B1^=f, B2^=f^j, B3^=B0 -> a5 之外的 d1
2704 localId = ts + 7位(k|65)大写字母 + md5 + crc32(前52字符)十进制前4位
3012 a6 = "w1.6" + base64(AES-CBC-PKCS7(gzip(按 vt.DFP 九字段投影的采集体)))
"""
import base64
import gzip
import hashlib
import json
import random
import struct
import time
import zlib
try:
from Crypto.Cipher import AES as _Cipher
except ImportError as _e: # 依赖边界，不做降级
raise ImportError("mtgsig 需要 pycryptodome：pip install pycryptodome") from _e
__all__ = ["MtgsigSigner", "calc_mtgsig", "decode_mtgsig", "decode_a6", "json_stringify",
"build_canonical", "build_a6", "build_dfp_request", "local_id", "new_profile",
"DEVICE_PROFILE", "FPV", "AES_KEY", "AES_IV", "ALPHABET"]
# ---------------------------------------------------------------- 基础工具
# jsguard.js:3352 自定义 base64 字母表
ALPHABET = "ZmserbBoHQtNP+wOcza/LpngG8yJq42KWYj0DSfdikx3VT16IlUAFM97hECvuRX5"
_UNRESERVED = set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_.!~*'()")
_CRC_POLY = 0xEDB88320
_CRC_FINAL = 306674911
_M = 1540483477 # murmur3 常量 0x5bd1e995
def _js_toint32(x):
x = int(x) % 0x100000000
return x - 0x100000000 if x >= 0x80000000 else x
def _js_touint32(x):
return int(x) % 0x100000000
undefined = object()
def _js_num(x):
"""JS Number -> 字符串：整值浮点不带 .0（31.0 -> "31"），与 JSON.stringify/ToString 一致"""
if isinstance(x, float):
if x != x or x in (float("inf"), float("-inf")):
return "null"
if x.is_integer() and abs(x) < 1e21:
return str(int(x))
return repr(x)
return str(x)
def _js_str(v):
"""JS ToString 语义（encodeURIComponent/字符串拼接收到的非字符串值走这套）"""
if v is None:
return "null"
if v is True:
return "true"
if v is False:
return "false"
if isinstance(v, (int, float)):
return _js_num(v)
if isinstance(v, dict):
return "[object Object]"
if isinstance(v, (list, tuple)):
return ",".join("" if e is None or e is undefined else _js_str(e) for e in v)
return v if isinstance(v, str) else str(v)
def json_stringify(o):
"""JS JSON.stringify：紧凑分隔、整值浮点写整数、undefined 键被丢弃"""
def conv(v):
if v is undefined:
return None
if isinstance(v, float):
return int(v) if v.is_integer() and abs(v) < 1e21 else v
if isinstance(v, dict):
return {k: conv(x) for k, x in v.items() if x is not undefined}
if isinstance(v, (list, tuple)):
return [conv(x) for x in v]
return v
return json.dumps(conv(o), ensure_ascii=False, separators=(",", ":"))
def encode_uri_component(s):
"""JS encodeURIComponent：仅不保留 A-Za-z0-9-_.!~*'()"""
out = []
for ch in _js_str(s):
if ch in _UNRESERVED:
out.append(ch)
else:
for b in ch.encode("utf-8"):
out.append("%%%02X" % b)
return "".join(out)
def fe(s):
"""jsguard fe()：encodeURIComponent 后再把 ! ' ( ) * 转义"""
return (encode_uri_component(s).replace("!", "%21").replace("'", "%27")
.replace("(", "%28").replace(")", "%29").replace("*", "%2A"))
def ce(s):
"""jsguard ce()：encodeURIComponent(s) 的逐字节数组（%XX 还原为字节）"""
e = encode_uri_component(s)
out = []
i = 0
while i < len(e):
if e[i] == "%":
out.append(int(e[i + 1:i + 3], 16))
i += 3
else:
out.append(ord(e[i]))
i += 1
return out
def re_(hexstr):
"""jsguard re()：十六进制串 -> 字节数组（奇数长度时末位单独解析）"""
out = []
i = 0
while i < len(hexstr):
out.append(int(hexstr[i:i + 2], 16))
i += 2
return out
def ie(x):
"""jsguard ie()：uint32 -> 4 字节大端"""
x = _js_touint32(x)
return [(x >> 24) & 255, (x >> 16) & 255, (x >> 8) & 255, x & 255]
def oe(bs):
return "".join("%02x" % b for b in bs)
_CRC = []
for _r in range(256):
_t = _r
for _ in range(8):
_t = (_t >> 1) ^ _CRC_POLY if _t & 1 else _t >> 1
_CRC.append(_t)
def Gn(data):
"""jsguard Gn()：CRC32 变体，末位异或 306674911"""
r = 0xFFFFFFFF
if isinstance(data, str):
for c in data:
r = _CRC[(r ^ ord(c)) & 0xFF] ^ (r >> 8)
r &= 0xFFFFFFFF
else:
for b in data:
r = _CRC[(r ^ b) & 0xFF] ^ (r >> 8)
r &= 0xFFFFFFFF
return _CRC_FINAL ^ r
def ge(data, seed):
"""jsguard ge()：murmur3-32 变体（长度混入种子、末尾再异或 0x5bd1e995）"""
n = len(data)
t = _js_toint32(seed) ^ n
c = 0
while n >= 4:
r = (data[c] & 255) | ((data[c + 1] & 255) << 8) | ((data[c + 2] & 255) << 16) | ((data[c + 3] & 255) << 24)
c += 4
r = _js_touint32(_M * (r & 0xFFFF) + ((_M * (r >> 16) & 0xFFFF) << 16))
t_word = _M * (t & 0xFFFF) + ((_M * (_js_touint32(t) >> 16) & 0xFFFF) << 16)
r = r ^ (r >> 24)
r = _js_touint32(_M * (r & 0xFFFF) + ((_M * (_js_touint32(r) >> 16) & 0xFFFF) << 16))
t = _js_toint32(t_word ^ r)
n -= 4
if n == 3:
t = _js_toint32(t ^ (data[c + 2] << 16))
t = _js_toint32(t ^ (data[c + 1] << 8))
t = _js_toint32(t ^ data[c])
t = _js_toint32(_M * (t & 0xFFFF) + ((_M * (_js_touint32(t) >> 16) & 0xFFFF) << 16))
elif n == 2:
t = _js_toint32(t ^ (data[c + 1] << 8))
t = _js_toint32(t ^ data[c])
t = _js_toint32(_M * (t & 0xFFFF) + ((_M * (_js_touint32(t) >> 16) & 0xFFFF) << 16))
elif n == 1:
t = _js_toint32(t ^ data[c])
t = _js_toint32(_M * (t & 0xFFFF) + ((_M * (_js_touint32(t) >> 16) & 0xFFFF) << 16))
t = _js_toint32(t ^ (_js_touint32(t) >> 13))
t = _M * (t & 0xFFFF) + ((_M * (_js_touint32(t) >> 16) & 0xFFFF) << 16)
t = _js_toint32(_js_touint32(t ^ (_js_touint32(t) >> 15)) ^ _M)
return t
def md5_words(data):
"""标准 MD5，返回 4 个 uint32 字（对应 jsguard 的 md5Array）"""
return list(struct.unpack("<4I", hashlib.md5(bytes(data)).digest()))
def le_bytes(words):
"""jsguard md5ToHex()+re()：uint32 字数组 <-> 小端字节数组"""
out = []
for w in words:
w &= 0xFFFFFFFF
out += [w & 0xFF, (w >> 8) & 0xFF, (w >> 16) & 0xFF, (w >> 24) & 0xFF]
return out
def b64_custom(bs):
"""jsguard se()/base64 VM：标准 base64 流程 + 自定义字母表"""
out = []
n = len(bs)
i = 0
while i + 3 <= n:
v = (bs[i] << 16) + (bs[i + 1] << 8) + bs[i + 2]
out.append(ALPHABET[(v >> 18) & 63] + ALPHABET[(v >> 12) & 63] + ALPHABET[(v >> 6) & 63] + ALPHABET[v & 63])
i += 3
rem = n - i
if rem == 1:
v = bs[i]
out.append(ALPHABET[v >> 2] + ALPHABET[(v << 4) & 63] + "==")
elif rem == 2:
v = (bs[i] << 8) + bs[i + 1]
out.append(ALPHABET[v >> 10] + ALPHABET[(v >> 4) & 63] + ALPHABET[(v << 2) & 63] + "=")
return "".join(out)
def rc4(key, text):
"""jsguard 内层 VM：KSA 多加了常数 31，PRGA 为标准 RC4"""
s = list(range(256))
j = 0
for i in range(256):
j = (j + s[i] + key[i % len(key)] + 31) % 256
s[i], s[j] = s[j], s[i]
out = []
i = j = 0
for ch in text:
i = (i + 1) % 256
j = (j + s[i]) % 256
s[i], s[j] = s[j], s[i]
out.append(ord(ch) ^ s[(s[i] + s[j]) % 256])
return out
# ---------------------------------------------------------------- 参数规范化
URL_RE = r"^(?:([A-Za-z]+):)?(\/{0,3})([0-9.\-A-Za-z]+)(?::(\d+))?(?:\/([^?#]*))?(?:\?([^#]*))?(?:#(.*))?$"
def parse_url(url):
"""还原 ue() 里的正则解析：只关心 path(5) 与 query(6)"""
import re
m = re.match(URL_RE, url)
if not m:
return "/", []
return (m.group(5) or ""), (m.group(6) or "")
def ne(query, keep_undefined=False):
"""jsguard ne()：query -> [(k, v)]"""
pairs = []
for part in query.split("&"):
kv = part.split("=")
if len(kv) < 2:
pairs.append([_decode(kv[0].replace("+", " ")), "undefined" if keep_undefined else ""])
else:
pairs.append([_decode(kv[0].replace("+", " ")), _decode(kv[1].replace("+", " "))])
return pairs
def _decode(s):
"""JS decodeURIComponent：连续 %XX 组成一个 UTF-8 段严格解码，非法即抛 URIError"""
out = []
i = 0
n = len(s)
while i < n:
if s[i] != "%":
out.append(s[i])
i += 1
continue
run = bytearray()
while i < n and s[i] == "%":
hx = s[i + 1:i + 3]
if len(hx) < 2 or any(c not in "0123456789abcdefABCDEF" for c in hx):
raise ValueError("URIError: malformed URI")
run.append(int(hx, 16))
i += 3
try:
out.append(run.decode("utf-8"))
except UnicodeDecodeError:
raise ValueError("URIError: malformed URI")
return "".join(out)
def te(pairs_out, src, encode=False):
"""jsguard te()：把 dict 或 [(k,v)] 追加为 [fe(k), fe(v)]"""
if encode:
items = (src.items() if isinstance(src, dict) else
list(enumerate(src)) if isinstance(src, (list, tuple)) else src)
for k, v in items:
if v is undefined:
pairs_out.append([fe(k), "undefined"])
elif v is None:
pairs_out.append([fe(k), "null"])
elif isinstance(v, (dict, list)):
pairs_out.append([fe(k), fe(json_stringify(v))])
else:
pairs_out.append([fe(k), fe(v)])
else:
for k, v in src:
pairs_out.append([fe(k), fe(v)])
def _pair_sort_key(p):
return (p[0], p[1])
def build_canonical(method, url, data, headers):
"""构造 z 字节数组（jsguard ue() 1011-1058 行）"""
method = (method or "GET").upper()
headers = headers or {}
path, query = parse_url(url)
path = "/" + path.lstrip("/") if path else "/"
g = ne(query) if query else [] # JS: b[6] && (g = ne(b[6])) —— 空 query 不解析
is_form = method != "GET" and _content_type_is_form(headers)
pairs = []
body_str = ""
if method == "GET":
if isinstance(data, (dict, list, tuple)) and len(data) > 0:
te(pairs, data, True)
if query and g:
keys = set(data) if isinstance(data, dict) else {str(i) for i in range(len(data))}
extra = {}
for k, v in ne(query, True): # JS: v[key]=value，重复键后者覆盖
if k not in keys:
extra[k] = v
te(pairs, extra, True)
else:
te(pairs, g)
else:
te(pairs, g)
if is_form:
if isinstance(data, str):
body_str = data
elif isinstance(data, dict):
body_str = "&".join(f"{encode_uri_component(k)}={encode_uri_component(v)}" for k, v in data.items())
elif isinstance(data, (list, tuple)):
body_str = "&".join(f"{i}={encode_uri_component(v)}" for i, v in enumerate(data))
pairs.sort(key=_pair_sort_key)
canon = method + " " + path + " " + "&".join(f"{k}={v}" for k, v in pairs)
z = ce(canon)
if not (is_form or method == "GET" or data is None):
raw = data if isinstance(data, str) else json_stringify(data)
z += ce(raw)[:16200]
if len(body_str) > 0:
z += ce(body_str)[:16200]
return z, canon
def _content_type_is_form(headers):
"""jsguard ue() 998-1006 行：只有 content-type 以 form-urlencoded 开头才为 True"""
found = False
for k, v in headers.items():
if k.lower() == "content-type":
found = True
if v and str(v).lower().startswith("application/x-www-form-urlencoded"):
return True
return False
# ---------------------------------------------------------------- 签名主体
def calc_mtgsig(method, url, data=None, headers=None, *, ts, dfpid, appid,
token="", qn=None, env=0, x0=3, version="1.2", route="", sig_extra="00000"):
"""生成 mtgsig 头。ts 必须为毫秒整数（对应 a2），qn 为指纹 JSON 字符串。
env 取 0（微信端常见值，可由 decode_mtgsig 从既有头反解核对）；qn 为 None 时用 _build_qn 造最小指纹体。
"""
o = _js_touint32(ts)
l = ie(o)
if qn is None:
qn = _build_qn(ts, appid, route, sig_extra)
# siua 头 16 字节
v = hashlib.md5(bytes(ce(token) + l)).hexdigest()
y = re_(v[:15])
y[7] = 255 & (env ^ Gn(l))
y += l
y += ie(_js_touint32(Gn(y)))
# a5 = base64( y ++ rc4(key=y, data=qn) )
enc = y + rc4(y, qn)
a5 = b64_custom(enc)
z, canon = build_canonical(method, url, data, headers)
x = ge(z, ts)
_u = ge(ce(a5), ts)
f = _js_touint32(_u)
a4 = oe(ie(x) + ie(_u) + le_bytes([x, _u, x ^ o, x ^ _u ^ o]))
sig = {"a1": version, "a2": ts, "a3": dfpid, "a4": a4, "a5": a5, "a6": token, "a7": appid, "x0": x0}
m_str = "".join([sig["a1"], str(ts), sig["a3"], sig["a4"], str(f), v, sig["a7"]])
b = md5_words(ce(m_str))
j = ((o << x0) & 0xFFFFFFFF) | ((o << (32 - x0)) & 0xFFFFFFFF)
b[0] ^= j
b[1] ^= f
b[2] ^= f ^ j
b[3] ^= b[0]
sig["d1"] = "".join("%02x" % c for c in le_bytes(b))
return sig, canon
def b64_custom_decode(s):
out, acc, bits = [], 0, 0
for ch in s.rstrip("="):
acc = acc * 64 + ALPHABET.index(ch)
bits += 6
if bits >= 8:
bits -= 8
out.append((acc >> bits) & 255)
return out
def decode_mtgsig(header_value):
"""把一个真实抓到的 mtgsig 头拆回原料：ts / dfpid / token / appid / env / qn 明文。
用于在离线端复刻同一台设备的签名上下文（对应 calc 的全部输入）。
"""
sig = json.loads(header_value) if isinstance(header_value, str) else dict(header_value)
raw = b64_custom_decode(sig["a5"])
y, cipher = raw[:16], raw[16:]
ts = sig["a2"]
l = ie(_js_touint32(ts))
env = (y[7] ^ Gn(l)) & 0xFF
qn = _rc4_decrypt(y, cipher)
return {"fields": sig, "ts": ts, "dfpid": sig["a3"], "token": sig.get("a6", ""),
"appid": sig.get("a7", ""), "env": env, "y": y, "qn": qn,
"x0": sig.get("x0", 3), "version": sig.get("a1", "1.2")}
def _rc4_decrypt(key, data):
"""RC4 对称：把密文字节映射回可 XOR 的字符序列（js 侧明文是 charCode 串）"""
s = list(range(256))
j = 0
for i in range(256):
j = (j + s[i] + key[i % len(key)] + 31) % 256
s[i], s[j] = s[j], s[i]
i = j = 0
out = []
for b in data:
i = (i + 1) % 256
j = (j + s[i]) % 256
s[i], s[j] = s[j], s[i]
out.append(chr(b ^ s[(s[i] + s[j]) % 256]))
return "".join(out)
class MtgsigSigner:
"""签名上下文：一台设备的 dfpid / appid / 指纹体 qn / env，之后按请求刷新 ts。
三种用法：
1) 复用一条已有头：MtgsigSigner.from_capture(mtgsig_str)
2) 复用已注册设备：MtgsigSigner.bootstrap(appid, dfpid="<上次拿到的 data.dfp>") —— 纯离线
3) 首次/无 dfpid：MtgsigSigner.bootstrap(appid) —— 会真发一次 /v1/wxdfpid 注册
"""
def __init__(self, appid, dfpid, qn=None, env=0, token="", version="1.2",
route="", server_time_diff=0, profile=None, openid="", accuracy="", sig_flags="00000"):
self.appid = appid
self.dfpid = dfpid
self.env = env
self.token = token
self.version = version
self.route = route
self.server_time_diff = server_time_diff
self.profile = profile
self.openid = openid
self.accuracy = accuracy
self.sig_flags = sig_flags
self._qn_fields = None
self.counter = 1
if qn is not None:
self._qn_fields = json.loads(qn)
self.counter = int(self._qn_fields.get("b8", 1))
@classmethod
def from_capture(cls, mtgsig_str, server_time_diff=0):
"""用一条已有的 mtgsig 建立签名上下文：qn 按字段还原，b8 计数继续自增。
qn 字段语义：b7=guard 初始化秒级时间戳（整个会话冻结）、
b1=getAccountInfoSync()、b6=openId、b8=签名调用次数、b12=appid、
b2=当前页 route、b9=生命周期标志位 5 元组、b11=定位精度 acc_hAcc_vAcc。
"""
d = decode_mtgsig(mtgsig_str)
return cls(appid=d["appid"], dfpid=d["dfpid"], env=d["env"], token=d["token"],
version=d["version"], qn=d["qn"], server_time_diff=server_time_diff)
@classmethod
def bootstrap(cls, appid, dfpid="", profile=None, openid="", session_id="", accuracy="",
server_time_diff=0, sig_flags="22122", timeout=10):
"""自建上下文（不依赖任何已有样本）：a6 走 i() 本地算，设备档案取常量 DEVICE_PROFILE（桌面端）。
dfpid 传入则直接复用（不发网络请求）；留空则现场注册一次 —— 会 POST
https://msp.meituan.com/v1/wxdfpid 并回填 a3 与 serverTimeDiff。
注册失败会抛异常，不做静默降级：a3 用本地 localId 兜底虽能签出格式合法的头，
但那是不被服务端认识的设备，失败应当让你看见。
"""
p = dict(profile or new_profile(appid))
p.setdefault("app", appid)
filetime = int(p.get("filetime") or time.time() * 1000)
system = p["system"]
lid = local_id(system.get("model"), system, filetime, openid)
los = system.get("LaunchOptionsSync") or {}
route = los.get("path", "") if isinstance(los, dict) else json.loads(los or "{}").get("path", "")
a6, n = build_a6(p, dfpid or lid, lid, filetime, filetime // 1000, session_id)
sig = cls(appid=appid, dfpid=dfpid or lid, env=0, token=a6, route=route,
server_time_diff=server_time_diff, profile=p, openid=openid,
accuracy=accuracy, sig_flags=sig_flags)
sig.filetime = filetime
sig.localid = lid
sig.session_id = session_id
sig.n = n
sig.device = {k: system.get(k) for k in ("platform", "brand", "model", "system", "SDKVersion", "version")}
if not dfpid:
sig.dfp_report(timeout=timeout)
return sig
def _qn(self, ts):
if self._qn_fields is not None:
fields = dict(self._qn_fields)
fields["b8"] = self.counter
return json.dumps(fields, ensure_ascii=False, separators=(",", ":"))
b7 = (getattr(self, "filetime", ts) or ts) // 1000
return _build_qn(b7 * 1000, self.appid, self.route, self.sig_flags,
self.openid, self.accuracy)
def dfp_request(self, err_stack=""):
"""生成 /v1/wxdfpid 注册请求体（换服务端 dfpid 用），dfp_report() 可直接发送。"""
p = self.profile or new_profile(self.appid)
filetime = int(getattr(self, "filetime", 0) or time.time() * 1000)
lid = getattr(self, "localid", "") or local_id(p["system"].get("model"), p["system"], filetime, self.openid)
return build_dfp_request(p, self.dfpid, lid, filetime, int(time.time() * 1000), err_stack)
def dfp_report(self, ua="", timeout=10):
"""真发送 /v1/wxdfpid，返回 (dfpid, serverTimeDiff) 并写入上下文。dfp_request() 为其离线构造。"""
import urllib.request
req = self.dfp_request()
headers = {"Content-Type": "application/json"}
if ua:
headers["User-Agent"] = ua
r = urllib.request.Request(req["url"], data=req["body"].encode("utf-8"), headers=headers, method="POST")
with urllib.request.urlopen(r, timeout=timeout) as resp:
js = json.loads(resp.read().decode("utf-8"))
d = js.get("data") or {}
if not d.get("dfp"):
raise ValueError("dfp 接口未返回 data.dfp: %s" % json.dumps(js, ensure_ascii=False)[:200])
now_ms = int(time.time() * 1000)
self.dfpid = d["dfp"]
self.server_time_diff = int(d.get("serverTimestamp", now_ms)) - now_ms
if self.profile is not None:
# a6 投影槽 1 就是 dfpid，注册后必须用新 dfpid 重新生成，否则 a3 与 a6 内不一致
self.token, self.n = build_a6(self.profile, self.dfpid, self.localid, self.filetime,
self.filetime // 1000, getattr(self, "session_id", ""))
return self.dfpid, self.server_time_diff
def sign(self, method, url, data=None, headers=None, ts=None):
"""返回 (mtgsig 头字符串, 规范化请求串)。ts 为毫秒时间戳，默认取当前时间。"""
base_ts = int(ts if ts is not None else time.time() * 1000) + self.server_time_diff
sig, canon = calc_mtgsig(method, url, data, headers, ts=base_ts, dfpid=self.dfpid,
appid=self.appid, token=self.token, qn=self._qn(base_ts),
env=self.env, version=self.version)
self.counter += 1
return json.dumps(sig, ensure_ascii=False, separators=(",", ":")), canon
# ---------------------------------------------------------- 指纹上报（a6 / dfp 注册）
# jsguard:2704 起 —— _n 采集体 + vt 模式投影 + sjcl AES-128-CBC(PKCS7) + gzip
AES_KEY = b"z7Jut6Ywr2Pe5Nhx" # jsguard:802 常量 1（hex-XOR 解出）
AES_IV = b"0807060504030201" # jsguard:802 常量 2
DFP_PREFIX = "WX__ver1.2.0_CCCC_" # jsguard:2704 Dn
FPV = "2.5.0"
DFP_URL = "https://msp.meituan.com/v1/wxdfpid" # jsguard:2696/2843
JSGUARD_TWEAK_POS = {"key": (2, 5, 8, 9), "iv": (3, 5, 6, 9, 10)}
_DFP_SCHEMA = ["app", "dfpid", "filetime", "fpv", "localid", "system", "timestamp", "ext", "sessionId"]
_SYSTEM_SCHEMA = ["accelerometer", "albumAuthorized", "BatteryInfo", "batteryLevel", "Beacons", "benchmarkLevel",
"bluetoothEnabled", "brand", "brightness", "cameraAuthorized", "compass", "deviceOrientation",
"devicePixelRatio", "enableDebug", "errMsg", "fontSizeSetting", "language", "LaunchOptionsSync",
"locationAuthorized", "locationEnabled", "locationReducedAccuracy", "microphoneAuthorized",
"model", "networkType", "notificationAlertAuthorized", "notificationAuthorized",
"notificationBadgeAuthorized", "notificationSoundAuthorized", "pixelRatio", "platform",
"safeArea", "screenHeight", "screenTop", "screenWidth", "SDKVersion", "statusBarHeight",
"system", "version", "wifiEnabled", "WifiInfo", "windowHeight", "windowWidth",
"screenRecord", "isPrivacy", "hasSystemProxy", "captureRecord"]
_SUB_SCHEMA = {
"BatteryInfo": ["errMsg", "isCharging", "level"],
"safeArea": ["left", "right", "top", "bottom", "width", "height"],
"WifiInfo": ["SSID", "BSSID", "autoJoined", "signalStrength", "justJoined", "secure", "frequency"],
}
def _pkcs7(data, block=16):
pad = block - len(data) % block
return data + bytes([pad]) * pad
def _unpkcs7(data):
return data[:-data[-1]]
def aes_encrypt_b64(raw, tweak=""):
"""jsguard Z(e)：sjcl AES-128-CBC + PKCS7 + 标准 base64（与 JS 侧双向互通已实测）"""
key, iv = _tweak_kv(tweak) if len(tweak) == 6 else (AES_KEY, AES_IV)
ct = _Cipher.new(key, _Cipher.MODE_CBC, iv).encrypt(_pkcs7(raw))
return base64.b64encode(ct).decode()
def aes_decrypt_b64(b64_str, tweak=""):
key, iv = _tweak_kv(tweak) if len(tweak) == 6 else (AES_KEY, AES_IV)
pt = _Cipher.new(key, _Cipher.MODE_CBC, iv).decrypt(base64.b64decode(b64_str + "=" * (-len(b64_str) % 4)))
return _unpkcs7(pt)
def _tweak_kv(tweak):
"""jsguard:814 —— 6 位码插入 key[2,5,8,9]（取 t0,t1,t2,t4）与 iv[3,5,6,9,10]（取 t0,t2,t3,t4,t5）"""
k = list(AES_KEY.decode())
for pos, ch in zip(JSGUARD_TWEAK_POS["key"], [tweak[i] for i in (0, 1, 2, 4)]):
k[pos] = ch
v = list(AES_IV.decode())
for pos, ch in zip(JSGUARD_TWEAK_POS["iv"], [tweak[i] for i in (0, 2, 3, 4, 5)]):
v[pos] = ch
return "".join(k).encode(), "".join(v).encode()
def project(obj, schema):
"""jsguard:3090 的模式投影：dict -> 按 schema 取值的数组（含三处特殊整形）"""
out = []
for key in schema:
v = obj.get(key)
if key == "LaunchOptionsSync" and v:
parsed = json.loads(v) if isinstance(v, str) else v
v = json.dumps({"path": parsed.get("path"), "scene": parsed.get("scene")},
ensure_ascii=False, separators=(",", ":"))
elif key == "accelerometer" and isinstance(v, list) and v:
v = [[s["x"], s["y"], s["z"]] for s in v if s.get("x") and s.get("y") and s.get("z")]
elif key in ("BatteryInfo", "WifiInfo", "safeArea") and isinstance(v, str):
v = json.loads(v)
if isinstance(v, dict) and key in _SUB_SCHEMA:
v = project(v, _SUB_SCHEMA[key])
out.append(v)
return out
def local_id(model, system, filetime, openid):
"""jsguard:2704 An()：ts + 7 位奇数字母 + md5 + crc32 十进制前 4 位"""
rand = ""
for _ in range(7):
rand += chr(int(random.random() * 25) | 65) # JS: Math.random()*(90-65) | 0+65 -> (k|65)
core = json_stringify({"model": model, "system": system, "timestamp": filetime, "openid": openid}).encode()
body = "" + str(filetime) + rand + hashlib.md5(core).hexdigest()
return body + str(zlib.crc32(body.encode()) & 0xFFFFFFFF)[:4]
def build_a6(profile, dfpid, localid, filetime, now_sec, session_id="", ext=None, openid=""):
"""jsguard:3012 i() -> ("w1.6" + base64(AES(gzip(project(_n, DFP)))), _n)"""
n = dict(profile)
n["app"] = profile.get("app", "")
n["dfpid"] = dfpid
n["localid"] = localid
n["filetime"] = filetime
n["fpv"] = FPV
n["timestamp"] = now_sec
n["ext"] = ext if ext is not None else profile.get("ext", [0, 1, 2, 0, 4])
n["sessionId"] = session_id
arr = project(n, _DFP_SCHEMA)
js = json_stringify(arr).encode()
return "w1.6" + aes_encrypt_b64(gzip.compress(js)), n
def decode_a6(token):
"""反向核对：解开任意 a6（本地自造或既有头里的），按 DFP 模式还原成 dict"""
raw = json.loads(gzip.decompress(aes_decrypt_b64(token[len("w1.6"):])))
out = {}
for key, val in zip(_DFP_SCHEMA, raw):
if isinstance(val, list) and key in _SUB_SCHEMA:
val = dict(zip(_SUB_SCHEMA[key], val))
out[key] = val
if isinstance(out.get("system"), list):
out["system"] = dict(zip(_SYSTEM_SCHEMA, out["system"]))
return out
def build_dfp_request(profile, dfpid, localid, filetime, now_ms, err_stack="", tweak_key=""):
"""jsguard:2821 Pn 的出站体：POST {DFP_URL}，data = "WX__ver1.2.0_CCCC_" + AES(json(_n'))"""
n = dict(profile)
n["ext"] = profile.get("ext", [0, 1, 2, 0, 4])
n["fsmode"] = profile.get("fsmode", [])
n["e"] = err_stack
n["app"] = profile.get("app")
n["dfpid"] = dfpid
n["localid"] = localid
n["filetime"] = filetime
n["fpv"] = FPV
n["timestamp"] = now_ms // 1000
n["reportTick"] = 0
body = {"jsgVersion": FPV,
"data": DFP_PREFIX + aes_encrypt_b64(json_stringify(n).encode(), tweak_key),
"jsdfpVersion": FPV,
"time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(now_ms / 1000)),
"os": "wechat"}
return {"url": DFP_URL, "method": "POST",
"headers": {"content-type": "application/json"},
"body": json.dumps(body, ensure_ascii=False, separators=(",", ":"))}
# 固定桌面端指纹（微信小程序 PC 端）。一次定好后写死，运行期不再随机；换端整块替换。
# 来源说明：brand/model/platform/system/SDKVersion/version/权限位/hasSystemProxy 是桌面微信端常见取值；
# 窗口几何(480x960)、亮度、scene 是自选值 —— PC 上这些由窗口大小和用户设置决定，没有标准答案。
DEVICE_PROFILE = {
"accelerometer": [], "albumAuthorized": True,
"BatteryInfo": {"errMsg": "getBatteryInfo:ok", "isCharging": True, "level": 100},
"batteryLevel": None, "Beacons": None, "benchmarkLevel": -1, "bluetoothEnabled": False,
"brand": "microsoft", "brightness": 0.6, "cameraAuthorized": True, "compass": [],
"deviceOrientation": None, "devicePixelRatio": 1.5, "enableDebug": False,
"errMsg": "getSystemInfo:ok", "fontSizeSetting": None, "language": "zh_CN",
"LaunchOptionsSync": {"path": "index/pages/mt/mt", "scene": 1256},
"locationAuthorized": True, "locationEnabled": True, "locationReducedAccuracy": None,
"microphoneAuthorized": True, "model": "microsoft", "networkType": "wifi",
"notificationAlertAuthorized": None, "notificationAuthorized": True,
"notificationBadgeAuthorized": None, "notificationSoundAuthorized": None,
"pixelRatio": 1.5, "platform": "windows",
"safeArea": {"left": 0, "right": 480, "top": 0, "bottom": 960, "width": 480, "height": 960},
"screenHeight": 960, "screenRecord": None, "screenTop": None, "screenWidth": 480,
"SDKVersion": "3.17.3", "statusBarHeight": 20, "system": "Windows 11 x64",
"version": "4.1.13.65", "wifiEnabled": True, "WifiInfo": None,
"windowHeight": 960, "windowWidth": 480, "captureRecord": "[]", "isPrivacy": 1, "hasSystemProxy": -1,
}
def new_profile(appid, system=None):
"""jsguard:2702 的 _n 采集体骨架"""
return {"system": dict(DEVICE_PROFILE if system is None else system), "fpv": FPV, "app": appid,
"ext": [0, 1, 2, 0, 4], "fsmode": []}
def _build_qn(ts, appid, route, sig_extra, openid="", accuracy=""):
"""qn 指纹体：字段顺序与采集端插入顺序一致（b7,b1[,b6],b8,b12,b2,b9[,b11]）"""
payload = {"b7": ts // 1000,
"b1": {"miniProgram": {"appId": appid, "envVersion": "release", "version": "10.39.1"}}}
if openid:
payload["b6"] = openid
payload["b8"] = 1
payload["b12"] = appid
payload["b2"] = route
payload["b9"] = sig_extra
if accuracy:
payload["b11"] = accuracy
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
if __name__ == "__main__":
import sys
APPID = "wx0000000000000000" # 换成你的小程序 appid
DFPID = sys.argv[1] if len(sys.argv) > 1 else "0" * 56 # 不给参数用占位 dfpid，自检不打网络
signer = MtgsigSigner.bootstrap(APPID, dfpid=DFPID)
print("dfpid :", signer.dfpid, "（%s）" % ("复用传入" if len(sys.argv) > 1 else "占位，未发请求"))
print("时差 :", signer.server_time_diff, "ms")
print("设备 :", json.dumps(signer.device, ensure_ascii=False))
# ---------- 案例 1：GET，参数全在 query，无 body ----------
get_url = ("https://wx.meituan.com/weapp/api/poi/v2/poi_detail?poi_id=123456®ion_id=6"
"&yodaReady=wx&csecplatform=3&csecappid=%s" % APPID)
head_get, canon_get = signer.sign("GET", get_url, headers={"Content-Type": "application/json"})
print("\n[GET ] canonical :", canon_get)
print("[GET ] mtgsig :", head_get)
# 真正发请求时（示例，不执行）：
# requests.get(get_url, headers={"mtgsig": head_get, "token": <登录态>, "openId": ..., "uuid": ...})
# ---------- 案例 2：POST，JSON body ----------
post_url = ("https://web.meituan.com/api/miniprogram/index/aggregate?riskLevel=71"
"&csecappid=%s&csecplatform=3&csecversion=3.0.2" % APPID)
post_body = {"ci": 673, "districtId": 2915, "lat": 31.037891, "lng": 117.08113499999999,
"scene": 1256, "platform": 1, "globalId": ""}
body_str = json_stringify(post_body) # 实际发送必须用这个串，和签名逐字节一致
head_post, canon_post = signer.sign("POST", post_url, post_body, {"Content-Type": "application/json"})
z_post, _ = build_canonical("POST", post_url, post_body, {"Content-Type": "application/json"})
print("\n[POST] canonical :", canon_post)
print("[POST] 实际入签 : 上面这段 + body 原文，合计 %d 字节（body 在 z 里，canonical 只显示前半）" % len(z_post))
print("[POST] body :", body_str)
print("[POST] mtgsig :", head_post)
# requests.post(post_url, data=body_str,
# headers={"mtgsig": head_post, "Content-Type": "application/json",
# "token": <登录态>, "openId": ..., "uuid": ..., "csecuuid": ...})
# ---------- 表单类 POST（content-type 决定 body 怎么进签名）----------
form_url = "https://epassport.meituan.com/authorize?yoda_req=1"
head_form, canon_form = signer.sign("POST", form_url, {"code": "abc 123", "state": "x/y"},
{"Content-Type": "application/x-www-form-urlencoded"})
print("\n[FORM] canonical :", canon_form)
# ---------- 自检 ----------
d = decode_mtgsig(json.loads(head_post))
print("\n自检: a3==signer.dfpid %s | a6 内 dfpid==a3 %s | qn 可解 %s | b8 计数 %s(每次 sign +1)" % (
d["dfpid"] == signer.dfpid,
decode_a6(d["token"])["dfpid"] == d["dfpid"],
d["qn"].startswith('{"b7"'),
json.loads(d["qn"])["b8"]))