网站：https://www.bet365.com
AES-128-CTR 数据加解密
Base64 外层传输
自定义 TLV 二进制协议
嵌套字段序列化
小端序整数编码
JavaScript 32 位运算兼容
FNV-1a 32 位哈希
利用混淆算法
请求校验字段编码
在逆向实时数据网站的过程中参数不断变化，参数利用浏览器指纹，以及检测 WebRTC 暴露的公网映射地址。
所以逆向这个参数的步骤就是通过 aiortc 创建 WebRTC DataChannel Offer，向指定 STUN 服务器收集 ICE 候选，然后返回包含候选信息的本地 SDP，并从中提取 srflx IPv6。解决这个检测 WebRTC 暴露的本机问题，然后再利用以上加密协议的原理进行逆向获取完整的参数，
RESOURCES
文章资源
代码文件
下载 · 19.5 KBglobal_encryption_generate
global_encryption_generate.py
import operator
from Crypto.Cipher import AES
from Crypto.Util import Counter
import base64
from typing import Dict, Any, Union
key = bytes([45, 67, 89, 12, 34, 56, 78, 90, 123, 234, 45, 67, 89, 12, 34, 56])
def aes_ctr_encrypt(plaintext,nonce,key=key):
# ctr = Counter.new(128, initial_value=int.from_bytes(nonce, byteorder='big'))
# cipher = AES.new(key, AES.MODE_CTR, counter=ctr)
# ciphertext = cipher.encrypt(plaintext)
# return ciphertext
ctr = Counter.new(128, initial_value=int.from_bytes(nonce, byteorder='big'))
cipher = AES.new(key, AES.MODE_CTR, counter=ctr)
ciphertext = cipher.encrypt(plaintext)
return ciphertext
def aes_ctr_decrypt(nonce, ciphertext,key=key):
ctr = Counter.new(128, initial_value=int.from_bytes(nonce, byteorder='big'))
cipher = AES.new(key, AES.MODE_CTR, counter=ctr)
plaintext = cipher.decrypt(ciphertext)
return plaintext
def decode__(b64_text):
decoded_bytes = base64.b64decode(b64_text)
ascii_byte_list = list(decoded_bytes)
x_net_header = ascii_byte_list[0] + ascii_byte_list[5] + 4
the_header = ascii_byte_list[:x_net_header]
nonce = ascii_byte_list[x_net_header:x_net_header + 16]
ciphertext = ascii_byte_list[(x_net_header + 16):]
return the_header, nonce, aes_ctr_decrypt(bytes(nonce), bytes(ciphertext))
def decode_byte_to_dict(bytestext):
def parse_data(data):
pos = 0
result = {}
while pos < len(data):
if pos + 1 > len(data):
break
key_length = data[pos]
pos += 1
if pos + key_length > len(data):
break
key = data[pos:pos + key_length].decode('utf-8')
pos += key_length
if pos + 2 > len(data):
break
value_length_part1 = data[pos]
value_length_part2 = data[pos + 1]
value_length = value_length_part1 + value_length_part2 * 256
pos += 2
if pos + value_length > len(data):
break
value = data[pos:pos + value_length]
pos += value_length
if ("i_c" not in result.keys() and key in ["d", "b", "p", "r", "q", "v", "w"]) or (key.startswith("i_") ):
value = int.from_bytes(value, byteorder='little')
else:
try:
value = value.decode('utf-8')
except:
pass
result[key] = value
return result
dicts = parse_data(bytestext)
if 'f' in dicts:
dicts['f'] = parse_data(dicts['f'])
return dicts
def jisuan(a1, a2, func):
# 将值限制在 32 位范围内
a1 = a1 & 0xFFFFFFFF
a2 = a2 & 0xFFFFFFFF
# 执行指定的位操作
result = func(a1, a2) & 0xFFFFFFFF
# 如果结果的最高位是 1，则这是一个负数
if result & 0x80000000:
result -= 0x100000000
return result
def right3(value, shift):
# js中value>>>shift
if value < 0:
value += 2 ** 32
# 进行右移操作
return (value >> shift) & (2 ** 32 - 1)
num1 = 1101195530
num2 = 1849170683
num3 = 1011462487
num4 = 1870552039
num5 = 2116561607
num6 = 1297822139
num7 = 495394314
num8 = 684556403
num9 = 1902634096
num10 = 37428903
num11 = 787769321
def encodechat(char, key):
f1 = jisuan(key, ord(char), operator.xor)
a2 = jisuan(f1, 1, operator.lshift)
a3 = right3(f1, 31)
a4 = jisuan(a2, a3, operator.or_)
f1 = jisuan(a4, num1, operator.xor)
f1 = forfunc(f1, num1)
f1 = forfunc(f1, num2)
f1 = forfunc(f1, num3)
f1 = forfunc(f1, num4)
f1 = forfunc(f1, num5)
f1 = forfunc(f1, num6)
f1 = forfunc(f1, num7)
f1 = forfunc(f1, num8)
a2 = jisuan(f1, 1, operator.lshift)
a3 = right3(f1, 31)
a4 = jisuan(a2, a3, operator.or_)
f1 = jisuan(a4, num2, operator.xor)
f1 = forfunc(f1, num2)
f1 = forfunc(f1, num3)
f1 = forfunc(f1, num4)
f1 = forfunc(f1, num5)
f1 = forfunc(f1, num6)
f1 = forfunc(f1, num7)
f1 = forfunc(f1, num8)
f1 = forfunc(f1, num9)
a2 = jisuan(f1, 1, operator.lshift)
a3 = right3(f1, 31)
a4 = jisuan(a2, a3, operator.or_)
f1 = jisuan(a4, num3, operator.xor)
f1 = forfunc(f1, num3)
f1 = forfunc(f1, num4)
f1 = forfunc(f1, num5)
f1 = forfunc(f1, num6)
f1 = forfunc(f1, num7)
f1 = forfunc(f1, num8)
f1 = forfunc(f1, num9)
f1 = forfunc(f1, num10)
a2 = jisuan(f1, 1, operator.lshift)
a3 = right3(f1, 31)
a4 = jisuan(a2, a3, operator.or_)
f1 = jisuan(a4, num4, operator.xor)
f1 = forfunc(f1, num4)
f1 = forfunc(f1, num5)
f1 = forfunc(f1, num6)
f1 = forfunc(f1, num7)
f1 = forfunc(f1, num8)
f1 = forfunc(f1, num9)
f1 = forfunc(f1, num10)
f1 = forfunc(f1, num11)
return f1
def forfunc(a5, keys):
a6 = jisuan(a5, 65535, operator.and_)
a7 = jisuan(a5, keys, operator.xor)
a8 = jisuan(a7, 65535, operator.and_)
a9 = jisuan(a6, a8, operator.mul)
a10 = jisuan(a9, 65535, operator.and_)
a11 = right3(a5, 16)
a12 = jisuan(a11, 65535, operator.and_)
a13 = jisuan(a5, keys, operator.xor)
a14 = right3(a13, 16)
a15 = jisuan(a12, a14, operator.mul)
a16 = jisuan(a15, 65535, operator.and_)
a17 = jisuan(a10, 16, operator.lshift)
a18 = a5 + a17
a19 = a18 + a16
a20 = right3(a19, 0)
a21 = jisuan(a20, 15, operator.and_)
a22 = a21 + 1
a23 = right3(a20, a22)
a24 = jisuan(a20, a23, operator.xor)
return a24
def to_signed_32bit(value):
# 将整数限制在 32 位范围内
value = value & 0xFFFFFFFF
# 如果最高位是1，将其解释为负数
if value & 0x80000000:
return value - 0x100000000
else:
return value
def hex_to_byte_list(hex_num):
# 确保输入为整数
if not isinstance(hex_num, int):
raise ValueError("Input must be an integer")
byte_list = [
(hex_num >> 24) & 0xFF, # 提取最高字节
(hex_num >> 16) & 0xFF, # 提取次高字节
(hex_num >> 8) & 0xFF, # 提取次低字节
hex_num & 0xFF # 提取最低字节
]
return byte_list
def fnv1a_32(data):
FNV_prime = 16777619
offset_basis = 2166136261
hash_value = offset_basis
for byte in data:
hash_value ^= byte
hash_value *= FNV_prime
# 确保哈希值保持在 32 位范围内
hash_value &= 0xFFFFFFFF
return hash_value
def in32(hash_value):
if hash_value > 2147483647:
return to_signed_32bit(hash_value)*-1
else:
return hash_value
def extendbytes(keystr, key):
return len(str(keystr)).to_bytes(1, byteorder='little') + keystr.encode() + len(key).to_bytes(2,byteorder='little') + key
def encode_dict_to_byte(dictcontent):
bts = bytearray()
d = dictcontent['d'].to_bytes(8, byteorder='little')
a = dictcontent['a'].encode()
b = dictcontent['b'].to_bytes(4, byteorder='little')
c = dictcontent['c'].encode()
e = dictcontent['e'].encode()
p = dictcontent['p'].to_bytes(8, byteorder='little')
g = dictcontent['g'].encode()
h = dictcontent['h'].encode()
n = dictcontent['n'].encode()
o = dictcontent['o'].encode()
q = dictcontent['q'].to_bytes(4, byteorder='little')
r = dictcontent['r'].to_bytes(4, byteorder='little')
s = dictcontent['s'].encode()
t = dictcontent['t'].encode()
u = dictcontent['u'].encode()
v = dictcontent['v'].to_bytes(1, byteorder='little')
x = dictcontent['x'].encode()
z = dictcontent['z'].encode()
f_i_c = dictcontent['f']['i_c'].to_bytes(1, byteorder='little')
f_i_e = dictcontent['f']['i_e'].to_bytes(8, byteorder='little')
f_i_n = dictcontent['f']['i_n'].to_bytes(1, byteorder='little')
f_i_u = dictcontent['f']['i_u'].to_bytes(4, byteorder='little')
f_i_r = dictcontent['f']['i_r'].to_bytes(1, byteorder='little')
f_i_cl = dictcontent['f']['i_cl'].to_bytes(1, byteorder='little')
f_i_ev = dictcontent['f']['i_ev'].to_bytes(8, byteorder='little')
f_i_ps = dictcontent['f']['i_ps'].to_bytes(4, byteorder='little')
f_uqid = dictcontent['f']['uqid'].encode() # \x00
f_fw = dictcontent['f']['fw'].encode()
f_ua = dictcontent['f']['ua'].encode()
f_rf = dictcontent['f']['rf'].encode()
f_i_l = dictcontent['f']['i_l'].to_bytes(1, byteorder='little')
f_z = dictcontent['f']['z'].encode()
f_x = dictcontent['f']['x'].encode()
f_i_z = dictcontent['f']['i_z'].to_bytes(1, byteorder='little')
f_pi = dictcontent['f']['pi'].encode()
f = bytearray()
f.extend(extendbytes("i_c", f_i_c))
f.extend(extendbytes("i_e", f_i_e))
f.extend(extendbytes("i_n", f_i_n))
f.extend(extendbytes("i_u", f_i_u))
f.extend(extendbytes("i_r", f_i_r))
f.extend(extendbytes("i_cl", f_i_cl))
f.extend(extendbytes("i_ev", f_i_ev))
f.extend(extendbytes("i_ps", f_i_ps))
f.extend(extendbytes("uqid", f_uqid))
f.extend(extendbytes("fw", f_fw))
f.extend(extendbytes("ua", f_ua))
f.extend(extendbytes("rf", f_rf))
f.extend(extendbytes("i_l", f_i_l))
f.extend(extendbytes("z", f_z))
f.extend(extendbytes("x", f_x))
f.extend(extendbytes("i_z", f_i_z))
f.extend(extendbytes("pi", f_pi))
bts.extend(extendbytes("d", d))
bts.extend(extendbytes("a", a))
bts.extend(extendbytes("b", b))
bts.extend(extendbytes("c", c))
bts.extend(extendbytes("e", e))
bts.extend(extendbytes("p", p))
bts.extend(extendbytes("g", g))
bts.extend(extendbytes("h", h))
bts.extend(extendbytes("n", n))
bts.extend(extendbytes("o", o))
bts.extend(extendbytes("q", q))
bts.extend(extendbytes("r", r))
bts.extend(extendbytes("s", s))
bts.extend(extendbytes("t", t))
bts.extend(extendbytes("u", u))
bts.extend(extendbytes("v", v))
bts.extend(extendbytes("x", x))
bts.extend(extendbytes("z", z))
bts.extend(len(str('f')).to_bytes(1, byteorder='little') + "f".encode() + len(f).to_bytes(2,byteorder='little') + f)
return bts
def encode_dict_byte(dictcontent: Dict[str, Any]) -> bytes:
"""
将字典数据编码为字节序列
"""
bts = bytearray()
# 编码各个字段
if 'd' in dictcontent:
d = dictcontent['d'].to_bytes(8, byteorder='little')
bts.extend(extend_bytes("d", d))
if 'a' in dictcontent:
a = dictcontent['a'].encode('utf-8')
bts.extend(extend_bytes("a", a))
if 'b' in dictcontent:
b = dictcontent['b'].to_bytes(4, byteorder='little')
bts.extend(extend_bytes("b", b))
if 'c' in dictcontent:
c = dictcontent['c'].encode('utf-8')
bts.extend(extend_bytes("c", c))
if 'e' in dictcontent:
e = dictcontent['e'].encode('utf-8')
bts.extend(extend_bytes("e", e))
if 'p' in dictcontent:
p = dictcontent['p'].to_bytes(4, byteorder='little') # 注意：示例中是4字节，不是8字节
bts.extend(extend_bytes("p", p))
if 'g' in dictcontent:
g = dictcontent['g'].encode('utf-8')
bts.extend(extend_bytes("g", g))
if 'h' in dictcontent:
h = dictcontent['h'].encode('utf-8')
bts.extend(extend_bytes("h", h))
if 'k' in dictcontent:
k = dictcontent['k'].encode('utf-8')
bts.extend(extend_bytes("k", k))
if 'n' in dictcontent:
n = dictcontent['n'].encode('utf-8')
bts.extend(extend_bytes("n", n))
if 'o' in dictcontent:
o = dictcontent['o'].encode('utf-8')
bts.extend(extend_bytes("o", o))
if 'j' in dictcontent:
j = dictcontent['j'].encode('utf-8')
bts.extend(extend_bytes("j", j))
if 'q' in dictcontent:
q = dictcontent['q'].to_bytes(4, byteorder='little')
bts.extend(extend_bytes("q", q))
if 'r' in dictcontent:
r = dictcontent['r'].to_bytes(4, byteorder='little')
bts.extend(extend_bytes("r", r))
if 's' in dictcontent:
s = dictcontent['s'].encode('utf-8')
bts.extend(extend_bytes("s", s))
if 't' in dictcontent:
t = dictcontent['t'].encode('utf-8')
bts.extend(extend_bytes("t", t))
if 'u' in dictcontent:
u = dictcontent['u'].encode('utf-8')
bts.extend(extend_bytes("u", u))
if 'v' in dictcontent:
v = dictcontent['v'].to_bytes(1, byteorder='little')
bts.extend(extend_bytes("v", v))
if 'w' in dictcontent:
w = dictcontent['w'].to_bytes(8, byteorder='little')
bts.extend(extend_bytes("w", w))
if 'x' in dictcontent:
x = dictcontent['x'].encode('utf-8')
bts.extend(extend_bytes("x", x))
if 'z' in dictcontent:
z = dictcontent['z'].encode('utf-8')
bts.extend(extend_bytes("z", z))
# 编码嵌套的f字段
if 'f' in dictcontent:
f_data = encode_f_value(dictcontent['f'])
bts.extend(len("f").to_bytes(1, byteorder='little'))
bts.extend("f".encode('latin1'))
bts.extend(len(f_data).to_bytes(2, byteorder='little'))
bts.extend(f_data)
return bytes(bts).decode('latin1')
def encode_f_value(f_content: Dict[str, Any]) -> bytes:
"""
编码嵌套的f字段
"""
f_bts = bytearray()
# 编码f字段中的各个子字段
if 'i_c' in f_content:
i_c = f_content['i_c'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_c", i_c))
if 'i_e' in f_content:
i_e = f_content['i_e'].to_bytes(8, byteorder='little')
f_bts.extend(extend_bytes("i_e", i_e))
if 'i_n' in f_content:
i_n = f_content['i_n'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_n", i_n))
if 'i_u' in f_content:
i_u = f_content['i_u'].to_bytes(2, byteorder='little')
f_bts.extend(extend_bytes("i_u", i_u))
if 'i_r' in f_content:
i_r = f_content['i_r'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_r", i_r))
if 'i_tf' in f_content:
i_tf = f_content['i_tf'].to_bytes(8, byteorder='little')
f_bts.extend(extend_bytes("i_tf", i_tf))
if 'i_bt' in f_content:
i_bt = f_content['i_bt'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_bt", i_bt))
if 'i_cl' in f_content:
i_cl = f_content['i_cl'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_cl", i_cl))
if 'i_ev' in f_content:
i_ev = f_content['i_ev'].to_bytes(8, byteorder='little')
f_bts.extend(extend_bytes("i_ev", i_ev))
if 'i_ps' in f_content:
i_ps = f_content['i_ps'].to_bytes(4, byteorder='little')
f_bts.extend(extend_bytes("i_ps", i_ps))
if 'uqid' in f_content:
uqid = f_content['uqid'].encode('utf-8')
f_bts.extend(extend_bytes("uqid", uqid))
if 'fw' in f_content:
fw = f_content['fw'].encode('utf-8')
f_bts.extend(extend_bytes("fw", fw))
if 'uv' in f_content:
uv = f_content['uv'].encode('utf-8')
f_bts.extend(extend_bytes("uv", uv))
if 'ur' in f_content:
ur = f_content['ur'].encode('utf-8')
f_bts.extend(extend_bytes("ur", ur))
if 'pr' in f_content:
pr = f_content['pr'].encode('utf-8')
f_bts.extend(extend_bytes("pr", pr))
if 'ua' in f_content:
ua = f_content['ua'].encode('utf-8')
f_bts.extend(extend_bytes("ua", ua))
if 'rf' in f_content:
rf = f_content['rf'].encode('utf-8')
f_bts.extend(extend_bytes("rf", rf))
if 'db' in f_content:
db = f_content['db'].encode('utf-8')
f_bts.extend(extend_bytes("db", db))
if 'db2' in f_content:
db2 = f_content['db2'].encode('utf-8')
f_bts.extend(extend_bytes("db2", db2))
if 'i_l' in f_content:
i_l = f_content['i_l'].to_bytes(4, byteorder='little')
f_bts.extend(extend_bytes("i_l", i_l))
if 'i_z' in f_content:
i_z = f_content['i_z'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_z", i_z))
if 'i_fd' in f_content:
i_fd = f_content['i_fd'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_fd", i_fd))
if 'i_kp' in f_content:
i_kp = f_content['i_kp'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_kp", i_kp))
if 'i_sc' in f_content:
i_sc = f_content['i_sc'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_sc", i_sc))
if 'i_sd' in f_content:
i_sd = f_content['i_sd'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_sd", i_sd))
if 'i_sf' in f_content:
i_sf = f_content['i_sf'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_sf", i_sf))
if 'i_ks' in f_content:
i_ks = f_content['i_ks'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_ks", i_ks))
if 'i_ams' in f_content:
i_ams = f_content['i_ams'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_ams", i_ams))
if 'i_tm' in f_content:
i_tm = f_content['i_tm'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_tm", i_tm))
if 'i_smm' in f_content:
i_smm = f_content['i_smm'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_smm", i_smm))
if 'i_abta' in f_content:
i_abta = f_content['i_abta'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_abta", i_abta))
if 'i_bsc' in f_content:
i_bsc = f_content['i_bsc'].to_bytes(8, byteorder='little')
f_bts.extend(extend_bytes("i_bsc", i_bsc))
if 'i_ct' in f_content:
i_ct = f_content['i_ct'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_ct", i_ct))
if 'i_rc' in f_content:
i_rc = f_content['i_rc'].to_bytes(1, byteorder='little')
f_bts.extend(extend_bytes("i_rc", i_rc))
if 'pi' in f_content:
pi = f_content['pi'].encode('utf-8')
f_bts.extend(extend_bytes("pi", pi))
if 'dm' in f_content:
dm = f_content['dm'].encode('utf-8')
f_bts.extend(extend_bytes("dm", dm))
return bytes(f_bts)
def extend_bytes(field_name: str, value: Union[bytes, bytearray]) -> bytes:
"""
将字段名和值编码为字节序列
格式：字段名长度(1字节) + 字段名 + 值长度(2字节小端序) + 值
"""
bts = bytearray()
# 字段名长度和字段名
bts.append(len(field_name))
bts.extend(field_name.encode('latin1'))
# 值长度和值
value_length = len(value)
bts.extend(value_length.to_bytes(2, byteorder='little'))
bts.extend(value)
return bytes(bts)
def extendbytes_to_bytes(keystr: str, value: bytes) -> bytes:
"""
编码单个字段: 1字节key长度 + key + 2字节value长度 + value
"""
return (
len(keystr).to_bytes(1, byteorder='little')
+ keystr.encode()
+ len(value).to_bytes(2, byteorder='little')
+ value
)
def encode_value(val):
"""
自动将 int/str 转换为 bytes
"""
if isinstance(val, int):
if val > 0xFFFFFFFF:
return val.to_bytes(8, 'little')
elif val > 0xFF:
return val.to_bytes(4, 'little')
else:
return val.to_bytes(1, 'little')
elif isinstance(val, str):
return val.encode()
elif isinstance(val, bytes):
return val
else:
return str(val).encode()
def encode_dict_to_bytearray(dictcontent: dict) -> bytearray:
bts = bytearray()
for k, v in dictcontent.items():
if isinstance(v, dict): # 子字典
sub = bytearray()
for kk, vv in v.items():
sub.extend(extendbytes_to_bytes(kk, encode_value(vv)))
bts.extend(extendbytes_to_bytes(k, sub))
else:
bts.extend(extendbytes_to_bytes(k, encode_value(v)))
return bts
代码文件
下载 · 28.3 KBglobal_get_sync_term
global_get_sync_term.py
import json
import re
from curl_cffi import requests
import time
import random
import secrets
import base64
import global_encryption_generate
def generate_p(text):# p=b+r+q+z+b+d+v
"""
生成函数，用于处理输入文本并返回处理后的结果
参数:
text (str): 需要处理的文本字符串
返回:
处理后的结果，通过unit_generate.right3方法对x1和0进行处理得到
"""
x1 = 0 # 初始化变量x1为0，用于在循环过程中累积处理结果
for i in text: # 遍历输入文本的每个字符
x1 = global_encryption_generate.encodechat(i, x1) # 对每个字符进行处理，并将结果累积到x1中
return global_encryption_generate.right3(x1, 0) # 使用unit_generate.right3方法处理最终的x1值，传入0作为第二个参数
def generate_nonce(d,b): # d,b
res=[]
for i in range(4):
str=f"0{d}{i}{b}"
strbytes=str.encode('utf-8')
res_num = global_encryption_generate.fnv1a_32(strbytes)
res=res+global_encryption_generate.hex_to_byte_list(global_encryption_generate.in32(res_num))
return res
def generate_X_net_sync_term(dicts,SST,header=[2, 188, 0, 4]):
# header=[3, 12, 0, 4]
x_header = header + list(base64.b64decode(SST))
nonce = generate_nonce(dicts['d'], dicts['b'])
# nonce =[11, 238, 53, 123, 42, 202, 7, 80, 124, 206, 214, 77, 64, 223, 81, 190]
ciphertext = list(global_encryption_generate.aes_ctr_encrypt(global_encryption_generate.encode_dict_to_bytearray(dicts),nonce))
return base64.b64encode(bytes(x_header+nonce+ciphertext)).decode()
class GetSyncTerm:
def __init__(self, session=None, home_url="https://www.bet365.com", useragent=''):
self.session = session or requests.Session()
self.useragent = useragent
self.headers = {'User-Agent': self.useragent}
self.home_url = home_url
self.configuration_path = {}
self.res_configs = {}
def generate_similar_ipv6(self, base_address):
"""
生成与给定 IPv6 地址格式相似的随机地址
参数:
base_address: 基础 IPv6 地址，用于提取模式
返回:
格式相似但值不同的 IPv6 地址
"""
parts = base_address.split(":")
new_parts = []
for part in parts:
if random.random() < 0.7: # 70% 的概率改变这部分
new_part = secrets.token_hex(2)
# 保持第一个字符相同（增加相似度）
if random.random() < 0.5:
new_part = part[0] + new_part[1:]
new_parts.append(new_part)
else:
new_parts.append(part) # 保留原部分
return ":".join(new_parts)
# 示例使用
# base_ip = "f054:747d:68cb:a835:f30e:747f:cba0:4861"
# similar_ips = [generate_similar_ipv6(base_ip) for _ in range(5)]
#
# print("原始地址:", base_ip)
# print("相似地址:")
# for ip in similar_ips:
# print(ip)
def advanced_graphics_info(self, ua_string):
"""
更高级的图形信息生成函数，包含更多硬件检测逻辑
"""
# 硬件数据库映射
gpu_mappings = {'AMD': {'patterns': [r'Renoir', r'Vega', r'Navi', r'Radeon.*Graphics'], 'models': ['renoir', 'vega', 'navi', 'radeon']}, 'Intel': {'patterns': [r'Iris.*Xe', r'UHD', r'HD Graphics'], 'models': ['xe', 'uhd', 'hd']}, 'NVIDIA': {'patterns': [r'RTX', r'GTX', r'GeForce'], 'models': ['rtx', 'gtx', 'geforce']}}
# 检测硬件
detected_vendor = "unknown"
detected_model = "unknown"
for vendor, data in gpu_mappings.items():
for pattern, model in zip(data['patterns'], data['models']):
if re.search(pattern, ua_string, re.IGNORECASE):
detected_vendor = vendor.lower()
detected_model = model
break
if detected_vendor != "unknown":
break
# 确定渲染器
if re.search(r'Windows', ua_string, re.IGNORECASE):
renderer = "angle"
elif re.search(r'Mac|iOS', ua_string, re.IGNORECASE):
renderer = "metal"
else:
renderer = "opengl"
# 版本信息（这里使用默认值，实际中可能需要更复杂的检测）
opengl_version = "4.6"
llvm_version = "15.0.7"
# 构建结果
if detected_model != "unknown":
return f"{renderer} ({detected_vendor}, {detected_model} ({detected_model} llvm {llvm_version}), opengl {opengl_version})"
else:
return f"{renderer} ({detected_vendor}, opengl {opengl_version})"
import random
def generate_graphics_parameter(self, ua_string=None):
"""
综合方法生成图形供应商参数
参数:
ua_string (str): 可选的User-Agent字符串
返回:
str: 图形供应商信息
"""
# 常见组合
common_combinations = ["Google Inc. (Intel)", "Google Inc. (AMD)", "Google Inc. (NVIDIA Corporation)", "Mozilla Foundation (Intel)", "Mozilla Foundation (AMD)", "Apple Inc. (Apple)", "Microsoft (Intel)"]
# 如果有UA字符串，尝试基于它生成
if ua_string:
if "Chrome" in ua_string and "Windows" in ua_string:
if "Intel" in ua_string:
return "Google Inc. (Intel)"
elif "AMD" in ua_string:
return "Google Inc. (AMD)"
else:
return random.choice(["Google Inc. (Intel)", "Google Inc. (AMD)", "Google Inc. (NVIDIA Corporation)"])
elif "Firefox" in ua_string:
return "Mozilla Foundation (Intel)"
elif "Safari" in ua_string and "Mac" in ua_string:
return "Apple Inc. (Apple)"
elif "Edge" in ua_string:
return "Microsoft (Intel)"
# 如果没有UA或无法确定，返回随机常见组合
return random.choice(common_combinations)
def generate_guid(self, ):
guid = []
for i in range(32):
# 生成0-15的随机数并转换为十六进制字符串
guid.append(random.randint(0, 15).__format__('x'))
# 在特定位置添加连字符
if i in (7, 11, 15, 19):
guid.append('-')
return ''.join(guid)
# 生成并打印GUID
# guid = generate_guid()
def generate_dict(self, dict, ramdom_float_num, ua,frist_time,sessionID,apipath,webpath ):
nowtime = int(time.time() * 1000)
dict['d'] = nowtime
dict['e'] = ua
dict['s'] = sessionID
dict['u'] = apipath
dict['c'] = webpath
dict['f']['i_e'] = frist_time
dict['f']['ua'] = ua
# dict['b'] = int(ramdom_float_num*2147483647)
# p=b+r+q+z+b+d+v
textp=f"{dict['b']}{dict['r']}{dict['q']}{dict['z']}{dict['b']}{dict['d']}{dict['v']}"
dict['p'] = self.generate_p(textp)
return dict
def generate_p(self, text):# p=b+r+q+z+b+d+v
x1 = 0
for i in text:
x1 = global_encryption_generate.encodechat(i, x1)
return global_encryption_generate.right3(x1,0)
def generate_nonce(self,d,b): # d,b
res=[]
for i in range(4):
str=f"0{d}{i}{b}"
strbytes=str.encode('utf-8')
res_num = global_encryption_generate.fnv1a_32(strbytes)
res=res+global_encryption_generate.hex_to_byte_list(global_encryption_generate.in32(res_num))
return res
def generate_X_net_sync_term(self,dicts,SST):
x_header = [2, 188, 0, 4] + list(base64.b64decode(SST))
nonce = self.generate_nonce(dicts['d'], dicts['b'])
ciphertext = list(global_encryption_generate.aes_ctr_encrypt(global_encryption_generate.encode_dict_to_byte(dicts),nonce))
return base64.b64encode(bytes(x_header+nonce+ciphertext)).decode()
def extract_ns_weblib_util(self, html_content):
# 正则表达式模式：匹配var ns_weblib_util = 后面的字典数据
# 考虑可能的空格和不同的闭合方式
pattern = r'var ns_weblib_util\s*=\s*({.*?});'
# 使用非贪婪匹配来捕获字典内容
match = re.search(pattern, html_content, re.DOTALL)
if match:
# 提取并返回匹配到的字典字符串
return match.group(1)
else:
return None
def return_configuration_path(self, response_text):
result = self.extract_ns_weblib_util(response_text)
configuration_path = json.loads(result.replace("'", "\""))
return configuration_path
def request_main_txt(self, session=None, origin_name='https://www.bet365.com', ua = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36'):
session = session or requests.Session(headers=self.headers, impersonate='chrome133a')
headerss = {'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
'accept-language': 'zh-CN,zh;q=0.9,ja;q=0.8', 'cache-control': 'no-cache',
'pragma': 'no-cache', 'priority': 'u=0, i', 'sec-ch-ua': '"Google Chrome";v="125", "Chromium";v="125", "Not.A/Brand";v="24"',
'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Linux"', 'sec-fetch-dest': 'document', 'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'none', 'sec-fetch-user': '?1', 'upgrade-insecure-requests': '1', 'user-agent':ua , }
response = session.get(origin_name, headers=headerss)
response_text = response.text
self.configuration_path = self.return_configuration_path(response_text)
return self.configuration_path
def get_WebsiteConfig(self, session):
configuration_path = self.request_main_txt(session)
res_configs = session.get(self.home_url + configuration_path["WebsiteConfig"]["SITE_CONFIG_LOCATION"]).json()
return res_configs
def return_res_configs(self, res_configs):
SST = res_configs['ns_weblib_util']['WebsiteConfig']['SST']
SESSION_ID = res_configs['flashvars']['SESSION_ID']
randomnum = random.random()
fisttime = int(time.time() * 1000)
return SESSION_ID, SST, randomnum, fisttime
def inits(self, configuration_path, res_configs, session=None):
SST = res_configs['ns_weblib_util']['WebsiteConfig']['SST']
SESSION_ID = res_configs['flashvars']['SESSION_ID']
randomnum = random.random()
fisttime = int(time.time() * 1000)
return SESSION_ID, SST, randomnum, fisttime
def leagua_data_dict(self,mian_url,u,type_, ctid='88', g='',h="I", origin='www.z1.bet365.com', api_str="", request_id='f047:7468:68c6:8134:f301:746a:cba0:5044'):
leagua_data = {}
if type_=="web":
leagua_data = {'d': 171231231,
'a': '1',
'b': 171231231,
'c': mian_url,
'e': u,
'p': 171231231,
'g': g,
'h': h,
'n': 'null',
'o': '88',
'q': 0,
'r': 0,
's': '59B4061EF0AC8D8BB18C821B14E167E8000003',
't': '-480',
'u': '/splashcontentapi/soccertab?lid=10&zid=0&pd=%23AS%23B1%23K%5E5%23&cid=88&cgid=1&ctid=88',
'v': 2,
'x': '',
'z': 'Asia/Shanghai',
'f': {
'i_c': 1,
'i_e': 171231231,
'i_n': 0,
'i_u': 15,
'i_r': 1,
'i_cl': 0,
'i_ev': 0,
'i_ps': 764416712,
'uqid': '',
'fw': '110',
'ua': u,
'rf': 'Direct',
# 'd': origin,
'i_l': 0,
'z': 't2',
'x': 's',
'i_z': 3,
'pi': '2'
}}
elif type_ == "api":
# request_id = self.generate_similar_ipv6('f09d:74c6:68cc:c881:f357:74c8:cba0:48aa')
# 'f09d:74c6:68cc:f836:f357:74c8:cba0:48aa'
# request_id = 'f09d:74c6:68cc:f836:f357:74c8:cba0:48aa'
# request_id = 'f09d:74c6:68cd:2445:f357:74c8:cba0:48aa'
# leagua_data = {'d': 171231231, 'a': '1', 'b': 171231231, 'c': mian_url, 'e': u, 'p': 171231231, 'g': g, "i": request_id, 'h': h, 'k': origin, 'n': 'null', 'o': ctid,'j': self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'), 'q': 0, 'r': 0, 's': '59B4061EF0AC8D8BB18C821B14E167E8000003', 't': '-480', 'u': '/splashcontentapi/soccertab?lid=10&zid=0&pd=%23AS%23B1%23K%5E5%23&cid=88&cgid=1&ctid=88', 'v': 2, 'x': '', 'z': 'Asia/Shanghai', 'f': {'i_c': 1, 'i_e': 171231231, 'i_n': 0, 'i_u': 15, 'i_r': 1, 'i_cl': 0, 'i_ev': 0, 'i_ps': 764416712, 'uqid': '', 'fw': '170', 'ua': u, 'rf': 'Direct', 'd': origin,
# 'i_l': 0, 'z': 't2', 'x': 's', 'i_z': 3, 'pi': '2'}}
leagua_data = { 'a': '1',
'b': 743047743,
'c': mian_url,
'd': 1758105079538,
'e': u,
'g': g,
'h': h,
'i': request_id,
'j': self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'),
'k': origin,
'n': 'null',
'o': ctid,
'p': 305921974,
'q': 0,
'r': 0,
's': 'AED2D8ACA55CCC5590ED5237BA651134000003',
't': '-480',
'u': api_str,
'v': 2,
'x': '',
'z': 'Asia/Shanghai',
'f': {'d': origin,
'db': '1',
'db2': '2',
'dm': '8',
'fw': '171',
'h': 'I',
'i_abta': 0,
'i_ams': 0,
'i_bsc': 0,
'i_bt': 0,
'i_c': 1,
'i_cl': 0,
'i_ct': 0,
'i_e': 1758105077652,
'i_ev': 0,
'i_fd': 0,
'i_kp': 0,
'i_ks': 0,
'i_l': 0,
'i_n': 0,
'i_ps': 907367449,
'i_r': 4,
'i_rc': 0,
'i_sc': 0,
'i_sd': 0,
'i_sf': 0,
'i_smm': 0,
'i_tf': 0,
'i_tm': 0,
'i_u': 185,
'i_z': 2,
'il': 'false',
'pi': '1',
'pr': '',
'rf': 'Direct',
'ua': u,
'ul': api_str,
'uqid': '',
'ur': self.advanced_graphics_info(u),
'uv': self.generate_graphics_parameter(u),
'x': 's',
'z': 'm2'},
}
# leagua_data = {'d': 1757668996686, 'a': '1', 'b': 1344743547, 'c': mian_url, 'e': u, 'p': 1344808334, 'g': '', "i": request_id,'h': 'I', 'k': origin, 'n': 'null', 'o': ctid, 'j': self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'), 'q': 272629826, 'r': 16778240, 's': '22D991EFACB84D449DF91AAEBCD8C284000003', 't': '-480', 'u': '/gamingservices/gam-client-gos-svcv1/gamingconfiguration?countryid=212&stateid=0&loggedin=0&platformtypeid=1&cid=212&languageid=2&statuscheck=1&cgid=2&ctid=212', 'v': 2, 'x': '', 'z': 'Asia/Shanghai',
# 'f': {'i_c': 2, 'i_e': 1757669000687, 'i_n': 0, 'i_u': 516, 'i_r': 2, 'i_tf': 0, 'i_bt': 0, 'i_cl': 0, 'i_ev': 0, 'i_ps': 1262838774, 'uqid': '', 'fw': '171', 'uv': 'Google Inc. (Intel)', 'ur': 'angle (intel, intel(r) iris(r) xe graphics (0x00009a49) direct3d11 vs_5_0 ps_5_0, d3d11)', 'pr': '', 'ua':u, 'rf': 'Direct', 'd': origin, 'db': '1', 'db2': '1', 'i_l': 272629826, 'h': 'K', 'z': 'm2', 'x': 's', 'i_z': 2, 'i_fd': 0, 'i_kp': 0, 'i_sc': 0, 'i_sd': 0, 'i_sf': 0, 'i_ks': 0, 'i_ams': 0, 'i_tm': 0, 'i_smm': 0, 'i_abta': 0, 'i_bsc': 0, 'i_ct': 0, 'i_rc': 0, 'pi': '1', 'dm': '8'}}
# leagua_data = {'d': 1757839669191, 'a': '1', 'b': 1708426658, 'c': mian_url, 'e': u, 'p': 171203974, 'g': '', 'h': h,
# leagua_data = {'d': 171231231, 'a': '1', 'b': 171231231, 'c': mian_url, 'e': u, 'p': 171231231, 'g': '', 'h': h,
# 'i': request_id,
# 'k': origin, 'n': 'null', 'o': ctid, 'j': self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'), 'q': 0, 'r': 0, 's': 'B1AE76BFE6B58A40B77EC669AEF27C3C000003', 't': '-480', 'u': '/splashcontentapi/soccertab?lid=2&zid=0&pd=%23AS%23B1%23K%5E5%23&cid=88&cgid=1&ctid=88', 'v': 2, 'x': '', 'z': 'Asia/Shanghai',
# 'f': {
# "q":f' ~ https://{origin}/sports-assets/sportsoc/FirebaseWebSDKModule/assets/firebase_iframe.html',
# 'i_c': 1, 'i_e': 1757839666948, 'i_n': 0, 'i_u': 87, 'i_r': 4, 'i_tf': 0, 'i_bt': 0, 'i_cl': 0, 'i_ev': 0, 'i_ps': 907367449, 'uqid': '', 'fw': '171', 'uv': 'Google Inc. (Intel)', 'ur': 'angle (intel, intel(r) iris(r) xe graphics (0x00009a49) direct3d11 vs_5_0 ps_5_0, d3d11)', 'pr': '', 'ua': u, 'rf': 'Direct', 'd': origin, 'db': '1', 'db2': '2', 'i_l': 0, 'h': h, 'z': 'm2', 'x': 's', 'i_z': 2, 'i_fd': 0, 'i_kp': 0, 'i_sc': 0, 'i_sd': 0, 'i_sf': 0, 'i_ks': 0, 'i_ams': 0, 'i_tm': 0, 'i_smm': 0, 'i_abta': 0, 'i_bsc': 0, 'i_ct': 0, 'i_rc': 0, 'pi': '1', 'dm': '8'}}
# leagua_data = {'d': 1758031985075, 'a': '1', 'b': 255293688, 'c': '#/AS/B1/K^5/', 'e': u, 'p': 505589040, 'g': '', 'h': h, 'i': 'f03e:7467:68c9:7072:f2f8:7469:cba0:484b', 'k': origin, 'n': 'null', 'o': ctid, 'j':self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'), 'q': 0, 'r': 0, 's': '2A59F77145006C3F9D3F6AAE5E75873E000003', 't': '-480', 'u': '/splashcontentapi/soccertab?lid=2&zid=0&pd=%23AS%23B1%23K%5E5%23&cid=88&cgid=1&ctid=88', 'v': 2, 'x': '', 'z': 'Asia/Shanghai',
# 'f': {'i_c': 2, 'i_e': 1758031986314, 'i_n': 0, 'i_u': 148, 'i_r': 3, 'i_tf': 0, 'i_bt': 0, 'i_cl': 0, 'i_ev': 0, 'i_ps': 907367449, 'uqid': '', 'fw': '171', 'uv': 'Google Inc. (Intel)', 'ur': 'angle (intel, intel(r) iris(r) xe graphics (0x00009a49) direct3d11 vs_5_0 ps_5_0, d3d11)', 'pr': '', 'ua': u, 'rf': 'Direct', 'd': origin, 'db': '1', 'db2': '2', 'i_l': 0, 'h': 'I', 'z': 'm2', 'x': 's', 'i_z': 2, 'i_fd': 0, 'i_kp': 0, 'i_sc': 0, 'i_sd': 0, 'i_sf': 0,
# 'i_ks': 0, 'i_ams': 0, 'i_tm': 0, 'i_smm': 0, 'i_abta': 0, 'i_bsc': 0, 'i_ct': 0, 'i_rc': 0, 'pi': '1', 'dm': '8'}}
elif type_ == "app":
leagua_data = {'a': '1', 'b': 1656656128,
'c': mian_url, 'd': 1760775263554, 'e': u,
'p': 133468807, 'g': g, 'h':h,
'i': request_id, 'j': self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'),
'k': origin, 'n': 'null', 'o': ctid, 'q': 0, 'r': 0, 's': '4037F87FA5604F8496B50F4AA92C102C000003', 't': '-480', 'u': api_str, 'v': 2, 'x': '', 'z': 'Asia/Shanghai',
'f': {'i_u': 162, 'q': f' ~ https://{origin}/sports-assets/sportsoc/FirebaseWebSDKModule/assets/firebase_iframe.html',
'i_r': 14, 'i_cl': 1, 'i_ps': 50339904, 'uqid': '', 'uv': self.generate_graphics_parameter(u), 'ur': self.advanced_graphics_info(u), 'ua': u,
'il': 'false', 'ul': api_str, 'd': 'www.z1.bet365.com', 'x': 's', 'i_z': 2, 'pi': '1'}}
return leagua_data
def get_SyncTerm(self, webpath, res_configs, ctid="88", api_str='', connectionID='', h="I", origin='www.z1.bet365.com', type_="web", header=None, request_id=""):
header = header or [2, 188, 0, 4]
self.res_configs = res_configs
# data = {'d': 1757841227203, 'a': '1', 'b': 1479666406, 'c': '#/AS/B1/K^5/', 'e': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36 Edg/140.0.0.0', 'p': 778743229, 'g': 'I3MRW36-AzWIbxNeAAMo', 'h': 'K', 'i': 'f047:7468:68c6:874c:f301:746a:cba0:5044', 'k': 'www.z1.bet365.com', 'n': 'null', 'o': '99', 'j': '199.5756', 'q': 0, 'r': 0, 's': '25D412EA6E443C9080A71824BA63A630000003', 't': '-480', 'u': '/zap/?uid=9162611288618789', 'v': 2, 'x': '', 'z': 'Asia/Shanghai', 'f': {'i_c': 1, 'i_e': 1757841226911, 'i_n': 0, 'i_u': 89, 'q': ' ~ https://www.z1.bet365.com/sports-assets/sportsoc/FirebaseWebSDKModule/assets/firebase_iframe.html', 'i_r': 2, 'i_tf': 0, 'i_bt': 0, 'i_cl': 0, 'i_ev': 0, 'i_ps': 954625300, 'uqid': '', 'fw': '171', 'uv': 'Google Inc. (Intel)', 'ur': 'angle (intel, intel(r) iris(r) xe graphics (0x00009a49) direct3d11 vs_5_0 ps_5_0, d3d11)', 'pr': '', 'ua': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36 Edg/140.0.0.0', 'rf': 'Direct', 'd': 'www.z1.bet365.com', 'db': '1', 'db2': '2', 'i_l': 0, 'h': 'K', 'z': 'm2', 'x': 's', 'i_z': 2, 'i_fd': 0, 'i_kp': 0, 'i_sc': 0, 'i_sd': 0, 'i_sf': 0, 'i_ks': 0, 'i_ams': 0, 'i_tm': 0, 'i_smm': 0, 'i_abta': 0, 'i_bsc': 0, 'i_ct': 0, 'i_rc': 0, 'pi': '1', 'dm': '8'}}
leagua_data = self.leagua_data_dict(webpath, self.useragent, type_, ctid=ctid, g=connectionID,h=h, origin=origin, api_str=api_str, request_id=request_id)
SESSION_ID, SST, randomnum, fisttime = self.inits(self.configuration_path, res_configs)
dicts = self.generate_dict(leagua_data, randomnum, self.useragent, fisttime, SESSION_ID, api_str,f'#{webpath.replace("#", "/")}')
sync_term = generate_X_net_sync_term(dicts, SST, header=header)
# print(dicts)
return sync_term
async def webrtc_via_socks_proxy(self, urls="s3.btsfapi.com"):
# 配置 ICE 服务器
ice_servers = [RTCIceServer(urls=f"stun:{urls}")]
configuration = RTCConfiguration(iceServers=ice_servers)
pc = RTCPeerConnection(configuration)
# 设置 ICE 候选收集回调
@pc.on("icecandidate")
def on_ice_candidate(candidate):
if candidate:
print(f"ICE 候选: {candidate.candidate}") # 解析候选信息获取 IP
# 创建数据通道
dc = pc.createDataChannel("proxy-channel")
@dc.on("open")
def on_open():
print("通过 SOCKS5 代理的数据通道已打开")
@dc.on("message")
def on_message(message):
print(f"通过代理收到消息: {message}")
# 创建 offer
offer = await pc.createOffer()
offer = await pc.createOffer()
await pc.setLocalDescription(offer)
sdp_text = pc.localDescription.sdp
parsed_sdp = self.parse_sdp(sdp_text)
# ip_ = self.extract_srflx_ipv6_from_sdp(sdp_text)[0]
port = parsed_sdp.get('media', {}).get('port')
# priority = [i.get('priority') for i in parsed_sdp.get('attributes', {}).get('candidates', []) if i.get('related_port') == port][0]
# numbers = [str(random.randint(1, 10)) for _ in range(9)]
# foundation = "".join(numbers)
# foundation = [i.get('foundation') for i in parsed_sdp.get('attributes', {}).get('candidates', []) if i.get('related_port') == port][0]
# 模拟添加 ICE 候选
# ice_candidate = RTCIceCandidate(component=1, foundation=foundation, ip=ip_, port=port, priority=priority, protocol="udp", type="host", sdpMid="1", relatedAddress=None, relatedPort=None, sdpMLineIndex=0)
# await pc.addIceCandidate(ice_candidate)
# print(f"Local SDP (通过 SOCKS5 代理):\n{pc.localDescription.sdp}")
# 模拟等待
# await asyncio.sleep(3)
sdp_text = pc.localDescription.sdp
await pc.close()
return sdp_text, self.extract_srflx_ipv6_from_sdp(pc.localDescription.sdp)
# def get_index_leaguas():
# webpath = "#AS#B1#K^5#"
# api = f"/splashcontentapi/soccertab?lid={langid}&zid=3&pd={urllib.parse.quote(webpath)}&cid=88&cgid=1&ctid=88"
# return req(webpath, api)
# def get_leaguas_by_country(self, PD_country):
# webpath = "#AS#B1#K^5#"
# api = f"/splashcontentapi/soccerpartial?lid={langid}&zid=3&pd=%23AS%23B1%23C1%23D1002%23{PD_country}%23I40%23O2%23&cid=88&cgid=1&ctid=88"
# return req(webpath, api)
#
# def get_matchs_by_leagua(PD_leaguas):
# webpath = f"#AC#B1#C1#D1002#{PD_leaguas}#G40#"
# api = f"/matchmarketscontentapi/markets?lid={langid}&zid=3&pd=%23AC%23B1%23C1%23D1002%23{PD_leaguas}%23G40%23&cid=88&cgid=1&ctid=88"
# return req(webpath, api)
#
# def req(self, webpath, api):
# dicts = generate_dict(leagua_data, randomnum, useragent, fisttime, SESSION_ID, api,
# f'#{webpath.replace("#", "/")}')
# X_net_sync_term = self.generate_X_net_sync_term(dicts, self.SST)
# session.headers.update({"X-Net-Sync-Term": X_net_sync_term})
# res = session.get(home_url + api)
# print(res.text)
# return res.text
# '#/AS/B1/K^5/'
# def get_index_leaguas():
# webpath = "#AS#B1#K^5#"
# api = f"/splashcontentapi/soccertab?lid={langid}&zid=3&pd={urllib.parse.quote(webpath)}&cid=88&cgid=1&ctid=88"
# return req(webpath, api)
#
# def get_leaguas_by_country(PD_country):
# webpath = "#AS#B1#K^5#"
# api = f"/splashcontentapi/soccerpartial?lid={langid}&zid=3&pd=%23AS%23B1%23C1%23D1002%23{PD_country}%23I40%23O2%23&cid=88&cgid=1&ctid=88"
# return req(webpath, api)
#
#
# def get_matchs_by_leagua(PD_leaguas):
# webpath = f"#AC#B1#C1#D1002#{PD_leaguas}#G40#"
# api = f"/matchmarketscontentapi/markets?lid={langid}&zid=3&pd=%23AC%23B1%23C1%23D1002%23{PD_leaguas}%23G40%23&cid=88&cgid=1&ctid=88"
# return req(webpath, api)
#
# if __name__ == '__main__':
# from parse_res import convert_to_dict
# langid = 10
# aaa=get_index_leaguas()
# print(aaa)
# print(convert_to_dict(aaa))
# aaa = get_leaguas_by_country("D1002")
# print(aaa)
# print(convert_to_dict(aaa))
# from hbmqtt.mqtt.packet import MQTTFixedHeader, PUBLISH, SUBSCRIBE, UNSUBSCRIBE, CONNECT
# from hbmqtt.mqtt import PublishPacket
# from hbmqtt.adapters import BufferReader
# async def parse_mqtt_packet(self, data):
# buffer = BufferReader(data)
# # 解析固定头部
# fixed_header = await MQTTFixedHeader.from_stream(buffer)
# # 根据包类型进行处理
# if fixed_header.packet_type == PUBLISH:
# packet = await PublishPacket.from_stream(buffer, fixed_header=fixed_header)
# return packet.payload.data
# elif fixed_header.packet_type == SUBSCRIBE:
# print("\033[1;34m This is a SUBSCRIBE packet. \033[0m")
# elif fixed_header.packet_type == UNSUBSCRIBE:
# print("\033[1;34m This is an UNSUBSCRIBE packet. \033[0m")
# elif fixed_header.packet_type == CONNECT:
# print("\033[1;34m This is a CONNECT packet. \033[0m")
# else:
# print("\033[1;34m This is an unknown packet type. \033[0m")
NEXT READ
评论交流
还没有公开评论。