引言
以前从来没去研究过安卓apk脱壳之类的，这次正好有机会和时间，所以也就上手研究了下，从入门到猝死，每天睡四五个小时，其他时间全部在研究，脑壳都给干冒烟了，主要是入手后脑子就停不下来，导致无法安心睡眠，不过期间学习到了挺多的知识的，只能说痛并快乐着。。。目的：
泡泡聊天APP（MosTalk）最近突然火起来了，同时也引来了很多炸群广告，用晚上开始全体群禁言的方式显得很呆，考虑写个脚本全自动检测所有群，发现炸群者直接踢出并撤回消息的辅助脚本，所以也就有了这次的研究。脱壳
查壳后发现用了最新的360加固，于是在脱壳上面废了几天时间，主要是不会IDA分析内存，所以只有各种查资料去找别人的工具了，最后通过 frida-unpack 成功脱壳，具体可以看上篇文章：初入APK逆向破解小记 这里就不过多阐述了。源码审计
脱壳后的代码是 Java 代码，拿着自己半吊子的功底在加上 open AI ，倒是也能做到审计了。思路1：通过python+frida hook 最新消息接受方法
泡泡聊天APP其实是存在WEB聊天页面的，只是它聊天是走的websocket，对websocket并不了解，所以不知道怎么监听到最新的群消息，只能想到这么个思路了。 尝试半天，总算写好了代码，测试了下发现存在个问题，会遗漏消息，并不能把所有的消息全部撤回; 在这里卡了一段时间，最开始以为是消息接受处理存在问题，所以又改成了接受消息存本地数据库，然后从数据库取出消息在处理，最后无奈发现因为python调用frida hook的原因，是必然会遗漏些频率高的消息的，不管用不用数据库。。。 虽然遗漏一些消息，但是只要能实现检测到频繁者，能撤回绝大部分消息并自动踢人，其实也算完工了，但是我这该死的完美主义，消息遗漏+配合模拟器ADB这种繁琐的方式，让我这个完美主义者不能忍，所以最终 PASS 掉了这个方法。 这里贴下当时的代码把，可能存在一些bug，毕竟是个半成品：import frida,json,re,time,datetime,requests,hashlib,threading,os,queue
import jaydebeapi
from collections import defaultdict
def md5_hash(text):
md5_hasher = hashlib.md5()
md5_hasher.update(text.encode('utf-8'))
hashed_text = md5_hasher.hexdigest()
return hashed_text
def md5Psw(account, pwd):
hashed_pwd = md5_hash( md5_hash(pwd) + account + 'MOSGRAM')
return hashed_pwd
def remove_special_characters(input_string):
try:
# 定义特殊字符的正则表达式模式
special_char_pattern = re.compile('[^\x00-\x7F]+')
# 使用正则表达式替换特殊字符为空字符串
cleaned_string = special_char_pattern.sub('', input_string)
return cleaned_string
except Exception as e:
print("[!] 异常发生:", e)
return input_string
def data_to_json(data):
try:
pattern = r'(\w+)\s*=\s*([^,]+)'
result = []
records = data.split('], [')
for record in records:
match = re.findall(pattern, record)
record_dict = {}
for key, value in match:
if value.startswith('{') and value.endswith('}'):
record_dict[key] = json.loads(value)
else:
record_dict[key] = value
result.append(record_dict)
parsed_data = json.dumps(result, indent=4)
return parsed_data
except json.JSONDecodeError as json_error:
print("[!] 数据转换失败: JSON 解析错误:", json_error)
except Exception as e:
print("[!] 数据转换失败:", e)
return None
def custom_request(method, url, json_data=None, headers=None):
try:
if method == 'POST':
response = requests.get(url, headers=headers)
elif method == 'POST':
response = requests.post(url, json=json_data, headers=headers)
elif method == 'DELETE':
response = requests.delete(url, json=json_data, headers=headers)
else:
raise ValueError("[!] 不支持的请求方法: " + method)
return response.json()
except requests.exceptions.RequestException as e:
print("[!] 网络请求异常:", e)
return None
def usb_device():
try:
# 连接安卓机上的frida-server
device = frida.get_usb_device(1000)
# 启动app
pid = device.spawn(["com.paopaotalk.im"])
device.resume(pid)
session = device.attach(pid)
return session
except:
exit("[!] 启动应用APP出错，重新执行脚本!")
# 消息回调函数，处理从脚本发送的消息
def on_message(message, payload):
try:
global call_back_message
if message['type'] == 'send':
call_back_message = message['payload']
return call_back_message
except:
exit("[!] 消息回调失败，重新执行脚本！")
def hook_function(session):
try:
global script
if script:
script.unload() # 释放之前的脚本
script_source = """
Java.perform(function()
{
var getVal = Java.use('v5.g');
var cons = getVal.l();
var send_it = cons.toString();
send(send_it);
});
"""
script = session.create_script(script_source)
script.on('message', on_message)
script.load()
return script
except:
exit("[!] 注入脚本失败，重新执行脚本！")
#登录
def login(username,pwd):
try:
print("[-] 正在登录账号...")
login_data_json = {"username": "aaa","password": "bbb","device_type":"WEB"}
login_data_json["username"] = username
login_data_json["password"] = md5Psw(username, pwd)
login_header_json = {"Content-Type":"application/json; charset=UTF-8","X-Requested-With":"XMLHttpRequest","App_id":"qiyunxin","timestamp": str(int(time.time() * 1000)),"sign": md5_hash(str(int(time.time() * 1000)) + "WEB"),}
logins = custom_request('POST', url='http://api.vvchat.im/userservices/v2/user/login', json_data=login_data_json, headers=login_header_json)
if logins['token'] and logins['open_id']:
print("[!] 登录账号成功！")
return logins
except:
exit("[!] 登录失败，重新执行脚本！")
#数据转换集合
def zhuanhuan(data):
data_json = json.loads(data_to_json(call_back_message))
return data_json
def tiren(chat_name, from_cust_id):
if chat_name in groups_list:
open_id = custom_request('GET', url=f"http://api.vvchat.im/groupapi/v1/users/{logins['open_id']}/groups/{groups_list[chat_name[1]]}/members?page_size=100000", headers={"Content-Type": "application/json; charset=UTF-8", "Authorization": logins['token']})
if open_id:
if chs['err_msg'] == 'OK':
print(f"[-] 群消息 {row[0]} 撤回成功！")
delete_data(f"DELETE FROM data WHERE msg_no = '{row[0]}' AND chat_name = '{row[1]}';")
else:
print(f"[!] 群消息 {row[0]} 撤回失败，原因：{chs['err_msg']} 请检查...")
delete_data(f"DELETE FROM data WHERE msg_no = '{row[0]}' AND chat_name = '{row[1]}';")
else:
msg_no_queue.put(row)
print(f"[!] 群消息 {row[0]} 撤回失败，已重新添加到列队！")
#持续不间断接受新数据并转存到列队
def run1():
global unique_data_set
while True:
hook_function(session)
if call_back_message:
call_back_message_json = zhuanhuan(remove_special_characters(call_back_message))
for item in call_back_message_json:
if groups in item.get("chat_name", "") and item.get("chat_type", "") == "2":
data = (item.get("chat_name", ""),item.get("msg_no", ""),item.get("from_cust_id", ""),item.get("content", ""),item.get("msg_time", ""))
if item.get("content", "") != '{0}开启群禁言' and item.get("content", "") != '{0}撤回了一条消息' and item.get("content", "") != '' and data not in unique_data_set:
ls_queue.put(data)
unique_data_set.add(data)
time.sleep(0.0001)
#从列表中提取数据判断是否存在
def run2():
while True:
with search_queue_lock:
while not ls_queue.empty():
data_to_insert = ls_queue.get() # 从队列取出数据
search_msg_no = query_data(f"select id from data where msg_no = '{data_to_insert[2]}'")
if not search_msg_no:
insert_queue.put(data_to_insert)
time.sleep(0.001)
#将insert_queue存储的最新发信信息插入到数据库中
def run3():
while True:
with search_queue_lock:
while not insert_queue.empty():
data_to_insert = insert_queue.get() # 从队列取出数据
print("[-] 新数据：",data_to_insert)
insert_data(data_to_insert)
time.sleep(0.001)
#搜索数据库，找出违规者所有发信MSG_NO编号，加入到列队中
def run4():
while True:
with search_queue_lock:
if msg_no_queue.empty() and insert_queue.empty() and ls_queue.empty():
#数据赛选，提取违规者发的全部信息 MSG_NO编号
select_query = """
WITH UniqueChatNames AS (
SELECT chat_name, from_cust_id FROM DATA
WHERE
id IN ( SELECT MIN(id) FROM
(SELECT id,chat_name,from_cust_id,content,SUBSTRING(msg_time, 1, 9) FROM DATA
WHERE
(chat_name,from_cust_id,content,SUBSTRING(msg_time, 1, 9)) IN (
SELECT chat_name,from_cust_id,content,SUBSTRING(msg_time, 1, 9) FROM DATA
GROUP BY chat_name,from_cust_id,content,SUBSTRING(msg_time, 1, 9)
HAVING COUNT(*) > 1
)
) grouped_data
GROUP BY chat_name, from_cust_id
)
) SELECT STRINGDECODE (d.msg_no), STRINGDECODE (d.chat_name)
FROM
(
SELECT *, ROW_NUMBER () OVER ( PARTITION BY chat_name, from_cust_id, msg_no ORDER BY id ) AS rn FROM DATA
) d JOIN UniqueChatNames subquery ON d.chat_name = subquery.chat_name AND d.from_cust_id = subquery.from_cust_id
WHERE
d.rn = 1;
"""
results = query_data(select_query)
if results:#列表类型
msg_no_queue.put(results)
print(f"[!] 已检测到 {len(results)} 条违规信息.")
time.sleep(0.001)
#提取msg_no编号，并撤回这些编号消息
def run5():
#登录
logins = login(username,pwd)
while True:
with search_queue_lock:
while not msg_no_queue.empty():
msg_no_list = msg_no_queue.get()
for row in msg_no_list:
if row[1] in groups_list:
data = {"msg_no": row[0],"msg_time": int(time.time() * 1000)}
chs = custom_request('POST', url=f"http://api.vvchat.im/groupapi/v1/users/{logins['open_id']}/groups/{groups_list[row[1]]}/revoke_msg", json_data=data, headers={"Content-Type": "application/json; charset=UTF-8", "Authorization": logins['token']})
if chs:
if chs['err_msg'] == 'OK':
print(f"[-] 群消息 {row[0]} 撤回成功！")
delete_data(f"DELETE FROM data WHERE msg_no = '{row[0]}' AND chat_name = '{row[1]}';")
else:
print(f"[!] 群消息 {row[0]} 撤回失败，原因：{chs['err_msg']} 请检查...")
delete_data(f"DELETE FROM data WHERE msg_no = '{row[0]}' AND chat_name = '{row[1]}';")
else:
msg_no_queue.put(row)
print(f"[!] 群消息 {row[0]} 撤回失败，已重新添加到列队！")
else:
exit("[!] 违规所在群 ID 获取失败, 代码中添加该群ID！")
time.sleep(0.001)
#5分钟清理一次数据库
def run6():
global unique_data_set
while True:
time.sleep(300) # 五分钟
unique_data_set.clear()
delete_data("TRUNCATE TABLE data;") #清空数据库
# 创建数据库连接
def create_connection():
conn = jaydebeapi.connect("org.h2.Driver", h2_jdbc_url, [h2_user, h2_password], h2_driver_path)
return conn
# 创建表
def create_table():
conn = create_connection()
create_table_sql = """
CREATE TABLE IF NOT EXISTS data (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
chat_name TEXT NOT NULL,
msg_no TEXT NOT NULL,
from_cust_id TEXT NOT NULL,
content TEXT NOT NULL,
msg_time BIGINT NOT NULL
)
"""
cursor = conn.cursor()
try:
cursor.execute(create_table_sql)
conn.commit()
except jaydebeapi.Error as e:
print("创建表异常:", e)
cursor.close()
return False
finally:
cursor.close()
return True
# 插入数据
def insert_data(data):
insert_query = '''
INSERT INTO data (chat_name, msg_no, from_cust_id, content, msg_time) VALUES (?, ?, ?, ?, ?);
'''
try:
conn = create_connection()
cursor = conn.cursor()
cursor.execute(insert_query, data)
conn.commit()
except jaydebeapi.Error as e:
print("[!] 插入失败:", e)
finally:
cursor.close()
# 查询数据
def query_data(sql):
try:
conn = create_connection()
cursor = conn.cursor()
cursor.execute(sql)
results = cursor.fetchall()
except jaydebeapi.Error as e:
print("[!] 查询失败:", e)
finally:
cursor.close()
return results
def delete_data(sql):
try:
conn = create_connection()
cursor = conn.cursor()
cursor.execute(sql)
conn.commit()
except jaydebeapi.Error as e:
print("[!] 查询失败:", e)
finally:
cursor.close()
if __name__ == "__main__":
################################################
username = '0086'
pwd = ''
groups = '关键词' #需要踢人的群名字统一关键词，也可以是一个字母符号，如果群名字不带这个则不会踢人
groups_list = {
"aaa测试3": "pUalcqhBTcdg44G_j9s5K-XCUF83FRHB",
"aaa测试1": "928JLh8AAA65RyryJLbTiPKnV0NJWXIi",
"aaa测试2": "Y26osJNBCaOMENv71XMT7CmOT1hYhee9"
}
################################################
# 全局变量
call_back_message = None
script = None # 用于存储已加载的脚本
unique_data_set = set()
# 创建队列
ls_queue = queue.Queue()
insert_queue = queue.Queue()
msg_no_queue = queue.Queue()
# 创建锁
search_queue_lock = threading.Lock()
# 连接参数
h2_jdbc_url = "jdbc:h2:./test"
h2_user = "test"
h2_password = "test"
h2_driver_path = "h2-1.4.200.jar"
if not os.path.exists(os.path.expanduser("./test.mv.db")):
# 数据库文件不存在，创建数据库和表
if create_table():
print("数据库表创建成功！")
else:
os.remove("./test.mv.db")
exit("[!] 数据表创建失败，退出代码！")
print("[-] 正在启动应用APP...")
session = usb_device()
time.sleep(8) #APP启动有加载时间
try:
print("[-] 检测状态： 正在检测信息...")
# 启动线程模拟函数的同时运行
threading.Thread(target=run1).start()
threading.Thread(target=run2).start()
threading.Thread(target=run3).start()
threading.Thread(target=run4).start()
threading.Thread(target=run5).start()
threading.Thread(target=run6).start()
except KeyboardInterrupt:
script.unload()
print("[-] 脚本已卸载，程序退出。")思路2：通过调用APP本地数据库
在审计源码过程中发现泡泡APP存在本地数据库，将最新的消息全部存在本地数据库中，当时以为终于找到出路了，通过sql语句调用能非常简单快速的达到我要的效果。 奈何困难总比办法多，在数据库上又卡住了； 数据库使用了sqlcipher 4.5.3 加密，在尝试用pysqlcipher3库打开sqlite数据库始终连不上，查阅半天资料发现pysqlcipher3库不支持sqlcipher 4.5.3 加密的sqlite数据库； 后又尝试本地编译sqlcipher失败，搭建环境各种报错，百度谷歌都快翻烂了都没解决，这里不得不吐槽下，为什么这么不容易入手的加密方式还这么出名和流行？就因为它难入手吗。。。 这里不得不提一下，后来换思路的时候才看到官方提供了windows编译的具体步骤教程，可惜当时已经对它彻底失望了，所以没去尝试，不过我感觉最终还是会编译失败。。。 后面通过 DB Browser 和 SQLiteStudio 2款sqlit数据库管理工具连上了数据库； 然后又查阅资料，python应该如何依赖这些工具来调用数据库呢？ 2款工具都不存在API，只能考虑命令行了； DB Browser 不存在Windows下命令行，不过可以在Linux下安装，所以使用 Termux 命令执行APP在安卓系统安装，结果发现源找不到库DB Browser的库，更换源也没辙，官方也没有提供Linux下编译安装包。 SQLiteStudio 直接不支持命令行，不过发现它可以直接通过adb的方式连接上APP中的数据库，但是因为没法通过python调用，所以最后pass掉了。思路3：从socket上面入手
最开始的时候还是满脑子盯着安卓代码研究，看它消息接受是怎么接受的，最后定位到了它socket接受消息的方法; 通过hook发现一个蛋疼的事情，每次消息接受scoket的IP端口都在变化，当时不懂是为什么，后听一哥们说是中转服务器，任意连一个就行了; 尝试使用python连接这些IP端口总是不成功，请教那哥们后那哥们告诉我用web端的websocket，他通过websocket网页在线连接测试成功连接上，所以我最后又将目光放在了WEB端的websocket上了。研究 websocket
通过各种查阅资料，去了解websocket的原理、流程等方面资料后，开始尝试用python连接； 连上以后并没有收到任何数据，隔一段时间连接就断开了，这也是我最开始不考虑webscoket的原因，因为不懂不了解； 后面尝试网上的发送心跳保持长连接活跃的方法，发现还是没屁用，收不到数据，20秒断开连接。 卡了几天后，开始研究WEB端的JS代码，看他的JS是如何连接处理websocket接受数据的， JS连接webscoket代码：https://im.mlskm.com/js/aiti-js-sdk-1.0.js 经过分析发现他连上webscoket后先构造登录数据包登录，登录后每20秒发送一次自定义构造的心跳数据包，通过查看它打印的控制台日志也是如此流程，这也是我收不到数据和断开连接的原因； 最后通过一点点根据JS代码来构造数据包，成功使用python连接登录并接受到了数据，本以为终于可以告别这次苦逼的研究了，奈何又遇到了新问题，20秒自动断开连接！！ 来回检查代码对比代码，发送心跳后服务端返回心跳正常，并且心跳发送也是10秒发一次，为什么20秒还是自动断开了？ 这个问题卡了我两三天，最后成功的时候才发现，python 的 websockets 库就是个垃圾，垃圾中的超级垃圾，坑死我了，我不知道它为什么会20秒断开，看看AI怎么评价 websockets 的：
使用 websocket 库成功搞定
最后把websockets库换成websocket库成功保持长连接活跃，也能持续接受新消息，只能说非常Nice，剩下的就是处理和优化了，贴上最终代码和截图：代码：
import queue
import websocket
import threading
import re
import base64
import random
import requests
import time
import json
import hashlib
from Crypto.Cipher import DES
from Crypto.Util.Padding import pad,unpad
#
# 群ID加密
#
def encryptCustId(custId):
custId = str(custId)
custIdArray = [int(digit) for digit in custId]
encrypted_message = encryptByDES(byteToString(custIdArray))
encrypted_message = encrypted_message.replace('/', '_').replace('+', '-')
return encrypted_message
#加密
def encryptByDES(message, key='klohjmz_'):
key = key.encode('utf-8')
iv = key
message = message.encode('utf-8')
cipher = DES.new(key, DES.MODE_CBC, iv)
ciphertext = cipher.encrypt(pad(message, DES.block_size))
encrypted_message = base64.b64encode(ciphertext).decode('utf-8')
return encrypted_message
#解密
def decryptByDES(ciphertext, key='klohjmz_'):
key_bytes = key.encode('utf-8')
iv_bytes = key.encode('utf-8')
cipher = DES.new(key_bytes, DES.MODE_CBC, iv_bytes)
decrypted = unpad(cipher.decrypt(base64.b64decode(ciphertext.replace('_', '/').replace('-', '+'))), DES.block_size)
return stringify(decrypted)
def stringify(a):
m = []
for t in a:
m.append(str(t & 15))
return ''.join(m)
def byteToString(arr):
if isinstance(arr, str):
return arr
str_ = ''
i = 0
while i < len(arr):
one = format(arr[i], '08b')
v = re.match(r'^1+?(?=0)', one)
if v and len(one) == 8:
bytes_length = len(v.group(0))
store = format(arr[i], '08b')[7 - bytes_length:]
for st in range(1, bytes_length):
store += format(arr[i + st], '08b')[2:]
str_ += chr(int(store, 2))
i += bytes_length - 1
else:
str_ += chr(arr[i])
i += 1
return str_
#
# 接受数据解析
#
def parse_number_data(data):
if len(data) == 8:
v1 = int.from_bytes(data[0:4], byteorder='little', signed=False)
v2 = int.from_bytes(data[4:8], byteorder='little', signed=False)
mp = 2 ** 32
x = v2
y = mp
v1n = v1
result = x * y + v1n
return str(result)
v = 0
for i, n in enumerate(data):
n = n if n >= 0 else n & 0xff
v += n * (256 ** i)
return v
def parse_string(arr):
return uint_to_string(bytearray(arr))
def uint_to_string(uint_array):
encoded_string = ''.join(chr(byte) for byte in uint_array)
decoded_string = bytes(encoded_string, 'latin-1').decode('utf-8')
return decoded_string
#
# 登录
#
pfcCounter = 0
def next_message_num():
global pfcCounter
pfcCounter += 1
pfc = pfcCounter
result = []
while pfc > 0:
temp = pfc
pfc = pfc // 256
result.append(temp - pfc * 256)
while len(result) < 8:
result.append(0)
result[3] = random.randint(1, 9)
result[4] = random.randint(1, 9)
result[5] = random.randint(1, 9)
result[6] = random.randint(1, 9)
result[7] = random.randint(1, 9)
return result
def current_times():
timestamp = int(time.time())
result = [0] * 8
i = 7
while timestamp > 0:
temp = timestamp
timestamp = timestamp // 256
result[i] = temp - timestamp * 256
i -= 1
return result
def calc_checksum(buffer, buffer_len):
if buffer is None:
return [0, 0, 0, 0]
tmp = [0, 0, 0, 0]
dest = [0, 0, 0, 0]
len = 0
offset = 0
while len < buffer_len:
if len + 4 > buffer_len:
offset = buffer_len - len
else:
offset = 4
for i in range(offset):
tmp[i] ^= buffer[len + i]
len += 4
for i in range(4):
dest[i] = tmp[i]
return dest
def split_cust_id(cust_id):
result = [0] * 8
i = 0
while cust_id > 0:
temp = cust_id
cust_id = cust_id // 256
result[i] = temp - cust_id * 256
i += 1
return result
def split_auth(auth):
result = [ord(char) for char in auth]
if len(result) < 32:
for i in range(len(result), 32):
result.append(0)
return result
def md5_hash(text):
md5_hasher = hashlib.md5()
md5_hasher.update(text.encode('utf-8'))
hashed_text = md5_hasher.hexdigest()
return hashed_text
def md5Psw(account, pwd):
hashed_pwd = md5_hash( md5_hash(pwd) + account + 'MOSGRAM')
return hashed_pwd
def login(username,pwd):
global open_id_qj,im_token,token
data = {
"username": f"{username}",
"password": md5Psw(username, pwd),
"device_type":"WEB"
}
headers = {
"Content-Type":"application/json; charset=UTF-8",
"X-Requested-With":"XMLHttpRequest",
"App_id":"qiyunxin",
"timestamp": str(int(time.time() * 1000)),
"sign": md5_hash(str(int(time.time() * 1000)) + "WEB"),
"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36"
}
try:
print('[信息] WEB端登录中...')
response = requests.post('http://api.vvchat.im/userservices/v2/user/login', headers=headers,data=json.dumps(data))
data = response.json()
if 'err_msg' in data:
exit('[信息] WEB端登录失败： %s！' % data['err_msg'])
elif 'open_id' in data:
print('[信息] WEB端成功！')
open_id_qj = data['open_id']
im_token = data['im_token']
token = data['token']
except requests.exceptions.RequestException as e:
print('[信息] WEB端登录出错：%s'%(e))
except Exception as e:
print('[信息] WEB端登录异常报错：%s'%(e))
def revoke_msg(msg_no,groups_id,group_name):
print(f"[巡查] 开始撤回群 {group_name} 消息编号 {msg_no} 消息！")
data = {"msg_no": msg_no,"msg_time": int(time.time()) * 1000}
headers = {"Content-Type":"application/json; charset=UTF-8","X-Requested-With":"XMLHttpRequest","App_id":"qiyunxin","User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36","Authorization":token}
try:
response = requests.post(f'http://api.vvchat.im/groupapi/v1/users/{open_id_qj}/groups/{groups_id}/revoke_msg', headers=headers,data=json.dumps(data))
result = response.json()
if result['err_msg'] == 'OK':
print(f"[巡查] 群 {group_name} 消息编号 {msg_no} 撤回成功！")
else:
print(f"[巡查] 群 {group_name} 消息编号 {msg_no} 撤回失败，原因：{result['err_msg']} 请检查...")
except requests.exceptions.RequestException as e:
print('[巡查] 群 %s 消息编号 %s 撤回出错：%s ,重试一次...'%(group_name,msg_no,e))
revoke_msg(msg_no,groups_id)
except Exception as e:
print('[巡查] 群 %s 消息编号 %s 撤回异常报错：%s ,重试一次...'%(group_name,msg_no,e))
revoke_msg(msg_no,groups_id)
def delete_member(from_cust_id,groups_id):
group_name = groups_name(groups_id)
print(f"[巡查] 开始剔除群 {group_name} 成员 {from_cust_id}")
from_cust_id_encrypt = encryptCustId(from_cust_id)
data = [{"name":"default","open_id": from_cust_id_encrypt,"cust_id": from_cust_id}]
headers = {"Content-Type":"application/json; charset=UTF-8","X-Requested-With":"XMLHttpRequest","App_id":"qiyunxin","User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36","Authorization":token}
try:
response = requests.delete(f'http://api.vvchat.im/groupapi/v1/users/{open_id_qj}/groups/{groups_id}/members', headers=headers,json=data)
result = response.json()
if result['err_msg'] == 'OK':
print(f"[巡查] 群 {group_name} 成员 {from_cust_id} 剔除成功！")
else:
print(f"[巡查] 群 {group_name} 成员 {from_cust_id} 剔除失败，原因：{result['err_msg']} 请检查...")
except requests.exceptions.RequestException as e:
print('[巡查] 群 %s 成员 %s 剔除出错：%s,重试一次...'%(group_name,from_cust_id,e))
delete_member(from_cust_id,groups_id)
except Exception as e:
print('[巡查] 群 %s 成员 %s 剔除异常报错：%s,重试一次...'%(group_name,from_cust_id,e))
delete_member(from_cust_id,groups_id)
def groups_name(groups_id):
headers = {"App_id":"qiyunxin","User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36","Authorization":token}
try:
response = requests.get(f'http://api.vvchat.im/groupapi/v1/users/{open_id_qj}/groups/{groups_id}', headers=headers)
result = response.json()
if result:
return result['group_name']
except requests.exceptions.RequestException as e:
print('[巡查] 获取群名称 出错：%s'%(e))
groups_name(groups_id)
except Exception as e:
print('[巡查] 获取群名称 异常报错：%s'%(e))
groups_name(groups_id)
#登录包
def wss_login_data(from_cust_id,im_token):
from_cust_id = split_cust_id(int(decryptByDES(from_cust_id)))
auth = split_auth(im_token)
msg_no = next_message_num()
buffer = []
msg_type = [1, 0]
body = []
body += login_type + from_cust_id + auth
checkCode = calc_checksum(body, len(body))
body += checkCode + end
buffer += start + length + version + msg_no + msg_type + body
len_body = len(body)
buffer[1] = len_body - ((len_body >> 16) << 16)
buffer[2] = len_body >> 8
buffer[3] = len_body >> 16
buffer[4] = 0
return buffer
#心跳包
def wss_xt_data(open_id):
buffer = []
from_cust_id = split_cust_id(int(decryptByDES(open_id)))
msg_no = next_message_num()
msg_type = [6, 0]
body = []
body += login_type + from_cust_id
checkCode = calc_checksum(body, len(body))
body += checkCode + end
buffer += start + length + version + msg_no + msg_type + body
len_body = len(body)
buffer[1] = len_body - ((len_body >> 16) << 16);
buffer[2] = len_body >> 8;
buffer[3] = len_body >> 16;
buffer[4] = 0;
return buffer
#消息检测
def check_and_clean_data():
global msg_queue
wg_list = []
msg_list = []
while True:
if not msg_queue.empty():
for _ in range(4):
item = msg_queue.get()
msg_list.append(item)
print("[巡查] 巡查检测中...")
matching_data = [item for item in msg_list if item[0] == msg_list[-1][0] and item[1] == msg_list[-1][1] and item[3] == msg_list[-1][3]]
if len(matching_data) >= 3:# 三者相同的存在大于等于三条
timestamps = [item[4] for item in matching_data] #获取时间字段
time_diffs = [int(timestamps[i + 1]) - int(timestamps[i]) for i in range(len(timestamps) - 1)]
for diff, data in zip(time_diffs, matching_data[1:]):
if diff == 0 or diff <= 3:
for x in msg_list:
if data[0] == x[0]:
wg_list.append(x)
matching_data = []
if wg_list:
# 使用集合去重，通过将子列表转换为元组
unique_list = list(set(tuple(sub_list) for sub_list in wg_list))
# 将元组转换回列表
unique_nested_list = [list(sub_tuple) for sub_tuple in unique_list]
from_cust_ids = set() # 用于存储去重后的 from_cust_id
for group in unique_nested_list:
if group: # 确保列表不为空
from_cust_ids.add((group[0], group[1]))
for user in from_cust_ids:
#踢人
print(f"[巡查] 检测到群： {groups_name(user[1])} 出现违规者：{user[0]}！")
delete_member(x[0],x[1])
for msg_no in msg_list:
for i in from_cust_ids:
if msg_no[0] == i[0] and msg_no[1] == i[1]:
#获取群名称
group_name = groups_name(msg_no[1])
#撤回
revoke_msg(msg_no[2],msg_no[1],group_name)
wg_list = []
msg_list = []
else:
print("[巡查] 消息未检测到频繁！")
#服务端传输数据类型判断
def eventReceiveMessage(ws,msg):
global msg_queue
msg_type = parse_number_data(msg[14:16])
status = parse_number_data(msg[25:29])
if msg_type == 2:
if status == 200:
print('[信息] websocket 端登录成功，开始监听信息...')
# 创建发送心跳包的线程
heartbeat_thread = threading.Thread(target=heartbeat, args=(ws,))
heartbeat_thread.daemon = True
heartbeat_thread.start()
# 创建消息处理线程
msg_thread = threading.Thread(target=check_and_clean_data)
msg_thread.daemon = True
msg_thread.start()
elif status == 401:
print('[信息] websocket 端登录失败!')
ws.close()
elif msg_type == 5:
chatType = parse_number_data(msg[16:17])
content = parse_string(msg[61:61+parse_number_data(msg[57:61])])
content_json = json.loads(content)['content']
msgNo = parse_number_data(msg[6:14])
groups_id = encryptCustId(parse_number_data(msg[25:33]))
msgTime = parse_number_data(msg[17:25])
from_cust_id = parse_number_data(msg[33:41])
data = [from_cust_id,groups_id,msgNo,content_json,msgTime]
if chatType == 2:
print(f"[消息] 发送者：{from_cust_id} 发自群：{groups_name(groups_id)} 发送时间：{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(int(msgTime)))}")
msg_queue.put(data)
elif msg_type == 6:
print("[信息] websocket 端心跳正常！")
else:
pass
def heartbeat(ws):
while True:
try:
print(f'[信息] 发送心跳推送...')
ws.send(bytes(wss_xt_data(open_id_qj)))
time.sleep(15)
except:
print("[信息] 心跳推送失败!")
ws.close()
def on_error(ws, error):
print(f"[错误] {error}")
def on_close(ws, close_status_code, close_msg):
print("[信息] websocket 端连接已断开，开始重新启动...")
run()
def on_message(ws, message):
eventReceiveMessage(ws,message)
def on_open(ws):
print("[信息] websocket 端已连接！")
login(username,password)
print("[信息] websocket 端登录中...")
ws.send(bytes(wss_login_data(open_id_qj,im_token)))
def run():
try:
# 创建WebSocket连接
print("[信息] websocket 端连接状态检测...")
ws = websocket.WebSocketApp(url, on_message=on_message, on_open=on_open,on_error=on_error,on_close=on_close)
# 启动WebSocket连接
ws.run_forever()
except:
print("[信息] websocket 端连接启动失败！")
if __name__ == "__main__":
global username,password
###############配置专区###############
username = '0086' #0086+手机号
password = ''
url = "wss://im.mlskm.com/ws8"
###############配置专区###############
#全局变量定义
msg_queue = queue.Queue()
start = [2]
version = [1]
end = [3]
length = [0, 0, 0, 0]
login_type = [2]
#启动
run()截图：
最终实验效果如上，不会遗漏消息，并且支持多个群管理和支持全自动化（掉线重连），只需要挂着脚本啥也不需要管，就非常OK，搞定这一瞬间感觉整个人都活了，总花费差不多两周，太不容易了。。。
总结
- 入手学习到了逆向脱壳，以及 app hook 方面的知识。
- 对安卓中的Java代码更加熟悉。
- 入手了 sqlite 数据库及 sqlcipher 加密方式方面的知识。
- 入手了socket、websocket方面的知识。
- 对python更加熟练。